import os
os.environ["MELUMIYACHAT_CLOUD_ENABLED"] = "0"

from fastapi.testclient import TestClient
from app.main import app
from app import db

c = TestClient(app)

# Basic customer app health.
r = c.get('/api/health')
assert r.status_code == 200
health = r.json()
assert health['ok'] is True
assert health['version'] == '0.9.15'
assert health['product'] == 'MelumiyaChat'
assert c.get('/').status_code == 200

# Customer build must have no Owner / private Test Lab HTTP surface.
blocked_get = [
    '/owner-desktop',
    '/api/providers',
    '/api/benchmarks',
    '/api/owner/summary',
    '/api/owner/support/tickets',
    '/api/owner/content',
    '/api/owner/models',
    '/api/owner/revenue/summary',
    '/api/owner/training/guided',
]
for path in blocked_get:
    rr = c.get(path)
    assert rr.status_code == 404, (path, rr.status_code, rr.text[:200])

blocked_post = [
    '/api/auth/claim-owner',
    '/api/benchmarks/compare',
    '/api/owner/support/tickets/fake/reply',
    '/api/owner/content',
    '/api/owner/training/runs',
    '/api/owner/marketing/influencers',
]
for path in blocked_post:
    rr = c.post(path, json={})
    assert rr.status_code == 404, (path, rr.status_code, rr.text[:200])

# Static customer shell must not expose private Owner/Test Lab navigation.
html = c.get('/').text
for forbidden in [
    'Owner Pocket', 'Model Test Lab', 'ownerPocketNav', 'view-owner', 'view-test',
    'Private Owner access', 'Open Owner Studio'
]:
    assert forbidden not in html, forbidden

# Persistent RP / continuity data remains intact.
convs = c.get('/api/conversations')
assert convs.status_code == 200 and convs.json(), 'Expected seeded prototype world'
cid = convs.json()[0]['id']
bundle = c.get(f'/api/conversations/{cid}/bundle')
assert bundle.status_code == 200
state = bundle.json()['state']
assert 'hard_canon' in state
assert 'scene' in state
assert 'world_bible' in state

# Customer model access remains available.
access = c.get('/api/models/available')
assert access.status_code == 200
model_data = access.json()
assert model_data['plan_tier'] == 'Free'
assert model_data['models']
assert model_data['usage']['total_limit'] == 500

# Discover remains available.
tags = c.get('/api/discover/tags')
assert tags.status_code == 200 and isinstance(tags.json(), list)
scenarios = c.get('/api/discover/scenarios')
assert scenarios.status_code == 200 and scenarios.json()
characters = c.get('/api/discover/characters')
assert characters.status_code == 200 and isinstance(characters.json(), list)

# Customer account/persona surfaces remain available.
profile = c.get('/api/profile/me')
assert profile.status_code == 200
assert profile.json()['plan_tier'] == 'Free'
personas = c.get('/api/personas')
assert personas.status_code == 200
assert 'items' in personas.json() and 'limit' in personas.json()

# Customer support create/list/get remains in the desktop app.
ticket = c.post('/api/help/tickets', json={
    'category': 'Bug',
    'subject': 'Customer-only self test',
    'body': 'Temporary local support test.'
})
assert ticket.status_code == 200, ticket.text
t = ticket.json()
assert t['status'] == 'Open'
ticket_id = t['id']
listed = c.get('/api/help/tickets')
assert listed.status_code == 200
assert any(str(x['id']) == str(ticket_id) for x in listed.json()['items'])
detail = c.get(f'/api/help/tickets/{ticket_id}')
assert detail.status_code == 200
assert detail.json()['case']['subject'] == 'Customer-only self test'
# Local mode deliberately has no two-way cloud reply transport; route exists and fails closed.
reply = c.post(f'/api/help/tickets/{ticket_id}/reply', json={'body': 'Temporary reply'})
assert reply.status_code == 503
# Remove the temporary local ticket.
db.execute('DELETE FROM support_tickets WHERE id=?', (ticket_id,))

# Customer creator flow remains available up to submission for review.
created = c.post('/api/creator/content', json={
    'item_type': 'character',
    'creator_name': 'You',
    'title': 'Customer-only temporary character',
    'summary': 'Temporary self-test record',
    'tags': ['self-test'],
    'payload': {'personality': ['calm']}
})
assert created.status_code == 200, created.text
item = created.json()
assert item['status'] == 'Draft'
submitted = c.post(f"/api/creator/content/{item['id']}/submit", json={})
assert submitted.status_code == 200
assert submitted.json()['status'] == 'Pending Review'
# There is intentionally no customer-side approval/review route.
assert c.post(f"/api/owner/content/{item['id']}/review", json={'action': 'Approve'}).status_code == 404
db.execute('DELETE FROM creator_content WHERE id=?', (item['id'],))

# Customer billing routes remain present without exposing Owner controls.
# In this local self-test cloud is deliberately disabled, so checkout must fail closed
# as service-unavailable rather than disappearing as a 404.
assert c.get('/api/billing/catalog').status_code == 503
assert c.post('/api/billing/checkout', json={'tier':'Plus','billing_period':'monthly'}).status_code == 503
assert c.post('/api/billing/refresh', json={}).status_code == 503

# Customer notifications / favorites / promotions remain routable.
assert c.get('/api/notifications').status_code == 200
assert c.get('/api/favorites').status_code == 200
assert c.get('/api/promos/validate?code=NO_SUCH_CODE').status_code in (200, 404)


# RP continuity bridge regression: World Bible + structured Memories must be
# real model inputs, and newly detected continuity must be readable on the next turn.
from app.services.context_builder import build_system_prompt, select_memories_for_context
from app.services.memory import apply_extracted_local
import json

bridge_cid = db.execute(
    "INSERT INTO conversations(title,persona_name,scenario_name) VALUES (?,?,?)",
    ("RP continuity bridge self-test", "Mariana", "Bridge Test"),
)
try:
    initial_world = {
        "schema_version": 3,
        "name": "Bridge Test World",
        "description": "Temporary continuity bridge regression world.",
        "background_facts": [
            {
                "stable_key": "background:el-paso",
                "fact": "BRIDGE_TEST: Xaden was raised in El Paso.",
                "source": "user-authored",
            }
        ],
        "story_timeline": [],
    }
    db.update_state(bridge_cid, "world_bible", initial_world)
    db.update_state(bridge_cid, "hard_canon", {})
    db.update_state(bridge_cid, "relationship", {})
    db.update_state(bridge_cid, "knowledge", {})
    db.update_state(bridge_cid, "scene", {"location": "Bridge room"})
    db.update_state(bridge_cid, "physical", {})

    db.execute(
        """INSERT INTO memories(
               conversation_id,memory_type,title,content,importance,confidence,active,
               what_happened,who_remembers,who_involved,story_when,story_where,
               emotional_weight,current_relevance,source_kind,original_context,stable_key,revision_history
           ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            bridge_cid, "Promise", "Silver key promise",
            "BRIDGE_TEST: Xaden promised to return the silver key.", "Critical", 1.0, 1,
            "BRIDGE_TEST: Xaden promised to return the silver key.",
            json.dumps(["Xaden"]), json.dumps(["Xaden", "Mariana"]),
            "last winter", "El Paso", "unresolved", "active", "User-created", "{}",
            "promise:xaden:return-silver-key", "[]",
        ),
    )
    # A minor filler memory should not outrank the Critical, active bridge memory.
    db.execute(
        """INSERT INTO memories(
               conversation_id,memory_type,title,content,importance,confidence,active,
               what_happened,who_remembers,who_involved,current_relevance,source_kind,stable_key,revision_history
           ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            bridge_cid, "Other", "Minor filler", "They once discussed the weather.", "Minor", 1.0, 1,
            "They once discussed the weather.", json.dumps(["Xaden"]), json.dumps(["Xaden"]),
            "active", "Auto-created from RP", "filler:weather", "[]",
        ),
    )
    # A superseded memory with overlapping terms must remain historical and not beat current truth.
    db.execute(
        """INSERT INTO memories(
               conversation_id,memory_type,title,content,importance,confidence,active,
               what_happened,who_remembers,who_involved,current_relevance,source_kind,stable_key,revision_history
           ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            bridge_cid, "Promise", "Old silver key state", "The silver key promise was impossible.", "Critical", 1.0, 1,
            "The silver key promise was impossible.", json.dumps(["Xaden"]), json.dumps(["Xaden"]),
            "superseded", "Auto-created from RP", "promise:xaden:old-silver-key-state", "[]",
        ),
    )

    conv = db.row("SELECT * FROM conversations WHERE id=?", (bridge_cid,))
    mem_rows = db.rows("SELECT * FROM memories WHERE conversation_id=? ORDER BY id DESC", (bridge_cid,))
    ranked = select_memories_for_context(mem_rows, "Xaden silver key El Paso", limit=10)
    assert ranked, "Expected bridge memories to be selected"
    assert ranked[0]["stable_key"] == "promise:xaden:return-silver-key", ranked

    prompt = build_system_prompt(conv, db.get_state(bridge_cid), mem_rows, "What did Xaden promise about the silver key in El Paso?")
    assert "BRIDGE_TEST: Xaden was raised in El Paso." in prompt
    assert "BRIDGE_TEST: Xaden promised to return the silver key." in prompt
    assert '"who_remembers": [' in prompt and '"Xaden"' in prompt
    assert "WORLD BIBLE / LINKED CHARACTER DOSSIERS (live structured reference)" in prompt
    assert "RETRIEVED MEMORIES (live character-aware history)" in prompt

    # Simulate the post-turn detector writing a new memory and new Bible facts.
    extracted = {
        "memories": [
            {
                "stable_key": "discovery:mariana:blue-door",
                "action": "create",
                "title": "Blue door discovery",
                "what_happened": "BRIDGE_TEST_AUTO: Mariana discovered the blue door opens only at midnight.",
                "who_remembers": ["Mariana", "Xaden"],
                "who_involved": ["Mariana", "Xaden"],
                "when": "tonight",
                "where": "Bridge room",
                "memory_type": "Discovery",
                "importance": "Important",
                "emotional_weight": "mixed",
                "current_relevance": "active",
                "confidence": 0.99,
            }
        ],
        "world_bible_updates": [
            {
                "section": "background_facts",
                "stable_key": "background:blue-door-rule",
                "action": "create",
                "item": {"fact": "BRIDGE_TEST_AUTO: The blue door opens only at midnight."},
                "confidence": 0.99,
                "conflicts_with_user_canon": False,
            },
            {
                "section": "story_timeline",
                "stable_key": "timeline:blue-door-discovery",
                "action": "create",
                "item": {"event": "BRIDGE_TEST_AUTO: Mariana and Xaden discovered the blue door's midnight rule."},
                "confidence": 0.99,
                "conflicts_with_user_canon": False,
            },
        ],
        "state_updates": {},
        "conflicts": [],
    }
    applied = apply_extracted_local(
        db, bridge_cid, extracted,
        source_message_id=None,
        user_text="We test the blue door.",
        assistant_text="The blue door reveals its midnight rule.",
    )
    assert applied["memories_created"] == 1, applied
    assert applied["world_bible_added"] >= 2, applied

    new_mem = db.row(
        "SELECT * FROM memories WHERE conversation_id=? AND stable_key=?",
        (bridge_cid, "discovery:mariana:blue-door"),
    )
    assert new_mem and new_mem["source_kind"] == "Auto-created from RP"
    assert json.loads(new_mem["who_remembers"]) == ["Mariana", "Xaden"]
    world_after = db.get_state(bridge_cid)["world_bible"]
    assert any(x.get("stable_key") == "background:blue-door-rule" for x in world_after["background_facts"])
    assert any(x.get("stable_key") == "timeline:blue-door-discovery" for x in world_after["story_timeline"])

    # The next RP prompt must immediately read the continuity that the prior turn wrote.
    mem_after = db.rows("SELECT * FROM memories WHERE conversation_id=? ORDER BY id DESC", (bridge_cid,))
    next_prompt = build_system_prompt(
        conv, db.get_state(bridge_cid), mem_after,
        "At midnight, what do Mariana and Xaden remember about the blue door?",
    )
    assert "BRIDGE_TEST_AUTO: Mariana discovered the blue door opens only at midnight." in next_prompt
    assert "BRIDGE_TEST_AUTO: The blue door opens only at midnight." in next_prompt

    # User-created memories are authoritative: auto extraction may not overwrite them.
    db.execute(
        """INSERT INTO memories(
               conversation_id,memory_type,title,content,importance,confidence,active,
               what_happened,who_remembers,who_involved,current_relevance,source_kind,stable_key,revision_history
           ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            bridge_cid, "Event", "Manual truth", "BRIDGE_PROTECTED: The ring is gold.", "Critical", 1.0, 1,
            "BRIDGE_PROTECTED: The ring is gold.", json.dumps(["Mariana"]), json.dumps(["Mariana"]),
            "active", "User-created", "object:ring:material", "[]",
        ),
    )
    protected_result = apply_extracted_local(
        db, bridge_cid,
        {
            "memories": [{
                "stable_key": "object:ring:material", "action": "update", "title": "Manual truth",
                "what_happened": "BRIDGE_BAD_OVERWRITE: The ring is silver.", "who_remembers": ["Mariana"],
                "who_involved": ["Mariana"], "memory_type": "Event", "importance": "Critical",
                "current_relevance": "active", "confidence": 0.99,
            }],
            "world_bible_updates": [], "state_updates": {}, "conflicts": [],
        },
        source_message_id=None, user_text="test", assistant_text="test",
    )
    protected = db.row(
        "SELECT * FROM memories WHERE conversation_id=? AND stable_key=? ORDER BY id DESC LIMIT 1",
        (bridge_cid, "object:ring:material"),
    )
    assert protected["what_happened"] == "BRIDGE_PROTECTED: The ring is gold."
    assert protected_result["conflicts"] >= 1

    # User-authored World Bible entries get the same protection.
    world_manual = db.get_state(bridge_cid)["world_bible"]
    world_manual.setdefault("important_objects", []).append({
        "stable_key": "object:manual-ring",
        "name": "Manual Ring",
        "details": "BRIDGE_PROTECTED_BIBLE: The manual ring is gold.",
        "source": "user-authored",
    })
    db.update_state(bridge_cid, "world_bible", world_manual)
    bible_protected_result = apply_extracted_local(
        db, bridge_cid,
        {
            "memories": [],
            "world_bible_updates": [{
                "section": "important_objects", "stable_key": "object:manual-ring", "action": "update",
                "item": {"name": "Manual Ring", "details": "BRIDGE_BAD_BIBLE_OVERWRITE: The manual ring is silver."},
                "confidence": 0.99, "conflicts_with_user_canon": False,
            }],
            "state_updates": {}, "conflicts": [],
        },
        source_message_id=None, user_text="test", assistant_text="test",
    )
    protected_world = db.get_state(bridge_cid)["world_bible"]
    manual_ring = next(x for x in protected_world["important_objects"] if x.get("stable_key") == "object:manual-ring")
    assert manual_ring["details"] == "BRIDGE_PROTECTED_BIBLE: The manual ring is gold."
    assert bible_protected_result["conflicts"] >= 1
finally:
    db.execute("DELETE FROM conversations WHERE id=?", (bridge_cid,))

print('PASS — MelumiyaChat Windows customer-only self-test')
print('Owner/Test Lab routes absent; customer RP, continuity, creator and support routes preserved.')
