merge: T70 LLM-merged group meta-summary

chat/db/connection.py

@@ -5,9 +5,9 @@ from pathlib import Path
 
 
 @contextmanager
-def open_db(path: Path):
+def open_db(path: Path, *, check_same_thread: bool = True):
     path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(path)
+    conn = sqlite3.connect(path, check_same_thread=check_same_thread)
     conn.execute("PRAGMA journal_mode=WAL")
     conn.execute("PRAGMA foreign_keys=ON")
     try:

chat/services/scene_summarize.py

@@ -334,26 +334,92 @@ async def apply_scene_close_summary(
             timeout_s=timeout_s,
         )
 
-    # Group node update: naive per-POV concat for v2. Only fires when
+    # Group node update: T70 runs a third classifier call to merge the
-    # both POVs ran (i.e. the guest is present) and a group_node row
+    # two per-POV summaries into a coherent group-level view + a brief
-    # exists for this chat.
+    # group-dynamic note. Falls back to the Phase 2 naive concat on
+    # classifier failure (see :func:`merge_group_summary`). Only fires
+    # when both POVs ran (i.e. the guest is present) and a group_node
+    # row exists for this chat.
     if guest_pov is not None and get_group_node(conn, chat_id) is not None:
         host_bot = get_bot(conn, host_bot_id) or {"name": host_bot_id}
         guest_bot = get_bot(conn, guest_bot_id) or {"name": guest_bot_id}
         host_name = host_bot.get("name", host_bot_id) or host_bot_id
         guest_name = guest_bot.get("name", guest_bot_id) or guest_bot_id
-        group_summary = (
+        merged = await merge_group_summary(
-            f"{host_name}: {host_pov.summary}\n\n"
+            client,
-            f"{guest_name}: {guest_pov.summary}"
+            classifier_model=classifier_model,
+            host_name=host_name,
+            host_pov_summary=host_pov.summary,
+            guest_name=guest_name,
+            guest_pov_summary=guest_pov.summary,
+            timeout_s=timeout_s,
         )
         append_and_apply(
             conn,
             kind="group_node_updated",
             payload={
                 "chat_id": chat_id,
-                "summary": group_summary,
+                "summary": merged.summary,
-                "dynamic": "",
+                "dynamic": merged.dynamic,
             },
         )
 
     return host_pov
+
+
+class GroupMetaSummary(BaseModel):
+    """Classifier output: a merged group-level view of a closed scene.
+
+    Defaults are an empty no-op so callers can use the schema's default
+    as a sentinel; in practice :func:`merge_group_summary` builds an
+    explicit naive-concat fallback rather than returning these defaults
+    directly so existing Phase 2 behavior is preserved on classifier
+    failure.
+    """
+
+    summary: str = ""
+    dynamic: str = ""
+
+
+_GROUP_MERGE_SYSTEM = (
+    "Given two per-POV scene summaries from a 3-entity scene (you + "
+    "host + guest), produce a coherent group-level summary capturing "
+    "the shared events as both witnesses experienced them, plus a "
+    "brief 'dynamic' note describing the trio's group dynamic during "
+    "the scene. Output strict JSON matching schema."
+)
+
+
+async def merge_group_summary(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    host_name: str,
+    host_pov_summary: str,
+    guest_name: str,
+    guest_pov_summary: str,
+    timeout_s: float = 30.0,
+) -> GroupMetaSummary:
+    """Merge two per-POV scene summaries into a coherent group-level
+    summary + group-dynamic note. Falls back to the naive concat (the
+    existing behavior) on classifier failure."""
+    user = (
+        f"{host_name} (host) POV summary:\n{host_pov_summary}\n\n"
+        f"{guest_name} (guest) POV summary:\n{guest_pov_summary}"
+    )
+    fallback = GroupMetaSummary(
+        summary=(
+            f"{host_name}: {host_pov_summary}\n\n"
+            f"{guest_name}: {guest_pov_summary}"
+        ),
+        dynamic="",
+    )
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_GROUP_MERGE_SYSTEM,
+        user=user,
+        schema=GroupMetaSummary,
+        default=fallback,
+        timeout_s=timeout_s,
+    )

chat/state/entities.py

@@ -48,6 +48,17 @@ def _apply_bot_reset(conn: Connection, e: Event) -> None:
             "SELECT id FROM chats WHERE host_bot_id = ?", (bot_id,)
         ).fetchall()
     ]
+    # T69: purge orphaned "you" activity rows pointing at containers in this
+    # bot's chats BEFORE the containers/chats themselves are deleted, otherwise
+    # the subqueries find nothing and the FK constraint on activity.container_id
+    # blocks the container delete.
+    conn.execute(
+        "DELETE FROM activity WHERE entity_id = 'you' "
+        "AND container_id IN (SELECT id FROM containers WHERE chat_id IN ("
+        "    SELECT id FROM chats WHERE host_bot_id = ?"
+        "))",
+        (bot_id,),
+    )
     for chat_id in chat_ids:
         conn.execute("DELETE FROM scenes WHERE chat_id = ?", (chat_id,))
         conn.execute("DELETE FROM containers WHERE chat_id = ?", (chat_id,))
@@ -74,8 +85,6 @@ def _apply_bot_reset(conn: Connection, e: Event) -> None:
         (bot_id,),
     )
     # NOTE: bots row itself is preserved (identity, kickoff_prose intact).
-    # NOTE: "you" activity (entity_id="you") may linger from a deleted chat;
-    # acceptable for v1 — Phase 1.5 cleanup if needed.
 
 
 def get_bot(conn: Connection, bot_id: str) -> dict | None:

chat/web/bots.py

@@ -1,10 +1,10 @@
 from __future__ import annotations
-import sqlite3
 from pathlib import Path
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import RedirectResponse, HTMLResponse
 from fastapi.templating import Jinja2Templates
 
+from chat.db.connection import open_db
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 from chat.state.entities import list_bots
@@ -19,15 +19,8 @@ REQUIRED_FIELDS = ("id", "name", "persona", "initial_relationship_to_you", "kick
 def get_conn(request: Request):
     settings = request.app.state.settings
     db_path: Path = settings.db_path
-    db_path.parent.mkdir(parents=True, exist_ok=True)
+    with open_db(db_path, check_same_thread=False) as conn:
-    conn = sqlite3.connect(db_path, check_same_thread=False)
-    conn.execute("PRAGMA journal_mode=WAL")
-    conn.execute("PRAGMA foreign_keys=ON")
-    try:
         yield conn
-        conn.commit()
-    finally:
-        conn.close()
 
 
 def _split_voice_samples(text: str) -> list[str]:

tests/test_open_db_threading.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+import sqlite3
+import threading
+
+from chat.db.connection import open_db
+
+
+def test_open_db_default_uses_check_same_thread_true(tmp_path):
+    """Default open_db must reject cross-thread use (safe default)."""
+    db = tmp_path / "t.db"
+    captured: list[BaseException | None] = []
+
+    with open_db(db) as conn:
+        conn.execute("CREATE TABLE t (x INTEGER)")
+
+        def worker():
+            try:
+                conn.execute("SELECT 1").fetchall()
+                captured.append(None)
+            except BaseException as e:  # noqa: BLE001
+                captured.append(e)
+
+        t = threading.Thread(target=worker)
+        t.start()
+        t.join()
+
+    assert len(captured) == 1
+    err = captured[0]
+    assert isinstance(err, sqlite3.ProgrammingError), (
+        f"expected sqlite3.ProgrammingError on cross-thread use, got {err!r}"
+    )
+
+
+def test_open_db_can_disable_check_same_thread(tmp_path):
+    """open_db(check_same_thread=False) must allow cross-thread use."""
+    db = tmp_path / "t.db"
+    captured: list[BaseException | None] = []
+    rows: list[object] = []
+
+    with open_db(db, check_same_thread=False) as conn:
+        conn.execute("CREATE TABLE t (x INTEGER)")
+        conn.execute("INSERT INTO t (x) VALUES (42)")
+
+        def worker():
+            try:
+                result = conn.execute("SELECT x FROM t").fetchall()
+                rows.extend(result)
+                captured.append(None)
+            except BaseException as e:  # noqa: BLE001
+                captured.append(e)
+
+        t = threading.Thread(target=worker)
+        t.start()
+        t.join()
+
+    assert captured == [None], f"worker thread raised: {captured}"
+    assert rows == [(42,)]

tests/test_per_pov_summary.py

@@ -636,8 +636,10 @@ async def test_close_with_guest_updates_both_edges(tmp_path):
 @pytest.mark.asyncio
 async def test_close_with_group_node_updates_group_summary(tmp_path):
     """When a group_node row exists, scene close emits group_node_updated
-    with a non-empty summary that mentions both bots' names (v2 naive
+    with a non-empty summary that mentions both bots' names. T70 swapped
-    concat of per-POV summaries)."""
+    the Phase 2 naive concat for an LLM-merged summary; this regression
+    test feeds bad-JSON merge responses so the helper falls back to the
+    original naive-concat shape, preserving the original assertions."""
     db = tmp_path / "t.db"
     apply_migrations(db)
     import chat.state.group_node  # noqa: F401  -- register handlers
@@ -660,7 +662,11 @@ async def test_close_with_group_node_updates_group_summary(tmp_path):
         _seed_two_bot_scene(conn, with_group_node=True)
         project(conn)
 
-        client = MockLLMClient(canned=[host_canned, guest_canned])
+        # 2 valid (host POV, guest POV) + 3 bad-JSON merge attempts ->
+        # merge_group_summary falls back to the naive concat default.
+        client = MockLLMClient(
+            canned=[host_canned, guest_canned, "bad1", "bad2", "bad3"]
+        )
         await apply_scene_close_summary(
             conn,
             client,
@@ -675,8 +681,167 @@ async def test_close_with_group_node_updates_group_summary(tmp_path):
         gn = get_group_node(conn, "chat_bot_a")
         assert gn is not None
         assert gn["summary"]  # non-empty
-        # Naive concat surfaces both bot names in the group summary.
+        # Naive-concat fallback surfaces both bot names in the group summary.
         assert "BotA" in gn["summary"]
         assert "BotB" in gn["summary"]
-        # Phase 2 v2 keeps dynamic empty (Phase 3 polishes).
+        # Naive-concat fallback keeps dynamic empty.
         assert gn["dynamic"] == ""
+
+
+# ---------------------------------------------------------------------------
+# T70: LLM-merged group meta-summary on scene close.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_group_summary_merges_per_pov_via_classifier_when_guest_present(
+    tmp_path,
+):
+    """With a guest present and a group_node row, scene close runs the
+    merge classifier as a third call after the two per-POV summarize_scene
+    calls; its output drives the group_node summary + dynamic fields."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    import chat.state.group_node  # noqa: F401  -- register handlers
+
+    host_canned = json.dumps(
+        {
+            "summary": "BotA appreciated the calm.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotA felt steady.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB found the room friendly.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotB warmed up.",
+        }
+    )
+    merge_canned = json.dumps(
+        {"summary": "merged group view", "dynamic": "warm rapport"}
+    )
+    with open_db(db) as conn:
+        _seed_two_bot_scene(conn, with_group_node=True)
+        project(conn)
+
+        # Canned-queue layout matches the production call order in
+        # apply_scene_close_summary: host POV summarize_scene runs first,
+        # then guest POV summarize_scene, then merge_group_summary.
+        client = MockLLMClient(
+            canned=[host_canned, guest_canned, merge_canned]
+        )
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        # All three canned entries consumed -> classifier ran exactly 3x.
+        assert client._canned == []
+
+        from chat.state.group_node import get_group_node
+
+        gn = get_group_node(conn, "chat_bot_a")
+        assert gn is not None
+        assert gn["summary"] == "merged group view"
+        assert gn["dynamic"] == "warm rapport"
+
+
+@pytest.mark.asyncio
+async def test_group_summary_falls_back_to_naive_concat_on_classifier_failure(
+    tmp_path,
+):
+    """If the merge classifier flaps (bad JSON across all 3 retries), the
+    helper falls back to the original Phase 2 naive concat shape and
+    leaves dynamic empty."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    import chat.state.group_node  # noqa: F401  -- register handlers
+
+    host_canned = json.dumps(
+        {
+            "summary": "BotA appreciated the calm.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotA felt steady.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB found the room friendly.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotB warmed up.",
+        }
+    )
+    with open_db(db) as conn:
+        _seed_two_bot_scene(conn, with_group_node=True)
+        project(conn)
+
+        # 2 valid POV summaries + 3 bad-JSON merge attempts trip the
+        # classifier's retry-then-default path; the default is the naive
+        # concat fallback.
+        client = MockLLMClient(
+            canned=[host_canned, guest_canned, "bad1", "bad2", "bad3"]
+        )
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        from chat.state.group_node import get_group_node
+
+        gn = get_group_node(conn, "chat_bot_a")
+        assert gn is not None
+        expected = (
+            "BotA: BotA appreciated the calm.\n\n"
+            "BotB: BotB found the room friendly."
+        )
+        assert gn["summary"] == expected
+        assert gn["dynamic"] == ""
+
+
+@pytest.mark.asyncio
+async def test_group_summary_skipped_when_no_guest(tmp_path):
+    """No-guest path: scene close does NOT invoke merge_group_summary
+    and emits no group_node_updated event. Confirms the existing
+    `if guest_bot_id is not None` gating at the call site."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    canned = json.dumps(
+        {
+            "summary": "BotA helped you talk through the deadline anxiety.",
+            "knowledge_facts": ["Deadline next Friday."],
+            "relationship_summary": "BotA leaned in supportively.",
+        }
+    )
+    with open_db(db) as conn:
+        _seed_single_bot_scene(conn)
+        project(conn)
+
+        # Only 1 canned entry; if merge_group_summary were called the
+        # MockLLMClient would IndexError on the empty queue.
+        client = MockLLMClient(canned=[canned])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        # Exactly the host POV call consumed, nothing else.
+        assert client._canned == []
+
+        # No group_node_updated event was emitted.
+        rows = conn.execute(
+            "SELECT 1 FROM event_log WHERE kind = 'group_node_updated'"
+        ).fetchall()
+        assert rows == []

tests/test_reset.py

@@ -292,6 +292,189 @@ def test_reset_clears_guest_reference_in_other_chats(client, tmp_path):
         ).fetchone()[0] == 1
 
 
+def test_reset_purges_orphaned_you_activity_rows(client, tmp_path):
+    """T69: when a bot's chats are deleted, "you" activity rows tied to those
+    chats' containers should also be purged (otherwise they linger orphaned)."""
+    db = tmp_path / "test.db"
+    with open_db(db) as conn:
+        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": "",
+            },
+        )
+        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",
+                "container_id": 1,
+                "posture": "standing",
+                "action": {"verb": "watching"},
+            },
+        )
+        project(conn)
+        # Sanity: the "you" activity row exists and points at the container.
+        assert conn.execute(
+            "SELECT COUNT(*) FROM activity WHERE entity_id = 'you'"
+        ).fetchone()[0] == 1
+
+    response = client.post(
+        "/bots/bot_a/reset",
+        data={"confirm_name": "BotA"},
+        follow_redirects=False,
+    )
+    assert response.status_code == 303
+
+    with open_db(db) as conn:
+        # The orphaned "you" activity row tied to bot_a's purged container is gone.
+        assert conn.execute(
+            "SELECT COUNT(*) FROM activity WHERE entity_id = 'you'"
+        ).fetchone()[0] == 0
+
+
+def test_reset_does_not_purge_you_activity_in_other_chats(client, tmp_path):
+    """T69: resetting bot_a must leave a "you" activity row pointing at
+    bot_b's container intact — only orphans from the reset bot's chats go."""
+    db = tmp_path / "test.db"
+    with open_db(db) as conn:
+        # bot_a + its chat + container.
+        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": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="container_created",
+            payload={
+                "chat_id": "chat_bot_a",
+                "name": "office",
+                "type": "workplace",
+                "properties": {},
+            },
+        )
+        # bot_b + its chat + container.
+        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": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="container_created",
+            payload={
+                "chat_id": "chat_bot_b",
+                "name": "kitchen",
+                "type": "home",
+                "properties": {},
+            },
+        )
+        # The activity table is keyed on entity_id (PRIMARY KEY), so only one
+        # "you" row exists at a time. Point it at bot_b's container so reset of
+        # bot_a should NOT touch it.
+        append_event(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "you",
+                "container_id": 2,  # kitchen, in chat_bot_b
+                "posture": "sitting",
+                "action": {"verb": "reading"},
+            },
+        )
+        project(conn)
+        # Sanity: the "you" activity row is in bot_b's container.
+        row = conn.execute(
+            "SELECT container_id FROM activity WHERE entity_id = 'you'"
+        ).fetchone()
+        assert row is not None and row[0] == 2
+
+    response = client.post(
+        "/bots/bot_a/reset",
+        data={"confirm_name": "BotA"},
+        follow_redirects=False,
+    )
+    assert response.status_code == 303
+
+    with open_db(db) as conn:
+        # The "you" activity in bot_b's container is preserved.
+        row = conn.execute(
+            "SELECT container_id FROM activity WHERE entity_id = 'you'"
+        ).fetchone()
+        assert row is not None
+        assert row[0] == 2
+
+
 def test_reset_purges_guest_memories_from_other_chats(client, tmp_path):
     db = tmp_path / "test.db"
     _seed_two_bots_with_guest_link(