Scope the debrief lease to agent kind so RunDebriefer and CautionDrafter coexist

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,22 +76,31 @@ 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:
         if event.event_type != _LEASE_EVENT_TYPE:
             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,55 +113,52 @@ 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
     triggering terminal event. `principal_id` is the
     `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 `<subscriber>.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"]

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,

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,

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 _: