From bc98e8be9bf7387130e1ead063081c6675feeac1 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Sat, 11 Jul 2026 18:23:29 +0300 Subject: [PATCH] Scope the debrief lease to agent kind so distinct agents coexist The lease-winner scan matched ANY DecisionDebriefRequested marker by terminal_event_id, so RunDebriefer and CautionDrafter, which both subscribe to all four terminal Run events, contended for one lease: whichever applied second saw the other as the winner and wrote a Conflicted audit Decision instead of doing its job. In the full app that left the second-registered agent systematically dead; the defect stayed invisible because every subscriber test drives one agent at a time. The caution_drafter call-site comment claimed independence the scan never implemented. The design memo's contention target is multi-agent pollution among agents doing the SAME job (rainbow-deploy variants of one debriefer, different LLM backends of one drafter), and the Agent aggregate's kind already names that job, so the marker now carries debriefer_kind (additive payload field) and the winner scan filters by it. Kindless legacy markers contend with every kind, preserving replay behavior for pre-scoping streams. Same-kind semantics are unchanged: variants race on stream version, same-agent retries stay idempotent via the uuid5 event_id. The one-shot append also became a bounded load-scan-append loop: with kinds independent, another kind's lease landing between load and append is now the COMMON version-bump, and treating it as a degenerate loss would drop real work. A non-competing bump retries (3 attempts); exhaustion keeps the old (False, None) exit. decided_by log values are now scan, append, and retry_exhaustion. Verified by test_apply_coexists_with_run_debriefer_on_same_terminal_event, which reproduces the contention on the pre-fix code (CautionDrafter wrote CautionDraftConflicted; both agents now write substantive Decisions), plus primitive-level pins for cross-kind independence, legacy wildcard, retry-past-non-competitor, and retry exhaustion. Co-Authored-By: Claude Fable 5 --- apps/api/src/cora/agent/_subscriber_lease.py | 157 ++++++++------- .../cora/agent/subscribers/caution_drafter.py | 13 +- .../cora/agent/subscribers/run_debriefer.py | 17 +- .../api/src/cora/run/aggregates/run/events.py | 13 ++ .../agent/test_caution_drafter_subscriber.py | 85 +++++++++ .../agent/test_run_debriefer_subscriber.py | 12 +- .../tests/unit/agent/test_subscriber_lease.py | 180 +++++++++++++++++- docs/architecture/modules/run/index.md | 4 +- docs/reference/glossary.md | 2 +- 9 files changed, 391 insertions(+), 92 deletions(-) diff --git a/apps/api/src/cora/agent/_subscriber_lease.py b/apps/api/src/cora/agent/_subscriber_lease.py index ec7d95c8778..9a5f937be7f 100644 --- a/apps/api/src/cora/agent/_subscriber_lease.py +++ b/apps/api/src/cora/agent/_subscriber_lease.py @@ -4,7 +4,13 @@ side-effecting subscribers (`RunDebrieferSubscriber`, `CautionDrafterSubscriber`, future agents) coordinate against concurrent same-Run debriefs by appending a `DecisionDebriefRequested` -lease marker to the Run stream BEFORE invoking the LLM. The append +lease marker to the Run stream BEFORE invoking the LLM. Contention is +scoped by agent KIND: only agents doing the same job compete (the +design memo's multi-agent pollution scenario is rainbow-deploy +variants of ONE debriefer, different LLM backends of ONE drafter). +Distinct kinds hold independent leases for the same terminal event; +RunDebriefer and CautionDrafter both act on every terminal Run event +and both write their own Decisions. The append uses the Run aggregate's current `expected_version`; first writer wins via the existing `UNIQUE(stream_type, stream_id, version)` optimistic-concurrency primitive on the events table. Losing @@ -40,6 +46,13 @@ _RUN_STREAM_TYPE = "Run" _LEASE_EVENT_TYPE = DecisionDebriefRequested.__name__ +_LEASE_APPEND_MAX_RETRIES = 3 +"""Bounded retries for the load-scan-append loop. A retry fires when +the stream version advanced under us for a reason that is NOT a +same-kind competitor: another kind's lease landing (the common case +now that kinds are independent) or a late-arriving Run-lifecycle +event. Beyond the bound the caller sees the degenerate +`(False, None)` loss and exits without a winner.""" def derive_lease_event_id( @@ -63,15 +76,21 @@ def derive_lease_event_id( def _find_lease_winner_for_terminal_event( stored_events: list[StoredEvent], terminal_event_id: UUID, + debriefer_kind: str, ) -> UUID | None: """Scan stored Run events for the lease winner for a terminal event. - Returns the `debriefer_agent_id` of the first matching - `DecisionDebriefRequested` event whose payload's - `terminal_event_id` matches; None if no such event exists. + Returns the `debriefer_agent_id` of the first + `DecisionDebriefRequested` event whose payload matches BOTH the + `terminal_event_id` and the caller's `debriefer_kind`; None if no + such event exists. Markers of a different kind are invisible here: + RunDebriefer and CautionDrafter do different jobs and must not + push each other into Conflicted decisions. A marker with no + recorded kind (pre-scoping legacy) matches every kind, preserving + the old all-contend semantics for historical streams. First-match-from-head: in a well-formed stream, at most one lease - per terminal event is ever appended successfully (the rest hit - ConcurrencyError and never persist). + per (terminal event, kind) is ever appended successfully (the rest + hit ConcurrencyError and never persist). """ target = str(terminal_event_id) for event in stored_events: @@ -79,6 +98,9 @@ def _find_lease_winner_for_terminal_event( continue if event.payload.get("terminal_event_id") != target: continue + marker_kind = event.payload.get("debriefer_kind") + if marker_kind is not None and marker_kind != debriefer_kind: + continue winner_raw = event.payload.get("debriefer_agent_id") if winner_raw is None: continue @@ -91,40 +113,37 @@ async def attempt_debrief_lease( *, run_id: UUID, debriefer_agent_id: UUID, + debriefer_kind: str, terminal_event: StoredEvent, occurred_at: datetime, command_name: str, ) -> tuple[bool, UUID | None]: - """Try to acquire the debrief lease for one Run terminal event. + """Try to acquire this kind's debrief lease for one Run terminal event. Returns: - `(True, None)` when the lease was acquired (this agent should proceed to the LLM call + write its substantive Decision). - - `(False, winner_agent_id)` when another agent already owns - the lease for this terminal event (this agent should write a - `DebriefConflicted` Decision and exit without an LLM call). + - `(False, winner_agent_id)` when another agent OF THE SAME KIND + already owns the lease for this terminal event (this agent + should write a `DebriefConflicted` Decision and exit without + an LLM call). Another kind's lease never surfaces here. `winner_agent_id` is None only on the rare case where the Run - stream version advanced between load and append for a reason - other than a competing lease (e.g., a late-arriving - `RunAddedToCampaign` event); the subscriber treats this as a - loss and exits without identifying a winner. + stream version keeps advancing for non-competing reasons + (other kinds' leases, a late-arriving `RunAddedToCampaign`) + past the retry bound; the subscriber treats this as a loss + and exits without identifying a winner. - Algorithm: + Algorithm, per attempt (bounded by `_LEASE_APPEND_MAX_RETRIES`): 1. Re-load Run stream (per Operation BC lazy-open pattern; the subscriber's earlier `load_run` snapshot may be stale). - 2. If a lease event for this `terminal_event_id` is already on + 2. If a same-kind lease event for this `terminal_event_id` is on the stream, return same-agent / cross-agent outcome based on the recorded `debriefer_agent_id`. 3. Otherwise attempt to append the lease with the freshly-read - `expected_version`. On `ConcurrencyError`, re-scan to - identify the winner. - - Single-attempt: the race is one-shot in practice because terminal - Run events are themselves terminal (no further Run-lifecycle - events follow in normal operation). The only retry case worth - handling would be a late-arriving non-lease Run event landing - between this helper's load + append, which is rare and surfaces - as `(False, None)` for the caller to log + exit. + `expected_version`. On `ConcurrencyError`, loop: the re-load + + re-scan either finds the same-kind winner (resolve) or sees + the version bump came from a non-competitor (another kind's + lease is the common case) and retries the append. The lease event carries `causation_id = terminal_event.event_id` so the Run stream's audit trail links the lease back to the @@ -132,14 +151,14 @@ async def attempt_debrief_lease( `debriefer_agent_id` (the Agent's identity-shared Actor.id). Emits structured log lines at every terminal branch: - - `lease.acquired` on clean acquisition (append succeeded) - - `lease.same_agent_replay` when initial scan returned own lease + - `lease.acquired` on acquisition (append succeeded) + - `lease.same_agent_replay` when the scan returned own lease (crash-recovery / projection re-fire path) - `lease.lost` on any loss path; `winning_agent_id` field is - the foreign agent id, or null when the version advanced for - a non-lease reason (the degenerate `(False, None)` path). - - `lease.conflict_on_append` precedes `lease.acquired` / - `lease.lost` when the append raced and we re-scanned. + the foreign same-kind agent id, or null on retry exhaustion + (the degenerate `(False, None)` path). + - `lease.conflict_on_append` precedes each retry when the + append raced and the loop re-loads + re-scans. The correlation_id is sourced from the terminal Run event so the lease lines join with the subscriber's `.start` / @@ -155,30 +174,17 @@ async def attempt_debrief_lease( run_id=str(run_id), terminal_event_id=str(terminal_event.event_id), debriefer_agent_id=str(debriefer_agent_id), + debriefer_kind=debriefer_kind, correlation_id=str(terminal_event.correlation_id), command_name=command_name, ) - stored, current_version = await event_store.load(_RUN_STREAM_TYPE, run_id) - - winner = _find_lease_winner_for_terminal_event(stored, terminal_event.event_id) - if winner is not None: - if winner == debriefer_agent_id: - log.info("lease.same_agent_replay", current_version=current_version) - return True, None - log.info( - "lease.lost", - winning_agent_id=str(winner), - decided_by="initial_scan", - current_version=current_version, - ) - return False, winner - lease_envelope = to_new_event( event_type=_LEASE_EVENT_TYPE, payload={ "run_id": str(run_id), "debriefer_agent_id": str(debriefer_agent_id), + "debriefer_kind": debriefer_kind, "terminal_event_id": str(terminal_event.event_id), "occurred_at": occurred_at.isoformat(), }, @@ -190,32 +196,47 @@ async def attempt_debrief_lease( principal_id=debriefer_agent_id, ) - try: - await event_store.append( - stream_type=_RUN_STREAM_TYPE, - stream_id=run_id, - expected_version=current_version, - events=[lease_envelope], + attempted_version: int | None = None + for _attempt in range(_LEASE_APPEND_MAX_RETRIES): + stored, current_version = await event_store.load(_RUN_STREAM_TYPE, run_id) + + winner = _find_lease_winner_for_terminal_event( + stored, terminal_event.event_id, debriefer_kind ) - except ConcurrencyError: - log.info("lease.conflict_on_append", attempted_version=current_version) - stored, post_version = await event_store.load(_RUN_STREAM_TYPE, run_id) - winner = _find_lease_winner_for_terminal_event(stored, terminal_event.event_id) - if winner == debriefer_agent_id: + if winner is not None: + if winner == debriefer_agent_id: + log.info("lease.same_agent_replay", current_version=current_version) + return True, None log.info( - "lease.acquired", decided_by="post_conflict_rescan", current_version=post_version + "lease.lost", + winning_agent_id=str(winner), + decided_by="scan", + current_version=current_version, ) - return True, None - log.info( - "lease.lost", - winning_agent_id=str(winner) if winner is not None else None, - decided_by="post_conflict_rescan", - current_version=post_version, - ) - return False, winner + return False, winner + + attempted_version = current_version + try: + await event_store.append( + stream_type=_RUN_STREAM_TYPE, + stream_id=run_id, + expected_version=current_version, + events=[lease_envelope], + ) + except ConcurrencyError: + log.info("lease.conflict_on_append", attempted_version=current_version) + continue - log.info("lease.acquired", decided_by="append", current_version=current_version + 1) - return True, None + log.info("lease.acquired", decided_by="append", current_version=current_version + 1) + return True, None + + log.info( + "lease.lost", + winning_agent_id=None, + decided_by="retry_exhaustion", + attempted_version=attempted_version, + ) + return False, None __all__ = ["attempt_debrief_lease", "derive_lease_event_id"] diff --git a/apps/api/src/cora/agent/subscribers/caution_drafter.py b/apps/api/src/cora/agent/subscribers/caution_drafter.py index c06f743d020..2ef39c51fea 100644 --- a/apps/api/src/cora/agent/subscribers/caution_drafter.py +++ b/apps/api/src/cora/agent/subscribers/caution_drafter.py @@ -92,6 +92,7 @@ ) from cora.agent.seed_caution_drafter import ( CAUTION_DRAFTER_AGENT_ID, + CAUTION_DRAFTER_AGENT_KIND, CAUTION_DRAFTER_AGENT_NAME, ) from cora.agent.subscribers._terminal_run_helpers import ( @@ -290,18 +291,18 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return - # Cross-agent lease (per [[project-run-debriefer-lease-design]]; + # Same-kind lease (per [[project-run-debriefer-lease-design]]; # mirrors RunDebriefer's gate verbatim). Reserved for future # multi-instance CautionDrafter variants (e.g., different LLM - # backends). The lease event_id includes CAUTION_DRAFTER_AGENT_ID - # in the uuid5 seed so the marker is independent of any - # RunDebriefer lease for the same terminal event; the two - # agents are intentionally distinct and both write their own - # Decisions. + # backends). Contention is scoped by debriefer_kind, so this + # lease is independent of any RunDebriefer lease for the same + # terminal event: the two agents do different jobs and both + # write their own Decisions. lease_acquired, winning_agent_id = await attempt_debrief_lease( self.event_store, run_id=run_id, debriefer_agent_id=CAUTION_DRAFTER_AGENT_ID, + debriefer_kind=CAUTION_DRAFTER_AGENT_KIND, terminal_event=event, occurred_at=event.occurred_at, command_name=_COMMAND_NAME, diff --git a/apps/api/src/cora/agent/subscribers/run_debriefer.py b/apps/api/src/cora/agent/subscribers/run_debriefer.py index 9ec53c86c1b..e0beb4e31ae 100644 --- a/apps/api/src/cora/agent/subscribers/run_debriefer.py +++ b/apps/api/src/cora/agent/subscribers/run_debriefer.py @@ -147,7 +147,11 @@ RunDebriefPayload, build_run_debrief_chat_request, ) -from cora.agent.seed import RUN_DEBRIEFER_AGENT_ID, RUN_DEBRIEFER_AGENT_NAME +from cora.agent.seed import ( + RUN_DEBRIEFER_AGENT_ID, + RUN_DEBRIEFER_AGENT_KIND, + RUN_DEBRIEFER_AGENT_NAME, +) from cora.agent.subscribers._terminal_run_helpers import ( extract_interrupted_at, extract_reason, @@ -396,17 +400,20 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return - # Cross-agent lease (per [[project-run-debriefer-lease-design]]): + # Same-kind lease (per [[project-run-debriefer-lease-design]]): # append a DecisionDebriefRequested marker to the Run stream # BEFORE the LLM call so a losing agent pays zero LLM cost. The # first writer wins via the existing UNIQUE(stream_type, - # stream_id, version) primitive; losing agents see the winner - # via the helper's re-load + emit a DebriefConflicted Decision - # on their own Decision stream for audit visibility. + # stream_id, version) primitive; a losing RunDebriefer VARIANT + # (rainbow-deploy sibling sharing this kind) sees the winner + # via the helper's re-load + emits a DebriefConflicted Decision + # on its own Decision stream for audit visibility. Other kinds + # (CautionDrafter) hold independent leases and never contend. lease_acquired, winning_agent_id = await attempt_debrief_lease( self.event_store, run_id=run_id, debriefer_agent_id=RUN_DEBRIEFER_AGENT_ID, + debriefer_kind=RUN_DEBRIEFER_AGENT_KIND, terminal_event=event, occurred_at=event.occurred_at, command_name=_COMMAND_NAME, diff --git a/apps/api/src/cora/run/aggregates/run/events.py b/apps/api/src/cora/run/aggregates/run/events.py index 99d43aed5cd..885ec4b61f2 100644 --- a/apps/api/src/cora/run/aggregates/run/events.py +++ b/apps/api/src/cora/run/aggregates/run/events.py @@ -292,6 +292,15 @@ class DecisionDebriefRequested: same agent's retries are idempotent (re-append fails on event_id UNIQUE) but different agents compete on stream version. + Contention is scoped by `debriefer_kind` (the Agent aggregate's + `kind`): only agents doing the SAME JOB compete (rainbow-deploy + variants of one debriefer, per the design memo's multi-agent + pollution scenario). Distinct kinds (RunDebriefer vs + CautionDrafter) hold independent leases for the same terminal + event and both write their own Decisions. `None` only on + pre-scoping markers, which contend with every kind (legacy + wildcard). + Audit-only on the Run aggregate: the evolver returns prior state unchanged. The lease's existence on the stream IS the lease. See [[project-run-debriefer-lease-design]] for the full design and @@ -303,6 +312,7 @@ class DecisionDebriefRequested: debriefer_agent_id: UUID terminal_event_id: UUID occurred_at: datetime + debriefer_kind: str | None = None @dataclass(frozen=True) @@ -833,12 +843,14 @@ def to_payload(event: RunEvent) -> dict[str, Any]: debriefer_agent_id=debriefer_agent_id, terminal_event_id=terminal_event_id, occurred_at=occurred_at, + debriefer_kind=debriefer_kind, ): return { "run_id": str(run_id), "debriefer_agent_id": str(debriefer_agent_id), "terminal_event_id": str(terminal_event_id), "occurred_at": occurred_at.isoformat(), + "debriefer_kind": debriefer_kind, } case _: # pragma: no cover # exhaustiveness guard assert_never(event) @@ -1072,6 +1084,7 @@ def _build_run_adjusted() -> RunAdjusted: debriefer_agent_id=UUID(payload["debriefer_agent_id"]), terminal_event_id=UUID(payload["terminal_event_id"]), occurred_at=datetime.fromisoformat(payload["occurred_at"]), + debriefer_kind=payload.get("debriefer_kind"), ), ) case _: 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 2ab2abd5730..7df49d7b79c 100644 --- a/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py +++ b/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py @@ -26,6 +26,7 @@ from cora.access.aggregates.actor import to_payload as actor_to_payload from cora.agent.seed_caution_drafter import ( CAUTION_DRAFTER_AGENT_ID, + CAUTION_DRAFTER_AGENT_KIND, CAUTION_DRAFTER_AGENT_NAME, ) from cora.agent.subscribers.caution_drafter import ( @@ -1238,6 +1239,7 @@ async def test_apply_writes_caution_draft_conflicted_when_another_agent_holds_le pre_acquired, _ = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=CAUTION_DRAFTER_AGENT_KIND, debriefer_agent_id=foreign_agent_id, terminal_event=event, occurred_at=event.occurred_at, @@ -1411,3 +1413,86 @@ async def test_apply_inference_recorder_failure_does_not_break_decision() -> Non decision = await load_decision(store, _derive_decision_id(event.event_id)) assert decision is not None assert decision.choice.value == "ProposeCaution" + + +# --------------------------------------------------------------------------- +# Coexistence with RunDebriefer on the same terminal event +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +async def test_apply_coexists_with_run_debriefer_on_same_terminal_event() -> None: + """The two agents do DIFFERENT jobs (debrief vs caution proposal), + so both must act on the same terminal Run event: the debrief lease + scopes contention to agents of the same KIND, and a RunDebriefer + lease must not push the CautionDrafter into a Conflicted decision + (nor vice versa).""" + from cora.agent.seed import RUN_DEBRIEFER_AGENT_ID + from cora.agent.subscribers.run_debriefer import ( + RunDebrieferSubscriber, + ) + from cora.agent.subscribers.run_debriefer import ( + _derive_decision_id as derive_debrief_decision_id, + ) + + store = InMemoryEventStore() + await _seed_caution_drafter_actor(store) + debrief_actor = ActorRegistered( + actor_id=RUN_DEBRIEFER_AGENT_ID, + occurred_at=_NOW, + kind=ActorKind.AGENT, + ) + await store.append( + stream_type="Actor", + stream_id=RUN_DEBRIEFER_AGENT_ID, + expected_version=0, + events=[ + to_new_event( + event_type=actor_event_type_name(debrief_actor), + payload=actor_to_payload(debrief_actor), + occurred_at=_NOW, + event_id=uuid4(), + command_name="SeedTestAgent", + correlation_id=_CORRELATION_ID, + causation_id=None, + principal_id=_PRINCIPAL_ID, + ) + ], + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + + canned_debrief = FakeLLMResponse( + parsed={ + "choice": "NominalCompletion", + "confidence": 0.92, + "reasoning": ( + "Synopsis: nominal single-Plan Run, completed cleanly. " + "What was supposed to happen matches what happened; no " + "parameter adjustments; nominal execution throughout." + ), + }, + stop_reason="tool_use", + model_id="claude-haiku-4-5", + ) + debriefer = RunDebrieferSubscriber( + event_store=store, + llm=FakeLLM(responses=[canned_debrief]), + logbook_mirror=None, + ) + caution_llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + caution = await _build_subscriber(store, caution_llm) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await debriefer.apply(event, conn=None) + await caution.apply(event, conn=None) + + debrief_decision = await load_decision(store, derive_debrief_decision_id(event.event_id)) + assert debrief_decision is not None + assert debrief_decision.choice.value == "NominalCompletion" + + caution_decision = await load_decision(store, _derive_decision_id(event.event_id)) + assert caution_decision is not None + assert caution_decision.choice.value == "NoAction" + assert len(caution_llm.received) == 1 diff --git a/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py b/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py index 4238767ca2d..9c5c2252c6b 100644 --- a/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py +++ b/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py @@ -26,6 +26,7 @@ from cora.agent.aggregates.agent import to_payload as agent_to_payload from cora.agent.seed import ( RUN_DEBRIEFER_AGENT_ID, + RUN_DEBRIEFER_AGENT_KIND, RUN_DEBRIEFER_AGENT_NAME, ) from cora.agent.subscribers._terminal_run_helpers import ( @@ -1383,10 +1384,11 @@ async def test_apply_appends_lease_event_to_run_stream_on_happy_path() -> None: @pytest.mark.unit async def test_apply_writes_debrief_conflicted_when_another_agent_holds_lease() -> None: - """When a different agent already holds the lease (lease event on - Run stream by a foreign debriefer_agent_id), the subscriber writes - DebriefConflicted on its own Decision stream WITHOUT invoking the - LLM. Per the design memo: losing agents pay zero LLM cost.""" + """When a different agent OF THE SAME KIND (a rainbow-deploy + RunDebriefer variant) already holds the lease, the subscriber + writes DebriefConflicted on its own Decision stream WITHOUT + invoking the LLM. Per the design memo: losing agents pay zero + LLM cost.""" from cora.agent._subscriber_lease import attempt_debrief_lease store = InMemoryEventStore() @@ -1401,6 +1403,7 @@ async def test_apply_writes_debrief_conflicted_when_another_agent_holds_lease() pre_acquired, _ = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=RUN_DEBRIEFER_AGENT_KIND, debriefer_agent_id=foreign_agent_id, terminal_event=event, occurred_at=event.occurred_at, @@ -1445,6 +1448,7 @@ async def test_apply_after_prior_lease_by_same_agent_proceeds_to_llm_and_writes_ pre_acquired, _ = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=RUN_DEBRIEFER_AGENT_KIND, debriefer_agent_id=RUN_DEBRIEFER_AGENT_ID, terminal_event=event, occurred_at=event.occurred_at, diff --git a/apps/api/tests/unit/agent/test_subscriber_lease.py b/apps/api/tests/unit/agent/test_subscriber_lease.py index afd147ac8b3..9e67e147522 100644 --- a/apps/api/tests/unit/agent/test_subscriber_lease.py +++ b/apps/api/tests/unit/agent/test_subscriber_lease.py @@ -32,6 +32,7 @@ _NOW = datetime(2026, 6, 4, 12, 0, 0, tzinfo=UTC) _COMMAND_NAME = "TestSubscriber" +_KIND = "RunDebriefer" def _terminal_event( @@ -152,6 +153,7 @@ async def test_attempt_debrief_lease_on_clean_stream_acquires_lease() -> None: success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -181,6 +183,7 @@ async def test_attempt_debrief_lease_prior_same_agent_lease_returns_success_idem await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -191,6 +194,7 @@ async def test_attempt_debrief_lease_prior_same_agent_lease_returns_success_idem success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -216,6 +220,7 @@ async def test_attempt_debrief_lease_with_prior_different_agent_lease_returns_wi await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=winner_id, terminal_event=terminal, occurred_at=_NOW, @@ -225,6 +230,7 @@ async def test_attempt_debrief_lease_with_prior_different_agent_lease_returns_wi success, observed_winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=loser_id, terminal_event=terminal, occurred_at=_NOW, @@ -253,6 +259,7 @@ async def test_attempt_debrief_lease_independent_terminal_events_both_acquire() a_success, _ = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal_a, occurred_at=_NOW, @@ -261,6 +268,7 @@ async def test_attempt_debrief_lease_independent_terminal_events_both_acquire() b_success, _ = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal_b, occurred_at=_NOW, @@ -313,6 +321,7 @@ async def test_attempt_debrief_lease_with_late_run_event_advancing_version_acqui success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -397,6 +406,7 @@ def _foreign_lease_envelope( run_id: UUID, debriefer_agent_id: UUID, terminal_event_id: UUID, + debriefer_kind: str = _KIND, ) -> NewEvent: """Build a `DecisionDebriefRequested` envelope as if a peer subscriber wrote it.""" return to_new_event( @@ -404,6 +414,7 @@ def _foreign_lease_envelope( payload={ "run_id": str(run_id), "debriefer_agent_id": str(debriefer_agent_id), + "debriefer_kind": debriefer_kind, "terminal_event_id": str(terminal_event_id), "occurred_at": _NOW.isoformat(), }, @@ -445,6 +456,7 @@ async def test_attempt_debrief_lease_concurrency_error_with_foreign_winner_retur success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=losing_id, terminal_event=terminal, occurred_at=_NOW, @@ -480,6 +492,7 @@ async def test_attempt_debrief_lease_concurrency_error_with_own_lease_returns_su success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -491,11 +504,11 @@ async def test_attempt_debrief_lease_concurrency_error_with_own_lease_returns_su @pytest.mark.unit -async def test_attempt_debrief_lease_concurrency_error_with_no_lease_winner_returns_none() -> None: - """Append raises `ConcurrencyError`; re-scan finds the version - advanced by a NON-lease event (e.g., `RunAddedToCampaign`); helper - returns `(False, None)` so the caller writes a `DebriefConflicted` - Decision with `winning_agent_id` unidentified.""" +async def test_attempt_debrief_lease_retries_past_non_competing_version_bump() -> None: + """Append raises `ConcurrencyError` because a NON-lease event + (e.g., `RunAddedToCampaign`) advanced the version; the bounded + retry loop re-loads and acquires on the next attempt instead of + giving up (a non-competitor is not a loss).""" from cora.run.aggregates.run.events import RunAddedToCampaign base = InMemoryEventStore() @@ -524,16 +537,165 @@ async def test_attempt_debrief_lease_concurrency_error_with_no_lease_winner_retu success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, command_name=_COMMAND_NAME, ) + assert success is True + assert winner is None + stored, _v = await store.load("Run", run_id) + lease_events = [e for e in stored if e.event_type == "DecisionDebriefRequested"] + assert len(lease_events) == 1 + + +class _AlwaysConflictingStore: + """Store whose every `append` raises `ConcurrencyError` (a stream + hot with non-competing writers), driving the retry loop to + exhaustion.""" + + def __init__(self, delegate: InMemoryEventStore) -> None: + self._delegate = delegate + + async def load(self, stream_type: str, stream_id: UUID) -> tuple[list[StoredEvent], int]: + return await self._delegate.load(stream_type, stream_id) + + async def append( + self, + stream_type: str, + stream_id: UUID, + expected_version: int, + events: Sequence[NewEvent], + ) -> int: + raise ConcurrencyError( + stream_type=stream_type, + stream_id=stream_id, + expected=expected_version, + actual=expected_version + 1, + ) + + async def append_streams( + self, + streams: Sequence[StreamAppend], + *, + conn: object | None = None, + ) -> dict[UUID, int]: + return await self._delegate.append_streams(streams, conn=conn) + + +@pytest.mark.unit +async def test_attempt_debrief_lease_returns_degenerate_loss_on_retry_exhaustion() -> None: + """Every append attempt conflicts without a same-kind winner ever + appearing; the bounded loop gives up with `(False, None)` so the + caller exits without an LLM call or an identified winner.""" + base = InMemoryEventStore() + run_id = uuid4() + await _seed_run(base, run_id) + terminal = _terminal_event(run_id=run_id) + + success, winner = await attempt_debrief_lease( + _AlwaysConflictingStore(base), + run_id=run_id, + debriefer_kind=_KIND, + debriefer_agent_id=uuid4(), + terminal_event=terminal, + occurred_at=_NOW, + command_name=_COMMAND_NAME, + ) + assert success is False assert winner is None +@pytest.mark.unit +async def test_attempt_debrief_lease_different_kinds_both_acquire_same_terminal_event() -> None: + """Kind scoping: agents doing DIFFERENT jobs (RunDebriefer vs + CautionDrafter) hold independent leases for the same terminal + event; the second kind acquires instead of losing to the first.""" + store = InMemoryEventStore() + run_id = uuid4() + await _seed_run(store, run_id) + terminal = _terminal_event(run_id=run_id) + + first_success, _ = await attempt_debrief_lease( + store, + run_id=run_id, + debriefer_kind="RunDebriefer", + debriefer_agent_id=uuid4(), + terminal_event=terminal, + occurred_at=_NOW, + command_name=_COMMAND_NAME, + ) + second_success, second_winner = await attempt_debrief_lease( + store, + run_id=run_id, + debriefer_kind="CautionDrafter", + debriefer_agent_id=uuid4(), + terminal_event=terminal, + occurred_at=_NOW, + command_name=_COMMAND_NAME, + ) + + assert first_success is True + assert second_success is True + assert second_winner is None + stored, _v = await store.load("Run", run_id) + lease_events = [e for e in stored if e.event_type == "DecisionDebriefRequested"] + assert len(lease_events) == 2 + + +@pytest.mark.unit +async def test_attempt_debrief_lease_legacy_kindless_marker_contends_with_every_kind() -> None: + """A pre-scoping lease marker (payload without `debriefer_kind`) + keeps the old all-contend semantics: any kind scanning the stream + sees it as the winner, so historical streams replay exactly as + they did before scoping.""" + store = InMemoryEventStore() + run_id = uuid4() + await _seed_run(store, run_id) + terminal = _terminal_event(run_id=run_id) + legacy_winner_id = uuid4() + + legacy_payload = { + "run_id": str(run_id), + "debriefer_agent_id": str(legacy_winner_id), + "terminal_event_id": str(terminal.event_id), + "occurred_at": _NOW.isoformat(), + } + await store.append( + stream_type="Run", + stream_id=run_id, + expected_version=1, + events=[ + to_new_event( + event_type="DecisionDebriefRequested", + payload=legacy_payload, + occurred_at=_NOW, + event_id=uuid4(), + command_name="LegacyAgent", + correlation_id=uuid4(), + causation_id=None, + principal_id=legacy_winner_id, + ) + ], + ) + + success, winner = await attempt_debrief_lease( + store, + run_id=run_id, + debriefer_kind="CautionDrafter", + debriefer_agent_id=uuid4(), + terminal_event=terminal, + occurred_at=_NOW, + command_name=_COMMAND_NAME, + ) + + assert success is False + assert winner == legacy_winner_id + + @pytest.mark.unit async def test_attempt_debrief_lease_skips_malformed_lease_payload_missing_winner_field() -> None: """A `DecisionDebriefRequested` event whose payload omits @@ -571,6 +733,7 @@ async def test_attempt_debrief_lease_skips_malformed_lease_payload_missing_winne success, winner = await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -603,6 +766,7 @@ async def test_attempt_debrief_lease_emits_acquired_log_on_clean_append() -> Non await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -633,6 +797,7 @@ async def test_attempt_debrief_lease_emits_same_agent_replay_log_on_own_prior_le await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -643,6 +808,7 @@ async def test_attempt_debrief_lease_emits_same_agent_replay_log_on_own_prior_le await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=agent_id, terminal_event=terminal, occurred_at=_NOW, @@ -670,6 +836,7 @@ async def test_attempt_debrief_lease_emits_lost_log_with_winner_on_cross_agent_l await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=winner_id, terminal_event=terminal, occurred_at=_NOW, @@ -680,6 +847,7 @@ async def test_attempt_debrief_lease_emits_lost_log_with_winner_on_cross_agent_l await attempt_debrief_lease( store, run_id=run_id, + debriefer_kind=_KIND, debriefer_agent_id=loser_id, terminal_event=terminal, occurred_at=_NOW, @@ -689,5 +857,5 @@ async def test_attempt_debrief_lease_emits_lost_log_with_winner_on_cross_agent_l lost = [e for e in logs if e.get("event") == "lease.lost"] assert len(lost) == 1 assert lost[0]["winning_agent_id"] == str(winner_id) - assert lost[0]["decided_by"] == "initial_scan" + assert lost[0]["decided_by"] == "scan" assert lost[0]["debriefer_agent_id"] == str(loser_id) diff --git a/docs/architecture/modules/run/index.md b/docs/architecture/modules/run/index.md index 310c6bbeca7..62f956a556a 100644 --- a/docs/architecture/modules/run/index.md +++ b/docs/architecture/modules/run/index.md @@ -111,7 +111,7 @@ stateDiagram-v2 | `RunObservationLogbookOpened` | `run_id`, `logbook_id`, `schema`, `occurred_at` | `append_observations` first write per Run (lazy open) | | `RunAddedToCampaign` | `run_id`, `campaign_id`, `occurred_at` | post-hoc Campaign membership write (see Campaign module) | | `RunRemovedFromCampaign` | `run_id`, `campaign_id`, `occurred_at` | post-hoc Campaign membership removal | -| `DecisionDebriefRequested` | `run_id`, `debriefer_agent_id`, `terminal_event_id`, `occurred_at` | appended by an Agent BC subscriber (RunDebriefer / CautionDrafter) before invoking its LLM; a per-(run, terminal-event, agent) lease so concurrent subscribers do not double-invoke; first write wins, audit-only | +| `DecisionDebriefRequested` | `run_id`, `debriefer_agent_id`, `debriefer_kind`, `terminal_event_id`, `occurred_at` | appended by an Agent BC subscriber (RunDebriefer / CautionDrafter) before invoking its LLM; a per-(run, terminal-event, kind) lease so same-kind subscribers do not double-invoke; distinct kinds hold independent leases; first same-kind write wins, audit-only | Individual reading rows do not emit per-row events on the Run stream; they are written directly to `entries_run_observations` via the `ObservationStore` port. The row's `event_id`, `correlation_id`, and `causation_id` constitute the audit trail without bloating the main event log. @@ -224,7 +224,7 @@ Clock skew between the sensor (`sampled_at`) and the handler (`occurred_at`) is | Decision | shared-id-with | `RunAdjusted.decided_by_decision_id` cites the Decision that justified a mid-flight adjustment; no existence check at write time (eventual-consistency stance) | | Calibration | reads-from | `Run.pinned_calibration_ids` is a frozen set of `CalibrationRevision.id`s captured at `start_run` and **immutable** for the life of the Run; every FSM transition preserves the set verbatim, and downstream consumers cite this set to answer "what calibration was this scan acquired against?" deterministically | | Agent | writes-to (Decision stream) | Terminal Run events (`RunCompleted`, `RunAborted`, `RunStopped`, `RunTruncated`) are subscribed by the RunDebriefer and CautionDrafter agents, each of which emits an advisory `Decision` per terminal Run | -| Agent | writes-to (Run stream, lease marker) | Each terminal-Run-event subscriber appends `DecisionDebriefRequested` to the Run stream BEFORE invoking its LLM as a per-(run, terminal-event, agent) lease primitive; first writer wins via the existing optimistic-concurrency constraint, losing agents emit a `DebriefConflicted` / `CautionDraftConflicted` audit Decision on their own Decision stream with zero LLM cost (no Run-state mutation) | +| Agent | writes-to (Run stream, lease marker) | Each terminal-Run-event subscriber appends `DecisionDebriefRequested` to the Run stream BEFORE invoking its LLM as a per-(run, terminal-event, kind) lease primitive; the first same-kind writer wins via the existing optimistic-concurrency constraint, a losing same-kind variant emits a `DebriefConflicted` / `CautionDraftConflicted` audit Decision on its own Decision stream with zero LLM cost (no Run-state mutation); distinct kinds never contend | | Agent | written-by (Run stream) | the deterministic RunSupervisor agent issues `hold_run` / `resume_run` / `stop_run` on an in-flight Run as facility beam is lost and returns, through the authorized command path; the resulting `RunHeld` / `RunResumed` / `RunStopped` carries `decided_by_decision_id` linking the agent's `RunSupervision` Decision and is byte-identical to an operator action. The gated wind-up (`resume_run`) re-checks the full start-safety envelope and only ever resumes a Run the supervisor itself held; it never starts a Run. The supervisor also carries shadow observe-only rules (run-liveness, plus signal-quality and signal-stall over the Run's observation channels) that log a would-flag and take no action. Off by default, and not a safety interlock (the floor PSS owns hard safety) | | Access | shared-id-with | Every Run event envelope carries `actor_id` for principal attribution; cross-module references are bare UUIDs and not verified at write time | diff --git a/docs/reference/glossary.md b/docs/reference/glossary.md index 836325271ac..174986a29ff 100644 --- a/docs/reference/glossary.md +++ b/docs/reference/glossary.md @@ -102,7 +102,7 @@ Watch-only (not adopted as a glossary term, see [Deferred](../stack/deferred.md) - **RunSupervisor.** Deterministic, active. A composition-root periodic loop that watches in-flight Runs and issues `hold_run` / `stop_run` through the authorized command path when facility beam is lost, recording a `RunSupervision` Decision. Wind-down only, off by default; not a safety interlock (the floor PSS owns hard safety). - **CautionPromoter.** Deterministic, active. Subscribes to registered `CautionProposal` Decisions and auto-registers a Caution from a high-confidence, Notice-only proposal (the same `CautionRegistered` write as the operator-initiated `promote_caution_proposal` slice), recording a `CautionPromotion` Decision. Notice-only and off by default; higher severities stay operator-gated. - **ClearanceExpirer.** Deterministic, active. A periodic loop that sweeps Active safety Clearances and issues `expire_clearance` on those whose `valid_until` has passed, recording a `ClearanceExpiry` Decision; byte-identical to an operator-driven expiry. -- **Debrief lease.** *(Agent BC subscriber pattern)* Cross-agent coordination primitive for terminal-Run-event subscribers (RunDebriefer, CautionDrafter, every future agent that LLM-debriefs a Run). Each subscriber appends a `DecisionDebriefRequested` event to the Run stream BEFORE invoking its LLM, using the existing `UNIQUE(stream_type, stream_id, version)` optimistic-concurrency primitive. First writer wins; losers emit a `DebriefConflicted` / `CautionDraftConflicted` audit Decision on their own Decision stream and exit with zero LLM cost. The lease event_id is `uuid5(run_id, f"lease:{terminal_event_id}:{agent_id}")` so per-agent retries are idempotent (same event_id collides) while cross-agent inserts compete on stream version (distinct event_ids, same expected_version). The lease event is audit-only with a no-op Run-evolver fold; it carries no authorization semantic and does not extend `Trust.Policy`. +- **Debrief lease.** *(Agent BC subscriber pattern)* Same-kind coordination primitive for terminal-Run-event subscribers (RunDebriefer, CautionDrafter, every future agent that LLM-debriefs a Run). Each subscriber appends a `DecisionDebriefRequested` event to the Run stream BEFORE invoking its LLM, using the existing `UNIQUE(stream_type, stream_id, version)` optimistic-concurrency primitive. Contention is scoped by the marker's `debriefer_kind` (the Agent aggregate's `kind`): only agents doing the same job compete, so rainbow-deploy variants of one debriefer race each other while distinct kinds hold independent leases for the same terminal event and each writes its own Decision. First same-kind writer wins; losers emit a `DebriefConflicted` / `CautionDraftConflicted` audit Decision on their own Decision stream and exit with zero LLM cost. The lease event_id is `uuid5(run_id, f"lease:{terminal_event_id}:{agent_id}")` so per-agent retries are idempotent (same event_id collides) while same-kind inserts compete on stream version (distinct event_ids, same expected_version). The lease event is audit-only with a no-op Run-evolver fold; it carries no authorization semantic and does not extend `Trust.Policy`. ## Calibration