docs: add T66-discovered consume_pending_meanwhile_digests backlog item

CLAUDE.md

@@ -213,3 +213,87 @@ New follow-ups discovered during Phase 2.5 execution. None are blocking; pick up
 - **`_witness_role_for` defensive coding (from T71 review)**: helper returns `"guest"` when `host_bot_id is None`, which is wrong for Phase-1 chats. Defensive: `return "host" if host_bot_id is None or speaker_bot_id == host_bot_id else "guest"`. Not exercised by current tests; harden as a precaution.
 - **Confidence type tightening (from T74 review)**: `chat/services/addressee.py::AddresseeDecision.confidence` could be typed as `Literal["high","medium","low"]` for stricter validation. Currently `str` with a comment.
 - **Scene-close-on-cancel UX revisit**: T74.3 pinned the existing behavior (close fires even on cancel). If real play-testing surfaces a regression, revisit.
+
+## Phase 3 status
+
+Phase 3 shipped end-to-end across 19 tasks (T49–T67). Events with full lifecycle, time skips, active threads, significance refinements, and meanwhile scenes are functional. Schema baseline is now version 11 (migrations 0009 events, 0010 threads, 0011 meanwhile_scenes). Test count grew from ~247 (Phase 2) to ~315 (+68 new tests across the wave).
+
+- **Wave 1 — schema + lifecycle handlers (parallel)**:
+  - **T49** `events` table + lifecycle handlers (`event_planned`, `event_started`, `event_completed`, `event_cancelled`, `event_expired`).
+  - **T50** `time_skip` event handlers (elision and jump variants).
+  - **T51** `threads` table + handlers (`thread_opened`, `thread_updated`, `thread_closed`).
+- **Wave 2 — detection / narration services (parallel)**:
+  - **T52** event-lifecycle detection service (planned→active→completed transitions inferred from narration).
+  - **T53** skip narration service (elision + jump prose).
+  - **T54** synthesized-memories service for jump skips (LLM-summarized intervening time).
+  - **T55** thread-detection service (open/update/close inferred from recent dialogue).
+- **Wave 3 — promotion + ranking (parallel)**:
+  - **T56** event-completion promotion service (objects → inventory, knowledge → edge knowledge, relationship deltas → edge summary; everything else stays in the closed event).
+  - **T57** significance-aware retrieval ranking — SQL-side `SIGNIFICANCE_RANK_BIAS` plus the existing Python composite re-rank.
+  - **T58** scene compression keeps key quotes when significance ≥ 2; thread emission piggybacks on scene close.
+- **Wave 4 — drawer UX (single)**:
+  - **T59** drawer additions: events panel, threads panel, skip controls.
+- **Wave 5a — prompt + turn flow integration (parallel)**:
+  - **T60** prompt assembly includes active events + open threads in the speaker's prompt.
+  - **T61** turn flow invokes event-detection + completion promotion alongside existing post-turn fan-out.
+- **Wave 5b — natural-language skip surface (single)**:
+  - **T62** classifier-driven skip command at the user-input layer; shared skip controllers extracted into `chat/web/skip.py`.
+- **Wave 6a — meanwhile schema (single)**:
+  - **T63** meanwhile-scene schema + state (scene config 4: host+guest, no "you").
+- **Wave 6b — meanwhile turn flow (parallel)**:
+  - **T64** meanwhile turn flow (host+guest, no "you" in the prompt or witness writes).
+  - **T65** meanwhile summary digest surfaces to the next "you"-present scene.
+- **Wave 7 — integration + docs (parallel)**:
+  - **T66** cross-feature integration tests covering events × skips × threads × meanwhile.
+  - **T67** documentation (this section).
+
+### Phase 3.5 / 4 backlog
+
+New follow-ups discovered during Phase 3 reviews and execution. None are blocking; pick up at any time.
+
+#### From T53 review
+
+- **`narrate_skip` `timeout_s` not piped through to `client.generate`**: parameter accepted but ignored. Fix: pass `timeout_s=timeout_s` to `client.generate(**...)`, or drop the parameter entirely if Featherless's client doesn't honor it.
+
+#### From T57 review
+
+- **`search_memories` docstring should mention SQL-side significance bias**: the function docstring still describes only the Python composite re-rank; add a one-line note about `SIGNIFICANCE_RANK_BIAS`.
+
+#### From T58 review
+
+- **Scene close re-close suffix bloat risk**: `_build_key_quotes_suffix` reads from `memories.pov_summary`. If a scene close runs twice, the second pass would read the rewritten text plus the previous "Key quotes:" suffix and append a second one. Either guard for double-suffix or source quotes from `event_log` `assistant_turn`/`user_turn` text instead.
+- **Thread detection transcript scoping**: `_read_recent_dialogue` returns chat-wide history with no `scene_id` filter (Phase 1 turns lack one). Feeding chat-wide history to `detect_threads` will misattribute threads to the closing scene when the scene boundary falls inside the last 50 turns. Scope by `scene_id` once turns carry it, or by `started_at` against scene-open timestamp.
+- **Swallowed exceptions in `detect_threads` try/except**: bare `Exception` swallows programmer errors silently. Log at debug level so silent regressions are recoverable.
+- **Scene close `closed_at` clock divergence**: T58 uses `datetime.now(timezone.utc).isoformat()` instead of chat-clock time. Diverges from chat-clock semantics elsewhere; revisit if event reconstructions need chat-clock ordering.
+- **Test coverage gaps in T58**: no test for 200-char quote truncation; no test for `thread_updated`/`thread_closed` candidate paths; no test for the `try/except` fallback.
+
+#### From T61 review
+
+- **Regenerate doesn't roll back lifecycle transitions from superseded turn**: `event_started`/`event_completed` rows from a superseded turn remain. Phase 3.5 should add a lifecycle-undo step. Caveat: regenerate-after-completion may double-emit promotion artifacts if the new text re-completes the same event.
+- **Asymmetry in event-detection ordering**: post_turn runs lifecycle BETWEEN interjection and scene-close; regenerate runs lifecycle at the END. Benign because regenerate has no scene-close path, but worth tidying.
+
+#### From T62 review
+
+- **Error-message prefix sniff for 404 vs 400 routing**: drawer skip routes use `str(exc).startswith("chat not found")` to distinguish 404 from 400. Fragile if error wording changes. Use a typed exception subclass.
+- **Skip command bypasses scene close detection**: a user typing "fade out, skip an hour" would skip without closing the scene. Acceptable for Phase 3 but worth noting.
+
+#### From T63 review
+
+- **`participants_json` JSON injection** (FIXED in T63 but worth noting in backlog as a "double-check other JSON-string-build sites" task): T63 originally used f-string interpolation; fixed to use `json.dumps`. Audit other state modules for similar patterns.
+
+#### From T64 review
+
+- **`record_meanwhile_memory` and `record_turn_memory_for_present` share private `_write_one_memory` helper**: minor DRY note; both helpers are similar enough that a unified API with a `you_present: bool` kwarg might be cleaner long-term.
+- **Stop button cancellation for meanwhile turns**: T64 fix-up registered tasks in `_in_flight_tasks`; verify the `/turns/cancel` endpoint actually cancels meanwhile streams (the test pins registration but not the cancel-from-route path).
+
+#### From cross-feature interactions discovered in Wave 6b merge
+
+- **Cross-feature canned-queue brittleness**: meanwhile-scene close test required a canned response for T65's digest call after T64+T65 merge. Future close-path additions will keep extending the queue; consider a structured fixture builder rather than positional canned arrays.
+
+#### From T66 integration tests
+
+- **`consume_pending_meanwhile_digests` is defined but NOT wired into `post_turn`**: the helper lives in `chat/services/prompt.py` (T65) but `chat/web/turns.py` never calls it. Meanwhile digests stay pending forever in production. Phase 3.5 should call the helper after the first you-turn following a meanwhile close — probably right after the assistant_turn lands but before the next prompt assembly. Pinned by `tests/test_phase3_integration.py::test_meanwhile_close_digest_surfaces_then_consumed` which currently calls the helper directly.
+
+#### Discovered during Phase 3 execution
+
+- **`_witness_role_for` defensive `host_bot_id is None`** (carry-over from Phase 2.5 T71 backlog) — still pending.

chat/db/migrations/0011_meanwhile_scenes.sql

@@ -0,0 +1,27 @@
+-- T63: Meanwhile scene support — extend scenes with a present-set discriminator
+-- and a parent link (you-scene -> meanwhile child), plus a pending-digest queue.
+--
+-- Existing scenes table (0007) has columns:
+--   id, chat_id, container_id, started_at, ended_at, significance,
+--   participants_json
+-- It has no `status` / `closed_at` columns. We treat `ended_at IS NULL` as
+-- "active" and `ended_at IS NOT NULL` as "closed" — consistent with the
+-- existing scene_opened/scene_closed handlers.
+
+ALTER TABLE scenes ADD COLUMN present_set_kind TEXT NOT NULL DEFAULT 'you_host';
+ALTER TABLE scenes ADD COLUMN parent_scene_id INTEGER;
+
+CREATE INDEX scenes_present_set_idx
+    ON scenes(chat_id, present_set_kind, ended_at);
+
+CREATE TABLE meanwhile_digest_pending (
+    id INTEGER PRIMARY KEY,
+    scene_id INTEGER NOT NULL,
+    chat_id TEXT NOT NULL,
+    summary TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT (datetime('now')),
+    consumed_at TEXT
+);
+
+CREATE INDEX meanwhile_digest_chat_idx
+    ON meanwhile_digest_pending(chat_id) WHERE consumed_at IS NULL;

chat/services/memory_write.py

@@ -176,3 +176,60 @@ def record_turn_memory_for_present(
             significance=significance,
         )
     return result
+
+
+def record_meanwhile_memory(
+    conn: Connection,
+    *,
+    chat_id: str,
+    host_bot_id: str,
+    guest_bot_id: str,
+    narrative_text: str,
+    scene_id: int | None = None,
+    chat_clock_at: str | None = None,
+    source: str = "direct",
+    significance: int = 1,
+) -> dict[str, tuple[int, int | None]]:
+    """Write per-POV ``memory_written`` events for a meanwhile turn (T64).
+
+    A meanwhile scene runs entirely between host + guest, with "you"
+    absent. Both bots are present witnesses, so each one gets a row with
+    witness flags ``[you=0, host=1, guest=1]`` — different from the
+    normal-turn ``record_turn_memory_for_present`` shape, which assumes
+    the user is always a witness (``witness_you=1``).
+
+    The ``guest_bot_id`` is required (a meanwhile scene by definition
+    has both bots) — callers passing ``None`` is a programming error.
+
+    Returns ``{bot_id: (event_id, memory_id)}`` mirroring
+    :func:`record_turn_memory_for_present` so downstream queues
+    (significance scoring) can pull memory ids without re-querying.
+    """
+    result: dict[str, tuple[int, int | None]] = {}
+    result[host_bot_id] = _write_one_memory(
+        conn,
+        owner_id=host_bot_id,
+        chat_id=chat_id,
+        narrative_text=narrative_text,
+        witness_you=0,
+        witness_host=1,
+        witness_guest=1,
+        scene_id=scene_id,
+        chat_clock_at=chat_clock_at,
+        source=source,
+        significance=significance,
+    )
+    result[guest_bot_id] = _write_one_memory(
+        conn,
+        owner_id=guest_bot_id,
+        chat_id=chat_id,
+        narrative_text=narrative_text,
+        witness_you=0,
+        witness_host=1,
+        witness_guest=1,
+        scene_id=scene_id,
+        chat_clock_at=chat_clock_at,
+        source=source,
+        significance=significance,
+    )
+    return result

chat/services/prompt.py

@@ -39,6 +39,7 @@ from chat.state.edges import get_edge, list_edges_for
 from chat.state.entities import get_bot, get_you
 from chat.state.events import list_active_events
 from chat.state.group_node import get_group_node
+from chat.state.meanwhile import list_pending_meanwhile_digests
 from chat.state.memory import search_memories
 from chat.state.threads import list_open_threads
 from chat.state.world import (
@@ -277,6 +278,31 @@ def _build_active_events_block(events: list[dict]) -> str | None:
     return "\n".join(lines)
 
 
+def _build_meanwhile_digests_block(digests: list[dict]) -> str | None:
+    """Render the ``Meanwhile while you were away:`` block for T65.
+
+    One bullet per pending digest, formatted as ``- {summary}`` with the
+    summary truncated to ~200 characters per spec. Returns ``None`` when
+    there are no pending digests so the caller can omit the entire block.
+
+    The block is rendered ONLY when the prompt is for a regular you-scene
+    (``present_set_kind != "host_guest"``); the caller filters before
+    constructing the digests list.
+    """
+    if not digests:
+        return None
+    lines = ["Meanwhile while you were away:"]
+    for d in digests:
+        summary = d.get("summary") or ""
+        if len(summary) > 200:
+            summary = summary[:199] + "…"
+        if summary:
+            lines.append(f"- {summary}")
+    if len(lines) == 1:
+        return None
+    return "\n".join(lines)
+
+
 def _build_open_threads_block(threads: list[dict]) -> str | None:
     """Render the ``Open threads:`` block for Phase 3 Task 60.
 
@@ -393,6 +419,7 @@ def assemble_narrative_prompt(
     budget_hard: int = 8000,
     encoding_name: str = "cl100k_base",
     guest_id: str | None = None,
+    present_set_kind: str = "you_host",
 ) -> list[Message]:
     """Assemble the narrative prompt for ``speaker_bot_id`` to respond.
 
@@ -431,6 +458,14 @@ def assemble_narrative_prompt(
     you = get_you(conn)
     addressee_id, addressee_name = _resolve_addressee(conn, addressee, you)
 
+    # T64: meanwhile-mode marker. When present_set_kind == "host_guest"
+    # the user ("you") is NOT a witness in the scene — bots speak only to
+    # each other. The local flag below is consumed by the activity-block
+    # builder (skip the "you" bullet entirely) and the other-edges filter
+    # (drop any speaker -> "you" rendering). Default "you_host" preserves
+    # the Phase 1/2/3 behavior for normal turns.
+    _exclude_you = present_set_kind == "host_guest"
+
     # ---- Build all components as text strings ------------------------------
 
     speaker_identity = _build_speaker_identity(bot)
@@ -453,6 +488,7 @@ def assemble_narrative_prompt(
     # header that Phase 2 T43 introduced (read by some LLMs as a
     # duplicate-section bug).
     you_activity: dict | None = None
+    if not _exclude_you:
         you_act = get_activity(conn, "you")
         if you_act is not None:
             you_activity = dict(you_act)
@@ -519,6 +555,32 @@ def assemble_narrative_prompt(
         list_open_threads(conn, chat_id)
     )
 
+    # SHOULD-tier meanwhile digest (Phase 3 / Task 65). Only surfaces
+    # when the prompt is for a regular you-scene (NOT for a meanwhile
+    # child scene — the absent player is the audience, not the bots
+    # currently mid-meanwhile). We distinguish via the chat's active
+    # scene's ``present_set_kind``; a missing scene row defaults to a
+    # you-scene render so the block can still surface during the
+    # post-meanwhile-close transition before the next scene opens.
+    #
+    # Consumption is INTENTIONALLY left to the post_turn flow (a
+    # ``consume_pending_meanwhile_digests`` helper, see below) rather
+    # than emitted inline here. Surfacing has no side-effects; the
+    # caller appends ``meanwhile_digest_consumed`` after the response
+    # streams. This keeps prompt assembly pure and deterministic — the
+    # Phase 1 invariant T29's regenerate flow relies on.
+    meanwhile_digests_block: str | None = None
+    active_scene_kind: str | None = None
+    if chat.get("active_scene_id"):
+        active_sc = get_scene(conn, chat["active_scene_id"])
+        if active_sc:
+            active_scene_kind = active_sc.get("present_set_kind")
+    if active_scene_kind != "host_guest":
+        pending_digests = list_pending_meanwhile_digests(conn, chat_id)
+        meanwhile_digests_block = _build_meanwhile_digests_block(
+            pending_digests
+        )
+
     container = None
     if chat.get("active_scene_id"):
         scene = get_scene(conn, chat["active_scene_id"])
@@ -530,9 +592,15 @@ def assemble_narrative_prompt(
         container = get_container(conn, scene["container_id"])
     scene_block = _build_scene_block(chat, container, scene)
 
-    # Other edges: speaker → non-addressee.
+    # Other edges: speaker → non-addressee. In meanwhile mode (host_guest)
+    # the speaker -> "you" edge is filtered out as well — "you" isn't
+    # part of the present set, so surfacing the speaker's relationship
+    # to the user inside a private bot-to-bot beat would leak context
+    # the bots aren't supposed to be drawing on right now.
     all_outgoing = list_edges_for(conn, speaker_bot_id)
     other_edges_raw = [e for e in all_outgoing if e.get("target_id") != addressee_id]
+    if _exclude_you:
+        other_edges_raw = [e for e in other_edges_raw if e.get("target_id") != "you"]
     for e in other_edges_raw:
         tid = e.get("target_id")
         if tid == "you":
@@ -616,6 +684,7 @@ def assemble_narrative_prompt(
         include_group_node: bool = True,
         include_active_events: bool = True,
         include_open_threads: bool = True,
+        include_meanwhile_digests: bool = True,
     ) -> tuple[str, int, list[dict]]:
         # dialogue: keep the last `dialogue_keep` turns verbatim; older
         # turns become an "earlier:" placeholder line.
@@ -653,6 +722,10 @@ def assemble_narrative_prompt(
             group_node_block if include_group_node else None,
             active_events_block if include_active_events else None,
             open_threads_block if include_open_threads else None,
+            (
+                meanwhile_digests_block
+                if include_meanwhile_digests else None
+            ),
             prev_block,
             memories_block,
             dialogue_block,
@@ -674,10 +747,12 @@ def assemble_narrative_prompt(
     include_group_node = group_node_block is not None
     include_active_events = active_events_block is not None
     include_open_threads = open_threads_block is not None
+    include_meanwhile_digests = meanwhile_digests_block is not None
 
     def _build(*, prev: bool, mem_k: int, dlg: int, other: bool,
                you_act: bool, guest_act: bool, group: bool,
-               events: bool, threads: bool) -> tuple[str, int]:
+               events: bool, threads: bool,
+               digests: bool) -> tuple[str, int]:
         body, total, _ = assemble(
             include_other_edges=other,
             include_previous_scene=prev,
@@ -688,6 +763,7 @@ def assemble_narrative_prompt(
             include_group_node=group,
             include_active_events=events,
             include_open_threads=threads,
+            include_meanwhile_digests=digests,
         )
         return body, total
 
@@ -696,6 +772,7 @@ def assemble_narrative_prompt(
         other=include_other, you_act=include_you_activity,
         guest_act=include_guest_activity, group=include_group_node,
         events=include_active_events, threads=include_open_threads,
+        digests=include_meanwhile_digests,
     )
 
     # If under soft, we're done.
@@ -731,6 +808,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -742,6 +820,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -753,6 +832,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -765,18 +845,32 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     # Drop SHOULD-tier extras in order:
-    #   1. open threads block  (T60: SHOULD-tier; least critical to the
-    #      speaker's immediate voice — drop first among SHOULD)
-    #   2. active events block (T60: same tier, drops next)
-    #   3. guest activity bullet (T71.2: bullet-level trim within the
+    #   1. meanwhile digests block (T65: SHOULD-tier; refers to a past
+    #      meanwhile scene — least critical to the speaker's immediate
+    #      voice, so dropped first among SHOULD)
+    #   2. open threads block  (T60: SHOULD-tier; least critical to the
+    #      speaker's immediate voice — drop next among SHOULD)
+    #   3. active events block (T60: same tier, drops next)
+    #   4. guest activity bullet (T71.2: bullet-level trim within the
     #      single ACTIVITIES: block — guest goes first per Task 43 spec)
-    #   4. group node block
-    #   5. you activity bullet (still SHOULD-tier; speaker bullet is the
+    #   5. group node block
+    #   6. you activity bullet (still SHOULD-tier; speaker bullet is the
     #      MUST-tier floor and never dropped)
-    #   6. other edges
+    #   7. other edges
+    if include_meanwhile_digests and total > budget_hard:
+        include_meanwhile_digests = False
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, you_act=include_you_activity,
+            guest_act=include_guest_activity, group=include_group_node,
+            events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
+        )
+
     if include_open_threads and total > budget_hard:
         include_open_threads = False
         body, total = _build(
@@ -784,6 +878,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if include_active_events and total > budget_hard:
@@ -793,6 +888,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if include_guest_activity and total > budget_hard:
@@ -802,6 +898,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if include_group_node and total > budget_hard:
@@ -811,6 +908,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if include_you_activity and total > budget_hard:
@@ -820,6 +918,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if include_other and total > budget_hard:
@@ -829,6 +928,7 @@ def assemble_narrative_prompt(
             other=include_other, you_act=include_you_activity,
             guest_act=include_guest_activity, group=include_group_node,
             events=include_active_events, threads=include_open_threads,
+            digests=include_meanwhile_digests,
         )
 
     if total > budget_hard:
@@ -854,4 +954,38 @@ def _emit(system_body: str, user_turn_prose: str | None) -> list[Message]:
     return msgs
 
 
-__all__ = ["assemble_narrative_prompt"]
+def consume_pending_meanwhile_digests(conn: Connection, chat_id: str) -> int:
+    """Mark every pending meanwhile digest for ``chat_id`` as consumed.
+
+    Called by the post_turn flow AFTER the assistant response streams,
+    once for the first you-turn that surfaced any pending digests. We
+    keep this side-effect out of :func:`assemble_narrative_prompt` so
+    prompt assembly stays pure (T29's regenerate flow rebuilds prompts
+    repeatedly without state mutation).
+
+    Returns the number of digests consumed (0 when none were pending).
+    """
+    from datetime import datetime, timezone
+
+    from chat.eventlog.log import append_and_apply
+
+    pending = list_pending_meanwhile_digests(conn, chat_id)
+    if not pending:
+        return 0
+    now = datetime.now(timezone.utc).isoformat()
+    for d in pending:
+        append_and_apply(
+            conn,
+            kind="meanwhile_digest_consumed",
+            payload={
+                "digest_id": d["id"],
+                "consumed_at": now,
+            },
+        )
+    return len(pending)
+
+
+__all__ = [
+    "assemble_narrative_prompt",
+    "consume_pending_meanwhile_digests",
+]

chat/services/scene_summarize.py

@@ -342,7 +342,7 @@ async def apply_scene_close_summary(
     from chat.state.entities import get_bot, get_you
     from chat.state.group_node import get_group_node
     from chat.state.threads import list_open_threads
-    from chat.state.world import get_chat
+    from chat.state.world import get_chat, get_scene
 
     you_entity = get_you(conn) or {"name": "you", "persona": ""}
     you_name = you_entity.get("name", "you") or "you"
@@ -350,6 +350,15 @@ async def apply_scene_close_summary(
     chat = get_chat(conn, chat_id) or {}
     guest_bot_id = chat.get("guest_bot_id")
 
+    # T65: detect meanwhile child scenes via the migration-0011
+    # ``present_set_kind`` column. The mechanism is a single field read
+    # — meanwhile scenes carry ``"host_guest"``, regular you-scenes
+    # carry the default ``"you_host"``. We read this once up front so
+    # both the dialogue source and the post-summary digest emission
+    # branches can reference it.
+    closing_scene = get_scene(conn, scene_id) or {}
+    is_meanwhile = closing_scene.get("present_set_kind") == "host_guest"
+
     dialogue = _read_recent_dialogue(conn, chat_id)
 
     # T58.1: build the "Key quotes:" suffix BEFORE the per-POV rewrites
@@ -415,6 +424,36 @@ async def apply_scene_close_summary(
             },
         )
 
+    # T65: when the closing scene was a meanwhile child (host_guest
+    # present set), generate an omniscient briefing for the absent
+    # "you" and queue it as a pending digest. We reuse summarize_scene
+    # with a narrator persona so the digest text is shaped by the same
+    # classifier — only the ``summary`` field is consumed downstream.
+    # Emitted AFTER per-POV summaries land so witness memories carry
+    # their own POV text first; this mirrors how group_node_updated is
+    # ordered relative to the per-POV writes above.
+    if is_meanwhile:
+        digest_pov = await summarize_scene(
+            client,
+            model=classifier_model,
+            bot_name="Narrator",
+            bot_persona=_MEANWHILE_DIGEST_PERSONA,
+            you_name=you_name,
+            prior_edge_summary="",
+            dialogue=dialogue,
+            timeout_s=timeout_s,
+        )
+        if digest_pov.summary:
+            append_and_apply(
+                conn,
+                kind="meanwhile_digest_created",
+                payload={
+                    "chat_id": chat_id,
+                    "scene_id": scene_id,
+                    "summary": digest_pov.summary,
+                },
+            )
+
     # T58.2: thread detection on close. Reuses the dialogue we already
     # gathered for per-POV summarization — same {speaker, text} shape
     # detect_threads expects. Failure-tolerant: classify() returns the
@@ -491,6 +530,18 @@ _GROUP_MERGE_SYSTEM = (
 )
 
 
+# T65: meanwhile-scene digest. The "you" player was absent during this
+# scene; the digest is a short neutral briefing they'll read on the next
+# you-scene resume. Reuses the ScenePOVSummary schema so the same
+# `summarize_scene` helper can be called with a different persona — only
+# the ``summary`` field is used downstream.
+_MEANWHILE_DIGEST_PERSONA = (
+    "an omniscient narrator briefing the absent player in 2-3 neutral "
+    "sentences on what happened while they were away — no editorializing, "
+    "no second-person address"
+)
+
+
 async def merge_group_summary(
     client: LLMClient,
     *,

chat/state/meanwhile.py

@@ -0,0 +1,164 @@
+"""Meanwhile-scene projection (T63).
+
+A "meanwhile" scene is a 2-bot scene where ``present_set = {host_bot_id,
+guest_bot_id}`` and "you" is absent. It runs alongside an active you-scene
+(its parent) so bots can have private interactions whose outcome later
+surfaces back to the you-scene as a pending digest.
+
+The underlying ``scenes`` table (migration 0007) has no explicit ``status``
+column; "active" is encoded as ``ended_at IS NULL`` and "closed" as
+``ended_at IS NOT NULL``. This module preserves that convention and adds
+two new columns introduced by migration 0011:
+
+- ``present_set_kind`` — ``'you_host'`` (default) for normal scenes,
+  ``'host_guest'`` for meanwhile child scenes.
+- ``parent_scene_id`` — the you-scene a meanwhile child hangs off of.
+
+Pending meanwhile digests live in their own table
+(``meanwhile_digest_pending``) and are consumed when their summary is
+surfaced in the next you-scene's prompt.
+"""
+from __future__ import annotations
+import json
+from sqlite3 import Connection
+
+from chat.eventlog.projector import on
+from chat.eventlog.log import Event
+
+
+@on("meanwhile_scene_started")
+def _apply_meanwhile_scene_started(conn: Connection, e: Event) -> None:
+    """Insert a new scenes row for the meanwhile child.
+
+    The caller supplies an explicit ``scene_id`` so subsequent events
+    (close, digest) can reference it without round-tripping through
+    ``lastrowid``.
+    """
+    p = e.payload
+    conn.execute(
+        "INSERT INTO scenes ("
+        "id, chat_id, started_at, ended_at, significance, "
+        "participants_json, present_set_kind, parent_scene_id"
+        ") VALUES (?, ?, ?, NULL, 0, ?, 'host_guest', ?)",
+        (
+            p["scene_id"],
+            p["chat_id"],
+            p.get("started_at"),
+            # participants_json mirrors the present_set: host + guest bot.
+            # json.dumps ensures bot ids with quotes/backslashes can't corrupt the JSON literal.
+            json.dumps([p["host_bot_id"], p["guest_bot_id"]]),
+            p["parent_scene_id"],
+        ),
+    )
+
+
+@on("meanwhile_scene_closed")
+def _apply_meanwhile_scene_closed(conn: Connection, e: Event) -> None:
+    """Mark the meanwhile scene closed by stamping ``ended_at``."""
+    p = e.payload
+    conn.execute(
+        "UPDATE scenes SET ended_at = ? "
+        "WHERE id = ? AND present_set_kind = 'host_guest' "
+        "AND ended_at IS NULL",
+        (p.get("closed_at"), p["scene_id"]),
+    )
+
+
+@on("meanwhile_digest_created")
+def _apply_meanwhile_digest_created(conn: Connection, e: Event) -> None:
+    """Queue a digest for surfacing to the next you-scene's prompt."""
+    p = e.payload
+    conn.execute(
+        "INSERT INTO meanwhile_digest_pending (scene_id, chat_id, summary) "
+        "VALUES (?, ?, ?)",
+        (p["scene_id"], p["chat_id"], p["summary"]),
+    )
+
+
+@on("meanwhile_digest_consumed")
+def _apply_meanwhile_digest_consumed(conn: Connection, e: Event) -> None:
+    """Mark a pending digest as consumed (idempotent on re-projection)."""
+    p = e.payload
+    conn.execute(
+        "UPDATE meanwhile_digest_pending SET consumed_at = ? "
+        "WHERE id = ? AND consumed_at IS NULL",
+        (p.get("consumed_at"), p["digest_id"]),
+    )
+
+
+def _scene_row_to_dict(row: tuple) -> dict:
+    """Shape a meanwhile-scene row.
+
+    ``status`` is derived from ``ended_at`` for callers that prefer the
+    higher-level vocabulary; ``closed_at`` aliases ``ended_at`` for the
+    same reason. The underlying column remains ``ended_at``.
+    """
+    ended_at = row[5]
+    return {
+        "id": row[0],
+        "chat_id": row[1],
+        "started_at": row[2],
+        "present_set_kind": row[3],
+        "parent_scene_id": row[4],
+        "closed_at": ended_at,
+        "status": "closed" if ended_at is not None else "active",
+    }
+
+
+def list_meanwhile_scenes(
+    conn: Connection, chat_id: str, status: str = "active"
+) -> list[dict]:
+    """Return meanwhile scenes for ``chat_id`` filtered by derived status."""
+    if status == "active":
+        ended_clause = "s.ended_at IS NULL"
+    elif status == "closed":
+        ended_clause = "s.ended_at IS NOT NULL"
+    else:
+        raise ValueError(f"unknown status: {status!r}")
+    rows = conn.execute(
+        "SELECT s.id, s.chat_id, s.started_at, s.present_set_kind, "
+        "s.parent_scene_id, s.ended_at "
+        "FROM scenes s "
+        "WHERE s.chat_id = ? AND s.present_set_kind = 'host_guest' "
+        f"AND {ended_clause} "
+        "ORDER BY s.id ASC",
+        (chat_id,),
+    ).fetchall()
+    return [_scene_row_to_dict(r) for r in rows]
+
+
+def get_parent_scene(conn: Connection, scene_id: int) -> dict | None:
+    """Given a meanwhile scene id, return its parent (you-scene) row."""
+    row = conn.execute(
+        "SELECT s.id, s.chat_id, s.started_at, s.present_set_kind, "
+        "s.parent_scene_id, s.ended_at "
+        "FROM scenes s JOIN scenes m ON m.parent_scene_id = s.id "
+        "WHERE m.id = ?",
+        (scene_id,),
+    ).fetchone()
+    if row is None:
+        return None
+    return _scene_row_to_dict(row)
+
+
+def list_pending_meanwhile_digests(
+    conn: Connection, chat_id: str
+) -> list[dict]:
+    """Return digests for ``chat_id`` that haven't been consumed yet."""
+    rows = conn.execute(
+        "SELECT id, scene_id, chat_id, summary, created_at "
+        "FROM meanwhile_digest_pending "
+        "WHERE chat_id = ? AND consumed_at IS NULL "
+        "ORDER BY id ASC",
+        (chat_id,),
+    ).fetchall()
+    return [
+        {
+            "id": r[0],
+            "scene_id": r[1],
+            "chat_id": r[2],
+            "summary": r[3],
+            "created_at": r[4],
+        }
+        for r in rows
+    ]

chat/web/meanwhile.py

@@ -0,0 +1,398 @@
+"""Meanwhile-mode turn controller (T64).
+
+A meanwhile scene is a private 2-bot scene running alongside an active
+you-scene (its parent). The user manually advances it by POSTing to the
+existing ``/chats/<id>/turns`` endpoint; the route detects an active
+meanwhile scene at the start of ``post_turn`` and dispatches here.
+
+Unlike the normal turn flow, "you" is NOT a witness to the scene. The
+controller mirrors ``post_turn`` shape but with:
+
+- Speaker alternation derived from the latest meanwhile ``assistant_turn``
+  scoped to this scene_id (host first, then alternating).
+- Prompt assembly with ``present_set_kind="host_guest"`` so the prompt
+  builder drops the "you" activity bullet and any speaker -> "you" edge.
+- Memory writes via ``record_meanwhile_memory`` — both bots get rows
+  with witness flags ``[you=0, host=1, guest=1]``.
+- State updates over exactly 2 directed pairs (host <-> guest); no
+  you-related pairs fire.
+- The ``assistant_turn`` payload carries ``meanwhile_scene_id`` and
+  ``present_set_kind="host_guest"`` so downstream filters (alternation
+  lookup, drawer rendering, scene-close detection) can scope to the
+  meanwhile slice without conflating it with the parent you-scene's
+  history.
+
+Scene-close detection for meanwhile scenes is not auto-fired here —
+T65 covers the close + digest pipeline. The controller's job ends
+after the post-turn classifier passes land.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from chat.config import Settings
+from chat.eventlog.log import append_and_apply, append_event
+from chat.llm.client import LLMClient
+from chat.services.memory_write import record_meanwhile_memory
+from chat.services.multi_state_update import compute_state_updates_for_present
+from chat.services.prompt import assemble_narrative_prompt
+from chat.services.turn_parse import parse_turn
+from chat.state.edges import get_edge
+from chat.state.entities import get_bot
+from chat.state.meanwhile import list_meanwhile_scenes
+from chat.state.world import get_chat
+from chat.web.pubsub import publish
+from chat.web.render import render_turn_html as _render_turn_html
+
+
+def _strip_ooc_for_prompt(parsed) -> str:
+    """Mirror of the helper in turns.py — concatenate non-OOC segments."""
+    keep = [s.text for s in parsed.segments if s.kind != "ooc"]
+    return " ".join(keep).strip()
+
+
+def _read_recent_meanwhile_dialogue(
+    conn, chat_id: str, scene_id: int, limit: int = 50
+) -> list[dict]:
+    """Return the meanwhile scene's prior turns shaped as
+    ``{"speaker": <id>, "text": <prose>}``.
+
+    Pulls ``user_turn`` rows for the chat (the user-side prose driving
+    this meanwhile scene rides through the same chat) plus only those
+    ``assistant_turn`` rows whose ``meanwhile_scene_id`` matches the
+    given scene id. Other meanwhile scenes on the same chat — and the
+    parent you-scene's assistant_turns — are excluded so the prompt
+    context stays scoped to the private beat.
+
+    Filters chat_id (and meanwhile_scene_id for assistant_turn) via
+    ``json_extract`` in SQL so SQLite stops at the first ``limit`` rows
+    that already match — avoids an unbounded scan as ``event_log``
+    grows. The user-side rows match on chat_id only since they aren't
+    tagged with a scene id (they ride the chat-wide log).
+    """
+    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') = ? "
+        "  AND ("
+        "    kind IN ('user_turn', 'user_turn_edit') "
+        "    OR json_extract(payload_json, '$.meanwhile_scene_id') = ?"
+        "  ) "
+        "ORDER BY id DESC LIMIT ?",
+        (chat_id, scene_id, limit),
+    )
+    rows = cur.fetchall()
+    rows.reverse()
+    out: list[dict] = []
+    for _row_id, kind, payload_json in rows:
+        p = json.loads(payload_json)
+        if kind in ("user_turn", "user_turn_edit"):
+            out.append({"speaker": "you", "text": p.get("prose", "")})
+        else:
+            out.append(
+                {
+                    "speaker": p.get("speaker_id", "bot"),
+                    "text": p.get("text", ""),
+                }
+            )
+    return out
+
+
+def _last_meanwhile_speaker(conn, chat_id: str, scene_id: int) -> str | None:
+    """Return the speaker_id of the latest assistant_turn linked to
+    ``scene_id`` for ``chat_id``, or ``None`` if no prior turn exists.
+
+    Used to alternate the speaker across consecutive meanwhile turns —
+    the OTHER bot speaks next. Pushes both filters into SQL via
+    ``json_extract`` and bounds with ``LIMIT 1`` so SQLite stops at the
+    first match instead of scanning the whole assistant_turn slice.
+    """
+    row = conn.execute(
+        "SELECT json_extract(payload_json, '$.speaker_id') AS speaker "
+        "FROM event_log "
+        "WHERE kind = 'assistant_turn' "
+        "  AND superseded_by IS NULL AND hidden = 0 "
+        "  AND json_extract(payload_json, '$.chat_id') = ? "
+        "  AND json_extract(payload_json, '$.meanwhile_scene_id') = ? "
+        "ORDER BY id DESC "
+        "LIMIT 1",
+        (chat_id, scene_id),
+    ).fetchone()
+    return row[0] if row else None
+
+
+async def process_meanwhile_turn(
+    conn,
+    client: LLMClient,
+    settings: Settings,
+    *,
+    chat_id: str,
+    prose: str,
+) -> dict:
+    """Run one meanwhile turn end-to-end.
+
+    Returns a small dict shape ``{"assistant_text": ..., "speaker_id":
+    ..., "scene_id": ..., "user_turn_id": ..., "assistant_event_id":
+    ...}`` so callers can introspect the produced beat (HTTP route maps
+    to a ``204``; future SSE rebroadcast may use the dict directly).
+
+    Raises ``ValueError`` when there is no active meanwhile scene on
+    ``chat_id`` — the caller (turns.py) only dispatches here after a
+    positive ``list_meanwhile_scenes`` lookup, but the defensive raise
+    keeps the controller usable in isolation.
+    """
+    chat = get_chat(conn, chat_id)
+    if chat is None:
+        raise ValueError(f"chat not found: {chat_id}")
+
+    scenes = list_meanwhile_scenes(conn, chat_id, status="active")
+    if not scenes:
+        raise ValueError(f"no active meanwhile scene on chat: {chat_id}")
+    scene = scenes[0]
+    scene_id = scene["id"]
+
+    host_bot_id = chat["host_bot_id"]
+    guest_bot_id = chat.get("guest_bot_id")
+    if guest_bot_id is None:
+        # A meanwhile scene without a guest is a schema violation —
+        # the projector requires both ids on meanwhile_scene_started.
+        raise ValueError(
+            f"meanwhile scene {scene_id} on chat {chat_id} lacks a guest"
+        )
+
+    host_bot = get_bot(conn, host_bot_id)
+    guest_bot = get_bot(conn, guest_bot_id)
+    if host_bot is None or guest_bot is None:
+        raise ValueError(
+            f"meanwhile bots missing: host={host_bot_id} guest={guest_bot_id}"
+        )
+
+    # 1. Parse the user prose with the classifier — same shape as the
+    # normal turn flow so OOC-stripping, segment-typing, etc. all work.
+    parsed = await parse_turn(
+        client, model=settings.classifier_model, prose=prose
+    )
+    prompt_prose = _strip_ooc_for_prompt(parsed)
+
+    # 2. Append user_turn — kept on the chat-wide log so the user can
+    # see their own prose in the timeline. Tagged with the meanwhile
+    # scene_id so future renderers can group it with the right scene.
+    user_turn_event_id = append_event(
+        conn,
+        kind="user_turn",
+        payload={
+            "chat_id": chat_id,
+            "prose": prose,
+            "segments": [s.model_dump() for s in parsed.segments],
+            "meanwhile_scene_id": scene_id,
+        },
+    )
+
+    # 3. Alternate the speaker. First turn -> host speaks; each
+    # subsequent turn -> the OTHER bot from the previous beat. Lookup
+    # is scoped by ``meanwhile_scene_id`` so unrelated assistant_turns
+    # on the same chat don't perturb the alternation.
+    last_speaker = _last_meanwhile_speaker(conn, chat_id, scene_id)
+    if last_speaker is None or last_speaker == guest_bot_id:
+        speaker_bot = host_bot
+        addressee_bot = guest_bot
+    else:
+        speaker_bot = guest_bot
+        addressee_bot = host_bot
+
+    # 4. Placeholder marker so SSE observers see "in flight". No
+    # projector handler is registered for this kind — it's transcript-
+    # only, same as the normal turn flow.
+    append_event(
+        conn,
+        kind="assistant_turn_started",
+        payload={
+            "chat_id": chat_id,
+            "speaker_id": speaker_bot["id"],
+            "user_turn_id": user_turn_event_id,
+            "meanwhile_scene_id": scene_id,
+        },
+    )
+
+    # 5. Build the narrative prompt. ``present_set_kind="host_guest"``
+    # tells the assembler to drop the "you" activity bullet and any
+    # speaker -> "you" edge — both irrelevant inside a private beat.
+    # Addressee is the OTHER bot, not "you".
+    recent_dialogue = _read_recent_meanwhile_dialogue(conn, chat_id, scene_id)
+    if recent_dialogue and recent_dialogue[-1].get("speaker") == "you":
+        recent_dialogue = recent_dialogue[:-1]
+    messages = assemble_narrative_prompt(
+        conn,
+        chat_id=chat_id,
+        speaker_bot_id=speaker_bot["id"],
+        addressee=addressee_bot["id"],
+        user_turn_prose=prompt_prose if prompt_prose else None,
+        recent_dialogue=recent_dialogue,
+        budget_soft=settings.narrative_budget_soft,
+        budget_hard=settings.narrative_budget_hard,
+        guest_id=guest_bot_id,
+        present_set_kind="host_guest",
+    )
+
+    # 6. Stream + accumulate. Same SSE pattern as the normal flow —
+    # tokens publish under the speaker's id so the UI can label the
+    # right bubble. Register the streaming task in the chat-keyed
+    # in-flight registry so POST /chats/<id>/turns/cancel can call
+    # ``.cancel()`` on it; without this, the Stop button is a no-op for
+    # meanwhile beats. We import the underscore name from turns.py
+    # deliberately — it's the same single-process registry the cancel
+    # route reads, and exposing it via a public alias would require
+    # touching every existing call site for no behavioural gain.
+    from chat.web.turns import _in_flight_tasks  # noqa: PLC0415
+
+    accumulated: list[str] = []
+    truncated = False
+    cancelled = False
+
+    async def _stream() -> None:
+        async for chunk in client.stream(
+            messages,
+            model=settings.narrative_model,
+            max_tokens=settings.narrative_max_tokens,
+            temperature=settings.narrative_temperature,
+        ):
+            accumulated.append(chunk)
+            await publish(
+                chat_id,
+                {
+                    "event": "token",
+                    "text": chunk,
+                    "speaker_id": speaker_bot["id"],
+                },
+            )
+
+    stream_task = asyncio.create_task(_stream())
+    _in_flight_tasks[chat_id] = stream_task
+    try:
+        await stream_task
+    except asyncio.CancelledError:
+        truncated = True
+        cancelled = True
+    except Exception:
+        truncated = True
+    finally:
+        # Always unregister so a subsequent turn can register a fresh
+        # task. Mirrors the cleanup in turns.py::post_turn.
+        _in_flight_tasks.pop(chat_id, None)
+
+    text = "".join(accumulated)
+
+    # 7. Append assistant_turn — tagged with meanwhile_scene_id so the
+    # next turn's alternation lookup can find it, and present_set_kind
+    # so downstream renderers / digesters can filter scope.
+    assistant_event_id = append_event(
+        conn,
+        kind="assistant_turn",
+        payload={
+            "chat_id": chat_id,
+            "speaker_id": speaker_bot["id"],
+            "text": text,
+            "truncated": truncated,
+            "user_turn_id": user_turn_event_id,
+            "meanwhile_scene_id": scene_id,
+            "present_set_kind": "host_guest",
+        },
+    )
+
+    # 8. Per-turn memory writes — both bots get a row with witness flags
+    # [you=0, host=1, guest=1]. Skipped on cancellation so we don't
+    # record memory for a partial beat the user never read.
+    if not cancelled and text.strip():
+        record_meanwhile_memory(
+            conn,
+            chat_id=chat_id,
+            host_bot_id=host_bot_id,
+            guest_bot_id=guest_bot_id,
+            narrative_text=text,
+            scene_id=scene_id,
+            chat_clock_at=chat.get("time"),
+        )
+
+        # 9. Post-turn state-update — exactly 2 directed pairs over the
+        # bot pair. No you-related pairs fire (you isn't present).
+        present_ids = [host_bot_id, guest_bot_id]
+        present_names = {
+            host_bot_id: host_bot["name"],
+            guest_bot_id: guest_bot["name"],
+        }
+        personas = {
+            host_bot_id: host_bot.get("persona") or "",
+            guest_bot_id: guest_bot.get("persona") or "",
+        }
+        prior_edges: dict[tuple[str, str], dict] = {}
+        for src in present_ids:
+            for tgt in present_ids:
+                if src == tgt:
+                    continue
+                edge = get_edge(conn, src, tgt) or {
+                    "affinity": 50,
+                    "trust": 50,
+                    "summary": "",
+                }
+                prior_edges[(src, tgt)] = edge
+
+        state_updates = await compute_state_updates_for_present(
+            client,
+            classifier_model=settings.classifier_model,
+            present_ids=present_ids,
+            present_names=present_names,
+            personas=personas,
+            prior_edges=prior_edges,
+            recent_dialogue=recent_dialogue,
+            timeout_s=settings.classifier_timeout_s,
+        )
+        last_at = chat.get("time")
+        for src_id, tgt_id, update in state_updates:
+            append_and_apply(
+                conn,
+                kind="edge_update",
+                payload={
+                    "source_id": src_id,
+                    "target_id": tgt_id,
+                    "chat_id": chat_id,
+                    "affinity_delta": update.affinity_delta,
+                    "trust_delta": update.trust_delta,
+                    "knowledge_facts": update.knowledge_facts,
+                    "last_interaction_at": last_at,
+                    "last_interaction_chat_id": chat_id,
+                },
+            )
+
+    # 10. SSE broadcast for the timeline UI — completion event + an HTML
+    # fragment for the HTMX SSE swap. Same pattern as the normal turn
+    # flow so the rendered transcript shows the meanwhile beat inline.
+    await publish(
+        chat_id,
+        {
+            "event": "assistant_turn_complete",
+            "speaker_id": speaker_bot["id"],
+            "text": text,
+            "truncated": truncated,
+        },
+    )
+    turn_html = _render_turn_html(speaker_bot["name"], text, role="bot")
+    await publish(chat_id, {"event": "turn_html", "data": turn_html})
+
+    if cancelled:
+        # Re-raise after the partial-turn has been recorded so callers
+        # see the cancel propagate (mirrors normal turn flow).
+        raise asyncio.CancelledError
+
+    return {
+        "assistant_text": text,
+        "speaker_id": speaker_bot["id"],
+        "scene_id": scene_id,
+        "user_turn_id": user_turn_event_id,
+        "assistant_event_id": assistant_event_id,
+    }
+
+
+__all__ = ["process_meanwhile_turn"]

chat/web/turns.py

@@ -72,9 +72,11 @@ from chat.services.turn_parse import ParsedTurn, parse_turn
 from chat.state.edges import get_edge
 from chat.state.entities import get_bot, get_you
 from chat.state.events import list_active_events
+from chat.state.meanwhile import list_meanwhile_scenes
 from chat.state.world import active_scene, get_chat, get_container
 from chat.web.bots import get_conn
 from chat.web.kickoff import get_llm_client
+from chat.web.meanwhile import process_meanwhile_turn
 from chat.web.pubsub import publish
 from chat.web.render import render_turn_html as _render_turn_html
 from chat.web.skip import _parse_iso_time, process_elision_skip
@@ -251,6 +253,29 @@ async def post_turn(
 
     settings = request.app.state.settings
 
+    # 0. Meanwhile-mode short-circuit (T64). When an active meanwhile
+    # scene is running on this chat, the turn flow is entirely between
+    # the two bots — "you" is absent. The meanwhile controller mirrors
+    # the post_turn shape but with no-you semantics: present_set_kind
+    # ``host_guest`` in the prompt assembler, ``record_meanwhile_memory``
+    # for witness flags, only 2 directed pairs in the state update, and
+    # the assistant_turn payload tagged with ``meanwhile_scene_id`` so
+    # alternation lookups can scope to this scene specifically. The
+    # T62 skip-intent dispatch and the regular narrative path below
+    # are skipped — a meanwhile beat is its own self-contained flow.
+    if list_meanwhile_scenes(conn, chat_id, status="active"):
+        try:
+            await process_meanwhile_turn(
+                conn,
+                client,
+                settings,
+                chat_id=chat_id,
+                prose=prose,
+            )
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc))
+        return Response(status_code=204)
+
     # 1. Parse turn (classifier).
     parsed = await parse_turn(
         client, model=settings.classifier_model, prose=prose

docs/plans/2026-04-26-v1-requirements-design.md

@@ -510,6 +510,8 @@ Written per witness when a scene closes. Different details, different interpreta
 
 ### Phase 3 — events, skips, threads
 
+**Status: shipped 2026-04-26** (T49–T67, 19 tasks across 8 waves; schema baseline now version 11; +68 tests). See "Phase 3 status" in CLAUDE.md for the per-task breakdown.
+
 - Events with lifecycles and scoped props.
 - Time skips: elision and jump.
 - Active threads.

tests/test_meanwhile_state.py

@@ -0,0 +1,269 @@
+from __future__ import annotations
+
+from chat.db.connection import open_db
+from chat.db.migrate import apply_migrations
+from chat.eventlog.log import append_event
+from chat.eventlog.projector import project
+import chat.state.entities  # registers handlers
+import chat.state.world  # registers handlers
+import chat.state.meanwhile  # registers handlers
+from chat.state.meanwhile import (
+    get_parent_scene,
+    list_meanwhile_scenes,
+    list_pending_meanwhile_digests,
+)
+from chat.state.world import active_scene
+
+
+def _bot_payload(bot_id: str, name: str) -> dict:
+    return {
+        "id": bot_id,
+        "name": name,
+        "persona": "thoughtful, observant",
+        "voice_samples": [],
+        "traits": [],
+        "backstory": "",
+        "initial_relationship_to_you": "coworker",
+        "kickoff_prose": "",
+    }
+
+
+def _chat_payload(chat_id: str = "chat_bot_a") -> dict:
+    return {
+        "id": chat_id,
+        "host_bot_id": "bot_a",
+        "guest_bot_id": "bot_b",
+        "initial_time": "2026-04-26T20:00:00+00:00",
+        "narrative_anchor": "Day 1 evening",
+        "weather": "clear",
+    }
+
+
+def test_meanwhile_started_creates_scene_with_correct_present_set_kind(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA")
+        )
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        # Parent (you-scene) — uses existing scene_opened handler. Will get
+        # the default present_set_kind='you_host' from the new column.
+        append_event(
+            conn,
+            kind="scene_opened",
+            payload={
+                "chat_id": "chat_bot_a",
+                "started_at": "2026-04-26T20:00:00+00:00",
+                "participants": ["you", "bot_a", "bot_b"],
+            },
+        )
+        # Now the meanwhile child scene — bot_a + bot_b only.
+        append_event(
+            conn,
+            kind="meanwhile_scene_started",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "parent_scene_id": 1,
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "started_at": "2026-04-26T20:05:00+00:00",
+            },
+        )
+        project(conn)
+
+        meanwhile_scenes = list_meanwhile_scenes(
+            conn, "chat_bot_a", status="active"
+        )
+        assert len(meanwhile_scenes) == 1
+        m = meanwhile_scenes[0]
+        assert m["id"] == 2
+        assert m["chat_id"] == "chat_bot_a"
+        assert m["status"] == "active"
+        assert m["present_set_kind"] == "host_guest"
+        assert m["parent_scene_id"] == 1
+        assert m["started_at"] == "2026-04-26T20:05:00+00:00"
+        assert m["closed_at"] is None
+
+        # Parent linkage helper.
+        parent = get_parent_scene(conn, 2)
+        assert parent is not None
+        assert parent["id"] == 1
+        assert parent["present_set_kind"] == "you_host"
+
+
+def test_meanwhile_closed_marks_scene_closed(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA")
+        )
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="scene_opened",
+            payload={
+                "chat_id": "chat_bot_a",
+                "started_at": "2026-04-26T20:00:00+00:00",
+                "participants": ["you", "bot_a", "bot_b"],
+            },
+        )
+        append_event(
+            conn,
+            kind="meanwhile_scene_started",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "parent_scene_id": 1,
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "started_at": "2026-04-26T20:05:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="meanwhile_scene_closed",
+            payload={
+                "scene_id": 2,
+                "closed_at": "2026-04-26T20:15:00+00:00",
+            },
+        )
+        project(conn)
+
+        assert list_meanwhile_scenes(conn, "chat_bot_a", status="active") == []
+        closed = list_meanwhile_scenes(conn, "chat_bot_a", status="closed")
+        assert len(closed) == 1
+        assert closed[0]["id"] == 2
+        assert closed[0]["status"] == "closed"
+        assert closed[0]["closed_at"] == "2026-04-26T20:15:00+00:00"
+
+
+def test_active_you_scene_can_coexist_with_active_meanwhile_scene(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA")
+        )
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="scene_opened",
+            payload={
+                "chat_id": "chat_bot_a",
+                "started_at": "2026-04-26T20:00:00+00:00",
+                "participants": ["you", "bot_a", "bot_b"],
+            },
+        )
+        append_event(
+            conn,
+            kind="meanwhile_scene_started",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "parent_scene_id": 1,
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "started_at": "2026-04-26T20:05:00+00:00",
+            },
+        )
+        project(conn)
+
+        # The you-scene is still the active scene from the chat's POV.
+        you_scene = active_scene(conn, "chat_bot_a")
+        assert you_scene is not None
+        assert you_scene["id"] == 1
+
+        # And the meanwhile child is independently active.
+        meanwhile_scenes = list_meanwhile_scenes(
+            conn, "chat_bot_a", status="active"
+        )
+        assert len(meanwhile_scenes) == 1
+        assert meanwhile_scenes[0]["id"] == 2
+        assert meanwhile_scenes[0]["present_set_kind"] == "host_guest"
+
+        # Cross-check via raw query: one row per present_set_kind, both unended.
+        rows = conn.execute(
+            "SELECT id, present_set_kind FROM scenes "
+            "WHERE chat_id = ? AND ended_at IS NULL ORDER BY id",
+            ("chat_bot_a",),
+        ).fetchall()
+        kinds = sorted(r[1] for r in rows)
+        assert kinds == ["host_guest", "you_host"]
+
+
+def test_meanwhile_digest_created_and_consumed(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA")
+        )
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="scene_opened",
+            payload={
+                "chat_id": "chat_bot_a",
+                "started_at": "2026-04-26T20:00:00+00:00",
+                "participants": ["you", "bot_a", "bot_b"],
+            },
+        )
+        append_event(
+            conn,
+            kind="meanwhile_scene_started",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "parent_scene_id": 1,
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "started_at": "2026-04-26T20:05:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="meanwhile_digest_created",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "summary": "BotA confides in BotB about the missing file.",
+            },
+        )
+        project(conn)
+
+        pending = list_pending_meanwhile_digests(conn, "chat_bot_a")
+        assert len(pending) == 1
+        digest_id = pending[0]["id"]
+        assert pending[0]["scene_id"] == 2
+        assert pending[0]["summary"].startswith("BotA confides")
+
+        # Use append_and_apply for the second beat: re-running project()
+        # would re-fire non-idempotent handlers (chat_created, scene_opened)
+        # whose INSERTs conflict on UNIQUE constraints.
+        from chat.eventlog.log import append_and_apply
+
+        append_and_apply(
+            conn,
+            kind="meanwhile_digest_consumed",
+            payload={
+                "digest_id": digest_id,
+                "consumed_at": "2026-04-26T20:30:00+00:00",
+            },
+        )
+
+        assert list_pending_meanwhile_digests(conn, "chat_bot_a") == []

tests/test_meanwhile_turn_flow.py

@@ -0,0 +1,572 @@
+"""Meanwhile-mode turn flow (T64).
+
+A meanwhile scene runs entirely between two bots — host + guest — with
+"you" absent. The user manually advances the scene by POSTing prose to
+the existing ``/chats/<id>/turns`` endpoint; the route detects the active
+meanwhile scene at the start of ``post_turn`` and dispatches to the
+``process_meanwhile_turn`` controller in ``chat/web/meanwhile.py``.
+
+Coverage:
+
+1. Memory writes for a meanwhile turn carry witness ``[you=0, host=1,
+   guest=1]`` for both the host's and the guest's per-POV memory rows.
+2. State updates after a meanwhile turn run for exactly 2 directed pairs
+   (host -> guest, guest -> host) — no you-related pairs fire.
+3. Speakers alternate across consecutive meanwhile turns: the host
+   speaks first (no prior meanwhile assistant_turn), the guest speaks
+   second (the prior turn's speaker was the host, so this turn's
+   speaker is the OTHER bot).
+4. Scene-close on a meanwhile scene writes per-POV summaries for host +
+   guest only — no "you" POV row is written, mirroring the no-you
+   present_set of the meanwhile scene.
+"""
+
+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
+import chat.state.meanwhile  # noqa: F401  (registers handlers)
+
+
+def _bot_payload(bot_id: str, name: str) -> dict:
+    return {
+        "id": bot_id,
+        "name": name,
+        "persona": f"persona for {name}",
+        "voice_samples": [],
+        "traits": [],
+        "backstory": "",
+        "initial_relationship_to_you": "",
+        "kickoff_prose": "...",
+    }
+
+
+def _seed_meanwhile_chat(db_path: Path) -> None:
+    """Seed two bots, you, a chat with both wired in, an open parent
+    you-scene, AND an active meanwhile child scene with bot_a + bot_b.
+
+    Edges are seeded for both directed pairs between bot_a and bot_b at
+    schema-default 50/50 so post-turn state-update writes land cleanly.
+    Activities for both bots are recorded so the prompt assembler has
+    something to render.
+    """
+    with open_db(db_path) as conn:
+        append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
+        append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(
+            conn,
+            kind="you_authored",
+            payload={"name": "Me", "pronouns": "they/them", "persona": ""},
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "guest_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_a",
+                "name": "office",
+                "type": "workplace",
+                "properties": {},
+            },
+        )
+        # Parent (you-scene) opens first.
+        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", "bot_b"],
+            },
+        )
+        # Meanwhile child scene — bot_a + bot_b only, parent linked.
+        append_event(
+            conn,
+            kind="meanwhile_scene_started",
+            payload={
+                "scene_id": 2,
+                "chat_id": "chat_bot_a",
+                "parent_scene_id": 1,
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "started_at": "2026-04-26T20:05:00+00:00",
+            },
+        )
+        # Seed both directed edges between the bots so state-update
+        # writes land on initialized rows.
+        for src, tgt in [("bot_a", "bot_b"), ("bot_b", "bot_a")]:
+            append_event(
+                conn,
+                kind="edge_update",
+                payload={
+                    "source_id": src,
+                    "target_id": tgt,
+                    "chat_id": "chat_bot_a",
+                    "knowledge_facts": [],
+                },
+            )
+        for entity_id, verb in [("bot_a", "listening"), ("bot_b", "talking")]:
+            append_event(
+                conn,
+                kind="activity_change",
+                payload={
+                    "entity_id": entity_id,
+                    "posture": "sitting",
+                    "action": {
+                        "verb": verb,
+                        "interruptible": True,
+                        "required_attention": "low",
+                        "expected_duration": "ongoing",
+                    },
+                    "attention": "",
+                    "holding": [],
+                    "status": {},
+                },
+            )
+        project(conn)
+
+
+def _override_llm(canned: list[str]) -> MockLLMClient:
+    from chat.web.kickoff import get_llm_client
+
+    mock = MockLLMClient(canned=list(canned))
+    app.dependency_overrides[get_llm_client] = lambda: mock
+    return mock
+
+
+def _zero_state() -> str:
+    return json.dumps(
+        {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
+    )
+
+
+@pytest.fixture
+def app_state_setup(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:
+        app.state.background_worker.enabled = False
+        yield c
+    app.dependency_overrides.clear()
+
+
+def test_meanwhile_turn_writes_memories_with_witness_0_1_1(
+    app_state_setup, tmp_path
+):
+    """A meanwhile turn writes one ``memory_written`` event per bot — host
+    and guest — with witness flags ``[you=0, host=1, guest=1]``. "You" is
+    not present in the scene, so the witness_you flag must be 0 for both
+    rows.
+
+    Canned queue (4 calls):
+      1. parse_turn (user prose classification)
+      2. narrative stream (host speaks first; no prior meanwhile turn)
+      3. state-update for bot_a -> bot_b
+      4. state-update for bot_b -> bot_a
+    """
+    _seed_meanwhile_chat(tmp_path / "test.db")
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "narration", "text": "they exchange a glance"}]}
+    )
+    canned = [
+        canned_parse,
+        "BotA leans in. *quietly* Tell me what you saw.",
+        _zero_state(),
+        _zero_state(),
+    ]
+    mock = _override_llm(canned)
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns",
+            data={"prose": "they exchange a glance"},
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log WHERE kind = 'memory_written' "
+            "ORDER BY id"
+        ).fetchall()
+        payloads = [json.loads(r[0]) for r in rows]
+
+    assert len(payloads) == 2
+    owners = sorted(p["owner_id"] for p in payloads)
+    assert owners == ["bot_a", "bot_b"]
+    for p in payloads:
+        assert p["witness_you"] == 0, p
+        assert p["witness_host"] == 1, p
+        assert p["witness_guest"] == 1, p
+
+
+def test_meanwhile_turn_emits_2_edge_updates_only(app_state_setup, tmp_path):
+    """A meanwhile turn runs state-update for exactly 2 directed pairs:
+    host -> guest and guest -> host. No you-related pairs fire.
+    """
+    _seed_meanwhile_chat(tmp_path / "test.db")
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "narration", "text": "they whisper"}]}
+    )
+    canned = [
+        canned_parse,
+        "BotA whispers. *softly* I noticed something today.",
+        _zero_state(),
+        _zero_state(),
+    ]
+    mock = _override_llm(canned)
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "they whisper"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        # Edge updates landed AFTER the assistant_turn (i.e. excluding
+        # the seed updates done before the request).
+        max_at = conn.execute(
+            "SELECT MAX(id) FROM event_log WHERE kind = 'assistant_turn'"
+        ).fetchone()[0]
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'edge_update' AND id > ? ORDER BY id",
+            (max_at,),
+        ).fetchall()
+        payloads = [json.loads(r[0]) for r in rows]
+
+    # Exactly 2 post-turn edge_update events.
+    assert len(payloads) == 2
+    pairs = sorted((p["source_id"], p["target_id"]) for p in payloads)
+    assert pairs == [("bot_a", "bot_b"), ("bot_b", "bot_a")]
+    # And NO you-related pair leaked in.
+    for p in payloads:
+        assert p["source_id"] != "you", p
+        assert p["target_id"] != "you", p
+
+
+def test_meanwhile_turn_alternates_speaker(app_state_setup, tmp_path):
+    """Successive meanwhile turns alternate which bot speaks.
+
+    The first turn has no prior meanwhile ``assistant_turn`` linked to
+    this scene, so the host speaks. The second turn finds the latest
+    such ``assistant_turn``'s speaker (the host) and picks the OTHER
+    bot, so the guest speaks. Each ``assistant_turn`` payload carries
+    ``meanwhile_scene_id`` so the alternation lookup is unambiguous.
+    """
+    _seed_meanwhile_chat(tmp_path / "test.db")
+    canned_parse_1 = json.dumps(
+        {"segments": [{"kind": "narration", "text": "they pause"}]}
+    )
+    canned_1 = [
+        canned_parse_1,
+        "BotA speaks first. *quietly*",
+        _zero_state(),
+        _zero_state(),
+    ]
+    mock = _override_llm(canned_1)
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "they pause"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    canned_parse_2 = json.dumps(
+        {"segments": [{"kind": "narration", "text": "and again"}]}
+    )
+    canned_2 = [
+        canned_parse_2,
+        "BotB replies. *thoughtfully*",
+        _zero_state(),
+        _zero_state(),
+    ]
+    mock = _override_llm(canned_2)
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "and again"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'assistant_turn' ORDER BY id"
+        ).fetchall()
+        payloads = [json.loads(r[0]) for r in rows]
+
+    assert len(payloads) == 2
+    # First turn — host speaks.
+    assert payloads[0]["speaker_id"] == "bot_a"
+    # Second turn — guest speaks (alternation).
+    assert payloads[1]["speaker_id"] == "bot_b"
+    # Both payloads tag this meanwhile scene id so the alternation
+    # lookup can scope to it specifically (not any other assistant_turn
+    # that might exist on the chat).
+    assert payloads[0]["meanwhile_scene_id"] == 2
+    assert payloads[1]["meanwhile_scene_id"] == 2
+    # Both also carry the present_set_kind discriminator for downstream
+    # filters (digest creation, drawer rendering).
+    assert payloads[0]["present_set_kind"] == "host_guest"
+    assert payloads[1]["present_set_kind"] == "host_guest"
+
+
+def test_meanwhile_scene_close_writes_per_pov_for_both_bots_only(
+    app_state_setup, tmp_path
+):
+    """When a meanwhile scene closes, per-POV summary rewrites land for
+    the host and the guest. No write fires for "you" — there is no
+    "you" memory store and no "you" POV in the meanwhile present set.
+    """
+    from chat.services.scene_summarize import apply_scene_close_summary
+    from chat.eventlog.log import append_and_apply
+
+    _seed_meanwhile_chat(tmp_path / "test.db")
+
+    # Run a meanwhile turn first so each bot has a memory row scoped to
+    # the meanwhile scene_id (=2). The per-POV rewrite targets these
+    # rows by ``scene_id``.
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "narration", "text": "they speak quietly"}]}
+    )
+    canned = [
+        canned_parse,
+        "BotA speaks. *quietly*",
+        _zero_state(),
+        _zero_state(),
+    ]
+    mock = _override_llm(canned)
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns",
+            data={"prose": "they speak quietly"},
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    # Close the meanwhile scene and run the close-summary pipeline.
+    # Two POV summaries (host + guest) — no "you" POV.
+    pov_payload_host = json.dumps(
+        {
+            "summary": "BotA reflects on the quiet moment with BotB.",
+            "knowledge_facts": [],
+            "relationship_summary": "",
+        }
+    )
+    pov_payload_guest = json.dumps(
+        {
+            "summary": "BotB notices BotA's reserved manner.",
+            "knowledge_facts": [],
+            "relationship_summary": "",
+        }
+    )
+    # T65 added a meanwhile digest summarize call after per-POV writes
+    # for meanwhile scenes. T58's thread detection is wrapped in try/except
+    # so its IndexError is swallowed gracefully.
+    digest_payload = json.dumps(
+        {
+            "summary": "While you were away, BotA and BotB talked quietly.",
+            "knowledge_facts": [],
+            "relationship_summary": "",
+        }
+    )
+    close_mock = MockLLMClient(
+        canned=[pov_payload_host, pov_payload_guest, digest_payload]
+    )
+
+    import asyncio as _asyncio
+
+    with open_db(tmp_path / "test.db") as conn:
+        # asyncio.run() can't nest under TestClient's loop, but the
+        # close pipeline is awaitable — drive it via a fresh loop here.
+        _loop = _asyncio.new_event_loop()
+        # Mark the meanwhile scene closed via the projector handler.
+        append_and_apply(
+            conn,
+            kind="meanwhile_scene_closed",
+            payload={
+                "scene_id": 2,
+                "closed_at": "2026-04-26T20:30:00+00:00",
+            },
+        )
+
+        # apply_scene_close_summary takes host_bot_id; here we tell it to
+        # operate on the meanwhile scene id (2). With no "you" memory
+        # row to rewrite (witness_you=0 means "you" doesn't have a
+        # memory for this scene), the call must produce per-POV writes
+        # ONLY for bot_a and bot_b.
+        try:
+            _loop.run_until_complete(
+                apply_scene_close_summary(
+                    conn,
+                    close_mock,
+                    classifier_model="x",
+                    chat_id="chat_bot_a",
+                    scene_id=2,
+                    host_bot_id="bot_a",
+                )
+            )
+        finally:
+            _loop.close()
+
+        # Per-POV memory rewrites: count manual_edits with target_kind
+        # ``memory_pov_summary`` whose target_id maps to a memory row
+        # scoped to scene 2.
+        edits = conn.execute(
+            "SELECT payload_json FROM event_log WHERE kind = 'manual_edit'"
+        ).fetchall()
+        pov_edits = []
+        for (raw,) in edits:
+            payload = json.loads(raw)
+            if payload.get("target_kind") != "memory_pov_summary":
+                continue
+            mem_row = conn.execute(
+                "SELECT owner_id, scene_id FROM memories WHERE id = ?",
+                (payload["target_id"],),
+            ).fetchone()
+            if mem_row is None or mem_row[1] != 2:
+                continue
+            pov_edits.append({"owner": mem_row[0], "new": payload["new_value"]})
+
+        # Verify the actual current pov_summary on each bot's memory row
+        # for scene 2 reflects the rewrite.
+        host_pov = conn.execute(
+            "SELECT pov_summary FROM memories WHERE owner_id = ? AND scene_id = ?",
+            ("bot_a", 2),
+        ).fetchone()
+        guest_pov = conn.execute(
+            "SELECT pov_summary FROM memories WHERE owner_id = ? AND scene_id = ?",
+            ("bot_b", 2),
+        ).fetchone()
+        # No "you" memory row should exist for the meanwhile scene —
+        # "you" was never a witness.
+        you_row = conn.execute(
+            "SELECT id FROM memories WHERE owner_id = 'you' AND scene_id = ?",
+            (2,),
+        ).fetchone()
+
+    # Exactly two memory_pov_summary rewrites — one per bot witness.
+    assert len(pov_edits) == 2
+    owners = sorted(e["owner"] for e in pov_edits)
+    assert owners == ["bot_a", "bot_b"]
+    assert host_pov is not None and "BotA reflects" in host_pov[0]
+    assert guest_pov is not None and "BotB notices" in guest_pov[0]
+    # No "you" POV row — meanwhile scenes don't surface a you-memory.
+    assert you_row is None
+
+
+def test_meanwhile_turn_registered_in_in_flight_tasks(
+    app_state_setup, tmp_path
+):
+    """A meanwhile turn registers its streaming task in the chat-keyed
+    ``_in_flight_tasks`` registry the cancel route reads from, and clears
+    the entry after the stream completes.
+
+    Without registration, ``POST /chats/<id>/turns/cancel`` would be a
+    silent no-op for meanwhile beats — the Stop button wouldn't actually
+    stop them. We pin the behaviour via a streaming mock that snapshots
+    ``_in_flight_tasks`` at the moment of its first yield (mid-flight),
+    then assert the entry is removed after the response returns.
+    """
+    from typing import AsyncIterator, Sequence
+
+    from chat.llm.client import Message
+    from chat.web.turns import _in_flight_tasks
+
+    _seed_meanwhile_chat(tmp_path / "test.db")
+
+    # Snapshot of (chat_id-present?, registered task object) captured
+    # at the first stream yield. The closure runs inside the streaming
+    # coroutine, so when it executes the task is alive and registered.
+    in_flight_snapshot: dict = {}
+
+    class _SnapshotMock(MockLLMClient):
+        async def stream(
+            self, messages: Sequence[Message], *, model: str, **params
+        ) -> AsyncIterator[str]:
+            text = self._canned.pop(0)
+            for i, ch in enumerate(text):
+                if i == 0:
+                    # Snapshot at first yield — the post_turn coroutine
+                    # is awaiting our generator and the streaming Task
+                    # is registered in _in_flight_tasks[chat_id].
+                    in_flight_snapshot["present"] = (
+                        "chat_bot_a" in _in_flight_tasks
+                    )
+                    in_flight_snapshot["task"] = _in_flight_tasks.get(
+                        "chat_bot_a"
+                    )
+                yield ch
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "narration", "text": "they exchange a glance"}]}
+    )
+    mock = _SnapshotMock(
+        canned=[
+            canned_parse,
+            "BotA leans in. *quietly*",
+            _zero_state(),
+            _zero_state(),
+        ]
+    )
+    from chat.web.kickoff import get_llm_client
+
+    app.dependency_overrides[get_llm_client] = lambda: mock
+    try:
+        # Pre-condition: registry is empty for this chat.
+        assert "chat_bot_a" not in _in_flight_tasks
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns",
+            data={"prose": "they exchange a glance"},
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+
+    # Mid-flight: the streaming task was present in the registry, and
+    # the captured value was an asyncio.Task (not None / not some other
+    # placeholder).
+    import asyncio
+
+    assert in_flight_snapshot.get("present") is True, (
+        "_in_flight_tasks was empty at first yield — meanwhile stream "
+        "isn't registering its task"
+    )
+    assert isinstance(in_flight_snapshot.get("task"), asyncio.Task)
+    # Post-flight: the entry has been cleaned up so the next turn (or
+    # the cancel route) doesn't see a stale task.
+    assert "chat_bot_a" not in _in_flight_tasks

tests/test_per_pov_summary.py

@@ -1095,3 +1095,326 @@ async def test_thread_detection_emits_events(tmp_path, monkeypatch):
         open_threads = list_open_threads(conn, "chat_bot_a")
         assert len(open_threads) == 1
         assert open_threads[0]["title"] == "Test thread"
+
+
+# ---------------------------------------------------------------------------
+# T65: meanwhile summary digest emitted on meanwhile-scene close, surfaced in
+# the next you-scene's prompt as a SHOULD-tier "Meanwhile while you were away:"
+# block, then consumed so it never re-renders.
+# ---------------------------------------------------------------------------
+
+
+def _seed_meanwhile_scene(conn) -> None:
+    """Seed a parent you-scene + a meanwhile child scene with one assistant
+    turn so apply_scene_close_summary has dialogue to summarize.
+
+    The meanwhile scene id is 2 (parent is scene 1). The meanwhile dialogue
+    is appended via assistant_turn events under chat_bot_a; the
+    _read_recent_dialogue helper picks them up by chat_id.
+    """
+    import chat.state.meanwhile  # noqa: F401  -- register handlers
+
+    append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
+    append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+    append_event(
+        conn,
+        kind="you_authored",
+        payload={"name": "Me", "pronouns": "they/them", "persona": "engineer"},
+    )
+    append_event(
+        conn,
+        kind="chat_created",
+        payload={
+            "id": "chat_bot_a",
+            "host_bot_id": "bot_a",
+            "guest_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_a",
+            "name": "office",
+            "type": "workplace",
+            "properties": {},
+        },
+    )
+    # Parent you-scene (scene_id=1).
+    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", "bot_b"],
+        },
+    )
+    # Meanwhile child scene (scene_id=2) — bot_a + bot_b only.
+    append_event(
+        conn,
+        kind="meanwhile_scene_started",
+        payload={
+            "scene_id": 2,
+            "chat_id": "chat_bot_a",
+            "parent_scene_id": 1,
+            "host_bot_id": "bot_a",
+            "guest_bot_id": "bot_b",
+            "started_at": "2026-04-26T20:05:00+00:00",
+        },
+    )
+    # Edges so per-POV apply has rows to update.
+    append_event(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "bot_a",
+            "target_id": "you",
+            "chat_id": "chat_bot_a",
+        },
+    )
+    append_event(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "bot_b",
+            "target_id": "you",
+            "chat_id": "chat_bot_a",
+        },
+    )
+    # One memory per witness in the meanwhile scene.
+    append_event(
+        conn,
+        kind="memory_written",
+        payload={
+            "owner_id": "bot_a",
+            "chat_id": "chat_bot_a",
+            "scene_id": 2,
+            "pov_summary": "Original raw narrative (host, meanwhile)",
+            "witness_you": 0,
+            "witness_host": 1,
+            "witness_guest": 1,
+            "significance": 1,
+        },
+    )
+    append_event(
+        conn,
+        kind="memory_written",
+        payload={
+            "owner_id": "bot_b",
+            "chat_id": "chat_bot_a",
+            "scene_id": 2,
+            "pov_summary": "Original raw narrative (guest, meanwhile)",
+            "witness_you": 0,
+            "witness_host": 1,
+            "witness_guest": 1,
+            "significance": 1,
+        },
+    )
+    # A bot-bot turn happens during the meanwhile scene.
+    append_event(
+        conn,
+        kind="assistant_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "speaker_id": "bot_a",
+            "text": "Did you hear what happened with the missing file?",
+            "truncated": False,
+            "user_turn_id": None,
+        },
+    )
+    append_event(
+        conn,
+        kind="assistant_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "speaker_id": "bot_b",
+            "text": "I have a theory but no proof yet.",
+            "truncated": False,
+            "user_turn_id": None,
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_meanwhile_close_creates_digest(tmp_path):
+    """When apply_scene_close_summary runs on a meanwhile scene
+    (present_set_kind == 'host_guest'), it emits a meanwhile_digest_created
+    event after the per-POV summaries land; the meanwhile_digest_pending
+    table then holds a row with non-empty summary text."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    host_canned = json.dumps(
+        {
+            "summary": "BotA confided in BotB about the missing file.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotA leaned on BotB.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB listened and offered to help investigate.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotB grew protective.",
+        }
+    )
+    digest_canned = json.dumps(
+        {
+            "summary": (
+                "While you were away, BotA confided in BotB about a "
+                "missing file; BotB offered to help quietly investigate."
+            ),
+            "knowledge_facts": [],
+            "relationship_summary": "",
+        }
+    )
+    no_threads = json.dumps({"candidates": []})
+    with open_db(db) as conn:
+        _seed_meanwhile_scene(conn)
+        project(conn)
+
+        # Order: host POV summary, guest POV summary, digest summary,
+        # thread detection.
+        client = MockLLMClient(
+            canned=[host_canned, guest_canned, digest_canned, no_threads]
+        )
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=2,
+            host_bot_id="bot_a",
+        )
+
+        # The meanwhile_digest_pending row was written.
+        from chat.state.meanwhile import list_pending_meanwhile_digests
+
+        pending = list_pending_meanwhile_digests(conn, "chat_bot_a")
+        assert len(pending) == 1
+        assert pending[0]["scene_id"] == 2
+        assert pending[0]["summary"]
+        assert "missing file" in pending[0]["summary"]
+
+        # And the meanwhile_digest_created event was logged.
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'meanwhile_digest_created'"
+        ).fetchall()
+        assert len(rows) == 1
+        payload = json.loads(rows[0][0])
+        assert payload["chat_id"] == "chat_bot_a"
+        assert payload["scene_id"] == 2
+        assert "missing file" in payload["summary"]
+
+
+def test_pending_digest_renders_in_you_scene_prompt(tmp_path):
+    """A pending meanwhile digest (created via direct event append) renders
+    as a 'Meanwhile while you were away:' SHOULD-tier block in the
+    assembled you-scene narrative prompt."""
+    from chat.eventlog.log import append_and_apply
+    from chat.services.prompt import assemble_narrative_prompt
+    import chat.state.meanwhile  # noqa: F401  -- register handlers
+    import chat.state.threads  # noqa: F401
+    import chat.state.events  # noqa: F401
+
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_single_bot_scene(conn)
+        project(conn)
+
+        digest_text = (
+            "While you were away, BotA confided in BotB about a missing file."
+        )
+        append_and_apply(
+            conn,
+            kind="meanwhile_digest_created",
+            payload={
+                "chat_id": "chat_bot_a",
+                "scene_id": 2,
+                "summary": digest_text,
+            },
+        )
+
+        msgs = assemble_narrative_prompt(
+            conn,
+            chat_id="chat_bot_a",
+            speaker_bot_id="bot_a",
+            recent_dialogue=[],
+            retrieved_memory_summaries=[],
+        )
+        body = msgs[0].content
+        assert "Meanwhile while you were away:" in body
+        assert digest_text in body
+
+
+def test_consumed_digest_does_not_render_again(tmp_path):
+    """After meanwhile_digest_consumed lands for a digest, reassembling the
+    you-scene prompt must NOT include that digest's text — the pending
+    list is filtered by ``consumed_at IS NULL``."""
+    from chat.eventlog.log import append_and_apply
+    from chat.services.prompt import assemble_narrative_prompt
+    import chat.state.meanwhile  # noqa: F401  -- register handlers
+    import chat.state.threads  # noqa: F401
+    import chat.state.events  # noqa: F401
+
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_single_bot_scene(conn)
+        project(conn)
+
+        digest_text = (
+            "While you were away, BotA confided in BotB about a missing file."
+        )
+        append_and_apply(
+            conn,
+            kind="meanwhile_digest_created",
+            payload={
+                "chat_id": "chat_bot_a",
+                "scene_id": 2,
+                "summary": digest_text,
+            },
+        )
+
+        # Sanity: it renders before consumption.
+        msgs = assemble_narrative_prompt(
+            conn,
+            chat_id="chat_bot_a",
+            speaker_bot_id="bot_a",
+            recent_dialogue=[],
+            retrieved_memory_summaries=[],
+        )
+        assert digest_text in msgs[0].content
+
+        # Look up the pending digest id, then consume it.
+        from chat.state.meanwhile import list_pending_meanwhile_digests
+
+        pending = list_pending_meanwhile_digests(conn, "chat_bot_a")
+        assert len(pending) == 1
+        digest_id = pending[0]["id"]
+
+        append_and_apply(
+            conn,
+            kind="meanwhile_digest_consumed",
+            payload={
+                "digest_id": digest_id,
+                "consumed_at": "2026-04-26T20:30:00+00:00",
+            },
+        )
+
+        msgs2 = assemble_narrative_prompt(
+            conn,
+            chat_id="chat_bot_a",
+            speaker_bot_id="bot_a",
+            recent_dialogue=[],
+            retrieved_memory_summaries=[],
+        )
+        body2 = msgs2[0].content
+        assert "Meanwhile while you were away:" not in body2
+        assert digest_text not in body2

tests/test_world.py

@@ -324,11 +324,11 @@ def test_get_scene_returns_none_for_missing(tmp_path):
         assert active_scene(conn, "chat_missing") is None
 
 
-def test_schema_version_after_migration_is_10(tmp_path):
+def test_schema_version_after_migration_is_11(tmp_path):
     db = tmp_path / "t.db"
     apply_migrations(db)
     with open_db(db) as conn:
         row = conn.execute(
             "SELECT value FROM meta WHERE key = 'schema_version'"
         ).fetchone()
-        assert int(row[0]) == 10
+        assert int(row[0]) == 11