"""Scene close on hard signals + manual override (T26). A small classifier service decides whether the user's prose narrates a "hard signal" that should close the active scene (container change, explicit "fade out" / "we're done here" patterns). Wired into the turn flow AFTER the assistant_turn so the bot's response is the final beat in the closing scene. The drawer also exposes a manual "Close scene" button that always fires a ``scene_closed`` event. Per Task 26 we DO NOT auto-open a new scene on close — the next interaction either lives in a fresh chat or operates without an active scene; the prompt assembler already tolerates ``active_scene == None``. """ from __future__ import annotations import json 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 from chat.llm.mock import MockLLMClient from chat.services.scene_close import detect_scene_close # --------------------------------------------------------------------------- # Service-level tests (no FastAPI involvement). # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_detect_scene_close_returns_decision(): canned = json.dumps( { "should_close": True, "reason": "container change", "new_container_hint": "park", } ) mock = MockLLMClient(canned=[canned]) decision = await detect_scene_close( mock, model="x", prose="we drove to the park", current_container_name="office", ) assert decision.should_close is True assert "container" in decision.reason @pytest.mark.asyncio async def test_detect_scene_close_default_on_failure(): """Two consecutive non-JSON returns trip the classifier's retry-then-default path; we should get the safe ``should_close=False`` fallback rather than crashing the turn flow.""" mock = MockLLMClient(canned=["nope", "still nope", "nope3"]) decision = await detect_scene_close( mock, model="x", prose="anything", current_container_name="office", ) assert decision.should_close is False # --------------------------------------------------------------------------- # HTTP integration: turn flow + manual close. # --------------------------------------------------------------------------- @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)) # Order of canned responses for one POST /turns: # 1. parse_turn classifier # 2. narrative streamer # 3. state_update bot->you # 4. state_update you->bot # 5. detect_scene_close (runs AFTER assistant_turn — see turns.py) # 6. summarize_scene (T27, runs only when scene-close fires) parse_canned = json.dumps( {"segments": [{"kind": "dialogue", "text": "hello"}]} ) narrative_canned = "BotA grins." state_update_canned = json.dumps( {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []} ) scene_close_canned = json.dumps( { "should_close": True, "reason": "container change", "new_container_hint": "park", } ) pov_summary_canned = json.dumps( { "summary": "BotA noticed you leaving the office.", "knowledge_facts": [], "relationship_summary": "BotA wonders where you're headed.", } ) from chat.web.kickoff import get_llm_client mock = MockLLMClient( canned=[ parse_canned, narrative_canned, state_update_canned, state_update_canned, scene_close_canned, pov_summary_canned, ] ) app.dependency_overrides[get_llm_client] = lambda: mock with TestClient(app) as c: # Same as other turn-flow tests: keep the async significance worker # off so it doesn't try to call Featherless with the test API key. app.state.background_worker.enabled = False yield c app.dependency_overrides.clear() def _seed(db_path: Path, *, with_scene: bool = True) -> None: """Seed enough state for a full turn flow plus an active scene.""" with open_db(db_path) as conn: append_event( conn, kind="bot_authored", payload={ "id": "bot_a", "name": "BotA", "persona": "thoughtful, observant", "voice_samples": [], "traits": [], "backstory": "", "initial_relationship_to_you": "", "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": "", }, ) append_event( conn, kind="container_created", payload={ "chat_id": "chat_bot_a", "name": "office", "type": "workplace", "properties": {}, }, ) append_event( conn, kind="activity_change", payload={ "entity_id": "you", "posture": "sitting", "action": { "verb": "thinking", "interruptible": True, "required_attention": "low", "expected_duration": "ongoing", }, "attention": "", "holding": [], "status": {}, }, ) append_event( conn, kind="activity_change", payload={ "entity_id": "bot_a", "posture": "standing", "action": { "verb": "watching", "interruptible": True, "required_attention": "low", "expected_duration": "ongoing", }, "attention": "", "holding": [], "status": {}, }, ) append_event( conn, kind="edge_update", payload={ "source_id": "bot_a", "target_id": "you", "chat_id": "chat_bot_a", "knowledge_facts": ["coworker"], }, ) append_event( conn, kind="edge_update", payload={ "source_id": "you", "target_id": "bot_a", "chat_id": "chat_bot_a", "knowledge_facts": [], }, ) if with_scene: 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"], }, ) project(conn) def test_post_turn_closes_scene_on_container_change(client, tmp_path): _seed(tmp_path / "test.db") response = client.post( "/chats/chat_bot_a/turns", data={"prose": "we drove to the park"} ) assert response.status_code == 204 with open_db(tmp_path / "test.db") as conn: # scene_closed event present. cur = conn.execute( "SELECT COUNT(*) FROM event_log WHERE kind = 'scene_closed'" ) assert cur.fetchone()[0] == 1 # Active scene cleared by the projector. from chat.state.world import active_scene assert active_scene(conn, "chat_bot_a") is None # Order: assistant_turn lands BEFORE scene_closed (the bot's reply is # the closing scene's final beat). cur = conn.execute( "SELECT kind FROM event_log " "WHERE kind IN ('assistant_turn', 'scene_closed') ORDER BY id" ) kinds = [r[0] for r in cur.fetchall()] assert kinds == ["assistant_turn", "scene_closed"] def test_manual_close_scene_button(client, tmp_path): _seed(tmp_path / "test.db") response = client.post("/chats/chat_bot_a/drawer/scene/close") assert response.status_code == 200 with open_db(tmp_path / "test.db") as conn: cur = conn.execute( "SELECT COUNT(*) FROM event_log WHERE kind = 'scene_closed'" ) assert cur.fetchone()[0] == 1 from chat.state.world import active_scene assert active_scene(conn, "chat_bot_a") is None def test_manual_close_400_when_no_active_scene(client, tmp_path): _seed(tmp_path / "test.db", with_scene=False) response = client.post("/chats/chat_bot_a/drawer/scene/close") assert response.status_code == 400