perf: read_recent_dialogue pushes chat-id filter into SQL (T90.1)

The previous implementation pulled the last N rows in SQL across all
chats and dropped foreign-chat rows in Python. With LIMIT N this could
return far fewer than N relevant rows when other chats had recent
activity. Push the chat_id filter into SQL via json_extract so LIMIT N
always returns N rows scoped to the requested chat.

Test: seeds two chats with 60 turns each interleaved; queries chat_a
with limit=50; asserts exactly 50 chat_a rows returned (was 0 prior to
the fix because chat_b's rows dominated the global tail).

chat/services/turn_common.py

@@ -54,14 +54,21 @@ def read_recent_dialogue(
     regenerate to drop the original assistant_turn from its prompt
     context window before that row has been marked superseded (the
     supersede UPDATE lands at the end so the new event_id is known).
+
+    T90.1: the chat_id filter is pushed into SQL via ``json_extract`` so
+    ``LIMIT N`` always returns N rows scoped to the requested chat. The
+    previous implementation filtered chat_id post-fetch in Python, which
+    let foreign-chat rows fill the LIMIT and yield fewer than N relevant
+    rows in busy multi-chat databases.
     """
     if exclude_event_id is None:
         cur = conn.execute(
             "SELECT id, kind, payload_json FROM event_log "
             "WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn') "
             "  AND superseded_by IS NULL AND hidden = 0 "
+            "  AND json_extract(payload_json, '$.chat_id') = ? "
             "ORDER BY id DESC LIMIT ?",
-            (limit,),
+            (chat_id, limit),
         )
     else:
         cur = conn.execute(
@@ -69,15 +76,14 @@ def read_recent_dialogue(
             "WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn') "
             "  AND id != ? "
             "  AND superseded_by IS NULL AND hidden = 0 "
+            "  AND json_extract(payload_json, '$.chat_id') = ? "
             "ORDER BY id DESC LIMIT ?",
-            (exclude_event_id, limit),
+            (exclude_event_id, chat_id, limit),
         )
     rows = list(reversed(cur.fetchall()))
     out: list[dict] = []
     for row_id, kind, payload_json in rows:
         p = json.loads(payload_json)
-        if p.get("chat_id") != chat_id:
-            continue
         if kind in ("user_turn", "user_turn_edit"):
             out.append(
                 {

tests/test_turn_common.py

@@ -186,6 +186,82 @@ def test_read_recent_dialogue_filters_superseded_and_other_chats(tmp_path):
     assert ut_id is not None
 
 
+def test_read_recent_dialogue_limit_respects_chat_scope(tmp_path):
+    """T90.1: ``read_recent_dialogue`` must push the chat_id filter into
+    SQL so that ``LIMIT N`` returns N rows scoped to the requested chat —
+    not N globally-recent rows that may then be filtered down to fewer in
+    Python.
+
+    Setup: two chats with 60 turns each, interleaved. With the old
+    post-fetch filter, ``LIMIT 50`` would pull 50 globally-recent rows
+    (most or all from chat_b — the most recent inserts) and then drop
+    chat_b ones via the Python check, yielding far fewer than 50 chat_a
+    rows. After the SQL pushdown, ``LIMIT 50`` should return exactly 50
+    chat_a rows.
+    """
+    db = tmp_path / "test.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        for chat_id, host_bot in (("chat_a", "bot_a"), ("chat_b", "bot_b")):
+            append_event(
+                conn,
+                kind="bot_authored",
+                payload={
+                    "id": host_bot,
+                    "name": host_bot,
+                    "persona": "...",
+                    "voice_samples": [],
+                    "traits": [],
+                    "backstory": "",
+                    "initial_relationship_to_you": "",
+                    "kickoff_prose": "",
+                },
+            )
+            append_event(
+                conn,
+                kind="chat_created",
+                payload={
+                    "id": chat_id,
+                    "host_bot_id": host_bot,
+                    "initial_time": "2026-04-26T20:00:00+00:00",
+                    "narrative_anchor": "Day 1",
+                    "weather": "",
+                },
+            )
+        # Interleave 60 user_turn rows in each chat — chat_b's go in last
+        # so they dominate the global tail.
+        for i in range(60):
+            append_event(
+                conn,
+                kind="user_turn",
+                payload={
+                    "chat_id": "chat_a",
+                    "prose": f"a-{i}",
+                    "segments": [],
+                },
+            )
+        for i in range(60):
+            append_event(
+                conn,
+                kind="user_turn",
+                payload={
+                    "chat_id": "chat_b",
+                    "prose": f"b-{i}",
+                    "segments": [],
+                },
+            )
+        project(conn)
+
+        out = read_recent_dialogue(conn, "chat_a", limit=50)
+
+    # All returned rows should belong to chat_a (texts a-* only).
+    assert len(out) == 50
+    for entry in out:
+        assert entry["text"].startswith("a-"), (
+            f"foreign chat row leaked: {entry!r}"
+        )
+
+
 def test_gather_prior_edges_fills_missing_with_default(tmp_path):
     """``gather_prior_edges`` returns one entry per directed pair across
     ``present_ids``. Missing rows fall back to the schema default