merge: T58 scene compression + thread emission on close

chat/db/migrations/0009_events.sql

@@ -0,0 +1,14 @@
+CREATE TABLE events (
+    id INTEGER PRIMARY KEY,
+    event_id TEXT NOT NULL UNIQUE,
+    chat_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'planned',
+    props_json TEXT NOT NULL DEFAULT '{}',
+    planned_for TEXT,
+    started_at TEXT,
+    completed_at TEXT,
+    created_at TEXT NOT NULL DEFAULT (datetime('now')),
+    updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX events_chat_idx ON events(chat_id, status);

chat/db/migrations/0010_threads.sql

@@ -0,0 +1,14 @@
+CREATE TABLE threads (
+    id INTEGER PRIMARY KEY,
+    thread_id TEXT NOT NULL UNIQUE,
+    chat_id TEXT NOT NULL,
+    title TEXT NOT NULL,
+    summary TEXT NOT NULL DEFAULT '',
+    status TEXT NOT NULL DEFAULT 'open',
+    opened_at TEXT NOT NULL DEFAULT (datetime('now')),
+    closed_at TEXT,
+    last_referenced_scene_id INTEGER,
+    created_at TEXT NOT NULL DEFAULT (datetime('now')),
+    updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX threads_chat_status_idx ON threads(chat_id, status);

chat/services/event_lifecycle.py

@@ -0,0 +1,72 @@
+"""Event-lifecycle detection (T52).
+
+After each turn, classify whether any active events transitioned
+(started, completed, cancelled). Conservative bias — most turns
+return empty. T61 turn flow appends one event_started/completed/
+cancelled per transition via append_and_apply.
+"""
+
+from __future__ import annotations
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class EventTransition(BaseModel):
+    event_id: str
+    new_status: str  # "active" | "completed" | "cancelled"
+    reason: str = ""
+
+
+class EventLifecycleDecision(BaseModel):
+    transitions: list[EventTransition] = Field(default_factory=list)
+
+
+_SYSTEM = (
+    "You decide whether any active events transitioned this turn. "
+    "STRONGLY default to empty transitions — most turns do NOT resolve "
+    "or start a known event. Output only transitions that the narrative "
+    "text clearly resolves or starts. Each transition MUST reference an "
+    "event_id from the active_events list. new_status is one of "
+    "'active' (planned -> active), 'completed', or 'cancelled'. "
+    "Output strict JSON matching the schema."
+)
+
+
+async def detect_event_transitions(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    narrative_text: str,
+    active_events: list[dict],   # [{event_id, kind, status, props}, ...]
+    timeout_s: float = 30.0,
+) -> EventLifecycleDecision:
+    """Classify event transitions for the latest turn. Empty active_events
+    short-circuits without an LLM call."""
+    if not active_events:
+        return EventLifecycleDecision()
+
+    user_lines = ["Active events:"]
+    for ev in active_events:
+        user_lines.append(
+            f"- event_id={ev['event_id']} kind={ev['kind']} "
+            f"status={ev['status']} props={ev.get('props', {})}"
+        )
+    user_lines.append("")
+    user_lines.append("Latest narrative:")
+    user_lines.append(narrative_text.strip())
+    user = "\n".join(user_lines)
+
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_SYSTEM,
+        user=user,
+        schema=EventLifecycleDecision,
+        default=EventLifecycleDecision(),
+        timeout_s=timeout_s,
+    )
+
+
+__all__ = ["EventTransition", "EventLifecycleDecision", "detect_event_transitions"]

chat/services/event_promotion.py

@@ -0,0 +1,149 @@
+"""Event-completion promotion (T56).
+
+When an event reaches ``status='completed'``, read its ``props_json``
+and emit promotion events into the appropriate state stores.
+Synchronous, no LLM. Skips when the event status is not ``completed``
+(cancelled / expired terminate the event without promoting).
+
+Props recognized:
+
+- ``acquired_objects: list[str]`` — emits a ``manual_edit`` with
+  ``target_kind="memory_pov_summary"`` per object on the host's memory
+  row, recording the acquisition. Phase 3 is a stub: it requires both
+  ``host_bot_id`` and ``host_memory_id`` (an existing memories.id) to
+  be present in props; missing either skips that object cleanly.
+  Phase 4 will introduce a real inventory schema.
+
+- ``knowledge_facts: list[{owner_id, target_id, fact}]`` — emits an
+  ``edge_update`` event on the directed ``owner_id -> target_id`` edge
+  with the fact appended to ``knowledge_facts``. The ``edge_update``
+  projector accepts ``knowledge_facts`` as a list and extends the
+  edge's stored ``knowledge_json``.
+
+- ``relationship_change: {summary, source_id, target_id}`` — emits a
+  ``manual_edit`` with ``target_kind="edge_summary"`` overwriting the
+  edge's ``summary`` field on the directed pair.
+
+Anything else stays in the closed event record (the projector kept
+the row; no further promotion).
+"""
+
+from __future__ import annotations
+
+from sqlite3 import Connection
+
+from chat.eventlog.log import append_and_apply
+from chat.state.events import get_event
+
+
+def promote_completed_event(
+    conn: Connection,
+    *,
+    event_id: str,
+    chat_id: str,
+    chat_clock_at: str | None,
+) -> dict:
+    """Read the completed event's props and emit promotion events.
+
+    Returns a dict of counts keyed by promoted artifact:
+    ``{"acquired_objects", "knowledge_facts", "relationship_change"}``.
+    Skips silently if the event row is missing or its status is not
+    ``completed`` — cancelled / expired events terminate without any
+    promotion.
+    """
+    counts = {
+        "acquired_objects": 0,
+        "knowledge_facts": 0,
+        "relationship_change": 0,
+    }
+
+    event = get_event(conn, event_id)
+    if event is None or event["status"] != "completed":
+        return counts
+
+    props = event.get("props") or {}
+
+    # acquired_objects: each becomes a memory_pov_summary edit (Phase 3
+    # stub). The manual_edit projector requires a valid memory rowid as
+    # ``target_id`` (it does ``int(target_id)``), so skip cleanly when
+    # neither a host_bot_id nor a host_memory_id is supplied.
+    host_bot_id = props.get("host_bot_id")
+    host_memory_id = props.get("host_memory_id")
+    for obj in props.get("acquired_objects", []) or []:
+        if host_bot_id is None or host_memory_id is None:
+            continue
+        append_and_apply(
+            conn,
+            kind="manual_edit",
+            payload={
+                "target_kind": "memory_pov_summary",
+                "target_id": host_memory_id,
+                "owner_id": host_bot_id,
+                "chat_id": chat_id,
+                "prior_value": "",
+                "new_value": f"Acquired: {obj}",
+                "source": "event_promotion",
+                "event_id": event_id,
+                "chat_clock_at": chat_clock_at,
+            },
+        )
+        counts["acquired_objects"] += 1
+
+    # knowledge_facts: each becomes an edge_update appending the fact.
+    for fact_entry in props.get("knowledge_facts", []) or []:
+        owner_id = fact_entry.get("owner_id")
+        target_id = fact_entry.get("target_id")
+        fact = fact_entry.get("fact", "")
+        if not owner_id or not target_id or not fact:
+            continue
+        append_and_apply(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": owner_id,
+                "target_id": target_id,
+                "chat_id": chat_id,
+                "affinity_delta": 0,
+                "trust_delta": 0,
+                "knowledge_facts": [fact],
+                "last_interaction_at": chat_clock_at,
+                "last_interaction_chat_id": chat_id,
+                "source": "event_promotion",
+                "event_id": event_id,
+            },
+        )
+        counts["knowledge_facts"] += 1
+
+    # relationship_change: edge_summary manual_edit on the directed pair.
+    # The manual_edit projector for ``edge_summary`` keys on a
+    # ``target_id`` dict ``{source_id, target_id}`` (see
+    # chat/state/manual_edit.py); we shape the payload to match.
+    rc = props.get("relationship_change") or {}
+    if rc:
+        source_id = rc.get("source_id")
+        rc_target_id = rc.get("target_id")
+        summary = rc.get("summary", "")
+        if source_id and rc_target_id and summary:
+            append_and_apply(
+                conn,
+                kind="manual_edit",
+                payload={
+                    "target_kind": "edge_summary",
+                    "target_id": {
+                        "source_id": source_id,
+                        "target_id": rc_target_id,
+                    },
+                    "chat_id": chat_id,
+                    "prior_value": "",
+                    "new_value": summary,
+                    "source": "event_promotion",
+                    "event_id": event_id,
+                    "chat_clock_at": chat_clock_at,
+                },
+            )
+            counts["relationship_change"] += 1
+
+    return counts
+
+
+__all__ = ["promote_completed_event"]

chat/services/scene_summarize.py

@@ -29,6 +29,8 @@ keeps moving.
 from __future__ import annotations
 
 import json
+import uuid
+from datetime import datetime, timezone
 from sqlite3 import Connection
 
 from pydantic import BaseModel, Field
@@ -167,6 +169,7 @@ async def _summarize_and_apply_for_witness(
     you_name: str,
     dialogue: list[dict],
     timeout_s: float,
+    key_quotes_suffix: str = "",
 ) -> ScenePOVSummary:
     """Run :func:`summarize_scene` for one bot witness and apply the
     three projected updates (memory pov_summary rewrite, edge summary
@@ -175,6 +178,10 @@ async def _summarize_and_apply_for_witness(
     Tolerant of missing pieces in the same way Phase 1 was: no memory
     row -> skip the rewrite; no edge row -> skip the edge_summary write
     (the empty-default classifier output simply yields no rewrites).
+
+    ``key_quotes_suffix`` is appended verbatim to the per-POV summary
+    text before the rewrite lands (T58.1) — empty string is the no-op
+    default for low-significance scenes.
     """
     from chat.state.edges import get_edge
     from chat.state.entities import get_bot
@@ -206,6 +213,7 @@ async def _summarize_and_apply_for_witness(
             # Empty default -> skip the memory rewrite; the seeded
             # per-turn pov_summary stays in place.
             continue
+        new_value = pov.summary + key_quotes_suffix
         append_and_apply(
             conn,
             kind="manual_edit",
@@ -213,7 +221,7 @@ async def _summarize_and_apply_for_witness(
                 "target_kind": "memory_pov_summary",
                 "target_id": int(memory_id),
                 "prior_value": prior_pov,
-                "new_value": pov.summary,
+                "new_value": new_value,
             },
         )
 
@@ -255,6 +263,40 @@ async def _summarize_and_apply_for_witness(
     return pov
 
 
+def _build_key_quotes_suffix(conn: Connection, scene_id: int) -> str:
+    """If the scene's max-turn-significance is >= 2, build the
+    "Key quotes:" suffix from the top-3 highest-significance memory rows
+    (per requirements §11.1). Otherwise return the empty string so the
+    per-POV summaries collapse fully (low-significance scenes lose all
+    raw text in favor of the classifier rewrite).
+
+    Quote source is each memory's current ``pov_summary`` — the raw
+    per-turn narrative seeded by T21, since this helper is called BEFORE
+    the per-POV rewrite. Texts are truncated to 200 chars to bound
+    memory row growth across many witnesses.
+    """
+    row = conn.execute(
+        "SELECT MAX(significance) FROM memories WHERE scene_id = ?",
+        (scene_id,),
+    ).fetchone()
+    max_sig = (row[0] if row else None) or 0
+    if max_sig < 2:
+        return ""
+    cur = conn.execute(
+        "SELECT pov_summary FROM memories WHERE scene_id = ? "
+        "ORDER BY significance DESC, id ASC LIMIT 3",
+        (scene_id,),
+    )
+    quotes = [
+        (r[0] or "")[:200]
+        for r in cur.fetchall()
+    ]
+    if not quotes:
+        return ""
+    lines = "\n".join(f'- "{q}"' for q in quotes)
+    return f"\n\nKey quotes:\n{lines}"
+
+
 async def apply_scene_close_summary(
     conn: Connection,
     client: LLMClient,
@@ -296,8 +338,10 @@ async def apply_scene_close_summary(
     """
     # Local imports to keep the module-level surface tight and avoid
     # any chance of a circular dep through chat.state.*.
+    from chat.services.thread_detection import detect_threads
     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
 
     you_entity = get_you(conn) or {"name": "you", "persona": ""}
@@ -308,6 +352,11 @@ async def apply_scene_close_summary(
 
     dialogue = _read_recent_dialogue(conn, chat_id)
 
+    # T58.1: build the "Key quotes:" suffix BEFORE the per-POV rewrites
+    # land — quote source is the raw seeded pov_summary text on each
+    # memory row, which the rewrite about to fire would clobber.
+    key_quotes_suffix = _build_key_quotes_suffix(conn, scene_id)
+
     host_pov = await _summarize_and_apply_for_witness(
         conn,
         client,
@@ -318,6 +367,7 @@ async def apply_scene_close_summary(
         you_name=you_name,
         dialogue=dialogue,
         timeout_s=timeout_s,
+        key_quotes_suffix=key_quotes_suffix,
     )
 
     guest_pov: ScenePOVSummary | None = None
@@ -332,6 +382,7 @@ async def apply_scene_close_summary(
             you_name=you_name,
             dialogue=dialogue,
             timeout_s=timeout_s,
+            key_quotes_suffix=key_quotes_suffix,
         )
 
     # Group node update: T70 runs a third classifier call to merge the
@@ -364,6 +415,56 @@ async def apply_scene_close_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
+    # empty default on retry-exhaustion, and the broad except below
+    # protects the close pipeline from any other classifier/mock flap.
+    try:
+        thread_result = await detect_threads(
+            client,
+            classifier_model=classifier_model,
+            scene_transcript=dialogue,
+            open_threads=list_open_threads(conn, chat_id),
+            timeout_s=timeout_s,
+        )
+    except Exception:
+        from chat.services.thread_detection import ThreadDetectionResult
+
+        thread_result = ThreadDetectionResult()
+    for cand in thread_result.candidates:
+        if cand.action == "open":
+            new_thread_id = f"thr_{uuid.uuid4().hex[:12]}"
+            append_and_apply(
+                conn,
+                kind="thread_opened",
+                payload={
+                    "thread_id": new_thread_id,
+                    "chat_id": chat_id,
+                    "title": cand.title,
+                    "summary": cand.summary,
+                },
+            )
+        elif cand.action == "update" and cand.existing_thread_id:
+            append_and_apply(
+                conn,
+                kind="thread_updated",
+                payload={
+                    "thread_id": cand.existing_thread_id,
+                    "summary": cand.summary,
+                    "last_referenced_scene_id": scene_id,
+                },
+            )
+        elif cand.action == "close" and cand.existing_thread_id:
+            append_and_apply(
+                conn,
+                kind="thread_closed",
+                payload={
+                    "thread_id": cand.existing_thread_id,
+                    "closed_at": datetime.now(timezone.utc).isoformat(),
+                },
+            )
+
     return host_pov
 
 

chat/services/skip_narration.py

@@ -0,0 +1,131 @@
+"""Skip narration service (T53).
+
+Generates brief transition prose for elision and jump skips.
+
+Skips come in two flavors that read very differently:
+
+* **Elision** — collapses an in-progress activity into its expected
+  end-state in 1-2 sentences, narrated from the speaker bot's POV.
+  Example: "skip ahead to when we arrive" while the characters are
+  driving — output describes pulling into the lot.
+* **Jump** — bridges a longer fiction-time delta ("next morning", "a
+  week later") in 2-3 sentences, setting the scene at the new time.
+
+Output is free-form prose, not structured JSON, so this service calls
+``client.generate`` directly rather than going through the classifier
+path used by, e.g., :mod:`chat.services.scene_summarize`. A
+deterministic template fallback fires on any LLM failure so the skip
+flow keeps moving even when the model is down — important because
+skips are a UI-blocking operation; we'd rather show a parenthetical
+sentence than hang the chat indefinitely.
+"""
+
+from __future__ import annotations
+
+from chat.llm.client import LLMClient, Message
+
+
+_ELISION_SYSTEM = (
+    "You write a brief 1-2 sentence transition that elides the time "
+    "between an in-progress activity and its expected end-state, "
+    "narrated from the speaker's POV. Keep it grounded and concrete. "
+    "Do not invent new events or characters."
+)
+
+_JUMP_SYSTEM = (
+    "You write a brief 2-3 sentence transition narrating a jump in "
+    "fiction time (e.g., 'next morning', 'a week later'), narrated "
+    "from the speaker's POV. Set the scene at the new time. Keep it "
+    "grounded — no invented major events. If a landing-state hint is "
+    "provided, weave it in naturally."
+)
+
+
+async def narrate_skip(
+    client: LLMClient,
+    *,
+    narrative_model: str,
+    skip_kind: str,
+    speaker_bot: dict,
+    you_name: str,
+    current_time: str,
+    new_time: str,
+    current_activity: str,
+    landing_state_hint: str = "",
+    timeout_s: float = 60.0,
+) -> str:
+    """Generate brief transition prose for a time skip.
+
+    ``skip_kind`` is ``"elision"`` or ``"jump"``; any other value short-
+    circuits to the deterministic fallback (defensive — callers
+    shouldn't be inventing new kinds without updating this service).
+
+    Returns plain text. Never raises: any LLM error, an empty/blank
+    result, or an unknown ``skip_kind`` falls back to a parenthetical
+    template like ``"(next morning: having coffee in the kitchen.)"``
+    so the skip UI always has *something* to render.
+    """
+    fallback = _build_fallback(
+        skip_kind=skip_kind,
+        new_time=new_time,
+        current_activity=current_activity,
+        landing_state_hint=landing_state_hint,
+    )
+
+    if skip_kind not in ("elision", "jump"):
+        return fallback
+
+    system = _ELISION_SYSTEM if skip_kind == "elision" else _JUMP_SYSTEM
+    user = (
+        f"Speaker: {speaker_bot.get('name', 'speaker')}\n"
+        f"Persona: {speaker_bot.get('persona', '')}\n"
+        f"Other party: {you_name}\n"
+        f"Current time: {current_time}\n"
+        f"New time: {new_time}\n"
+        f"Current activity: {current_activity}\n"
+    )
+    if landing_state_hint:
+        user += f"Landing state hint: {landing_state_hint}\n"
+
+    try:
+        result = await client.generate(
+            [
+                Message(role="system", content=system),
+                Message(role="user", content=user),
+            ],
+            model=narrative_model,
+            max_tokens=200,
+            temperature=0.7,
+        )
+        text = (result or "").strip()
+        if not text:
+            return fallback
+        return text
+    except Exception:
+        # Any failure — network blip, timeout, mock raising in tests —
+        # collapses to the deterministic template so the skip pipeline
+        # is never blocked on the LLM being available.
+        return fallback
+
+
+def _build_fallback(
+    *,
+    skip_kind: str,
+    new_time: str,
+    current_activity: str,
+    landing_state_hint: str,
+) -> str:
+    """Deterministic parenthetical narration used when the LLM fails.
+
+    Both flavors render the same shape today: ``(<new_time>:
+    <detail>.)``. They're separated as branches to make it easy to
+    diverge later (e.g. an elision-specific template) without churning
+    the call site or the public signature.
+    """
+    detail = landing_state_hint or current_activity or "moments later"
+    if skip_kind == "elision":
+        return f"({new_time}: {detail}.)"
+    return f"({new_time}: {detail}.)"
+
+
+__all__ = ["narrate_skip"]

chat/services/synthesized_memories.py

@@ -0,0 +1,74 @@
+"""Synthesized-memories service (T54).
+
+When the user jump-skips with 'anything notable happen?' prose, parse
+that prose into 1-N synthesized memories per present bot. Each memory
+carries source="synthesized" and reliability=0.7 (lower than direct).
+Caller (T62 skip flow) writes the memories via record_turn_memory_for_present.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class SynthesizedMemory(BaseModel):
+    text: str
+    significance: int = 1   # 0..3, default 1
+    affinity_delta: int = 0
+    trust_delta: int = 0
+
+
+class SynthesizedDigest(BaseModel):
+    memories: list[SynthesizedMemory] = Field(default_factory=list)
+
+
+_SYSTEM = (
+    "You parse a short user-supplied prose describing 'anything notable' "
+    "that happened during a time skip into 1-N synthesized memories from "
+    "a single bot's POV. Each memory has: text (one factual sentence "
+    "from that bot's perspective), significance (0-3, default 1; only "
+    "use 2 or 3 for genuinely scene-level or relationship-altering "
+    "events), affinity_delta and trust_delta (-10..+10, default 0; "
+    "use small adjustments only when prose explicitly describes a shift). "
+    "Empty/whitespace prose returns an empty memories list. Output "
+    "strict JSON matching the schema."
+)
+
+
+async def synthesize_memories(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    prose: str,
+    bot_name: str,
+    bot_persona: str,
+    you_name: str,
+    timeout_s: float = 30.0,
+) -> SynthesizedDigest:
+    """Parse 'anything notable' prose into structured memories from a
+    single bot's POV. Empty/whitespace prose short-circuits to an
+    empty digest (no LLM call)."""
+    if not prose or not prose.strip():
+        return SynthesizedDigest()
+
+    user = (
+        f"Bot: {bot_name}\n"
+        f"Persona: {bot_persona}\n"
+        f"Other party: {you_name}\n\n"
+        f"Prose:\n{prose.strip()}"
+    )
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_SYSTEM,
+        user=user,
+        schema=SynthesizedDigest,
+        default=SynthesizedDigest(),
+        timeout_s=timeout_s,
+    )
+
+
+__all__ = ["SynthesizedMemory", "SynthesizedDigest", "synthesize_memories"]

chat/services/thread_detection.py

@@ -0,0 +1,89 @@
+"""Thread-detection service (T55).
+
+On scene close, classify the transcript into thread open/update/close
+candidates. Returns ThreadCandidate list; caller (T58 scene compression)
+emits one thread_opened/thread_updated/thread_closed event per candidate.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class ThreadCandidate(BaseModel):
+    action: str  # "open" | "update" | "close"
+    title: str = ""        # required for "open"; ignored otherwise
+    summary: str = ""
+    existing_thread_id: str | None = None  # required for "update" / "close"
+
+
+class ThreadDetectionResult(BaseModel):
+    candidates: list[ThreadCandidate] = Field(default_factory=list)
+
+
+_SYSTEM = (
+    "You analyze a closed scene's transcript to identify narrative "
+    "threads (unresolved arcs, dangling questions, promises made, "
+    "open obligations). Choose actions:\n"
+    "- 'open': a NEW thread the scene introduced. Provide title (short "
+    "noun phrase) + summary (one sentence).\n"
+    "- 'update': an EXISTING open thread that the scene developed. "
+    "Provide existing_thread_id + new summary.\n"
+    "- 'close': an EXISTING open thread that the scene resolved. "
+    "Provide existing_thread_id; summary may capture the resolution.\n"
+    "Conservative bias: most scenes do NOT open new threads. Only "
+    "produce candidates when the transcript clearly justifies them. "
+    "Output strict JSON matching the schema."
+)
+
+
+async def detect_threads(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    scene_transcript: list[dict],   # [{speaker, text}, ...]
+    open_threads: list[dict],       # [{thread_id, title, summary}, ...]
+    timeout_s: float = 30.0,
+) -> ThreadDetectionResult:
+    """Classify scene close into thread open/update/close candidates."""
+    if not scene_transcript:
+        return ThreadDetectionResult()
+
+    transcript_lines = [
+        f"{turn.get('speaker', 'unknown')}: {turn.get('text', '')}"
+        for turn in scene_transcript
+    ]
+    threads_lines = []
+    if open_threads:
+        threads_lines.append("Currently open threads:")
+        for t in open_threads:
+            threads_lines.append(
+                f"- thread_id={t['thread_id']} "
+                f"title={t.get('title', '')} "
+                f"summary={t.get('summary', '')}"
+            )
+    else:
+        threads_lines.append("No currently open threads.")
+
+    user = (
+        "Scene transcript:\n"
+        + "\n".join(transcript_lines)
+        + "\n\n"
+        + "\n".join(threads_lines)
+    )
+
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_SYSTEM,
+        user=user,
+        schema=ThreadDetectionResult,
+        default=ThreadDetectionResult(),
+        timeout_s=timeout_s,
+    )
+
+
+__all__ = ["ThreadCandidate", "ThreadDetectionResult", "detect_threads"]

chat/state/events.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+import json
+from sqlite3 import Connection
+
+from chat.eventlog.projector import on
+from chat.eventlog.log import Event
+
+
+_TERMINAL_STATUSES = {"completed", "cancelled", "expired"}
+
+
+@on("event_planned")
+def _apply_event_planned(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "INSERT OR IGNORE INTO events "
+        "(event_id, chat_id, kind, status, props_json, planned_for) "
+        "VALUES (?, ?, ?, 'planned', ?, ?)",
+        (
+            p["event_id"],
+            p["chat_id"],
+            p["kind"],
+            json.dumps(p.get("props", {})),
+            p.get("planned_for"),
+        ),
+    )
+
+
+@on("event_started")
+def _apply_event_started(conn: Connection, e: Event) -> None:
+    p = e.payload
+    # Idempotent: only transition from non-terminal status.
+    conn.execute(
+        "UPDATE events SET status = 'active', started_at = ?, updated_at = datetime('now') "
+        "WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
+        (p.get("started_at"), p["event_id"]),
+    )
+
+
+@on("event_completed")
+def _apply_event_completed(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE events SET status = 'completed', completed_at = ?, updated_at = datetime('now') "
+        "WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
+        (p.get("completed_at"), p["event_id"]),
+    )
+
+
+@on("event_cancelled")
+def _apply_event_cancelled(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE events SET status = 'cancelled', completed_at = ?, updated_at = datetime('now') "
+        "WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
+        (p.get("completed_at"), p["event_id"]),
+    )
+
+
+@on("event_expired")
+def _apply_event_expired(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE events SET status = 'expired', completed_at = ?, updated_at = datetime('now') "
+        "WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
+        (p.get("completed_at"), p["event_id"]),
+    )
+
+
+def get_event(conn: Connection, event_id: str) -> dict | None:
+    row = conn.execute(
+        "SELECT event_id, chat_id, kind, status, props_json, planned_for, "
+        "started_at, completed_at, created_at, updated_at "
+        "FROM events WHERE event_id = ?",
+        (event_id,),
+    ).fetchone()
+    if not row:
+        return None
+    return {
+        "event_id": row[0],
+        "chat_id": row[1],
+        "kind": row[2],
+        "status": row[3],
+        "props": json.loads(row[4]),
+        "planned_for": row[5],
+        "started_at": row[6],
+        "completed_at": row[7],
+        "created_at": row[8],
+        "updated_at": row[9],
+    }
+
+
+def list_active_events(conn: Connection, chat_id: str) -> list[dict]:
+    rows = conn.execute(
+        "SELECT event_id, chat_id, kind, status, props_json, planned_for, "
+        "started_at, completed_at, created_at, updated_at "
+        "FROM events WHERE chat_id = ? AND status IN ('planned','active') "
+        "ORDER BY id ASC",
+        (chat_id,),
+    ).fetchall()
+    return [
+        {
+            "event_id": r[0], "chat_id": r[1], "kind": r[2], "status": r[3],
+            "props": json.loads(r[4]),
+            "planned_for": r[5], "started_at": r[6], "completed_at": r[7],
+            "created_at": r[8], "updated_at": r[9],
+        }
+        for r in rows
+    ]
+
+
+def list_events_in_status(conn: Connection, chat_id: str, status: str) -> list[dict]:
+    rows = conn.execute(
+        "SELECT event_id, chat_id, kind, status, props_json, planned_for, "
+        "started_at, completed_at, created_at, updated_at "
+        "FROM events WHERE chat_id = ? AND status = ? ORDER BY id ASC",
+        (chat_id, status),
+    ).fetchall()
+    return [
+        {
+            "event_id": r[0], "chat_id": r[1], "kind": r[2], "status": r[3],
+            "props": json.loads(r[4]),
+            "planned_for": r[5], "started_at": r[6], "completed_at": r[7],
+            "created_at": r[8], "updated_at": r[9],
+        }
+        for r in rows
+    ]

chat/state/memory.py

@@ -94,6 +94,14 @@ def get_pinned(conn: Connection, owner_id: str) -> list[dict]:
 _SIGNIFICANCE_WEIGHT = 0.3
 _RECENCY_WEIGHT = 0.5
 
+# T57 (Phase 3, §11.1): significance multiplier applied to the SQL ORDER BY in
+# ``search_memories`` so that the FTS over-fetch already prefers
+# higher-significance rows for tied / near-tied BM25 ranks. Module-level so it
+# can be tuned without a code change. BM25 ``rank`` is lower-is-better, so the
+# bias is *subtracted* from rank in the ASC ordering — equivalent to multiplying
+# a higher-is-better score by a positive constant per the spec wording.
+SIGNIFICANCE_RANK_BIAS = 0.5
+
 
 def search_memories(
     conn: Connection,
@@ -137,10 +145,15 @@ def search_memories(
         "JOIN memories m ON m.id = memories_fts.rowid "
         f"WHERE m.owner_id = ? AND m.{witness_col} = 1 "
         "AND memories_fts MATCH ? "
-        "ORDER BY memories_fts.rank "
+        # T57: significance multiplier biases the FTS over-fetch order. BM25
+        # ``rank`` is lower-is-better, so subtracting ``significance * BIAS``
+        # surfaces higher-significance rows above lower-significance rows with
+        # equal/near-equal match strength. Equivalent to ``score × constant``
+        # per §11.1 once the rank is inverted to a higher-is-better score.
+        "ORDER BY (memories_fts.rank - m.significance * ?) ASC "
         "LIMIT ?"
     )
-    cur = conn.execute(sql, (owner_id, query, over_fetch))
+    cur = conn.execute(sql, (owner_id, query, SIGNIFICANCE_RANK_BIAS, over_fetch))
     rows = cur.fetchall()
     if not rows:
         return []

chat/state/threads.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+from sqlite3 import Connection
+
+from chat.eventlog.projector import on
+from chat.eventlog.log import Event
+
+
+@on("thread_opened")
+def _apply_thread_opened(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "INSERT OR IGNORE INTO threads "
+        "(thread_id, chat_id, title, summary, status) "
+        "VALUES (?, ?, ?, ?, 'open')",
+        (
+            p["thread_id"],
+            p["chat_id"],
+            p["title"],
+            p.get("summary", ""),
+        ),
+    )
+
+
+@on("thread_updated")
+def _apply_thread_updated(conn: Connection, e: Event) -> None:
+    p = e.payload
+    # Idempotent: closed threads ignore subsequent updates.
+    conn.execute(
+        "UPDATE threads SET summary = ?, last_referenced_scene_id = ?, "
+        "updated_at = datetime('now') "
+        "WHERE thread_id = ? AND status = 'open'",
+        (
+            p.get("summary", ""),
+            p.get("last_referenced_scene_id"),
+            p["thread_id"],
+        ),
+    )
+
+
+@on("thread_closed")
+def _apply_thread_closed(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE threads SET status = 'closed', closed_at = ?, "
+        "updated_at = datetime('now') "
+        "WHERE thread_id = ? AND status = 'open'",
+        (p.get("closed_at"), p["thread_id"]),
+    )
+
+
+def get_thread(conn: Connection, thread_id: str) -> dict | None:
+    row = conn.execute(
+        "SELECT thread_id, chat_id, title, summary, status, "
+        "opened_at, closed_at, last_referenced_scene_id, "
+        "created_at, updated_at "
+        "FROM threads WHERE thread_id = ?",
+        (thread_id,),
+    ).fetchone()
+    if not row:
+        return None
+    return {
+        "thread_id": row[0],
+        "chat_id": row[1],
+        "title": row[2],
+        "summary": row[3],
+        "status": row[4],
+        "opened_at": row[5],
+        "closed_at": row[6],
+        "last_referenced_scene_id": row[7],
+        "created_at": row[8],
+        "updated_at": row[9],
+    }
+
+
+def list_open_threads(conn: Connection, chat_id: str) -> list[dict]:
+    rows = conn.execute(
+        "SELECT thread_id, chat_id, title, summary, status, "
+        "opened_at, closed_at, last_referenced_scene_id, "
+        "created_at, updated_at "
+        "FROM threads WHERE chat_id = ? AND status = 'open' "
+        "ORDER BY id ASC",
+        (chat_id,),
+    ).fetchall()
+    return [
+        {
+            "thread_id": r[0], "chat_id": r[1], "title": r[2],
+            "summary": r[3], "status": r[4],
+            "opened_at": r[5], "closed_at": r[6],
+            "last_referenced_scene_id": r[7],
+            "created_at": r[8], "updated_at": r[9],
+        }
+        for r in rows
+    ]
+
+
+def list_threads(conn: Connection, chat_id: str, status: str | None = None) -> list[dict]:
+    if status is None:
+        rows = conn.execute(
+            "SELECT thread_id, chat_id, title, summary, status, "
+            "opened_at, closed_at, last_referenced_scene_id, "
+            "created_at, updated_at "
+            "FROM threads WHERE chat_id = ? ORDER BY id ASC",
+            (chat_id,),
+        ).fetchall()
+    else:
+        rows = conn.execute(
+            "SELECT thread_id, chat_id, title, summary, status, "
+            "opened_at, closed_at, last_referenced_scene_id, "
+            "created_at, updated_at "
+            "FROM threads WHERE chat_id = ? AND status = ? "
+            "ORDER BY id ASC",
+            (chat_id, status),
+        ).fetchall()
+    return [
+        {
+            "thread_id": r[0], "chat_id": r[1], "title": r[2],
+            "summary": r[3], "status": r[4],
+            "opened_at": r[5], "closed_at": r[6],
+            "last_referenced_scene_id": r[7],
+            "created_at": r[8], "updated_at": r[9],
+        }
+        for r in rows
+    ]

chat/state/world.py

@@ -29,6 +29,34 @@ def _apply_chat_created(conn: Connection, e: Event) -> None:
     )
 
 
+@on("time_skip_elision")
+def _apply_time_skip_elision(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE chat_state SET time = ? WHERE chat_id = ?",
+        (p["new_time"], p["chat_id"]),
+    )
+
+
+@on("time_skip_jump")
+def _apply_time_skip_jump(conn: Connection, e: Event) -> None:
+    p = e.payload
+    chat_id = p["chat_id"]
+    conn.execute(
+        "UPDATE chat_state SET time = ? WHERE chat_id = ?",
+        (p["new_time"], chat_id),
+    )
+    if p.get("reset_activity", False):
+        # Activity rows are keyed by entity_id with a container_id FK.
+        # Each chat owns its containers, so deleting activity rows whose
+        # container_id belongs to this chat clears every present entity.
+        conn.execute(
+            "DELETE FROM activity "
+            "WHERE container_id IN (SELECT id FROM containers WHERE chat_id = ?)",
+            (chat_id,),
+        )
+
+
 @on("guest_added")
 def _apply_guest_added(conn: Connection, e: Event) -> None:
     p = e.payload

tests/test_event_lifecycle.py

@@ -0,0 +1,103 @@
+"""Tests for the event-lifecycle detection service (T52).
+
+Per Phase 3, after each narrated turn we ask a classifier whether any
+active events transitioned (started, completed, cancelled). The bias is
+strongly toward an empty result — most turns do NOT resolve or start a
+known event, and the turn-flow caller (T61) only appends an
+event_started/completed/cancelled record when this service yields one.
+
+These tests cover:
+
+* The classifier returning a single transition is honored end-to-end.
+* An empty ``active_events`` list short-circuits before any LLM call,
+  so callers that hold no live events pay zero classifier cost.
+* Three rounds of malformed JSON exhaust ``classify``'s retries and we
+  fall back to the empty default — graceful degradation per §3.3.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.event_lifecycle import (
+    EventLifecycleDecision,
+    detect_event_transitions,
+)
+
+
+@pytest.mark.asyncio
+async def test_detects_one_transition_happy_path():
+    canned = json.dumps(
+        {
+            "transitions": [
+                {
+                    "event_id": "evt_1",
+                    "new_status": "completed",
+                    "reason": "they arrived at the park",
+                }
+            ]
+        }
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await detect_event_transitions(
+        mock,
+        classifier_model="x",
+        narrative_text="They walked through the park gate, finally there.",
+        active_events=[
+            {
+                "event_id": "evt_1",
+                "kind": "date_at_park",
+                "status": "active",
+                "props": {},
+            }
+        ],
+    )
+    assert isinstance(result, EventLifecycleDecision)
+    assert len(result.transitions) == 1
+    assert result.transitions[0].event_id == "evt_1"
+    assert result.transitions[0].new_status == "completed"
+    assert result.transitions[0].reason == "they arrived at the park"
+
+
+@pytest.mark.asyncio
+async def test_empty_active_events_short_circuits_without_classifier_call():
+    """No active events -> no classifier call.
+
+    The mock has an empty canned list; any ``generate`` call would raise
+    ``IndexError`` from ``list.pop(0)``. The test passing proves the
+    short-circuit holds.
+    """
+    mock = MockLLMClient(canned=[])
+    result = await detect_event_transitions(
+        mock,
+        classifier_model="x",
+        narrative_text="Just a quiet moment between them.",
+        active_events=[],
+    )
+    assert isinstance(result, EventLifecycleDecision)
+    assert result.transitions == []
+
+
+@pytest.mark.asyncio
+async def test_classifier_failure_returns_empty_default():
+    """``classify`` retries 3 times; after all fail it returns the empty
+    default so the turn flow keeps moving (§3.3 graceful degradation)."""
+    mock = MockLLMClient(canned=["bad", "bad", "bad"])
+    result = await detect_event_transitions(
+        mock,
+        classifier_model="x",
+        narrative_text="Some text the classifier will choke on.",
+        active_events=[
+            {
+                "event_id": "evt_1",
+                "kind": "date_at_park",
+                "status": "active",
+                "props": {},
+            }
+        ],
+    )
+    assert isinstance(result, EventLifecycleDecision)
+    assert result.transitions == []

tests/test_event_promotion.py

@@ -0,0 +1,256 @@
+"""Tests for the event-completion promotion service (T56).
+
+When an event reaches ``status='completed'``, the orchestrator promotes
+structured artifacts the event carried (``acquired_objects``,
+``knowledge_facts``, ``relationship_change``) into the appropriate
+state stores via downstream events. Cancelled / expired events do NOT
+promote — the closed event row is left in place but no follow-on
+events fire.
+"""
+
+from __future__ import annotations
+
+import json
+
+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
+from chat.services.event_promotion import promote_completed_event
+from chat.state.edges import get_edge
+import chat.state.edges  # noqa: F401  - register edge_update handler
+import chat.state.entities  # noqa: F401  - register handlers
+import chat.state.events  # noqa: F401  - register events handlers
+import chat.state.manual_edit  # noqa: F401  - register manual_edit handler
+import chat.state.world  # noqa: F401  - register handlers
+
+
+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 _seed_chat(conn) -> None:
+    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())
+
+
+def _seed_event(
+    conn,
+    *,
+    event_id: str,
+    props: dict,
+    terminal_kind: str = "event_completed",
+) -> None:
+    """Append event_planned, then a terminal transition (default completed)."""
+    append_event(
+        conn,
+        kind="event_planned",
+        payload={
+            "event_id": event_id,
+            "chat_id": "chat_bot_a",
+            "kind": "story_event",
+            "props": props,
+            "planned_for": "2026-04-30T18:00:00+00:00",
+        },
+    )
+    append_event(
+        conn,
+        kind=terminal_kind,
+        payload={
+            "event_id": event_id,
+            "completed_at": "2026-04-30T20:00:00+00:00",
+        },
+    )
+    project(conn)
+
+
+def _max_event_id(conn) -> int:
+    return conn.execute("SELECT COALESCE(MAX(id), 0) FROM event_log").fetchone()[0]
+
+
+def _events_after(conn, after_id: int, kind: str) -> list[dict]:
+    rows = conn.execute(
+        "SELECT id, kind, payload_json FROM event_log "
+        "WHERE id > ? AND kind = ? ORDER BY id ASC",
+        (after_id, kind),
+    ).fetchall()
+    return [
+        {"id": r[0], "kind": r[1], "payload": json.loads(r[2])} for r in rows
+    ]
+
+
+def test_empty_props_no_op(tmp_path):
+    """Completed event with empty props produces no promotion events."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        _seed_event(conn, event_id="evt_empty", props={})
+
+        before = _max_event_id(conn)
+        counts = promote_completed_event(
+            conn,
+            event_id="evt_empty",
+            chat_id="chat_bot_a",
+            chat_clock_at="2026-04-30T20:00:00+00:00",
+        )
+
+        assert counts == {
+            "acquired_objects": 0,
+            "knowledge_facts": 0,
+            "relationship_change": 0,
+        }
+        # No new edge_update or manual_edit rows after the promote call.
+        assert _events_after(conn, before, "edge_update") == []
+        assert _events_after(conn, before, "manual_edit") == []
+
+
+def test_knowledge_facts_emits_edge_update(tmp_path):
+    """A knowledge_facts entry promotes to an edge_update on the directed edge."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        _seed_event(
+            conn,
+            event_id="evt_kf",
+            props={
+                "knowledge_facts": [
+                    {
+                        "owner_id": "bot_a",
+                        "target_id": "you",
+                        "fact": "Maya prefers tea over coffee",
+                    }
+                ]
+            },
+        )
+
+        before = _max_event_id(conn)
+        counts = promote_completed_event(
+            conn,
+            event_id="evt_kf",
+            chat_id="chat_bot_a",
+            chat_clock_at="2026-04-30T20:00:00+00:00",
+        )
+
+        assert counts["knowledge_facts"] == 1
+        assert counts["acquired_objects"] == 0
+        assert counts["relationship_change"] == 0
+
+        # An edge_update event landed in the event_log AFTER the promote call.
+        new_edge_updates = _events_after(conn, before, "edge_update")
+        assert len(new_edge_updates) == 1
+        payload = new_edge_updates[0]["payload"]
+        assert payload["source_id"] == "bot_a"
+        assert payload["target_id"] == "you"
+        assert payload["knowledge_facts"] == ["Maya prefers tea over coffee"]
+
+        # And the projected edge has the fact applied.
+        edge = get_edge(conn, "bot_a", "you")
+        assert edge is not None
+        assert "Maya prefers tea over coffee" in edge["knowledge"]
+
+
+def test_relationship_change_emits_manual_edit(tmp_path):
+    """A relationship_change promotes to a manual_edit edge_summary."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        _seed_event(
+            conn,
+            event_id="evt_rc",
+            props={
+                "relationship_change": {
+                    "source_id": "bot_a",
+                    "target_id": "you",
+                    "summary": "they're now dating",
+                }
+            },
+        )
+
+        before = _max_event_id(conn)
+        counts = promote_completed_event(
+            conn,
+            event_id="evt_rc",
+            chat_id="chat_bot_a",
+            chat_clock_at="2026-04-30T20:00:00+00:00",
+        )
+
+        assert counts["relationship_change"] == 1
+        assert counts["knowledge_facts"] == 0
+        assert counts["acquired_objects"] == 0
+
+        new_manual_edits = _events_after(conn, before, "manual_edit")
+        # Filter to edge_summary only — Phase 3 stub may also emit
+        # memory_pov_summary entries for acquired_objects, but here there
+        # are none.
+        edge_summary_edits = [
+            m for m in new_manual_edits
+            if m["payload"].get("target_kind") == "edge_summary"
+        ]
+        assert len(edge_summary_edits) == 1
+        payload = edge_summary_edits[0]["payload"]
+        assert payload["target_kind"] == "edge_summary"
+        assert payload["target_id"] == {"source_id": "bot_a", "target_id": "you"}
+        assert payload["new_value"] == "they're now dating"
+
+
+def test_cancelled_event_does_not_promote(tmp_path):
+    """Cancelled events have promotable props ignored — no follow-on events."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        _seed_event(
+            conn,
+            event_id="evt_canx",
+            props={
+                "knowledge_facts": [
+                    {"owner_id": "bot_a", "target_id": "you", "fact": "x"}
+                ],
+                "relationship_change": {
+                    "source_id": "bot_a",
+                    "target_id": "you",
+                    "summary": "ignored",
+                },
+            },
+            terminal_kind="event_cancelled",
+        )
+
+        before = _max_event_id(conn)
+        counts = promote_completed_event(
+            conn,
+            event_id="evt_canx",
+            chat_id="chat_bot_a",
+            chat_clock_at="2026-04-30T20:00:00+00:00",
+        )
+
+        assert counts == {
+            "acquired_objects": 0,
+            "knowledge_facts": 0,
+            "relationship_change": 0,
+        }
+        assert _events_after(conn, before, "edge_update") == []
+        assert _events_after(conn, before, "manual_edit") == []

tests/test_events_state.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+from chat.db.connection import open_db
+from chat.db.migrate import apply_migrations
+from chat.eventlog.log import append_and_apply, append_event
+from chat.eventlog.projector import project
+import chat.state.entities  # registers handlers
+import chat.state.world  # registers handlers
+import chat.state.group_node  # registers handlers
+import chat.state.events  # registers handlers
+from chat.state.events import (
+    get_event,
+    list_active_events,
+    list_events_in_status,
+)
+
+
+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 _seed_chat(conn) -> None:
+    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())
+
+
+def test_event_planned_creates_row(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        append_event(
+            conn,
+            kind="event_planned",
+            payload={
+                "event_id": "evt_abc",
+                "chat_id": "chat_bot_a",
+                "kind": "date_at_park",
+                "props": {"location": "park"},
+                "planned_for": "2026-04-30T18:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        ev = get_event(conn, "evt_abc")
+        assert ev is not None
+        assert ev["event_id"] == "evt_abc"
+        assert ev["chat_id"] == "chat_bot_a"
+        assert ev["kind"] == "date_at_park"
+        assert ev["status"] == "planned"
+        assert ev["props"]["location"] == "park"
+        assert ev["planned_for"] == "2026-04-30T18:00:00+00:00"
+        assert ev["started_at"] is None
+        assert ev["completed_at"] is None
+
+
+def test_event_started_then_completed_updates_status(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        append_event(
+            conn,
+            kind="event_planned",
+            payload={
+                "event_id": "evt_abc",
+                "chat_id": "chat_bot_a",
+                "kind": "date_at_park",
+                "props": {},
+                "planned_for": "2026-04-30T18:00:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_started",
+            payload={
+                "event_id": "evt_abc",
+                "started_at": "2026-04-30T18:01:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_completed",
+            payload={
+                "event_id": "evt_abc",
+                "completed_at": "2026-04-30T20:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        ev = get_event(conn, "evt_abc")
+        assert ev is not None
+        assert ev["status"] == "completed"
+        assert ev["started_at"] == "2026-04-30T18:01:00+00:00"
+        assert ev["completed_at"] == "2026-04-30T20:00:00+00:00"
+
+
+def test_event_cancelled_terminal_subsequent_transitions_ignored(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        append_event(
+            conn,
+            kind="event_planned",
+            payload={
+                "event_id": "evt_abc",
+                "chat_id": "chat_bot_a",
+                "kind": "date_at_park",
+                "props": {},
+                "planned_for": "2026-04-30T18:00:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_cancelled",
+            payload={
+                "event_id": "evt_abc",
+                "completed_at": "2026-04-30T17:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        ev = get_event(conn, "evt_abc")
+        assert ev is not None
+        assert ev["status"] == "cancelled"
+        assert ev["completed_at"] == "2026-04-30T17:00:00+00:00"
+
+        # Subsequent event_started must be no-oped because status is terminal.
+        # Use append_and_apply so we apply ONLY this new event without
+        # replaying earlier non-idempotent handlers (e.g. chat_created).
+        append_and_apply(
+            conn,
+            kind="event_started",
+            payload={
+                "event_id": "evt_abc",
+                "started_at": "2026-04-30T18:01:00+00:00",
+            },
+        )
+
+        ev2 = get_event(conn, "evt_abc")
+        assert ev2 is not None
+        assert ev2["status"] == "cancelled"
+        assert ev2["started_at"] is None
+        # completed_at unchanged from the cancelled transition
+        assert ev2["completed_at"] == "2026-04-30T17:00:00+00:00"
+
+
+def test_list_active_events_filters_to_planned_and_active(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_chat(conn)
+        # Four events: one planned, one active, one completed, one cancelled.
+        for ev_id, kind in [
+            ("evt_planned", "date_at_park"),
+            ("evt_active", "movie_night"),
+            ("evt_done", "dinner"),
+            ("evt_canx", "trip"),
+        ]:
+            append_event(
+                conn,
+                kind="event_planned",
+                payload={
+                    "event_id": ev_id,
+                    "chat_id": "chat_bot_a",
+                    "kind": kind,
+                    "props": {},
+                    "planned_for": "2026-04-30T18:00:00+00:00",
+                },
+            )
+        append_event(
+            conn,
+            kind="event_started",
+            payload={
+                "event_id": "evt_active",
+                "started_at": "2026-04-30T18:01:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_started",
+            payload={
+                "event_id": "evt_done",
+                "started_at": "2026-04-30T18:01:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_completed",
+            payload={
+                "event_id": "evt_done",
+                "completed_at": "2026-04-30T20:00:00+00:00",
+            },
+        )
+        append_event(
+            conn,
+            kind="event_cancelled",
+            payload={
+                "event_id": "evt_canx",
+                "completed_at": "2026-04-30T17:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        active = list_active_events(conn, "chat_bot_a")
+        active_ids = {e["event_id"] for e in active}
+        assert active_ids == {"evt_planned", "evt_active"}
+
+        completed = list_events_in_status(conn, "chat_bot_a", "completed")
+        assert [e["event_id"] for e in completed] == ["evt_done"]
+
+        cancelled = list_events_in_status(conn, "chat_bot_a", "cancelled")
+        assert [e["event_id"] for e in cancelled] == ["evt_canx"]

tests/test_memory_search.py

@@ -125,3 +125,37 @@ def test_search_invalid_witness_role_raises(tmp_path):
     with open_db(db) as conn:
         with pytest.raises(ValueError):
             search_memories(conn, "bot_a", "invalid_role", "anything", k=4)
+
+
+def test_higher_significance_outranks_equal_rank(tmp_path):
+    """T57: significance multiplier biases the SQL ORDER BY.
+
+    Two memories with IDENTICAL FTS-matching text yield (effectively) equal
+    BM25 ranks. The significance bias applied in the SQL ORDER BY must
+    surface the higher-significance row first.
+    """
+    db = tmp_path / "t.db"
+    _seed(
+        db,
+        memory_specs=[
+            # Identical pov_summary text -> FTS BM25 rank is the same for both.
+            {"pov_summary": "she swore an oath", "significance": 0},
+            {"pov_summary": "she swore an oath", "significance": 3},
+        ],
+    )
+    with open_db(db) as conn:
+        out = search_memories(conn, "bot_a", "host", "oath", k=5)
+        assert len(out) == 2
+        # Higher significance wins despite tied FTS rank.
+        assert out[0]["significance"] == 3
+        assert out[1]["significance"] == 0
+
+
+def test_significance_bias_is_constant_module_level():
+    """T57: pin ``SIGNIFICANCE_RANK_BIAS`` as a tunable module-level numeric."""
+    from chat.state.memory import SIGNIFICANCE_RANK_BIAS
+
+    assert isinstance(SIGNIFICANCE_RANK_BIAS, (int, float))
+    # Must be non-negative -- a negative bias would invert the desired
+    # "higher significance ranks higher" semantics.
+    assert SIGNIFICANCE_RANK_BIAS >= 0

tests/test_per_pov_summary.py

@@ -504,13 +504,15 @@ async def test_close_with_no_guest_matches_phase1(tmp_path):
             "relationship_summary": "BotA leaned in supportively.",
         }
     )
+    no_threads = json.dumps({"candidates": []})
     with open_db(db) as conn:
         _seed_single_bot_scene(conn)
         project(conn)
 
-        # canned has 2 entries to detect any over-call; the assertion below
+        # 1 host-POV entry + 1 thread-detection entry (T58.2) + 1 spare
-        # confirms only one was consumed.
+        # to detect any over-call. Assertion below confirms exactly two
-        client = MockLLMClient(canned=[canned, canned])
+        # were consumed.
+        client = MockLLMClient(canned=[canned, no_threads, canned])
         await apply_scene_close_summary(
             conn,
             client,
@@ -520,8 +522,8 @@ async def test_close_with_no_guest_matches_phase1(tmp_path):
             host_bot_id="bot_a",
         )
 
-        # Exactly one classifier call -> exactly one canned entry consumed,
+        # Host POV + thread detection -> exactly two canned entries
-        # leaving the second untouched.
+        # consumed, leaving the spare untouched.
         assert len(client._canned) == 1
 
         # Host memory rewritten with the per-POV summary content.
@@ -845,3 +847,251 @@ async def test_group_summary_skipped_when_no_guest(tmp_path):
             "SELECT 1 FROM event_log WHERE kind = 'group_node_updated'"
         ).fetchall()
         assert rows == []
+
+
+# ---------------------------------------------------------------------------
+# T58: significance-driven quote retention + thread detection on close.
+# ---------------------------------------------------------------------------
+
+
+def _seed_single_bot_scene_no_memory(conn) -> None:
+    """Like ``_seed_single_bot_scene`` but skips the memory_written event so
+    callers can seed memories with custom significance / text themselves."""
+    append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
+    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",
+            "initial_time": "2026-04-26T20:00:00+00:00",
+            "narrative_anchor": "Day 1",
+            "weather": "",
+        },
+    )
+    append_event(
+        conn,
+        kind="container_created",
+        payload={
+            "chat_id": "chat_bot_a",
+            "name": "office",
+            "type": "workplace",
+            "properties": {},
+        },
+    )
+    append_event(
+        conn,
+        kind="scene_opened",
+        payload={
+            "chat_id": "chat_bot_a",
+            "container_id": 1,
+            "started_at": "2026-04-26T20:00:00+00:00",
+            "participants": ["you", "bot_a"],
+        },
+    )
+    append_event(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "bot_a",
+            "target_id": "you",
+            "chat_id": "chat_bot_a",
+        },
+    )
+    append_event(
+        conn,
+        kind="user_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "prose": "Quick chat about the deadline",
+            "segments": [],
+        },
+    )
+    append_event(
+        conn,
+        kind="assistant_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "speaker_id": "bot_a",
+            "text": "It's going to be okay.",
+            "truncated": False,
+            "user_turn_id": 1,
+        },
+    )
+
+
+def _seed_memory(conn, *, pov_summary: str, significance: int) -> None:
+    append_event(
+        conn,
+        kind="memory_written",
+        payload={
+            "owner_id": "bot_a",
+            "chat_id": "chat_bot_a",
+            "scene_id": 1,
+            "pov_summary": pov_summary,
+            "witness_you": 1,
+            "witness_host": 1,
+            "witness_guest": 0,
+            "significance": significance,
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_low_significance_scene_omits_quotes(tmp_path):
+    """When the scene's max-turn-significance is < 2, the per-POV summary
+    rewrite collapses fully — no "Key quotes:" suffix is appended."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    canned = json.dumps(
+        {
+            "summary": "BotA had a low-key chat with you.",
+            "knowledge_facts": [],
+            "relationship_summary": "Nothing major shifted.",
+        }
+    )
+    no_threads = json.dumps({"candidates": []})
+    with open_db(db) as conn:
+        _seed_single_bot_scene_no_memory(conn)
+        _seed_memory(conn, pov_summary="Maya rambled about coffee", significance=1)
+        _seed_memory(conn, pov_summary="Maya glanced at the clock", significance=0)
+        project(conn)
+
+        client = MockLLMClient(canned=[canned, no_threads])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        rows = conn.execute(
+            "SELECT pov_summary FROM memories WHERE scene_id = 1"
+        ).fetchall()
+        assert rows
+        for (pov,) in rows:
+            assert "Key quotes:" not in pov
+            assert "BotA had a low-key chat" in pov
+
+
+@pytest.mark.asyncio
+async def test_high_significance_scene_includes_top_3_quotes(tmp_path):
+    """When max-turn-significance is >= 2, each per-POV summary text gains
+    a "Key quotes:" suffix listing the top-3 highest-significance memory
+    rows verbatim, ordered by (significance DESC, id ASC)."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    canned = json.dumps(
+        {
+            "summary": "BotA had a heavy talk with you.",
+            "knowledge_facts": [],
+            "relationship_summary": "Things shifted.",
+        }
+    )
+    no_threads = json.dumps({"candidates": []})
+    with open_db(db) as conn:
+        _seed_single_bot_scene_no_memory(conn)
+        # Insertion order matches id ASC. Top-3 by (sig DESC, id ASC):
+        # quote 1 (sig 3) -> quote 2 (sig 2, lower id) -> quote 4 (sig 2,
+        # higher id). quote 3 (sig 1) is dropped.
+        _seed_memory(conn, pov_summary="Maya quote one", significance=3)
+        _seed_memory(conn, pov_summary="Maya quote two", significance=2)
+        _seed_memory(conn, pov_summary="Maya quote three", significance=1)
+        _seed_memory(conn, pov_summary="Maya quote four", significance=2)
+        project(conn)
+
+        client = MockLLMClient(canned=[canned, no_threads])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        rows = conn.execute(
+            "SELECT pov_summary FROM memories WHERE scene_id = 1"
+        ).fetchall()
+        assert rows
+        for (pov,) in rows:
+            assert "Key quotes:" in pov
+            assert '"Maya quote one"' in pov
+            assert '"Maya quote two"' in pov
+            assert '"Maya quote four"' in pov
+            # The sig-1 quote falls outside the top-3 cap.
+            assert '"Maya quote three"' not in pov
+            # Ordering: sig 3 first, then the two sig-2s by id ASC.
+            i_one = pov.index('"Maya quote one"')
+            i_two = pov.index('"Maya quote two"')
+            i_four = pov.index('"Maya quote four"')
+            assert i_one < i_two < i_four
+
+
+@pytest.mark.asyncio
+async def test_thread_detection_emits_events(tmp_path, monkeypatch):
+    """On scene close, ``detect_threads`` is invoked and each "open"
+    candidate yields a ``thread_opened`` event with a fresh thread_id."""
+    from chat.services import thread_detection as td_mod
+
+    canned = json.dumps(
+        {
+            "summary": "BotA noticed something unresolved.",
+            "knowledge_facts": [],
+            "relationship_summary": "Tension lingered.",
+        }
+    )
+
+    async def fake_detect_threads(client, **kwargs):
+        return td_mod.ThreadDetectionResult(
+            candidates=[
+                td_mod.ThreadCandidate(
+                    action="open",
+                    title="Test thread",
+                    summary="A test",
+                    existing_thread_id=None,
+                ),
+            ]
+        )
+
+    monkeypatch.setattr(td_mod, "detect_threads", fake_detect_threads)
+
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_single_bot_scene(conn)
+        project(conn)
+
+        client = MockLLMClient(canned=[canned])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log WHERE kind = 'thread_opened'"
+        ).fetchall()
+        assert len(rows) == 1
+        payload = json.loads(rows[0][0])
+        assert payload["title"] == "Test thread"
+        assert payload["summary"] == "A test"
+        assert payload["chat_id"] == "chat_bot_a"
+        assert payload["thread_id"].startswith("thr_")
+
+        # The threads-table projection ran via append_and_apply.
+        from chat.state.threads import list_open_threads
+
+        open_threads = list_open_threads(conn, "chat_bot_a")
+        assert len(open_threads) == 1
+        assert open_threads[0]["title"] == "Test thread"

tests/test_skip_narration.py

@@ -0,0 +1,117 @@
+"""Skip narration service tests (T53).
+
+The skip-narration service generates short transition prose between an
+in-progress moment and a post-skip moment. Two flavors:
+
+* ``elision`` — collapses an in-progress activity to its expected
+  end-state in 1-2 sentences (e.g. "skip ahead to when we arrive").
+* ``jump`` — bridges a longer fiction-time delta in 2-3 sentences
+  (e.g. "next morning", "a week later").
+
+Output is free-form prose, not structured JSON, so the service goes
+through ``client.generate`` directly rather than the classifier path.
+A deterministic template fallback fires on any LLM failure so the skip
+flow never blocks even when the model is down.
+"""
+
+from __future__ import annotations
+
+from typing import AsyncIterator, Sequence
+
+import pytest
+
+from chat.llm.client import Message
+from chat.llm.mock import MockLLMClient
+from chat.services.skip_narration import narrate_skip
+
+
+_SPEAKER = {
+    "id": "bot1",
+    "name": "Aria",
+    "persona": "thoughtful, observant",
+}
+
+
+@pytest.mark.asyncio
+async def test_narrate_elision_returns_classifier_output():
+    canned = (
+        "She closes her laptop and slings her bag over her shoulder. "
+        "The office shrinks behind her as she steps into the late "
+        "afternoon light."
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await narrate_skip(
+        mock,
+        narrative_model="x",
+        skip_kind="elision",
+        speaker_bot=_SPEAKER,
+        you_name="Me",
+        current_time="3:42 PM",
+        new_time="5:10 PM",
+        current_activity="finishing up at her desk",
+        landing_state_hint="walking out into the parking lot",
+    )
+    assert "office" in result or result == canned
+
+
+@pytest.mark.asyncio
+async def test_narrate_jump_returns_classifier_output():
+    canned = (
+        "Morning light spills through the kitchen window. The coffee "
+        "maker hums. She's already at the table, scrolling her phone."
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await narrate_skip(
+        mock,
+        narrative_model="x",
+        skip_kind="jump",
+        speaker_bot=_SPEAKER,
+        you_name="Me",
+        current_time="late evening",
+        new_time="next morning",
+        current_activity="winding down for the night",
+        landing_state_hint="having coffee in the kitchen",
+    )
+    assert result
+    lower = result.lower()
+    assert "morning" in lower or "coffee" in lower
+
+
+class _RaisingMock:
+    """Mock LLMClient whose ``generate`` always raises.
+
+    ``MockLLMClient.generate`` raises ``IndexError`` once the canned
+    list is empty, but the test wants a clear, unambiguous failure
+    regardless of canned-list state, so we ship a tiny dedicated mock
+    instead.
+    """
+
+    async def generate(
+        self, messages: Sequence[Message], *, model: str, **params
+    ) -> str:
+        raise RuntimeError("LLM is down")
+
+    async def stream(
+        self, messages: Sequence[Message], *, model: str, **params
+    ) -> AsyncIterator[str]:
+        raise RuntimeError("LLM is down")
+        yield  # pragma: no cover - make this a generator
+
+
+@pytest.mark.asyncio
+async def test_narrate_falls_back_on_generation_failure():
+    new_time = "next morning"
+    result = await narrate_skip(
+        _RaisingMock(),
+        narrative_model="x",
+        skip_kind="jump",
+        speaker_bot=_SPEAKER,
+        you_name="Me",
+        current_time="late evening",
+        new_time=new_time,
+        current_activity="winding down for the night",
+        landing_state_hint="having coffee in the kitchen",
+    )
+    # Fallback template includes the new_time so callers can see *what*
+    # we skipped to even when the LLM never answered.
+    assert new_time in result

tests/test_synthesized_memories.py

@@ -0,0 +1,98 @@
+"""Tests for the synthesized-memories service (T54).
+
+When the user jump-skips ("a week later") they are prompted "anything
+notable happen?" If they answer with prose, this service parses it into
+1-N synthesized memories per present bot. Each memory carries
+``source="synthesized"`` and ``reliability=0.7`` (the caller — T62 skip
+flow — applies those tags when persisting; this service just produces
+the structured digest).
+
+These tests cover:
+
+* The happy path: a canned classifier response parses cleanly into a
+  populated :class:`SynthesizedDigest` with one memory.
+* Empty prose short-circuits before any classifier call — the mock has
+  no canned responses, so an accidental call would raise
+  ``IndexError``.
+* Classifier failure (3 bad responses, exhausting :func:`classify`'s
+  retry budget) falls back to an empty default digest.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.synthesized_memories import (
+    SynthesizedDigest,
+    SynthesizedMemory,
+    synthesize_memories,
+)
+
+
+@pytest.mark.asyncio
+async def test_synthesize_parses_canned_prose():
+    canned = json.dumps(
+        {
+            "memories": [
+                {
+                    "text": "Maya started a new pottery class.",
+                    "significance": 1,
+                    "affinity_delta": 0,
+                    "trust_delta": 0,
+                }
+            ]
+        }
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await synthesize_memories(
+        mock,
+        classifier_model="x",
+        prose="we saw each other at her pottery class once",
+        bot_name="Maya",
+        bot_persona="warm potter, mid-30s",
+        you_name="Sam",
+    )
+    assert isinstance(result, SynthesizedDigest)
+    assert len(result.memories) == 1
+    mem = result.memories[0]
+    assert isinstance(mem, SynthesizedMemory)
+    assert mem.text == "Maya started a new pottery class."
+    assert mem.significance == 1
+    assert mem.affinity_delta == 0
+    assert mem.trust_delta == 0
+
+
+@pytest.mark.asyncio
+async def test_empty_prose_returns_empty_digest():
+    """Empty prose short-circuits — the classifier must not be called."""
+    mock = MockLLMClient(canned=[])
+    result = await synthesize_memories(
+        mock,
+        classifier_model="x",
+        prose="",
+        bot_name="Maya",
+        bot_persona="warm potter, mid-30s",
+        you_name="Sam",
+    )
+    assert result == SynthesizedDigest()
+    assert result.memories == []
+
+
+@pytest.mark.asyncio
+async def test_classifier_failure_returns_empty_default():
+    """Three bad responses exhaust the classifier's retry budget; the
+    service then returns the empty default digest."""
+    mock = MockLLMClient(canned=["bad", "bad", "bad"])
+    result = await synthesize_memories(
+        mock,
+        classifier_model="x",
+        prose="we saw each other at her pottery class once",
+        bot_name="Maya",
+        bot_persona="warm potter, mid-30s",
+        you_name="Sam",
+    )
+    assert result == SynthesizedDigest()
+    assert result.memories == []

tests/test_thread_detection.py

@@ -0,0 +1,128 @@
+"""Tests for the thread-detection service (T55).
+
+On scene close, the transcript is classified to detect open threads
+(unresolved arcs, dangling questions, promises made). The service can
+also signal **update** to an existing thread when the scene developed
+it, or **close** when the scene resolved it.
+
+These tests cover:
+
+* A new thread the scene introduced — action="open" with a fresh title.
+* An update to an existing thread — action="update" with
+  ``existing_thread_id`` referencing the prior thread.
+* Classifier failure — three bad responses degrade to an empty
+  candidates list (graceful degradation, §3.3).
+* Empty transcript short-circuits before any classifier call.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.thread_detection import (
+    ThreadCandidate,
+    ThreadDetectionResult,
+    detect_threads,
+)
+
+
+@pytest.mark.asyncio
+async def test_detects_new_thread_open():
+    canned = json.dumps(
+        {
+            "candidates": [
+                {
+                    "action": "open",
+                    "title": "Maya's job hunt",
+                    "summary": "Maya is looking for a new job",
+                    "existing_thread_id": None,
+                }
+            ]
+        }
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await detect_threads(
+        mock,
+        classifier_model="x",
+        scene_transcript=[
+            {"speaker": "Maya", "text": "I need to find a new job soon."},
+            {"speaker": "Sam", "text": "What kind of role are you looking for?"},
+        ],
+        open_threads=[],
+    )
+    assert isinstance(result, ThreadDetectionResult)
+    assert len(result.candidates) == 1
+    cand = result.candidates[0]
+    assert isinstance(cand, ThreadCandidate)
+    assert cand.action == "open"
+    assert cand.title == "Maya's job hunt"
+    assert cand.summary == "Maya is looking for a new job"
+    assert cand.existing_thread_id is None
+
+
+@pytest.mark.asyncio
+async def test_detects_update_to_existing_thread():
+    canned = json.dumps(
+        {
+            "candidates": [
+                {
+                    "action": "update",
+                    "title": "",
+                    "summary": "Maya interviewed at Acme today",
+                    "existing_thread_id": "thr_jobhunt",
+                }
+            ]
+        }
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await detect_threads(
+        mock,
+        classifier_model="x",
+        scene_transcript=[
+            {"speaker": "Maya", "text": "I had the Acme interview today."},
+            {"speaker": "Sam", "text": "How did it go?"},
+        ],
+        open_threads=[
+            {
+                "thread_id": "thr_jobhunt",
+                "title": "Maya's job hunt",
+                "summary": "Maya is looking for a new job",
+            }
+        ],
+    )
+    assert len(result.candidates) == 1
+    cand = result.candidates[0]
+    assert cand.action == "update"
+    assert cand.existing_thread_id == "thr_jobhunt"
+    assert cand.summary == "Maya interviewed at Acme today"
+
+
+@pytest.mark.asyncio
+async def test_classifier_failure_returns_empty():
+    """Three malformed classifier responses → empty candidates list."""
+    mock = MockLLMClient(canned=["not json", "still not json", "{bad"])
+    result = await detect_threads(
+        mock,
+        classifier_model="x",
+        scene_transcript=[
+            {"speaker": "Maya", "text": "Anything could happen here."},
+        ],
+        open_threads=[],
+    )
+    assert result.candidates == []
+
+
+@pytest.mark.asyncio
+async def test_empty_transcript_short_circuits():
+    """Empty transcript short-circuits — classifier must not be called."""
+    mock = MockLLMClient(canned=[])
+    result = await detect_threads(
+        mock,
+        classifier_model="x",
+        scene_transcript=[],
+        open_threads=[],
+    )
+    assert result.candidates == []

tests/test_threads_state.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from chat.db.connection import open_db
+from chat.db.migrate import apply_migrations
+from chat.eventlog.log import append_and_apply, append_event
+from chat.eventlog.projector import project
+import chat.state.entities  # registers handlers
+import chat.state.world  # registers handlers
+import chat.state.threads  # registers handlers
+from chat.state.threads import get_thread, list_open_threads
+
+
+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_thread_opened_creates_row(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="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="thread_opened",
+            payload={
+                "thread_id": "thr_abc",
+                "chat_id": "chat_bot_a",
+                "title": "Maya's job hunt",
+                "summary": "Maya is looking for a new job",
+            },
+        )
+        project(conn)
+
+        t = get_thread(conn, "thr_abc")
+        assert t is not None
+        assert t["thread_id"] == "thr_abc"
+        assert t["chat_id"] == "chat_bot_a"
+        assert t["title"] == "Maya's job hunt"
+        assert t["summary"] == "Maya is looking for a new job"
+        assert t["status"] == "open"
+        assert t["closed_at"] is None
+        assert t["last_referenced_scene_id"] is None
+
+
+def test_thread_updated_changes_summary_and_last_referenced(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="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="thread_opened",
+            payload={
+                "thread_id": "thr_abc",
+                "chat_id": "chat_bot_a",
+                "title": "Maya's job hunt",
+                "summary": "Maya is looking for a new job",
+            },
+        )
+        append_event(
+            conn,
+            kind="thread_updated",
+            payload={
+                "thread_id": "thr_abc",
+                "summary": "Maya landed an interview at a startup",
+                "last_referenced_scene_id": 42,
+            },
+        )
+        project(conn)
+
+        t = get_thread(conn, "thr_abc")
+        assert t is not None
+        assert t["summary"] == "Maya landed an interview at a startup"
+        assert t["last_referenced_scene_id"] == 42
+        assert t["status"] == "open"
+
+
+def test_thread_closed_terminal(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="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="thread_opened",
+            payload={
+                "thread_id": "thr_abc",
+                "chat_id": "chat_bot_a",
+                "title": "Maya's job hunt",
+                "summary": "Maya is looking for a new job",
+            },
+        )
+        append_event(
+            conn,
+            kind="thread_closed",
+            payload={
+                "thread_id": "thr_abc",
+                "closed_at": "2026-04-26T21:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        t = get_thread(conn, "thr_abc")
+        assert t is not None
+        assert t["status"] == "closed"
+        assert t["closed_at"] == "2026-04-26T21:00:00+00:00"
+
+        # Subsequent updates to a closed thread are no-ops.
+        append_and_apply(
+            conn,
+            kind="thread_updated",
+            payload={
+                "thread_id": "thr_abc",
+                "summary": "should not be applied",
+            },
+        )
+
+        t2 = get_thread(conn, "thr_abc")
+        assert t2 is not None
+        assert t2["summary"] == "Maya is looking for a new job"
+        assert t2["status"] == "closed"
+
+
+def test_list_open_threads_filters_to_open_only(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="chat_created", payload=_chat_payload())
+        for tid, title in [
+            ("thr_1", "Arc 1"),
+            ("thr_2", "Arc 2"),
+            ("thr_3", "Arc 3"),
+        ]:
+            append_event(
+                conn,
+                kind="thread_opened",
+                payload={
+                    "thread_id": tid,
+                    "chat_id": "chat_bot_a",
+                    "title": title,
+                    "summary": "",
+                },
+            )
+        append_event(
+            conn,
+            kind="thread_closed",
+            payload={
+                "thread_id": "thr_3",
+                "closed_at": "2026-04-26T21:00:00+00:00",
+            },
+        )
+        project(conn)
+
+        open_threads = list_open_threads(conn, "chat_bot_a")
+        assert len(open_threads) == 2
+        ids = {t["thread_id"] for t in open_threads}
+        assert ids == {"thr_1", "thr_2"}

tests/test_time_skip_handlers.py

@@ -0,0 +1,132 @@
+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.world  # registers handlers
+from chat.state.world import get_activity, get_chat
+
+
+def _chat_payload(**overrides):
+    payload = {
+        "id": "chat_bot_a",
+        "host_bot_id": "bot_a",
+        "guest_bot_id": None,
+        "initial_time": "2026-04-26T20:00:00+00:00",
+        "narrative_anchor": "Day 1 evening",
+        "weather": "clear",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _container_payload(**overrides):
+    payload = {
+        "chat_id": "chat_bot_a",
+        "name": "office",
+        "type": "workplace",
+        "properties": {
+            "public": True,
+            "moving": False,
+            "audible_range": "normal",
+            "slots": [],
+        },
+        "parent_id": None,
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _activity_payload(**overrides):
+    payload = {
+        "entity_id": "bot_a",
+        "container_id": 1,
+        "slot": "desk_chair",
+        "posture": "sitting",
+        "action": {"verb": "writing email"},
+        "attention": "the screen",
+        "holding": ["pen"],
+        "status": {"hungry": False},
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _seed_events(conn):
+    """Append seed events but do NOT project — caller appends more then projects once."""
+    append_event(conn, kind="chat_created", payload=_chat_payload())
+    append_event(conn, kind="container_created", payload=_container_payload())
+    append_event(conn, kind="activity_change", payload=_activity_payload())
+
+
+def test_elision_advances_chat_clock_only(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_events(conn)
+        append_event(conn, kind="time_skip_elision", payload={
+            "chat_id": "chat_bot_a",
+            "new_time": "2026-04-26T20:30:00+00:00",
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat["time"] == "2026-04-26T20:30:00+00:00"
+
+        # Activity row preserved with the same fields it was seeded with.
+        a = get_activity(conn, "bot_a")
+        assert a is not None
+        assert a["entity_id"] == "bot_a"
+        assert a["container_id"] == 1
+        assert a["slot"] == "desk_chair"
+        assert a["posture"] == "sitting"
+        assert a["action"] == {"verb": "writing email"}
+        assert a["attention"] == "the screen"
+        assert a["holding"] == ["pen"]
+        assert a["status"] == {"hungry": False}
+
+
+def test_jump_with_reset_clears_activity(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_events(conn)
+        append_event(conn, kind="time_skip_jump", payload={
+            "chat_id": "chat_bot_a",
+            "new_time": "2026-04-27T08:00:00+00:00",
+            "reset_activity": True,
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat["time"] == "2026-04-27T08:00:00+00:00"
+
+        count = conn.execute(
+            "SELECT COUNT(*) FROM activity "
+            "WHERE container_id IN (SELECT id FROM containers WHERE chat_id = ?)",
+            ("chat_bot_a",),
+        ).fetchone()[0]
+        assert count == 0
+        assert get_activity(conn, "bot_a") is None
+
+
+def test_jump_without_reset_preserves_activity(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_events(conn)
+        append_event(conn, kind="time_skip_jump", payload={
+            "chat_id": "chat_bot_a",
+            "new_time": "2026-04-27T08:00:00+00:00",
+            "reset_activity": False,
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat["time"] == "2026-04-27T08:00:00+00:00"
+
+        a = get_activity(conn, "bot_a")
+        assert a is not None
+        assert a["posture"] == "sitting"
+        assert a["action"]["verb"] == "writing email"

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_8(tmp_path):
+def test_schema_version_after_migration_is_10(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]) == 8
+        assert int(row[0]) == 10