diff --git a/apps/api/src/cora/agent/prompts/__init__.py b/apps/api/src/cora/agent/prompts/__init__.py index 82bba041aa4..f61662a8ce3 100644 --- a/apps/api/src/cora/agent/prompts/__init__.py +++ b/apps/api/src/cora/agent/prompts/__init__.py @@ -38,7 +38,8 @@ KNOWN_PROMPT_TEMPLATES: dict[UUID, str] = { RUN_DEBRIEF_PROMPT_TEMPLATE_ID: "RunDebrief v1: terminal-Run AAR narrative + advisory choice", CAUTION_DRAFTER_PROMPT_TEMPLATE_ID: ( - "CautionDrafter v1: terminal-Run Caution proposal with 5-choice verdict" + "CautionDrafter v1: terminal-Run Caution proposal with 5-choice verdict; " + "payload now also carries the terminal snapshot's frame-tally capture_progress" ), } diff --git a/apps/api/src/cora/agent/prompts/caution_drafter.py b/apps/api/src/cora/agent/prompts/caution_drafter.py index 8c2ac8e30fc..6ce25292847 100644 --- a/apps/api/src/cora/agent/prompts/caution_drafter.py +++ b/apps/api/src/cora/agent/prompts/caution_drafter.py @@ -55,11 +55,26 @@ ## Read scope (v1) -v1 reads: terminal Run event + Run aggregate state + existing +v1 reads: terminal Run event (including its frame-tally snapshot, when +the terminal event carries one) + Run aggregate state + existing Active Cautions for the target (via `CautionLookup` port). Deferred to v2 per design memo: RunDebriefer's prior Decision for the same Run (needs `DecisionLookup` port; deferred until pilot UX surfaces need). +Named boundary, not a TODO: this agent has no recurrence signal across +Runs on the same target. It reads one terminal event at a time, so it +cannot tell "this Run was short" from "this Asset is short on a third +of its Runs, by a consistent margin." Originating a Caution from a +repeated shortfall therefore cannot happen in v1; a shortfall alone +stays `NoAction` (see the system prompt's "Frame-count context" +section). A pilot observation grounds this, measured 2026-08-21: every +recorded completion declares the same expectation of 1541 saved +frames; 616 met it exactly and 364 fell short by about 11 frames on +average. A shortfall there is common and systematic, not exceptional, +which is exactly why a single occurrence is weak evidence and why a +cross-Run recurrence signal, once it can be sourced without a new +lookup/port, is the next slice. + ## Structured output schema JSON Schema with five fields, four always-required + one @@ -404,6 +419,41 @@ If no Active Caution matches, propose new (pick the severity tier). +## Frame-count context (capture_progress) + +The input payload may carry `capture_progress`: the terminal snapshot's +frame tallies for a witnessed capture. Not every Run is a witnessed +capture; when the field is absent, draw no conclusion from its absence. + +A `frames_saved` shortfall against `frames_saved_expected` is a fact about +ONE RUN. RunDebriefer already records exactly that fact on its own +Decision, as `DataSuspect`; do not duplicate it here. A Caution is a claim +about an ASSET, and an asset-level claim needs a pattern, which a single +Run cannot establish. This payload carries one Run, so it does not carry +sufficient evidence to originate a Caution from a shortfall. + +A shortfall alone must never produce a proposal. `NoAction` is the +correct verdict for an isolated shortfall. + +Refusing here discards nothing. The tallies stay on the terminal event +permanently, and RunDebriefer's verdict is a durable Decision; the +occurrence is not lost by declining to propose from it. + +Exception: when an `existing_cautions` entry for the same target already +describes frame loss or detector trouble, a fresh shortfall is +corroboration that the condition persists, and `ProposeSupersede` is +available under the "Lookback then propose" rule above. This is a +secondary path, not the main mechanism: reach for it only when the +existing Caution already names the pattern, never as a way to originate +one from a shortfall alone. + +Never compare a saved count with a collected count. They come from +different instruments and their totals are not the same quantity, so a +difference between them means nothing. Each count is only ever comparable +with its own `_expected` pair (`frames_saved` against +`frames_saved_expected`; `frames_collected` against +`frames_collected_expected`). + ## Categories (closed 6-value set) Pick the ONE that most narrowly fits: @@ -493,6 +543,20 @@ class CautionDrafterPayload: `informed_by_decision_id` is reserved for v2 when DecisionLookup ports ship; v1 always None. + + `capture_progress` mirrors `RunDebriefPayload`'s field of the same + name exactly: the terminal snapshot's frame tallies + (`frames_saved` / `frames_saved_expected`, `frames_collected` / + `frames_collected_expected`), plus `reading_age_seconds_before_terminal` + when the tallies carry a parseable timestamp pair, or `None` for a + Run with no witnessed-capture snapshot. Absence is ordinary, not a + fault: not every Run is a witnessed capture. See + `extract_capture_progress` in `_terminal_run_helpers` for the + extraction and the full provenance discussion. Unlike RunDebriefer, + which reports a shortfall as a fact about the one Run it just + watched, CautionDrafter's job is to decide whether the ASSET + warrants a standing advisory; see the system prompt's "Frame-count + context" section for the current framing of that distinction. """ terminal_event_type: str @@ -510,6 +574,7 @@ class CautionDrafterPayload: interrupted_at: str | None candidate_targets: tuple[CandidateTarget, ...] = field(default_factory=tuple) existing_cautions: tuple[ExistingCaution, ...] = field(default_factory=tuple) + capture_progress: dict[str, int] | None = None def build_caution_drafter_chat_request( @@ -580,6 +645,7 @@ def _payload_to_json_safe(payload: CautionDrafterPayload) -> dict[str, Any]: } for ec in payload.existing_cautions ], + "capture_progress": payload.capture_progress, } diff --git a/apps/api/src/cora/agent/subscribers/caution_drafter.py b/apps/api/src/cora/agent/subscribers/caution_drafter.py index b93cb512b69..fa7d81bbb0b 100644 --- a/apps/api/src/cora/agent/subscribers/caution_drafter.py +++ b/apps/api/src/cora/agent/subscribers/caution_drafter.py @@ -99,6 +99,7 @@ CAUTION_DRAFTER_AGENT_NAME, ) from cora.agent.subscribers._terminal_run_helpers import ( + extract_capture_progress, extract_interrupted_at, extract_reason, ) @@ -492,6 +493,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: interrupted_at=interrupted_at, candidate_targets=candidate_targets, existing_cautions=existing_cautions, + capture_progress=extract_capture_progress(event), ) # The Agent's declared model, not the module default: that # declaration is what `define_agent` gated against the approved diff --git a/apps/api/tests/unit/agent/test_caution_drafter_prompt.py b/apps/api/tests/unit/agent/test_caution_drafter_prompt.py index 01883d32863..8a72ed71b4f 100644 --- a/apps/api/tests/unit/agent/test_caution_drafter_prompt.py +++ b/apps/api/tests/unit/agent/test_caution_drafter_prompt.py @@ -168,6 +168,59 @@ def test_payload_with_no_candidate_targets_serialises_empty_list() -> None: assert parsed["existing_cautions"] == [] +@pytest.mark.unit +def test_build_request_carries_capture_progress_when_snapshot_present() -> None: + """The frame tallies must reach the user message JSON verbatim so the + LLM can read them under the system prompt's "Frame-count context" + section.""" + payload = _payload( + capture_progress={ + "frames_saved": 1528, + "frames_saved_expected": 1541, + "frames_collected": 1541, + "frames_collected_expected": 1541, + "reading_age_seconds_before_terminal": 13, + } + ) + request = build_caution_drafter_chat_request(payload) + json_blob = request.user_message.text[request.user_message.text.index("{") :] + parsed = json.loads(json_blob) + assert parsed["capture_progress"] == { + "frames_saved": 1528, + "frames_saved_expected": 1541, + "frames_collected": 1541, + "frames_collected_expected": 1541, + "reading_age_seconds_before_terminal": 13, + } + + +@pytest.mark.unit +def test_build_request_capture_progress_defaults_to_none() -> None: + """A Run with no witnessed-capture snapshot must still build a valid + request, with `capture_progress` travelling as JSON `null` rather + than being omitted.""" + request = build_caution_drafter_chat_request(_payload()) + json_blob = request.user_message.text[request.user_message.text.index("{") :] + parsed = json.loads(json_blob) + assert parsed["capture_progress"] is None + + +@pytest.mark.unit +def test_system_prompt_forbids_originating_from_a_lone_shortfall() -> None: + """Pin the load-bearing invariant, not a section title: a single Run's + shortfall must never by itself produce a Caution proposal. A future + reword of the section is fine; this rule collapsing back into a + per-run alert is not.""" + assert "A shortfall alone must never produce a proposal" in CAUTION_DRAFTER_SYSTEM_PROMPT + + +@pytest.mark.unit +def test_system_prompt_forbids_comparing_saved_and_collected_counts() -> None: + """Pin the other load-bearing invariant: saved and collected tallies + come from different instruments and are never mutually comparable.""" + assert "Never compare a saved count with a collected count" in CAUTION_DRAFTER_SYSTEM_PROMPT + + @pytest.mark.unit def test_decision_context_constant_matches_design_lock() -> None: """The context value used by the subscriber is `CautionProposal`.""" diff --git a/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py b/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py index 7f46930963b..18390d25597 100644 --- a/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py +++ b/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py @@ -12,6 +12,7 @@ # pyright: reportPrivateUsage=false, reportUnknownMemberType=false +import json from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, cast from uuid import UUID, uuid4, uuid5 @@ -44,6 +45,7 @@ from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.ports import ( AlwaysQuietCautionLookup, + CautionLookupResult, FakeLLM, FakeLLMResponse, LLMServerError, @@ -223,8 +225,16 @@ def _terminal_event( event_type: str, run_id: UUID, reason: str | None = None, + capture_progress_snapshot: dict[str, object] | None = None, ) -> StoredEvent: - """Build a StoredEvent for a terminal Run event.""" + """Build a StoredEvent for a terminal Run event. + + `capture_progress_snapshot`, when given, is merged into the payload + by hand (rather than via the typed `CaptureProgressSnapshot` + dataclass) so tests can assemble the exact raw shape + `extract_capture_progress` reads, mirroring + `test_run_debriefer_subscriber._progress_event`. + """ domain: Any if event_type == "RunCompleted": domain = RunCompleted(run_id=run_id, occurred_at=_LATER, observed_at=None) @@ -234,6 +244,9 @@ def _terminal_event( else: msg = f"unsupported event type for fixture: {event_type}" raise ValueError(msg) + payload = run_to_payload(domain) + if capture_progress_snapshot is not None: + payload = {**payload, "capture_progress_snapshot": capture_progress_snapshot} return StoredEvent( position=1, event_id=UUID("01900000-0000-7000-8000-00000000ff01"), @@ -242,7 +255,7 @@ def _terminal_event( version=2, event_type=event_type, schema_version=1, - payload=run_to_payload(domain), + payload=payload, correlation_id=_CORRELATION_ID, causation_id=None, occurred_at=_LATER, @@ -390,6 +403,196 @@ async def test_apply_emits_caution_proposal_decision_on_run_aborted() -> None: assert decision.inputs["confidence_band"] == "medium" +# --------------------------------------------------------------------------- +# Capture-progress frame tallies (extract_capture_progress -> payload) +# --------------------------------------------------------------------------- + + +class _FixedCautionLookup: + """`CautionLookup` stub returning a canned list of existing cautions. + + Sibling of `AlwaysQuietCautionLookup` for tests that need + `find_active_in_scope` to answer non-empty, so the + shortfall-plus-existing-Caution corroboration path can be pinned. + """ + + def __init__(self, results: list[CautionLookupResult]) -> None: + self._results = results + + async def find_active_in_scope( + self, + *, + asset_ids: frozenset[UUID], + procedure_ids: frozenset[UUID], + min_severity: str = "Caution", + ) -> list[CautionLookupResult]: + _ = (asset_ids, procedure_ids, min_severity) + return self._results + + async def find_retired_for_target( + self, + *, + target_kind: str, + target_id: UUID, + category: str, + authored_by: UUID, + ) -> list[CautionLookupResult]: + _ = (target_kind, target_id, category, authored_by) + return [] + + +_SHORTFALL_SNAPSHOT: dict[str, object] = { + "saved_count": 1528, + "saved_total": 1541, + "collected_count": 1541, + "collected_total": 1541, + "saved_at": "2026-05-17T14:46:47+00:00", +} + + +@pytest.mark.unit +async def test_apply_passes_capture_progress_to_llm_payload_when_snapshot_present() -> None: + """The terminal event's frame tallies must reach the built request's + user message, mirroring RunDebriefer's `extract_capture_progress` + wiring. Before this, CautionDrafter never saw a shortfall at all.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm) + event = _terminal_event( + event_type="RunCompleted", + run_id=run_id, + capture_progress_snapshot=_SHORTFALL_SNAPSHOT, + ) + + await subscriber.apply(event, conn=None) + + assert len(llm.received) == 1 + user_text = llm.received[0].user_message.text + json_blob = user_text[user_text.index("{") :] + parsed = json.loads(json_blob) + # No `reading_age_seconds_before_terminal` key: the fixture's terminal + # event carries `observed_at=None`, so `_reading_lag_seconds` has no + # second timestamp to subtract and the key is omitted rather than + # defaulted to zero. + assert parsed["capture_progress"] == { + "frames_saved": 1528, + "frames_saved_expected": 1541, + "frames_collected": 1541, + "frames_collected_expected": 1541, + } + + +@pytest.mark.unit +async def test_apply_passes_none_capture_progress_when_snapshot_absent() -> None: + """A Run with no witnessed-capture snapshot (the ordinary case, not a + fault) must still build a valid request, with `capture_progress` + travelling as `None`.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert len(llm.received) == 1 + user_text = llm.received[0].user_message.text + json_blob = user_text[user_text.index("{") :] + parsed = json.loads(json_blob) + assert parsed["capture_progress"] is None + + +@pytest.mark.unit +async def test_apply_sends_shortfall_alongside_matching_existing_caution() -> None: + """One documented path: a shortfall corroborated by an existing + Active Caution describing frame loss on the same target. Pins the + payload the LLM sees (both the tallies and the existing Caution) + rather than any verdict, per the "cannot assert what the model + chooses" constraint.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + matching_caution_id = uuid4() + caution_lookup = _FixedCautionLookup( + [ + CautionLookupResult( + caution_id=matching_caution_id, + target_kind="Asset", + target_id=_ASSET_ID, + category="Wiring", + severity="Caution", + text_excerpt="detector drops frames intermittently under load", + workaround_excerpt="re-arm the file writer before long scans", + ) + ] + ) + subscriber = CautionDrafterSubscriber( + event_store=store, + llm=llm, + caution_lookup=caution_lookup, + ) + event = _terminal_event( + event_type="RunCompleted", + run_id=run_id, + capture_progress_snapshot=_SHORTFALL_SNAPSHOT, + ) + + await subscriber.apply(event, conn=None) + + assert len(llm.received) == 1 + user_text = llm.received[0].user_message.text + json_blob = user_text[user_text.index("{") :] + parsed = json.loads(json_blob) + assert parsed["capture_progress"]["frames_saved"] == 1528 + assert parsed["capture_progress"]["frames_saved_expected"] == 1541 + assert len(parsed["existing_cautions"]) == 1 + assert parsed["existing_cautions"][0]["caution_id"] == str(matching_caution_id) + assert parsed["existing_cautions"][0]["text_excerpt"] == ( + "detector drops frames intermittently under load" + ) + + +@pytest.mark.unit +async def test_apply_sends_shortfall_absent_existing_caution() -> None: + """The other documented path: a shortfall with nothing in + `existing_cautions` to corroborate it. Per the guidance, this is + the case where the prompt tells the model `NoAction` remains + correct; this test pins that the payload still carries the tallies + with an empty `existing_cautions` list, not that the model must + refuse.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm) # AlwaysQuietCautionLookup + event = _terminal_event( + event_type="RunCompleted", + run_id=run_id, + capture_progress_snapshot=_SHORTFALL_SNAPSHOT, + ) + + await subscriber.apply(event, conn=None) + + assert len(llm.received) == 1 + user_text = llm.received[0].user_message.text + json_blob = user_text[user_text.index("{") :] + parsed = json.loads(json_blob) + assert parsed["capture_progress"]["frames_saved"] == 1528 + assert parsed["existing_cautions"] == [] + + # --------------------------------------------------------------------------- # NoAction path (the most common outcome per design) # ---------------------------------------------------------------------------