فهرست منبع

Changed triage behavior

Efren Yevale Varela 1 ماه پیش
والد
کامیت
4dd96d5181
6فایلهای تغییر یافته به همراه152 افزوده شده و 75 حذف شده
  1. 6 5
      docs/CONFIGURATION.md
  2. 22 15
      docs/TRIAGE.md
  3. 12 9
      src/stt_daemon/assistant/service.py
  4. 26 24
      src/stt_daemon/assistant/triage.py
  5. 58 13
      tests/test_assistant_service.py
  6. 28 9
      tests/test_triage.py

+ 6 - 5
docs/CONFIGURATION.md

@@ -122,9 +122,9 @@ window               = 500                    # Utterances available per classif
 state_char_budget    = 3000                 # Char cap (~750 tokens) for the condensed state sent to Laya
 state_char_budget    = 3000                 # Char cap (~750 tokens) for the condensed state sent to Laya
 predict_max_tokens   = 1024                 # Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room)
 predict_max_tokens   = 1024                 # Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room)
 confidence_field     = "answer_confidence"  # Gating confidence: "answer_confidence" (calibrated max(p)) | "confidence"
 confidence_field     = "answer_confidence"  # Gating confidence: "answer_confidence" (calibrated max(p)) | "confidence"
-escalate_threshold   = 0.6                  # noul answers at or above this p escalate to extraction
-choice_min_confidence = 0.85                # choice answers below this confidence escalate
-extract_enabled      = true                 # Cascade to the LLM JSON extraction when signals escalate
+escalate_threshold   = 0.6                  # noul answers at or above this p are flagged (⚠ marker only)
+choice_min_confidence = 0.85                # choice answers below this confidence are flagged (⚠ marker only)
+extract_enabled      = true                 # Master switch for the LLM JSON extraction after every Laya pass
 extract_max_tokens   = 512                  # Extraction reply token cap
 extract_max_tokens   = 512                  # Extraction reply token cap
 extract_timeout_s    = 30.0                 # Extraction request timeout
 extract_timeout_s    = 30.0                 # Extraction request timeout
 device               = ""                   # "" = auto -> None (Laya picks the multilingual checkpoint's default device)
 device               = ""                   # "" = auto -> None (Laya picks the multilingual checkpoint's default device)
@@ -189,8 +189,9 @@ can never disagree. A legacy `max_prompt_tokens` key is ignored with a warning.
 `[triage]` adds a display-only signal panel. On each settled turn in any mode, independent of
 `[triage]` adds a display-only signal panel. On each settled turn in any mode, independent of
 chat, the assistant classifies the last `window` transcribed utterances (from the STT stream,
 chat, the assistant classifies the last `window` transcribed utterances (from the STT stream,
 not the chat history) with a local Laya **multilingual** checkpoint on a dedicated
 not the chat history) with a local Laya **multilingual** checkpoint on a dedicated
-in-process lane, then, when signals escalate, asks the `[llm]` lane for a structured JSON
-extraction (`extract_enabled`). `device = ""` means auto: it maps to `None`, so Laya picks the
+in-process lane, then always asks the `[llm]` lane for a structured JSON
+extraction (`extract_enabled`; on a Laya error the prompt carries transcript state only).
+`device = ""` means auto: it maps to `None`, so Laya picks the
 default device for the multilingual checkpoint. `preload = ["multilingual"]` is the fixed default
 default device for the multilingual checkpoint. `preload = ["multilingual"]` is the fixed default
 and `default_checkpoint = "multilingual"`; the multilingual checkpoint is pinned for every call
 and `default_checkpoint = "multilingual"`; the multilingual checkpoint is pinned for every call
 (there is no `.en` English-only variant). `predict_max_tokens` is the Laya token budget (the
 (there is no `.en` English-only variant). `predict_max_tokens` is the Laya token budget (the

+ 22 - 15
docs/TRIAGE.md

@@ -2,9 +2,8 @@
 
 
 The triage panel is a display-only signal readout in the assistant TUI. On each settled
 The triage panel is a display-only signal readout in the assistant TUI. On each settled
 turn (no chat mode required) the assistant classifies the most recent transcribed
 turn (no chat mode required) the assistant classifies the most recent transcribed
-utterances with a local Laya **multilingual** model running in-process, and, when the
-signals look strong enough, asks the existing `[llm]` lane for one structured JSON
-extraction. Both results land in a
+utterances with a local Laya **multilingual** model running in-process, and then always
+asks the existing `[llm]` lane for one structured JSON extraction. Both results land in a
 right-hand pane. Nothing here triggers a mode, a command, or a tool: the panel only shows
 right-hand pane. Nothing here triggers a mode, a command, or a tool: the panel only shows
 what the model inferred.
 what the model inferred.
 
 
@@ -38,9 +37,11 @@ the assistant.
 settled turn (any mode)
 settled turn (any mode)
     -> newest transcribed-utterance window (window entries, condensed under state_char_budget)
     -> newest transcribed-utterance window (window entries, condensed under state_char_budget)
     -> Laya multilingual classification (in-process lane)
     -> Laya multilingual classification (in-process lane)
-    -> signals  ------------------------------------------------+
-    -> should_escalate? and extract_enabled?                    |
+    -> signals (emitted; flagged per should_escalate) ----------+
+    -> extract_enabled and an extract function present?         |
          -> one triage-extract job on the LLM InferenceQueue    |
          -> one triage-extract job on the LLM InferenceQueue    |
+              (all normalized signals in the prompt; on a Laya  |
+               error the prompt carries transcript state only)  |
               -> parsed JSON merged as content ------------------+
               -> parsed JSON merged as content ------------------+
 ```
 ```
 
 
@@ -63,13 +64,19 @@ settled turn (any mode)
    - `choice`: the gating confidence is `< choice_min_confidence`
    - `choice`: the gating confidence is `< choice_min_confidence`
    - `score`: `expected >= (levels - 1) * 0.5`, where `levels` is the number of score
    - `score`: `expected >= (levels - 1) * 0.5`, where `levels` is the number of score
      criteria
      criteria
-5. **Extraction.** When the signals escalated, extraction is enabled, and an extract
+   Escalation only drives the per-item `flagged` flag and the `⚠` marker in the panel;
+   it no longer gates extraction.
+5. **Extraction.** After every Laya pass, when extraction is enabled and an extract
    function is configured, the service submits **one** `triage-extract` job to the existing
    function is configured, the service submits **one** `triage-extract` job to the existing
-   LLM `InferenceQueue`. If the queue is full the job is refused with `submit_if_room`: no
-   queued chat job is ever evicted, and the signals simply stand. The worker re-checks the
-   generation before and after the blocking LLM call, parses the reply (strict JSON, fences
-   stripped, lists capped), and merges it as `content` over the signals. A parse failure
-   keeps the signals and emits no `content`.
+   LLM `InferenceQueue`. The extraction prompt carries the condensed transcript state plus
+   **all** normalized signals in order (rendered as words under a `Signals:` header); when
+   there are no signal lines the signals block is omitted and the prompt is state plus the
+   JSON-only instruction. On a Laya error (no payload) the LLM still runs, with transcript
+   state only and no `signals` event emitted. If the queue is full the job is refused with
+   `submit_if_room`: no queued chat job is ever evicted, and the signals simply stand. The
+   worker re-checks the generation before and after the blocking LLM call, parses the reply
+   (strict JSON, fences stripped, lists capped), and merges it as `content` over the
+   signals. A parse failure keeps the signals and emits no `content`.
 6. **Reset.** `clear` bumps the triage generation, clears the stored signals and the
 6. **Reset.** `clear` bumps the triage generation, clears the stored signals and the
    ambient buffer, drops any pending lane request and any queued `triage-extract` job, and
    ambient buffer, drops any pending lane request and any queued `triage-extract` job, and
    emits the `cleared` stage. An in-flight result that lands after the bump is suppressed.
    emits the `cleared` stage. An in-flight result that lands after the bump is suppressed.
@@ -210,10 +217,10 @@ file.
 | `window` | `500` | Recent transcribed utterances available per classification; the char budget is the real limiter. |
 | `window` | `500` | Recent transcribed utterances available per classification; the char budget is the real limiter. |
 | `state_char_budget` | `3000` | Character cap (~750 tokens) for the condensed state sent to Laya. |
 | `state_char_budget` | `3000` | Character cap (~750 tokens) for the condensed state sent to Laya. |
 | `predict_max_tokens` | `1024` | Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room). |
 | `predict_max_tokens` | `1024` | Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room). |
-| `confidence_field` | `"answer_confidence"` | Which Laya confidence gates escalation: `answer_confidence` (calibrated `max(p)`) or `confidence` (normalized entropy). |
-| `escalate_threshold` | `0.6` | A `noul` answer at or above this `p` escalates to extraction. |
-| `choice_min_confidence` | `0.85` | A `choice` answer below this confidence escalates. |
-| `extract_enabled` | `true` | Cascade to the LLM JSON extraction when signals escalate. |
+| `confidence_field` | `"answer_confidence"` | Which Laya confidence feeds the per-item `flagged`/`⚠` marker: `answer_confidence` (calibrated `max(p)`) or `confidence` (normalized entropy). |
+| `escalate_threshold` | `0.6` | A `noul` answer at or above this `p` is flagged (`⚠` marker only). |
+| `choice_min_confidence` | `0.85` | A `choice` answer below this confidence is flagged (`⚠` marker only). |
+| `extract_enabled` | `true` | Master switch for the LLM JSON extraction, which runs after every Laya pass. |
 | `extract_max_tokens` | `512` | Token cap for the extraction reply. |
 | `extract_max_tokens` | `512` | Token cap for the extraction reply. |
 | `extract_timeout_s` | `30.0` | Timeout for the extraction request. |
 | `extract_timeout_s` | `30.0` | Timeout for the extraction request. |
 | `device` | `""` | Empty string means auto, which maps to `None` so Laya picks the checkpoint's default device. Any other value is passed through. |
 | `device` | `""` | Empty string means auto, which maps to `None` so Laya picks the checkpoint's default device. Any other value is passed through. |

+ 12 - 9
src/stt_daemon/assistant/service.py

@@ -681,21 +681,24 @@ class AssistantService:
         self._laya_lane.submit(state, questions, self._last_transcript_lang, generation)
         self._laya_lane.submit(state, questions, self._last_transcript_lang, generation)
 
 
     def _on_triage_result(self, generation, payload) -> None:
     def _on_triage_result(self, generation, payload) -> None:
-        if self._triage is None or payload is None:
+        if self._triage is None:
             return
             return
         with self._lock:
         with self._lock:
             if generation != self._triage_generation:
             if generation != self._triage_generation:
                 return
                 return
+            window = self._last_triage_window
+        signals = None
+        if payload is not None:
             signals = self._triage.normalize(payload)
             signals = self._triage.normalize(payload)
             if not signals:
             if not signals:
                 return
                 return
-            window = self._last_triage_window
-            self.last_triage = signals
-        self._emit(TRIAGE_EVENT, stage="signals", triage=signals)
-        if (
-            signals.get("escalated")
-            and self._triage_extract_fn is not None
-            and getattr(self.cfg.triage, "extract_enabled", True)
+            with self._lock:
+                if generation != self._triage_generation:
+                    return
+                self.last_triage = signals
+            self._emit(TRIAGE_EVENT, stage="signals", triage=signals)
+        if self._triage_extract_fn is not None and getattr(
+            self.cfg.triage, "extract_enabled", True
         ):
         ):
             job = InferenceJob(
             job = InferenceJob(
                 name="triage-extract",
                 name="triage-extract",
@@ -724,7 +727,7 @@ class AssistantService:
         with self._lock:
         with self._lock:
             if generation != self._triage_generation:
             if generation != self._triage_generation:
                 return
                 return
-            merged = {**signals, "content": parsed}
+            merged = {**(signals or {}), "content": parsed}
             self.last_triage = merged
             self.last_triage = merged
         self._emit(TRIAGE_EVENT, stage="content", triage=merged)
         self._emit(TRIAGE_EVENT, stage="content", triage=merged)
 
 

+ 26 - 24
src/stt_daemon/assistant/triage.py

@@ -340,30 +340,32 @@ class TriageAnalyzer:
         return "%s: flagged" % (qid,)
         return "%s: flagged" % (qid,)
 
 
     def build_extraction_prompt(self, window, signals) -> str:
     def build_extraction_prompt(self, window, signals) -> str:
-        """Render condensed window + flagged signals as words + JSON-only instruction."""
-        state_text = self.build_state(window)
-        items = signals.get("items", {}) if isinstance(signals, dict) else {}
-        order = signals.get("order", []) if isinstance(signals, dict) else []
-        if not isinstance(items, dict):
-            items = {}
-        if not isinstance(order, list):
-            order = []
-        flagged = []
-        for qid in order:
-            item = items.get(qid)
-            if isinstance(item, dict) and item.get("flagged") is True:
-                try:
-                    flagged.append("- " + self._signal_words(qid, item))
-                except Exception:
-                    continue
-        return (
-            "Recent user utterances:\n"
-            + state_text
-            + "\n\nFlagged signals:\n"
-            + ("\n".join(flagged) if flagged else "none")
-            + "\n\n"
-            + _EXTRACTION_INSTRUCTION
-        )
+        """Render condensed window + all signals as words + JSON-only instruction."""
+        try:
+            state_text = self.build_state(window)
+        except Exception:
+            state_text = ""
+        lines = []
+        try:
+            items = signals.get("items", {}) if isinstance(signals, dict) else {}
+            order = signals.get("order", []) if isinstance(signals, dict) else []
+            if not isinstance(items, dict):
+                items = {}
+            if not isinstance(order, list):
+                order = []
+            for qid in order:
+                item = items.get(qid)
+                if isinstance(item, dict):
+                    try:
+                        lines.append("- " + self._signal_words(qid, item))
+                    except Exception:
+                        continue
+        except Exception:
+            lines = []
+        base = "Recent user utterances:\n" + state_text + "\n\n"
+        if lines:
+            return base + "Signals:\n" + "\n".join(lines) + "\n\n" + _EXTRACTION_INSTRUCTION
+        return base + _EXTRACTION_INSTRUCTION
 
 
     def parse_extraction(self, text) -> dict | None:
     def parse_extraction(self, text) -> dict | None:
         """Parse fenced-or-bare JSON extraction; cap lists/strings; None on failure."""
         """Parse fenced-or-bare JSON extraction; cap lists/strings; None on failure."""

+ 58 - 13
tests/test_assistant_service.py

@@ -518,7 +518,9 @@ class FakeTriage:
         return signals
         return signals
 
 
     def build_extraction_prompt(self, window, signals):
     def build_extraction_prompt(self, window, signals):
-        self.prompt_calls.append((tuple(window), dict(signals)))
+        self.prompt_calls.append(
+            (tuple(window), dict(signals) if isinstance(signals, dict) else signals)
+        )
         return "EXTRACT"
         return "EXTRACT"
 
 
     def parse_extraction(self, text):
     def parse_extraction(self, text):
@@ -832,6 +834,46 @@ def test_on_triage_result_ignores_none_payload():
     assert _triage_events(emitted) == []
     assert _triage_events(emitted) == []
 
 
 
 
+def test_none_payload_enqueues_state_only_extract():
+    service, emitted, clock, triage, lane = _triage_service(
+        extract_fn=lambda prompt: "RAW"
+    )
+    service.toggle_chat()
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+
+    service._on_triage_result(gen, None)
+
+    assert service.last_triage is None
+    assert [e.get("stage") for e in _triage_events(emitted)] == []
+    assert [j.name for j in service._inference._queue] == ["chat", "triage-extract"]
+    job = [j for j in service._inference._queue if j.name == "triage-extract"][0]
+    job.run()
+
+    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
+    assert len(contents) == 1
+    assert contents[0]["triage"] == {"content": triage.parse_extraction("RAW")}
+    assert service.last_triage == contents[0]["triage"]
+    assert triage.prompt_calls == [(service._last_triage_window, None)]
+
+
+def test_none_payload_without_extract_fn_enqueues_nothing():
+    service, emitted, clock, _, lane = _triage_service()
+    service.toggle_chat()
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+
+    service._on_triage_result(gen, None)
+
+    assert service.last_triage is None
+    assert _triage_events(emitted) == []
+    assert [j.name for j in service._inference._queue] == ["chat"]
+
+
 def test_on_triage_result_ignores_empty_signals():
 def test_on_triage_result_ignores_empty_signals():
     service, emitted, clock, _, lane = _triage_service(
     service, emitted, clock, _, lane = _triage_service(
         triage=FakeTriage(normalized={})
         triage=FakeTriage(normalized={})
@@ -869,7 +911,7 @@ def test_escalation_enqueues_extract_job():
     assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
     assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
 
 
 
 
-def test_no_extract_without_escalation():
+def test_extract_runs_without_escalation():
     service, _, clock, _, lane = _triage_service(extract_fn=lambda prompt: "RAW")
     service, _, clock, _, lane = _triage_service(extract_fn=lambda prompt: "RAW")
     service.toggle_chat()
     service.toggle_chat()
     service.handle_event(_transcript(clock, "all calm"))
     service.handle_event(_transcript(clock, "all calm"))
@@ -882,8 +924,8 @@ def test_no_extract_without_escalation():
     )
     )
 
 
     assert service.last_triage is not None
     assert service.last_triage is not None
-    assert service._inference.depth() == 1
-    assert [j.name for j in service._inference._queue] == ["chat"]
+    assert service._inference.depth() == 2
+    assert [j.name for j in service._inference._queue] == ["chat", "triage-extract"]
 
 
 
 
 def test_no_extract_when_disabled_in_config():
 def test_no_extract_when_disabled_in_config():
@@ -1093,7 +1135,7 @@ def test_triage_end_to_end_signals_content_cleared():
     assert prompts == []
     assert prompts == []
     assert service.last_triage is not None
     assert service.last_triage is not None
     assert service.last_triage["escalated"] is False
     assert service.last_triage["escalated"] is False
-    assert [j.name for j in service._inference._queue] == ["chat"]
+    assert [j.name for j in service._inference._queue] == ["chat", "triage-extract"]
 
 
     triage._escalate = True
     triage._escalate = True
     service.handle_event(_transcript(clock, "we might churn"))
     service.handle_event(_transcript(clock, "we might churn"))
@@ -1106,19 +1148,21 @@ def test_triage_end_to_end_signals_content_cleared():
     assert [e.get("stage") for e in _triage_events(emitted)] == ["signals", "signals"]
     assert [e.get("stage") for e in _triage_events(emitted)] == ["signals", "signals"]
     assert [j.name for j in service._inference._queue] == [
     assert [j.name for j in service._inference._queue] == [
         "chat",
         "chat",
+        "triage-extract",
         "chat",
         "chat",
         "triage-extract",
         "triage-extract",
     ]
     ]
-    job = [j for j in service._inference._queue if j.name == "triage-extract"][0]
-    job.run()
+    for job in [j for j in service._inference._queue if j.name == "triage-extract"]:
+        job.run()
 
 
     contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
     contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
-    assert len(contents) == 1
-    assert len(prompts) == 1
-    assert contents[0]["triage"]["escalated"] is True
-    assert contents[0]["triage"]["content"] == triage.parse_extraction("x")
-    assert service.last_triage == contents[0]["triage"]
-    assert service.status()["triage"] == contents[0]["triage"]
+    assert len(contents) == 2
+    assert len(prompts) == 2
+    assert contents[0]["triage"]["escalated"] is False
+    assert contents[1]["triage"]["escalated"] is True
+    assert contents[1]["triage"]["content"] == triage.parse_extraction("x")
+    assert service.last_triage == contents[1]["triage"]
+    assert service.status()["triage"] == contents[1]["triage"]
 
 
     service.clear()
     service.clear()
 
 
@@ -1131,6 +1175,7 @@ def test_triage_end_to_end_signals_content_cleared():
         "signals",
         "signals",
         "signals",
         "signals",
         "content",
         "content",
+        "content",
         "cleared",
         "cleared",
     ]
     ]
 
 

+ 28 - 9
tests/test_triage.py

@@ -357,12 +357,29 @@ def test_build_extraction_prompt_words_not_dump(tmp_path):
     prompt = analyzer.build_extraction_prompt(["please cancel my account"], _signals(noul_p=0.8))
     prompt = analyzer.build_extraction_prompt(["please cancel my account"], _signals(noul_p=0.8))
     assert "please cancel my account" in prompt
     assert "please cancel my account" in prompt
     assert "churn_risk" in prompt  # flagged signal named in words
     assert "churn_risk" in prompt  # flagged signal named in words
-    assert "category" not in prompt  # unflagged signals omitted
+    assert "category" in prompt  # unflagged signals included too
+    assert "severity" in prompt
+    assert "Signals:" in prompt
+    assert "Flagged signals:" not in prompt
     assert "{'kind'" not in prompt and '{"kind"' not in prompt  # no raw dict dump
     assert "{'kind'" not in prompt and '{"kind"' not in prompt  # no raw dict dump
     for key in ("summary", "opportunities", "doubts", "action_items", "risks"):
     for key in ("summary", "opportunities", "doubts", "action_items", "risks"):
         assert key in prompt
         assert key in prompt
 
 
 
 
+def test_build_extraction_prompt_state_only_when_no_signals(tmp_path):
+    analyzer = _analyzer(tmp_path, _score_questions())
+    for signals in (None, {}, {"order": [], "items": {}}):
+        prompt = analyzer.build_extraction_prompt(["hello there"], signals)
+        assert "hello there" in prompt
+        assert "Signals:" not in prompt
+        assert "Flagged signals:" not in prompt
+        assert "summary" in prompt  # instruction keys still present
+        assert "opportunities" in prompt
+    prompt = analyzer.build_extraction_prompt(["hello there"], None)
+    assert "Signals:" not in prompt
+    assert "\nnone" not in prompt
+
+
 def _extraction_doc(**overrides):
 def _extraction_doc(**overrides):
     doc = {
     doc = {
         "summary": "user wants to cancel",
         "summary": "user wants to cancel",
@@ -620,15 +637,15 @@ def test_end_to_end_triage_flow():
     assert lang == "es"
     assert lang == "es"
     assert gen1 == 0
     assert gen1 == 0
 
 
-    # 2. non-escalating result -> signals only, no extraction job queued
+    # 2. non-escalating result -> signals plus one state+signals extraction job
     service._on_triage_result(gen1, _answers(noul_p=0.1))
     service._on_triage_result(gen1, _answers(noul_p=0.1))
     assert [e["stage"] for e in triage_events()] == ["signals"]
     assert [e["stage"] for e in triage_events()] == ["signals"]
     assert triage_events()[0]["triage"]["escalated"] is False
     assert triage_events()[0]["triage"]["escalated"] is False
     assert service.last_triage is not None
     assert service.last_triage is not None
     assert service.last_triage["escalated"] is False
     assert service.last_triage["escalated"] is False
-    assert extract_jobs() == []
+    assert len(extract_jobs()) == 1
 
 
-    # 3. escalating turn -> second submit, then exactly one extract job
+    # 3. escalating turn -> second submit, then a second extract job
     service.handle_event(
     service.handle_event(
         {"type": events.TRANSCRIPT, "text": "quiero cancelar mi suscripcion", "lang": "es"}
         {"type": events.TRANSCRIPT, "text": "quiero cancelar mi suscripcion", "lang": "es"}
     )
     )
@@ -640,13 +657,14 @@ def test_end_to_end_triage_flow():
     assert [e["stage"] for e in triage_events()] == ["signals", "signals"]
     assert [e["stage"] for e in triage_events()] == ["signals", "signals"]
     assert triage_events()[1]["triage"]["escalated"] is True
     assert triage_events()[1]["triage"]["escalated"] is True
     jobs = extract_jobs()
     jobs = extract_jobs()
-    assert len(jobs) == 1
+    assert len(jobs) == 2
 
 
-    # 4. run the extraction job directly (deterministic, no worker thread)
-    jobs[0].run()
+    # 4. run the extraction jobs directly (deterministic, no worker thread)
+    for job in jobs:
+        job.run()
     content_events = [e for e in triage_events() if e["stage"] == "content"]
     content_events = [e for e in triage_events() if e["stage"] == "content"]
-    assert len(content_events) == 1
-    merged = content_events[0]["triage"]
+    assert len(content_events) == 2
+    merged = content_events[1]["triage"]
     assert merged["content"]["summary"] == "s"
     assert merged["content"]["summary"] == "s"
     assert merged["escalated"] is True  # signals preserved under the content
     assert merged["escalated"] is True  # signals preserved under the content
     assert merged["order"] == triage_events()[1]["triage"]["order"]
     assert merged["order"] == triage_events()[1]["triage"]["order"]
@@ -678,6 +696,7 @@ def test_end_to_end_triage_flow():
         "signals",
         "signals",
         "signals",
         "signals",
         "content",
         "content",
+        "content",
         "cleared",
         "cleared",
     ]
     ]
     assert view.triage is None
     assert view.triage is None