merge: T55 thread-detection service

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/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"]

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_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 == []