dohertj2/chat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: scene close on hard signals with manual override

Browse Source
This commit is contained in:
Joseph Doherty
2026-04-26 13:46:14 -04:00
parent db3005fc17
commit 0997562e75
7 changed files with 482 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_memory_write.py
+7
View File
@@ -156,6 +156,12 @@ def client(tmp_path, monkeypatch):
canned_state_update = json.dumps(
{"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
)
# T26 scene-close detection runs after the state-update pass. ``_seed_full``
# below doesn't open a scene so the classifier call is short-circuited in
# turns.py — but the canned slot stays in place to document the order.
canned_scene_close = json.dumps(
{"should_close": False, "reason": "no signal"}
)
from chat.web.kickoff import get_llm_client
@@ -165,6 +171,7 @@ def client(tmp_path, monkeypatch):
canned_response,
canned_state_update,
canned_state_update,
canned_scene_close,
]
)
app.dependency_overrides[get_llm_client] = lambda: mock
tests/test_scene_close.py
+278
View File
@@ -0,0 +1,278 @@
"""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"])
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)
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",
}
)
from chat.web.kickoff import get_llm_client
mock = MockLLMClient(
canned=[
parse_canned,
narrative_canned,
state_update_canned,
state_update_canned,
scene_close_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
tests/test_turn_flow.py
+8
View File
@@ -43,6 +43,13 @@ def client(tmp_path, monkeypatch):
canned_state_update = json.dumps(
{"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
)
# T26 scene-close detection runs after the state-update pass. These
# tests don't seed an active scene so the classifier is short-circuited
# in turns.py — but the canned slot is harmless to leave in place,
# and adding it documents the order even when the call doesn't fire.
canned_scene_close = json.dumps(
{"should_close": False, "reason": "no signal"}
)
# Import here so env vars are visible to the dependency lookup.
from chat.web.kickoff import get_llm_client
@@ -53,6 +60,7 @@ def client(tmp_path, monkeypatch):
canned_response,
canned_state_update,
canned_state_update,
canned_scene_close,
]
)
app.dependency_overrides[get_llm_client] = lambda: mock