merge: T61 per-turn event-lifecycle detection + completion promotion

chat/services/prompt.py

@@ -37,8 +37,10 @@ import tiktoken
 from chat.llm.client import Message
 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.memory import search_memories
+from chat.state.threads import list_open_threads
 from chat.state.world import (
     active_scene,
     get_activity,
@@ -227,6 +229,76 @@ def _build_group_node_block(group_node: dict | None) -> str | None:
     return "\n".join(lines)
 
 
+def _props_excerpt(props: dict | None, limit: int = 80) -> str:
+    """Return a one-line excerpt of an event's ``props`` dict.
+
+    Renders ``key=value`` pairs separated by ", " (deterministic by dict
+    insertion order) and truncates to ~``limit`` characters with a
+    trailing ellipsis. Returns empty string for falsy/empty props so the
+    caller can omit the line entirely.
+    """
+    if not props:
+        return ""
+    pieces: list[str] = []
+    for k, v in props.items():
+        pieces.append(f"{k}={v}")
+    rendered = ", ".join(pieces)
+    if len(rendered) > limit:
+        # Reserve 1 char for the ellipsis so the total never exceeds limit.
+        rendered = rendered[: max(0, limit - 1)] + "…"
+    return rendered
+
+
+def _build_active_events_block(events: list[dict]) -> str | None:
+    """Render the ``Active events:`` block for Phase 3 Task 60.
+
+    One bullet per event. The sub-label depends on status:
+      - ``planned`` → ``(planned for {planned_for})``
+      - ``active``  → ``(active, started_at={started_at})``
+    A second indented line carries a one-line excerpt of the event's
+    ``props`` (truncated ~80 chars) when non-empty. Returns ``None`` when
+    there are no active events so the caller can omit the entire block.
+    """
+    if not events:
+        return None
+    lines = ["Active events:"]
+    for ev in events:
+        kind = ev.get("kind") or "?"
+        status = ev.get("status") or ""
+        if status == "active":
+            started = ev.get("started_at") or ""
+            lines.append(f"- {kind} (active, started_at={started})")
+        else:
+            planned = ev.get("planned_for") or ""
+            lines.append(f"- {kind} (planned for {planned})")
+        excerpt = _props_excerpt(ev.get("props"))
+        if excerpt:
+            lines.append(f"  {excerpt}")
+    return "\n".join(lines)
+
+
+def _build_open_threads_block(threads: list[dict]) -> str | None:
+    """Render the ``Open threads:`` block for Phase 3 Task 60.
+
+    One bullet per thread, formatted as ``- {title}: {summary}`` with the
+    summary truncated to ~120 characters. Returns ``None`` when there are
+    no open threads so the caller can omit the entire block.
+    """
+    if not threads:
+        return None
+    lines = ["Open threads:"]
+    for t in threads:
+        title = t.get("title") or "?"
+        summary = t.get("summary") or ""
+        if len(summary) > 120:
+            summary = summary[:119] + "…"
+        if summary:
+            lines.append(f"- {title}: {summary}")
+        else:
+            lines.append(f"- {title}")
+    return "\n".join(lines)
+
+
 def _closing_instruction(speaker_name: str, addressee_name: str) -> str:
     return (
         f"Continue the scene as {speaker_name}, in their voice, responding "
@@ -436,6 +508,17 @@ def assemble_narrative_prompt(
             if required.issubset(members):
                 group_node_block = _build_group_node_block(gn)
 
+    # SHOULD-tier active events + open threads (Phase 3 / Task 60).
+    # Auto-detect both from the chat_id per the Phase 2 T43 precedent —
+    # no new function parameter. Both blocks are omit-when-empty so a
+    # Phase 1 chat with no events/threads renders identically to before.
+    active_events_block = _build_active_events_block(
+        list_active_events(conn, chat_id)
+    )
+    open_threads_block = _build_open_threads_block(
+        list_open_threads(conn, chat_id)
+    )
+
     container = None
     if chat.get("active_scene_id"):
         scene = get_scene(conn, chat["active_scene_id"])
@@ -531,6 +614,8 @@ def assemble_narrative_prompt(
         include_you_activity: bool = True,
         include_guest_activity: bool = True,
         include_group_node: bool = True,
+        include_active_events: bool = True,
+        include_open_threads: bool = True,
     ) -> tuple[str, int, list[dict]]:
         # dialogue: keep the last `dialogue_keep` turns verbatim; older
         # turns become an "earlier:" placeholder line.
@@ -566,6 +651,8 @@ def assemble_narrative_prompt(
             scene_block,
             activity_block,
             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,
             prev_block,
             memories_block,
             dialogue_block,
@@ -585,9 +672,12 @@ def assemble_narrative_prompt(
     include_you_activity = you_activity is not None
     include_guest_activity = guest_activity is not None
     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
 
     def _build(*, prev: bool, mem_k: int, dlg: int, other: bool,
-               you_act: bool, guest_act: bool, group: bool) -> tuple[str, int]:
+               you_act: bool, guest_act: bool, group: bool,
+               events: bool, threads: bool) -> tuple[str, int]:
         body, total, _ = assemble(
             include_other_edges=other,
             include_previous_scene=prev,
@@ -596,6 +686,8 @@ def assemble_narrative_prompt(
             include_you_activity=you_act,
             include_guest_activity=guest_act,
             include_group_node=group,
+            include_active_events=events,
+            include_open_threads=threads,
         )
         return body, total
 
@@ -603,6 +695,7 @@ def assemble_narrative_prompt(
         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,
     )
 
     # If under soft, we're done.
@@ -637,6 +730,7 @@ def assemble_narrative_prompt(
             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,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -647,6 +741,7 @@ def assemble_narrative_prompt(
             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,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -657,6 +752,7 @@ def assemble_narrative_prompt(
             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,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -668,21 +764,44 @@ def assemble_narrative_prompt(
             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,
         )
 
     # Drop SHOULD-tier extras in order:
-    #   1. guest activity bullet (T71.2: bullet-level trim within the
+    #   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
     #      single ACTIVITIES: block — guest goes first per Task 43 spec)
-    #   2. group node block
-    #   3. you activity bullet (still SHOULD-tier; speaker bullet is the
+    #   4. group node block
+    #   5. you activity bullet (still SHOULD-tier; speaker bullet is the
     #      MUST-tier floor and never dropped)
-    #   4. other edges
+    #   6. other edges
+    if include_open_threads and total > budget_hard:
+        include_open_threads = 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,
+        )
+
+    if include_active_events and total > budget_hard:
+        include_active_events = 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,
+        )
+
     if include_guest_activity and total > budget_hard:
         include_guest_activity = 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,
         )
 
     if include_group_node and total > budget_hard:
@@ -691,6 +810,7 @@ def assemble_narrative_prompt(
             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,
         )
 
     if include_you_activity and total > budget_hard:
@@ -699,6 +819,7 @@ def assemble_narrative_prompt(
             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,
         )
 
     if include_other and total > budget_hard:
@@ -707,6 +828,7 @@ def assemble_narrative_prompt(
             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,
         )
 
     if total > budget_hard:

chat/services/regenerate.py

@@ -73,12 +73,15 @@ from sqlite3 import Connection
 
 from chat.config import Settings
 from chat.eventlog.log import append_and_apply, append_event
+from chat.services.event_lifecycle import detect_event_transitions
+from chat.services.event_promotion import promote_completed_event
 from chat.services.interjection import detect_interjection
 from chat.services.memory_write import record_turn_memory_for_present
 from chat.services.multi_state_update import compute_state_updates_for_present
 from chat.services.prompt import assemble_narrative_prompt
 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.world import active_scene, get_chat
 from chat.web.pubsub import publish
 from chat.web.render import render_turn_html
@@ -617,6 +620,67 @@ async def regenerate_assistant_turn(
                 (new_assistant_event_id, original_interjection_event_id),
             )
 
+    # 10. Event-lifecycle detection (Phase 3, T61). Mirrors the post_turn
+    # block: classify whether any active events transitioned in the
+    # regenerated narrative and append the corresponding event_started /
+    # event_completed / event_cancelled. ``promote_completed_event``
+    # runs inline after a completion so promotion artifacts land in the
+    # same regenerate path.
+    #
+    # Phase 3.5 follow-up: when a regenerate replaces a turn that had
+    # already produced event transitions, those original transitions are
+    # NOT undone here. The superseded ``assistant_turn`` group keeps its
+    # prior ``event_started`` / ``event_completed`` events in the log
+    # (they remain projected onto the events table). Phase 3.5 will add
+    # an "undo lifecycle" step to roll back the prior transitions before
+    # re-classifying the regenerated text. For v3 we accept that a
+    # regenerate-after-completion will double-emit promotion artifacts
+    # if the new text re-completes the same event — narratively rare,
+    # and a true fix needs the lifecycle-undo pass.
+    new_active_events = list_active_events(conn, chat_id)
+    if new_active_events:
+        lifecycle_decision = await detect_event_transitions(
+            client,
+            classifier_model=settings.classifier_model,
+            narrative_text=new_text,
+            active_events=new_active_events,
+            timeout_s=settings.classifier_timeout_s,
+        )
+        for transition in lifecycle_decision.transitions:
+            if transition.new_status == "active":
+                append_and_apply(
+                    conn,
+                    kind="event_started",
+                    payload={
+                        "event_id": transition.event_id,
+                        "started_at": chat.get("time"),
+                    },
+                )
+            elif transition.new_status == "completed":
+                append_and_apply(
+                    conn,
+                    kind="event_completed",
+                    payload={
+                        "event_id": transition.event_id,
+                        "completed_at": chat.get("time"),
+                    },
+                )
+                promote_completed_event(
+                    conn,
+                    event_id=transition.event_id,
+                    chat_id=chat_id,
+                    chat_clock_at=chat.get("time"),
+                )
+            elif transition.new_status == "cancelled":
+                append_and_apply(
+                    conn,
+                    kind="event_cancelled",
+                    payload={
+                        "event_id": transition.event_id,
+                        "completed_at": chat.get("time"),
+                    },
+                )
+
     return new_text
 
 

chat/web/turns.py

@@ -57,6 +57,8 @@ from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from chat.eventlog.log import append_and_apply, append_event
 from chat.services.addressee import detect_addressee
 from chat.services.background import SignificanceJob
+from chat.services.event_lifecycle import detect_event_transitions
+from chat.services.event_promotion import promote_completed_event
 from chat.services.interjection import detect_interjection
 from chat.services.memory_write import record_turn_memory_for_present
 from chat.services.multi_state_update import compute_state_updates_for_present
@@ -67,6 +69,7 @@ from chat.services.scene_summarize import apply_scene_close_summary
 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.world import active_scene, get_chat, get_container
 from chat.web.bots import get_conn
 from chat.web.kickoff import get_llm_client
@@ -654,6 +657,76 @@ async def post_turn(
                         )
                     )
 
+    # 8a. Event-lifecycle detection (Phase 3, T61). Runs after the post-turn
+    # classifier passes (memory write + state update + optional
+    # interjection) and BEFORE scene-close detection. The classifier reads
+    # ``primary_text`` against the chat's currently-active events and
+    # returns a (usually empty) list of transitions. Each transition lands
+    # an ``event_started`` / ``event_completed`` / ``event_cancelled``
+    # event via ``append_and_apply`` so the events projection updates
+    # synchronously. A completion is followed inline by
+    # ``promote_completed_event`` so any structured artifacts the event
+    # carries (knowledge_facts, relationship_change, acquired_objects)
+    # land in state in the same turn — see chat/services/event_promotion.
+    #
+    # ``detect_event_transitions`` short-circuits when ``active_events``
+    # is empty (per T52), so chats without active events don't pay a
+    # classifier round-trip and existing fixtures need no extra canned
+    # slots.
+    active_events = list_active_events(conn, chat_id)
+    if active_events:
+        lifecycle_decision = await detect_event_transitions(
+            client,
+            classifier_model=settings.classifier_model,
+            narrative_text=primary_text,
+            active_events=active_events,
+            timeout_s=settings.classifier_timeout_s,
+        )
+        for transition in lifecycle_decision.transitions:
+            if transition.new_status == "active":
+                append_and_apply(
+                    conn,
+                    kind="event_started",
+                    payload={
+                        "event_id": transition.event_id,
+                        "started_at": chat.get("time"),
+                    },
+                )
+            elif transition.new_status == "completed":
+                append_and_apply(
+                    conn,
+                    kind="event_completed",
+                    payload={
+                        "event_id": transition.event_id,
+                        "completed_at": chat.get("time"),
+                    },
+                )
+                # Run promotion inline so the artifact-emitting events
+                # (edge_update / manual_edit) land synchronously after
+                # the completion. ``promote_completed_event`` is
+                # synchronous (no await) and skips silently when the
+                # event row's status isn't 'completed' — a safety net
+                # for races, not expected to trigger in practice.
+                promote_completed_event(
+                    conn,
+                    event_id=transition.event_id,
+                    chat_id=chat_id,
+                    chat_clock_at=chat.get("time"),
+                )
+            elif transition.new_status == "cancelled":
+                append_and_apply(
+                    conn,
+                    kind="event_cancelled",
+                    payload={
+                        "event_id": transition.event_id,
+                        "completed_at": chat.get("time"),
+                    },
+                )
+            # Any other ``new_status`` value falls through silently —
+            # the lifecycle service constrains the schema to the three
+            # valid transitions, and a defensive no-op here keeps the
+            # turn flow tolerant of unexpected outputs.
+
     # 9. Scene-close detection (Plan §7.2, T26). Runs AFTER assistant_turn
     # and the optional interjection so the bots' responses are part of
     # the closing scene's final beat — closing before narrative would

tests/test_prompt.py

@@ -12,12 +12,14 @@ import pytest
 
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
-from chat.eventlog.log import append_event
+from chat.eventlog.log import append_and_apply, append_event
 from chat.eventlog.projector import project
 import chat.state.entities  # noqa: F401  (registers handlers)
 import chat.state.edges  # noqa: F401
 import chat.state.memory  # noqa: F401
 import chat.state.world  # noqa: F401
+import chat.state.events  # noqa: F401
+import chat.state.threads  # noqa: F401
 from chat.llm.client import Message
 from chat.services.prompt import assemble_narrative_prompt
 
@@ -761,3 +763,92 @@ def test_assemble_with_tight_budget_drops_guest_activity_first(tmp_path):
     import tiktoken
     enc = tiktoken.get_encoding("cl100k_base")
     assert len(enc.encode(body)) <= 340
+
+
+# ---------------------------------------------------------------------------
+# Task 60: Active events + open threads in prompt assembly
+# ---------------------------------------------------------------------------
+
+
+def test_assemble_with_no_events_or_threads_omits_blocks(tmp_path):
+    """Regression: with the basic 2-entity scenario (no events seeded, no
+    threads seeded), the assembled prompt must NOT contain the
+    ``Active events:`` or ``Open threads:`` headers — both blocks are
+    omit-when-empty."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_basic(conn)
+        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 "Active events:" not in body
+    assert "Open threads:" not in body
+
+
+def test_assemble_with_active_events_renders_block(tmp_path):
+    """Seed a planned event then transition it to active; the assembled
+    prompt should render the ``Active events:`` block listing the event
+    by kind."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_basic(conn)
+        # event_planned then event_started → status="active". Use
+        # append_and_apply because _seed_basic already projected; calling
+        # project() again would replay every prior event (and trip
+        # UNIQUE constraints on chat_created etc.).
+        append_and_apply(conn, kind="event_planned", payload={
+            "event_id": "evt_park",
+            "chat_id": "chat_bot_a",
+            "kind": "date_at_park",
+            "props": {"location": "Riverside Park", "vibe": "casual"},
+            "planned_for": "2026-04-30T18:00:00+00:00",
+        })
+        append_and_apply(conn, kind="event_started", payload={
+            "event_id": "evt_park",
+            "started_at": "2026-04-30T18:05:00+00:00",
+        })
+        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 "Active events:" in body
+    assert "date_at_park" in body
+
+
+def test_assemble_with_open_thread_renders_block(tmp_path):
+    """Seed a single open thread; the assembled prompt should render the
+    ``Open threads:`` block listing the thread by title."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_basic(conn)
+        # _seed_basic already projected; use append_and_apply for the
+        # post-seed event so we don't re-trigger UNIQUE constraint
+        # collisions on the prior chat_created/etc. events.
+        append_and_apply(conn, kind="thread_opened", payload={
+            "thread_id": "thr_job",
+            "chat_id": "chat_bot_a",
+            "title": "Maya's job hunt",
+            "summary": "Maya is looking for a new job",
+        })
+        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 "Open threads:" in body
+    assert "Maya's job hunt" in body

tests/test_turn_flow.py

@@ -19,7 +19,7 @@ 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.log import append_and_apply, append_event
 from chat.eventlog.projector import project
 from chat.llm.mock import MockLLMClient
 
@@ -896,3 +896,244 @@ def test_interjection_enqueues_significance_job(app_state_setup, tmp_path):
     # The two narrative texts should be the two streamed beats.
     narrative_texts = sorted(job.narrative_text for job in captured_jobs)
     assert narrative_texts == ["Interjection beat!", "Primary beat."]
+
+
+# ---------------------------------------------------------------------------
+# Phase 3 (T61) — per-turn event-lifecycle detection + completion promotion.
+#
+# After the post-turn classifier passes (memory write, state update,
+# interjection check) and BEFORE scene-close detection, ``post_turn``
+# calls :func:`detect_event_transitions`. Each transition becomes one
+# of ``event_started`` / ``event_completed`` / ``event_cancelled``. A
+# completed event is followed inline by ``promote_completed_event`` so
+# the props it carries (knowledge_facts, etc.) land in state
+# synchronously.
+#
+# When no active events are seeded the classifier short-circuits without
+# an LLM call (per T52) — the canned queue therefore needs ZERO extra
+# slots in that case.
+# ---------------------------------------------------------------------------
+
+
+def test_turn_with_event_transition_appends_started_event(
+    app_state_setup, tmp_path
+):
+    """A planned event becomes active when the classifier reports a
+    ``new_status='active'`` transition for that event_id.
+
+    Canned queue (5 calls — single-bot, no scene seeded):
+      1. parse_turn
+      2. narrative stream
+      3. state-update bot_a -> you
+      4. state-update you -> bot_a
+      5. detect_event_transitions -> 1 transition (active)
+    """
+    _seed(tmp_path / "test.db")
+    # Seed a planned event so list_active_events returns 1 row. Use
+    # append_and_apply so we don't re-replay the prior chat_created event
+    # (whose handler is INSERT-not-IGNORE and would 409 on replay).
+    with open_db(tmp_path / "test.db") as conn:
+        append_and_apply(
+            conn,
+            kind="event_planned",
+            payload={
+                "event_id": "evt_1",
+                "chat_id": "chat_bot_a",
+                "kind": "story_event",
+                "props": {},
+                "planned_for": "2026-04-30T18:00:00+00:00",
+            },
+        )
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "dialogue", "text": "they arrived"}]}
+    )
+    canned_event_decision = json.dumps(
+        {
+            "transitions": [
+                {
+                    "event_id": "evt_1",
+                    "new_status": "active",
+                    "reason": "they arrived",
+                }
+            ]
+        }
+    )
+    mock = _override_llm(
+        [
+            canned_parse,
+            "They walk in.",
+            _zero_state(),
+            _zero_state(),
+            canned_event_decision,
+        ]
+    )
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "they arrived"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    # All 5 canned slots consumed.
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        # event_started landed in event_log.
+        rows = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'event_started' ORDER BY id"
+        ).fetchall()
+        assert len(rows) == 1
+        started_payload = json.loads(rows[0][0])
+        assert started_payload["event_id"] == "evt_1"
+        assert started_payload["started_at"] == "2026-04-26T20:00:00+00:00"
+
+        # The events projection row reflects the active status.
+        ev_row = conn.execute(
+            "SELECT status, started_at FROM events WHERE event_id = ?",
+            ("evt_1",),
+        ).fetchone()
+        assert ev_row is not None
+        assert ev_row[0] == "active"
+        assert ev_row[1] == "2026-04-26T20:00:00+00:00"
+
+
+def test_turn_with_event_completion_runs_promotion(app_state_setup, tmp_path):
+    """An active event with knowledge_facts in props completes; the
+    inline call to ``promote_completed_event`` emits the corresponding
+    ``edge_update``.
+    """
+    _seed(tmp_path / "test.db")
+    # Seed: planned -> started so the event is currently active. Props
+    # carry a knowledge_fact that promotion will turn into an edge_update.
+    # Use append_and_apply (not project) to avoid re-replaying chat_created.
+    with open_db(tmp_path / "test.db") as conn:
+        append_and_apply(
+            conn,
+            kind="event_planned",
+            payload={
+                "event_id": "evt_2",
+                "chat_id": "chat_bot_a",
+                "kind": "story_event",
+                "props": {
+                    "knowledge_facts": [
+                        {
+                            "owner_id": "bot_a",
+                            "target_id": "you",
+                            "fact": "Maya likes pottery",
+                        }
+                    ]
+                },
+                "planned_for": "2026-04-30T18:00:00+00:00",
+            },
+        )
+        append_and_apply(
+            conn,
+            kind="event_started",
+            payload={
+                "event_id": "evt_2",
+                "started_at": "2026-04-30T19:00:00+00:00",
+            },
+        )
+
+    # Snapshot the max event_log id so we can assert on rows AFTER the turn.
+    with open_db(tmp_path / "test.db") as conn:
+        before_id = conn.execute(
+            "SELECT COALESCE(MAX(id), 0) FROM event_log"
+        ).fetchone()[0]
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "dialogue", "text": "we wrap it up"}]}
+    )
+    canned_event_decision = json.dumps(
+        {
+            "transitions": [
+                {
+                    "event_id": "evt_2",
+                    "new_status": "completed",
+                    "reason": "wrapped",
+                }
+            ]
+        }
+    )
+    mock = _override_llm(
+        [
+            canned_parse,
+            "They wrap it up.",
+            _zero_state(),
+            _zero_state(),
+            canned_event_decision,
+        ]
+    )
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "we wrap it up"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        # event_completed landed.
+        completed_rows = conn.execute(
+            "SELECT id, payload_json FROM event_log "
+            "WHERE kind = 'event_completed' AND id > ? ORDER BY id",
+            (before_id,),
+        ).fetchall()
+        assert len(completed_rows) == 1
+        completed_payload = json.loads(completed_rows[0][1])
+        assert completed_payload["event_id"] == "evt_2"
+        completed_id = completed_rows[0][0]
+
+        # promote_completed_event ran inline AFTER event_completed: the
+        # follow-on edge_update carries the knowledge fact and is tagged
+        # with source=event_promotion.
+        promo_rows = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'edge_update' AND id > ? ORDER BY id",
+            (completed_id,),
+        ).fetchall()
+        promo_facts: list[str] = []
+        for (payload_json,) in promo_rows:
+            p = json.loads(payload_json)
+            if p.get("source") == "event_promotion":
+                promo_facts.extend(p.get("knowledge_facts") or [])
+
+    assert "Maya likes pottery" in promo_facts
+
+
+def test_turn_with_no_active_events_skips_classifier(app_state_setup, tmp_path):
+    """When no active events are seeded, ``detect_event_transitions``
+    short-circuits without an LLM call (per T52). The canned queue must
+    therefore have ZERO event-detection slots — same shape as the
+    Phase 2 no-guest baseline.
+    """
+    _seed(tmp_path / "test.db")
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "dialogue", "text": "hello"}]}
+    )
+    # Only 4 slots: parse + narrative + 2 state-updates. NO extra slot for
+    # event-detection — non-existent active_events causes the helper to
+    # short-circuit before pulling from the queue.
+    mock = _override_llm(
+        [canned_parse, "Hi there.", _zero_state(), _zero_state()]
+    )
+    try:
+        response = app_state_setup.post(
+            "/chats/chat_bot_a/turns", data={"prose": "hello"}
+        )
+        assert response.status_code == 204
+    finally:
+        app.dependency_overrides.clear()
+    # Queue fully drained — no canned slot was consumed by event detection.
+    assert mock._canned == []
+
+    with open_db(tmp_path / "test.db") as conn:
+        for kind in ("event_started", "event_completed", "event_cancelled"):
+            count = conn.execute(
+                "SELECT COUNT(*) FROM event_log WHERE kind = ?", (kind,)
+            ).fetchone()[0]
+            assert count == 0, f"expected zero {kind} events, got {count}"