from __future__ import annotations from pathlib import Path import pytest from fastapi.testclient import TestClient from chat.app import app from chat.db.connection import open_db from chat.eventlog.log import append_event from chat.eventlog.projector import project @pytest.fixture def client(tmp_path, monkeypatch): cfg = tmp_path / "config.toml" cfg.write_text('featherless_api_key = "test"\n') monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg)) db = tmp_path / "test.db" monkeypatch.setenv("CHAT_DB_PATH", str(db)) with TestClient(app) as c: if hasattr(app.state, "background_worker"): app.state.background_worker.enabled = False yield c def _seed_bot_with_state(db: Path) -> None: """Seed a bot plus a chat, container, scene, edge, memory, and activity row.""" with open_db(db) as conn: append_event( conn, kind="bot_authored", payload={ "id": "bot_a", "name": "BotA", "persona": "thoughtful, observant", "voice_samples": [], "traits": ["shy"], "backstory": "", "initial_relationship_to_you": "coworker", "kickoff_prose": "you stay late", }, ) append_event( conn, kind="chat_created", payload={ "id": "chat_bot_a", "host_bot_id": "bot_a", "initial_time": "2026-04-26T20:00:00+00:00", "narrative_anchor": "Day 1", "weather": "", }, ) append_event( conn, kind="container_created", payload={ "chat_id": "chat_bot_a", "name": "office", "type": "workplace", "properties": {}, }, ) append_event( conn, kind="scene_opened", payload={ "chat_id": "chat_bot_a", "container_id": 1, "started_at": "2026-04-26T20:00:00+00:00", "participants": ["you", "bot_a"], }, ) append_event( conn, kind="edge_update", payload={ "source_id": "bot_a", "target_id": "you", "chat_id": "chat_bot_a", "affinity_delta": 5, "trust_delta": 2, }, ) append_event( conn, kind="memory_written", payload={ "owner_id": "bot_a", "chat_id": "chat_bot_a", "pov_summary": "Talked about her sister", "witness_you": 1, "witness_host": 1, "witness_guest": 0, "significance": 2, }, ) append_event( conn, kind="activity_change", payload={ "entity_id": "bot_a", "posture": "sitting", "action": {"verb": "writing"}, }, ) project(conn) def test_reset_purges_state_but_preserves_identity(client, tmp_path): _seed_bot_with_state(tmp_path / "test.db") response = client.post( "/bots/bot_a/reset", data={"confirm_name": "BotA"}, follow_redirects=False, ) assert response.status_code == 303 assert response.headers["location"] == "/bots" with open_db(tmp_path / "test.db") as conn: # Identity preserved. bot = conn.execute( "SELECT id, name, kickoff_prose, initial_relationship_to_you " "FROM bots WHERE id = 'bot_a'" ).fetchone() assert bot is not None assert bot[1] == "BotA" assert bot[2] == "you stay late" assert bot[3] == "coworker" # State purged. assert conn.execute( "SELECT COUNT(*) FROM chats WHERE host_bot_id = 'bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM scenes WHERE chat_id = 'chat_bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM containers WHERE chat_id = 'chat_bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM chat_state WHERE chat_id = 'chat_bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM edges WHERE source_id = 'bot_a' OR target_id = 'bot_a'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM activity WHERE entity_id = 'bot_a'" ).fetchone()[0] == 0 # Event log records the bot_reset event. assert conn.execute( "SELECT COUNT(*) FROM event_log WHERE kind = 'bot_reset'" ).fetchone()[0] == 1 def test_reset_400_when_confirm_name_mismatch(client, tmp_path): _seed_bot_with_state(tmp_path / "test.db") response = client.post( "/bots/bot_a/reset", data={"confirm_name": "WrongName"}, follow_redirects=False, ) assert response.status_code == 400 def test_reset_404_when_bot_missing(client): response = client.post( "/bots/no_such/reset", data={"confirm_name": "Anything"}, follow_redirects=False, ) assert response.status_code == 404 def test_bot_list_renders_reset_form(client, tmp_path): _seed_bot_with_state(tmp_path / "test.db") response = client.get("/bots") assert response.status_code == 200 assert "Reset" in response.text assert "confirm_name" in response.text def _seed_two_bots_with_guest_link( db: Path, *, extra_events: list[dict] | None = None ) -> None: """Seed bot_a + bot_b, each hosting their own chat, with bot_b a guest in chat_bot_a. ``extra_events`` is appended after the guest_added event and projected together with the rest of the seed (so handlers run only once per event). """ with open_db(db) as conn: # bot_a + its chat append_event( conn, kind="bot_authored", payload={ "id": "bot_a", "name": "BotA", "persona": "thoughtful", "voice_samples": [], "traits": [], "backstory": "", "initial_relationship_to_you": "coworker", "kickoff_prose": "", }, ) append_event( conn, kind="chat_created", payload={ "id": "chat_bot_a", "host_bot_id": "bot_a", "initial_time": "2026-04-26T20:00:00+00:00", "narrative_anchor": "Day 1", "weather": "", }, ) # bot_b + its own chat append_event( conn, kind="bot_authored", payload={ "id": "bot_b", "name": "BotB", "persona": "curious", "voice_samples": [], "traits": [], "backstory": "", "initial_relationship_to_you": "friend", "kickoff_prose": "", }, ) append_event( conn, kind="chat_created", payload={ "id": "chat_bot_b", "host_bot_id": "bot_b", "initial_time": "2026-04-26T20:00:00+00:00", "narrative_anchor": "Day 1", "weather": "", }, ) # bot_b joins chat_bot_a as a guest. append_event( conn, kind="guest_added", payload={ "chat_id": "chat_bot_a", "guest_bot_id": "bot_b", }, ) for ev in extra_events or []: append_event(conn, kind=ev["kind"], payload=ev["payload"]) project(conn) def test_reset_clears_guest_reference_in_other_chats(client, tmp_path): db = tmp_path / "test.db" _seed_two_bots_with_guest_link(db) # Sanity-check the seed: bot_b is the guest in bot_a's chat. from chat.state.world import get_chat with open_db(db) as conn: assert get_chat(conn, "chat_bot_a")["guest_bot_id"] == "bot_b" assert get_chat(conn, "chat_bot_b") is not None response = client.post( "/bots/bot_b/reset", data={"confirm_name": "BotB"}, follow_redirects=False, ) assert response.status_code == 303 with open_db(db) as conn: # The guest reference in bot_a's chat is cleared. chat_a = get_chat(conn, "chat_bot_a") assert chat_a is not None assert chat_a["guest_bot_id"] is None # bot_b's own chat is gone (Phase 1 host purge behavior). assert get_chat(conn, "chat_bot_b") is None # bot_a is untouched. assert conn.execute( "SELECT COUNT(*) FROM bots WHERE id = 'bot_a'" ).fetchone()[0] == 1 def test_reset_purges_guest_memories_from_other_chats(client, tmp_path): db = tmp_path / "test.db" _seed_two_bots_with_guest_link( db, extra_events=[ # bot_b is a guest in chat_bot_a and remembers things from there. { "kind": "memory_written", "payload": { "owner_id": "bot_b", "chat_id": "chat_bot_a", "pov_summary": "Met BotA; she was tense.", "witness_you": 1, "witness_host": 1, "witness_guest": 1, "significance": 3, }, }, # And a memory from bot_b's own chat for good measure. { "kind": "memory_written", "payload": { "owner_id": "bot_b", "chat_id": "chat_bot_b", "pov_summary": "A quiet evening at home.", "witness_you": 1, "witness_host": 1, "witness_guest": 0, "significance": 1, }, }, ], ) with open_db(db) as conn: # Sanity: bot_b owns 2 memories pre-reset, one in each chat. assert conn.execute( "SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b'" ).fetchone()[0] == 2 response = client.post( "/bots/bot_b/reset", data={"confirm_name": "BotB"}, follow_redirects=False, ) assert response.status_code == 303 with open_db(db) as conn: # ALL of bot_b's memories are gone, including the cross-chat one in chat_bot_a. assert conn.execute( "SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b'" ).fetchone()[0] == 0 assert conn.execute( "SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b' AND chat_id = 'chat_bot_a'" ).fetchone()[0] == 0