From 9de1820c4d8be7e063dc7e2d23e0c3f4a1fe8110 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 20 Aug 2026 17:49:00 -0500 Subject: [PATCH] Let a deployment designate which Agent each LLM subscriber acts as RunDebrieferSubscriber and CautionDrafterSubscriber hardcoded the seeded singleton Agent id, and both seeded Agents declare provider=anthropic. Every LLM adapter refuses a request whose model_ref.provider is not its own, so on a deployment where api.anthropic.com is unreachable (2-BM's controls network has no internet) the autonomous path is structurally dead: LLM_ENABLED=true would defer every completed Run instead of debriefing it. The on-demand regenerate_run_debrief path already lets an operator name the Agent; the subscribers did not. Add run_debriefer_agent_id / caution_drafter_agent_id settings (UUID | None, default None = the seeded singleton, so nothing changes on upgrade). Thread the id through both subscribers as a keyword-only constructor argument so the class stays unit-testable without Settings. Validation lives on the per-apply gate path, gated on designation being explicit: a designated Agent that doesn't exist or is the wrong kind skips with a named reason; the seeded default stays exempt from the existence check the way regenerate_run_debrief already exempts it. caution_drafter.py never served the Agent's declared model (unlike run_debriefer.py, which already did); fixed so designating a CautionDrafter Agent that declares a reachable provider actually changes what gets served. report_designated_agents(deps), called from main.py after both singleton seeds, logs which Agent is effective for each subscriber and warns on a provider mismatch. It is a report, never a gate: the adapter already refuses the call at request time and is the authority, so a second gate here would re-decide a verdict that already exists. The warning exists only because that refusal otherwise arrives per-event, hours later, as a deferred Decision that never names the mismatch. ModelRef stays immutable; no update_agent_model_ref command is added. The approved-model catalog gate is not re-checked at designation (define_agent already checked it). Co-Authored-By: Claude Sonnet 5 --- .env.example | 15 ++ apps/api/src/cora/agent/__init__.py | 3 +- apps/api/src/cora/agent/_subscribers.py | 66 +++++- .../cora/agent/subscribers/caution_drafter.py | 82 +++++-- .../cora/agent/subscribers/run_debriefer.py | 67 ++++-- apps/api/src/cora/api/main.py | 7 + apps/api/src/cora/infrastructure/config.py | 15 ++ apps/api/tests/unit/agent/_helpers.py | 2 + .../test_agent_subscribers_registration.py | 178 ++++++++++++++- .../agent/test_caution_drafter_subscriber.py | 214 +++++++++++++++++- .../agent/test_run_debriefer_subscriber.py | 187 ++++++++++++++- apps/api/tests/unit/test_settings.py | 13 ++ docs/deployments/2-bm/llm_debrief.md | 30 ++- 13 files changed, 826 insertions(+), 53 deletions(-) diff --git a/.env.example b/.env.example index 4922ddb9dd7..e76ec2bad69 100644 --- a/.env.example +++ b/.env.example @@ -182,6 +182,21 @@ LOG_LEVEL=INFO # alone must now also set LLM_ENABLED=true, or those two stop registering. # Nothing crashes; a boot warning names the switch and /readyz reports "off". # +# RUN_DEBRIEFER_AGENT_ID / CAUTION_DRAFTER_AGENT_ID let a deployment +# designate WHICH Agent each LLM subscriber acts as, instead of always +# acting as the seeded singleton. Unset (the default) means the seeded +# singleton, so nothing changes on upgrade. The named Agent must already +# exist: define it first through the gated `POST /agents` (`define_agent`) +# path, and its declared model must be reachable by the configured +# LLM_PROVIDER (`anthropic` / `argo` / `local`, see `Settings.llm_provider`), +# or every call the subscriber makes will be refused by the adapter at +# request time. This matters for a deployment whose network cannot reach +# the seeded agents' declared provider (eg. `anthropic` from a controls +# network with no internet): define an Agent against a reachable provider +# (`argo` or `local`) and name it here. +# RUN_DEBRIEFER_AGENT_ID= +# CAUTION_DRAFTER_AGENT_ID= +# # NOTE this is a DIFFERENT axis from actuation (CONTROL_WRITES_ENABLED and # COMPUTE_SUBSTRATE above). Those bound what CORA can MOVE; these bound the # LLM path. A facility told "read-only" hears the first and would still diff --git a/apps/api/src/cora/agent/__init__.py b/apps/api/src/cora/agent/__init__.py index 9b42eb58994..0f6074f5749 100644 --- a/apps/api/src/cora/agent/__init__.py +++ b/apps/api/src/cora/agent/__init__.py @@ -29,7 +29,7 @@ from cora.agent._pricing_bridge import refresh_language_model_pricing from cora.agent._projections import register_agent_projections -from cora.agent._subscribers import register_agent_subscribers +from cora.agent._subscribers import register_agent_subscribers, report_designated_agents from cora.agent.aggregates.agent import load_agent from cora.agent.build_llm import build_llm from cora.agent.errors import ( @@ -135,6 +135,7 @@ "register_agent_routes", "register_agent_subscribers", "register_agent_tools", + "report_designated_agents", "seed_authority_revocation_holder_agent", "seed_calibration_watcher_agent", "seed_campaign_watcher_agent", diff --git a/apps/api/src/cora/agent/_subscribers.py b/apps/api/src/cora/agent/_subscribers.py index 3a6dcd5b004..365bd764d77 100644 --- a/apps/api/src/cora/agent/_subscribers.py +++ b/apps/api/src/cora/agent/_subscribers.py @@ -62,13 +62,27 @@ the `dismiss_event_in_reaction` slice. Further widening (a separate `ReactionWorker` with its own pool budget) stays deferred behind the next named triggers: 3rd Reaction OR first wedged-bookmark incident. + +## Subscriber agent designation + +`settings.run_debriefer_agent_id` / `settings.caution_drafter_agent_id` +let a deployment designate a non-seeded Agent for either LLM +subscriber to act as (a deployment whose configured `llm_provider` +cannot reach the seeded agents' declared provider defines its own +Agent and names it here). `report_designated_agents(deps)`, called +from `cora.api.main` after both singleton seeds, logs which Agent is +effective for each subscriber and warns on a provider mismatch; it is +a report, not a gate, so it never blocks boot. """ from __future__ import annotations from typing import TYPE_CHECKING +from cora.agent.aggregates.agent import load_agent from cora.agent.build_llm import llm_unwired_reason +from cora.agent.seed import RUN_DEBRIEFER_AGENT_ID +from cora.agent.seed_caution_drafter import CAUTION_DRAFTER_AGENT_ID from cora.agent.subscribers.authority_revocation_holder import ( make_authority_revocation_holder_subscriber, ) @@ -161,4 +175,54 @@ def register_agent_subscribers(registry: ProjectionRegistry, deps: Kernel) -> No ) -__all__ = ["register_agent_subscribers"] +async def report_designated_agents(deps: Kernel) -> None: + """Log which Agent each LLM subscriber will act as, and whether its + declared provider matches `settings.llm_provider`. + + Called from `cora.api.main` AFTER both singleton seeds, so the + default (unset designation) case always resolves an Agent. This is + a REPORT, never a gate: it does not skip, raise, or refuse boot on a + mismatch. The LLM adapter already refuses a request whose + `model_ref.provider` isn't its own and is the authority on whether + the pairing works; a second gate here would re-decide a verdict + that already exists. The warning exists only because the adapter's + refusal arrives per-event, hours later, as a deferred Decision that + never names the mismatch -- an operator reading the boot log should + see it up front instead. + + A designated-but-missing Agent (misconfiguration) logs a warning + and moves on; the subscriber's own per-apply gate is what actually + skips work for that case. + """ + for subscriber_name, agent_id in ( + ("run_debriefer", deps.settings.run_debriefer_agent_id or RUN_DEBRIEFER_AGENT_ID), + ("caution_drafter", deps.settings.caution_drafter_agent_id or CAUTION_DRAFTER_AGENT_ID), + ): + agent = await load_agent(deps.event_store, agent_id) + if agent is None: + _log.warning( + "agent_subscriber.designated_agent_not_found", + subscriber=subscriber_name, + agent_id=str(agent_id), + ) + continue + _log.info( + "agent_subscriber.designated_agent", + subscriber=subscriber_name, + agent_id=str(agent_id), + agent_name=agent.name.value, + agent_kind=agent.kind.value, + provider=agent.model_ref.provider, + model=agent.model_ref.model, + ) + if agent.model_ref.provider != deps.settings.llm_provider: + _log.warning( + "agent_subscriber.designated_agent_provider_mismatch", + subscriber=subscriber_name, + agent_id=str(agent_id), + agent_provider=agent.model_ref.provider, + configured_llm_provider=deps.settings.llm_provider, + ) + + +__all__ = ["register_agent_subscribers", "report_designated_agents"] diff --git a/apps/api/src/cora/agent/subscribers/caution_drafter.py b/apps/api/src/cora/agent/subscribers/caution_drafter.py index 0fc084e8cdc..483d60ce928 100644 --- a/apps/api/src/cora/agent/subscribers/caution_drafter.py +++ b/apps/api/src/cora/agent/subscribers/caution_drafter.py @@ -81,6 +81,7 @@ from cora.access.aggregates.actor import load_actor from cora.agent._budget_gate import find_allocation_breach, find_budget_breach +from cora.agent._model_ref import to_port_model_ref from cora.agent._subscriber_lease import attempt_debrief_lease from cora.agent.aggregates.agent import AgentStatus, load_agent from cora.agent.prompts import ( @@ -90,6 +91,7 @@ ExistingCaution, build_caution_drafter_chat_request, ) +from cora.agent.prompts.caution_drafter import DEFAULT_CAUTION_DRAFTER_MODEL from cora.agent.seed_caution_drafter import ( CAUTION_DRAFTER_AGENT_ID, CAUTION_DRAFTER_AGENT_KIND, @@ -184,9 +186,11 @@ class CautionDrafterSubscriber: satisfies the `Reaction` Protocol structurally. Holds references to the LLM port, event store, and CautionLookup - port. The Decision's `actor_id` is the seeded CautionDrafter - Agent's id (== that agent's Actor.id per 8f-a's identity-sharing - invariant). + port. The Decision's `actor_id` is the CautionDrafter Agent this + subscriber acts as (== that agent's Actor.id per 8f-a's + identity-sharing invariant): the seeded singleton by default, or a + deployment-designated Agent when `settings.caution_drafter_agent_id` + names one (see `_agent_id`). `batch_size = 1` for the same reason as RunDebriefer: the apply path includes a slow LLM round-trip, so holding the bookmark @@ -208,11 +212,17 @@ def __init__( inference_recorder: InferenceRecorder | None = None, spend_lookup: SpendLookup | None = None, allocation_lookup: AllocationLookup | None = None, + agent_id: UUID = CAUTION_DRAFTER_AGENT_ID, ) -> None: self.event_store = event_store self.llm = llm self.caution_lookup = caution_lookup self.signer = signer + # Which Agent this subscriber acts as. Defaults to the seeded + # singleton so the class stays unit-testable without Settings; + # `make_caution_drafter_subscriber` passes the deployment's + # `settings.caution_drafter_agent_id` designation when set. + self._agent_id = agent_id # Defaults to the no-op recorder so direct test construction stays # inert; production wiring passes the Kernel's recorder via # `make_caution_drafter_subscriber`. @@ -267,19 +277,21 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: # Pre-load the Agent's Actor + revocation gate (mirrors # RunDebriefer verbatim). - actor = await load_actor(self.event_store, CAUTION_DRAFTER_AGENT_ID) + actor = await load_actor(self.event_store, self._agent_id) if actor is None: + # No Agent fold to name here (the Actor itself is missing), + # so the log carries the id only -- a bare `agent_name` + # constant would misname a designated Agent under + # designation. log.warning( "caution_drafter.skip.agent_actor_missing", - agent_id=str(CAUTION_DRAFTER_AGENT_ID), - agent_name=CAUTION_DRAFTER_AGENT_NAME, + agent_id=str(self._agent_id), ) return if not actor.active: log.warning( "caution_drafter.skip.agent_actor_deactivated", - agent_id=str(CAUTION_DRAFTER_AGENT_ID), - agent_name=CAUTION_DRAFTER_AGENT_NAME, + agent_id=str(self._agent_id), ) return @@ -287,12 +299,37 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: # acts; Suspended, Deprecated, and not-yet-promoted Defined all # skip. A missing Agent stream stays permissive. The Agent fold # also carries the declared budget the post-lease gate reads. - agent = await load_agent(self.event_store, CAUTION_DRAFTER_AGENT_ID) + agent = await load_agent(self.event_store, self._agent_id) + + # Designation validation (mirrors RunDebriefer). Gated on + # `is_designated` so the seeded default stays exempt from the + # existence check, exactly as `regenerate_run_debrief` exempts it + # on the on-demand path: an explicitly named Agent is a + # deliberate choice and gets checked; the approved-model catalog + # gate is NOT re-checked here (`define_agent` already checked it). + is_designated = self._agent_id != CAUTION_DRAFTER_AGENT_ID + if is_designated: + if agent is None: + log.warning( + "caution_drafter.skip.designated_agent_missing", + agent_id=str(self._agent_id), + ) + return + if agent.kind.value != CAUTION_DRAFTER_AGENT_KIND: + log.warning( + "caution_drafter.skip.designated_agent_wrong_kind", + agent_id=str(self._agent_id), + agent_name=agent.name.value, + expected_kind=CAUTION_DRAFTER_AGENT_KIND, + actual_kind=agent.kind.value, + ) + return + if agent is not None and agent.status is not AgentStatus.VERSIONED: log.warning( "caution_drafter.skip.agent_not_versioned", - agent_id=str(CAUTION_DRAFTER_AGENT_ID), - agent_name=CAUTION_DRAFTER_AGENT_NAME, + agent_id=str(self._agent_id), + agent_name=agent.name.value, agent_status=str(agent.status), ) return @@ -307,7 +344,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: lease_acquired, winning_agent_id = await attempt_debrief_lease( self.event_store, run_id=run_id, - debriefer_agent_id=CAUTION_DRAFTER_AGENT_ID, + debriefer_agent_id=self._agent_id, debriefer_kind=CAUTION_DRAFTER_AGENT_KIND, terminal_event=event, occurred_at=event.occurred_at, @@ -455,7 +492,19 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: candidate_targets=candidate_targets, existing_cautions=existing_cautions, ) - request = build_caution_drafter_chat_request(payload) + # The Agent's declared model, not the module default: that + # declaration is what `define_agent` gated against the approved + # catalog, so serving anything else makes the gate decorative. + # `agent` is None only when the Agent stream was never seeded + # (the branch above tolerates it), and the default stands in. + request = build_caution_drafter_chat_request( + payload, + model_ref=( + to_port_model_ref(agent.model_ref) + if agent is not None + else DEFAULT_CAUTION_DRAFTER_MODEL + ), + ) try: response = await self.llm.chat(request) @@ -500,6 +549,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: await self._record_inference( decision_id=decision_id, actor=actor, + agent_name=agent.name.value if agent is not None else CAUTION_DRAFTER_AGENT_NAME, request=request, response=response, terminal_event=event, @@ -511,6 +561,7 @@ async def _record_inference( *, decision_id: UUID, actor: Actor, + agent_name: str, request: LLMChatRequest, response: LLMResponse, terminal_event: StoredEvent, @@ -540,8 +591,8 @@ async def _record_inference( request_max_tokens=request.max_output_tokens, request_temperature=request.temperature, request_top_p=request.top_p, - agent_id=str(CAUTION_DRAFTER_AGENT_ID), - agent_name=CAUTION_DRAFTER_AGENT_NAME, + agent_id=str(self._agent_id), + agent_name=agent_name, ) try: await self.inference_recorder.record( @@ -913,6 +964,7 @@ def make_caution_drafter_subscriber(deps: Kernel) -> CautionDrafterSubscriber: inference_recorder=deps.inference_recorder, spend_lookup=deps.spend_lookup, allocation_lookup=deps.allocation_lookup, + agent_id=deps.settings.caution_drafter_agent_id or CAUTION_DRAFTER_AGENT_ID, ) diff --git a/apps/api/src/cora/agent/subscribers/run_debriefer.py b/apps/api/src/cora/agent/subscribers/run_debriefer.py index ff4b70e6d40..77ea9b012d7 100644 --- a/apps/api/src/cora/agent/subscribers/run_debriefer.py +++ b/apps/api/src/cora/agent/subscribers/run_debriefer.py @@ -269,8 +269,10 @@ class RunDebrieferSubscriber: it extends) structurally. Holds references to the LLM port and event store. The Decision's - `actor_id` is the seeded RunDebriefer Agent's id (== that agent's - Actor.id per 8f-a's identity-sharing invariant). + `actor_id` is the RunDebriefer Agent this subscriber acts as (== + that agent's Actor.id per 8f-a's identity-sharing invariant): the + seeded singleton by default, or a deployment-designated Agent when + `settings.run_debriefer_agent_id` names one (see `_agent_id`). `name`, `subscribed_event_types`, and `batch_size` are plain class-level constants (matches the wider Subscriber convention; @@ -299,11 +301,17 @@ def __init__( inference_recorder: InferenceRecorder | None = None, spend_lookup: SpendLookup | None = None, allocation_lookup: AllocationLookup | None = None, + agent_id: UUID = RUN_DEBRIEFER_AGENT_ID, ) -> None: self.event_store = event_store self.llm = llm self.logbook_mirror = logbook_mirror self.signer = signer + # Which Agent this subscriber acts as. Defaults to the seeded + # singleton so the class stays unit-testable without Settings; + # `make_run_debriefer_subscriber` passes the deployment's + # `settings.run_debriefer_agent_id` designation when set. + self._agent_id = agent_id # Defaults to the no-op recorder so direct test construction (and any # caller that omits it) stays inert; production wiring passes the # Kernel's recorder via `make_run_debriefer_subscriber`. @@ -360,12 +368,15 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: # `actor_id` to exist in Access BC). If the agent isn't seeded # (bootstrap not yet run, deployment misconfigured), short-circuit # without writing -- the operator needs to fix the seed. - actor = await load_actor(self.event_store, RUN_DEBRIEFER_AGENT_ID) + actor = await load_actor(self.event_store, self._agent_id) if actor is None: + # No Agent fold to name here (the Actor itself is missing), + # so the log carries the id only -- a bare `agent_name` + # constant would misname a designated Agent under + # designation. log.warning( "run_debriefer.skip.agent_actor_missing", - agent_id=str(RUN_DEBRIEFER_AGENT_ID), - agent_name=RUN_DEBRIEFER_AGENT_NAME, + agent_id=str(self._agent_id), ) return @@ -379,8 +390,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: if not actor.active: log.warning( "run_debriefer.skip.agent_actor_deactivated", - agent_id=str(RUN_DEBRIEFER_AGENT_ID), - agent_name=RUN_DEBRIEFER_AGENT_NAME, + agent_id=str(self._agent_id), ) return @@ -393,12 +403,40 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: # behavior for the NEXT terminal event; skipped work items are # not replayed. The Agent fold also carries the declared budget # the post-lease gate below reads. - agent = await load_agent(self.event_store, RUN_DEBRIEFER_AGENT_ID) + agent = await load_agent(self.event_store, self._agent_id) + + # Designation validation. Gated on `is_designated` (an explicit + # deployment setting, not the seeded default) because the seeded + # default is exempt from the existence check the same way + # `regenerate_run_debrief` exempts it: the apply path already + # tolerates an Actor-only legacy deployment, but an explicitly + # named Agent is a deliberate choice and gets checked. The + # approved-model catalog gate is NOT re-checked here: `define_agent` + # already checked it, and a second authority could disagree with + # the first. + is_designated = self._agent_id != RUN_DEBRIEFER_AGENT_ID + if is_designated: + if agent is None: + log.warning( + "run_debriefer.skip.designated_agent_missing", + agent_id=str(self._agent_id), + ) + return + if agent.kind.value != RUN_DEBRIEFER_AGENT_KIND: + log.warning( + "run_debriefer.skip.designated_agent_wrong_kind", + agent_id=str(self._agent_id), + agent_name=agent.name.value, + expected_kind=RUN_DEBRIEFER_AGENT_KIND, + actual_kind=agent.kind.value, + ) + return + if agent is not None and agent.status is not AgentStatus.VERSIONED: log.warning( "run_debriefer.skip.agent_not_versioned", - agent_id=str(RUN_DEBRIEFER_AGENT_ID), - agent_name=RUN_DEBRIEFER_AGENT_NAME, + agent_id=str(self._agent_id), + agent_name=agent.name.value, agent_status=str(agent.status), ) return @@ -415,7 +453,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: lease_acquired, winning_agent_id = await attempt_debrief_lease( self.event_store, run_id=run_id, - debriefer_agent_id=RUN_DEBRIEFER_AGENT_ID, + debriefer_agent_id=self._agent_id, debriefer_kind=RUN_DEBRIEFER_AGENT_KIND, terminal_event=event, occurred_at=event.occurred_at, @@ -585,6 +623,7 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: await self._record_inference( decision_id=decision_id, actor=actor, + agent_name=agent.name.value if agent is not None else RUN_DEBRIEFER_AGENT_NAME, request=request, response=response, terminal_event=event, @@ -613,6 +652,7 @@ async def _record_inference( *, decision_id: UUID, actor: Actor, + agent_name: str, request: LLMChatRequest, response: LLMResponse, terminal_event: StoredEvent, @@ -647,8 +687,8 @@ async def _record_inference( request_max_tokens=request.max_output_tokens, request_temperature=request.temperature, request_top_p=request.top_p, - agent_id=str(RUN_DEBRIEFER_AGENT_ID), - agent_name=RUN_DEBRIEFER_AGENT_NAME, + agent_id=str(self._agent_id), + agent_name=agent_name, ) try: await self.inference_recorder.record( @@ -931,6 +971,7 @@ def make_run_debriefer_subscriber(deps: Kernel) -> RunDebrieferSubscriber: inference_recorder=deps.inference_recorder, spend_lookup=deps.spend_lookup, allocation_lookup=deps.allocation_lookup, + agent_id=deps.settings.run_debriefer_agent_id or RUN_DEBRIEFER_AGENT_ID, ) diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index d355eb94f30..ba6577e38b6 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -58,6 +58,7 @@ register_agent_routes, register_agent_subscribers, register_agent_tools, + report_designated_agents, seed_authority_revocation_holder_agent, seed_calibration_watcher_agent, seed_campaign_watcher_agent, @@ -1054,6 +1055,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: await seed_run_debriefer_agent(deps) # same shape for CautionDrafter. await seed_caution_drafter_agent(deps) + # Report which Agent each LLM subscriber will act as (the + # seeded singleton, or a deployment designation) and warn on + # a provider mismatch. Must run AFTER both seeds above so the + # default (unset designation) case always resolves an Agent; + # a report, never a gate, so it never refuses boot. + await report_designated_agents(deps) # LanguageModel catalog entries for the fleet's three default # models, born Defined AND Approved so the define_agent gate # never refuses the shipped fleet on a fresh deployment. diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index 50970927ab5..e9c43895cee 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -255,6 +255,21 @@ class Settings(BaseSettings): # for the entry, which is what makes the envelope source-agnostic. llm_provider: Literal["anthropic", "argo", "local"] = "anthropic" + # `run_debriefer_agent_id` / `caution_drafter_agent_id` let a deployment + # designate WHICH Agent each LLM subscriber acts as, instead of always + # acting as the seeded singleton (`RUN_DEBRIEFER_AGENT_ID` / + # `CAUTION_DRAFTER_AGENT_ID`). Unset (None, the default) means the seeded + # singleton, so nothing changes on upgrade. The named Agent must already + # exist, defined through the gated `define_agent` path: this setting only + # SELECTS among Agents that already passed that gate, the same way + # `run_initiator_plan_id` below selects among Plans rather than + # declaring one. A deployment whose configured `llm_provider` cannot + # reach the seeded agents' declared provider (eg. `anthropic` from a + # controls network with no internet) defines its own Agent against a + # reachable provider and names it here. + run_debriefer_agent_id: UUID | None = None + caution_drafter_agent_id: UUID | None = None + # Bought-through-gateway path. Argo authenticates with a bare ANL # domain username in the API-key position, so there is no issued # credential to rotate; it is held as a SecretStr anyway because it diff --git a/apps/api/tests/unit/agent/_helpers.py b/apps/api/tests/unit/agent/_helpers.py index 8bdec320d4a..0c80a2d2548 100644 --- a/apps/api/tests/unit/agent/_helpers.py +++ b/apps/api/tests/unit/agent/_helpers.py @@ -226,6 +226,7 @@ async def seed_versioned_agent( monthly_usd_cap: float | None = None, daily_token_cap: int | None = None, model_ref: ModelRef | None = None, + kind: str = "RunDebriefer", ) -> None: """Seed Defined then Versioned, leaving the Agent at stream version 2.""" await seed_defined_agent( @@ -238,6 +239,7 @@ async def seed_versioned_agent( monthly_usd_cap=monthly_usd_cap, daily_token_cap=daily_token_cap, model_ref=model_ref, + kind=kind, ) versioned = AgentVersioned(agent_id=agent_id, version="v1", occurred_at=versioned_at) await store.append( diff --git a/apps/api/tests/unit/agent/test_agent_subscribers_registration.py b/apps/api/tests/unit/agent/test_agent_subscribers_registration.py index facb4c0b188..3d877296f6c 100644 --- a/apps/api/tests/unit/agent/test_agent_subscribers_registration.py +++ b/apps/api/tests/unit/agent/test_agent_subscribers_registration.py @@ -1,13 +1,20 @@ """Unit tests for register_agent_subscribers.""" -# pyright: reportUnknownMemberType=false +# pyright: reportPrivateUsage=false, reportUnknownMemberType=false from datetime import UTC, datetime +from typing import Any +from uuid import UUID, uuid4 import pytest import structlog.testing -from cora.agent import register_agent_subscribers +from cora.agent import register_agent_subscribers, report_designated_agents +from cora.agent.seed import RUN_DEBRIEFER_AGENT_ID, RUN_DEBRIEFER_AGENT_KIND +from cora.agent.seed_caution_drafter import CAUTION_DRAFTER_AGENT_ID +from cora.agent.subscribers.caution_drafter import CautionDrafterSubscriber +from cora.agent.subscribers.run_debriefer import RunDebrieferSubscriber +from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.infrastructure.config import Settings from cora.infrastructure.deps import make_inmemory_kernel from cora.infrastructure.ports import ( @@ -17,6 +24,13 @@ FixedIdGenerator, ) from cora.infrastructure.projection.registry import ProjectionRegistry +from tests.unit.agent._helpers import seed_versioned_agent + +_CORRELATION_ID = UUID("01900000-0000-7000-8000-00000009900a") +_PRINCIPAL_ID = UUID("01900000-0000-7000-8000-000000099001") +_NOW = datetime(2026, 5, 17, 14, 0, 0, tzinfo=UTC) +_DESIGNATED_RUN_DEBRIEFER_ID = UUID("01900000-0000-7000-8000-0000cccc0001") +_DESIGNATED_CAUTION_DRAFTER_ID = UUID("01900000-0000-7000-8000-0000cccc0002") def _kernel( @@ -24,10 +38,17 @@ def _kernel( llm: object | None, caution_promoter_enabled: bool = False, llm_enabled: bool = False, + llm_provider: str = "anthropic", + run_debriefer_agent_id: UUID | None = None, + caution_drafter_agent_id: UUID | None = None, + event_store: object | None = None, ) -> object: settings = Settings( # type: ignore[call-arg] caution_promoter_enabled=caution_promoter_enabled, llm_enabled=llm_enabled, + llm_provider=llm_provider, # type: ignore[arg-type] + run_debriefer_agent_id=run_debriefer_agent_id, + caution_drafter_agent_id=caution_drafter_agent_id, ) return make_inmemory_kernel( settings=settings, @@ -35,6 +56,7 @@ def _kernel( id_generator=FixedIdGenerator([]), authz=AllowAllAuthorize(), llm=llm, # type: ignore[arg-type] + event_store=event_store, # type: ignore[arg-type] ) @@ -160,3 +182,155 @@ def test_skip_warning_names_the_credential_when_the_switch_is_on() -> None: assert "ANTHROPIC_API_KEY is not configured" in reason assert "LLM_ENABLED is true" in reason + + +# --------------------------------------------------------------------------- +# Subscriber agent designation +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_run_debriefer_designation_setting_threads_into_subscriber() -> None: + """`settings.run_debriefer_agent_id` reaches the constructed subscriber.""" + registry = ProjectionRegistry() + kernel = _kernel(llm=FakeLLM(), run_debriefer_agent_id=_DESIGNATED_RUN_DEBRIEFER_ID) + + register_agent_subscribers(registry, kernel) # type: ignore[arg-type] + + subscriber = registry.get("run_debriefer") + assert isinstance(subscriber, RunDebrieferSubscriber) + assert subscriber._agent_id == _DESIGNATED_RUN_DEBRIEFER_ID + + +@pytest.mark.unit +def test_run_debriefer_unset_designation_uses_seeded_singleton() -> None: + """Unset means the seeded singleton, so nothing changes on upgrade.""" + registry = ProjectionRegistry() + kernel = _kernel(llm=FakeLLM()) + + register_agent_subscribers(registry, kernel) # type: ignore[arg-type] + + subscriber = registry.get("run_debriefer") + assert isinstance(subscriber, RunDebrieferSubscriber) + assert subscriber._agent_id == RUN_DEBRIEFER_AGENT_ID + + +@pytest.mark.unit +def test_caution_drafter_designation_setting_threads_into_subscriber() -> None: + """`settings.caution_drafter_agent_id` reaches the constructed subscriber.""" + registry = ProjectionRegistry() + kernel = _kernel(llm=FakeLLM(), caution_drafter_agent_id=_DESIGNATED_CAUTION_DRAFTER_ID) + + register_agent_subscribers(registry, kernel) # type: ignore[arg-type] + + subscriber = registry.get("caution_drafter") + assert isinstance(subscriber, CautionDrafterSubscriber) + assert subscriber._agent_id == _DESIGNATED_CAUTION_DRAFTER_ID + + +@pytest.mark.unit +def test_caution_drafter_unset_designation_uses_seeded_singleton() -> None: + """Unset means the seeded singleton, so nothing changes on upgrade.""" + registry = ProjectionRegistry() + kernel = _kernel(llm=FakeLLM()) + + register_agent_subscribers(registry, kernel) # type: ignore[arg-type] + + subscriber = registry.get("caution_drafter") + assert isinstance(subscriber, CautionDrafterSubscriber) + assert subscriber._agent_id == CAUTION_DRAFTER_AGENT_ID + + +# --------------------------------------------------------------------------- +# report_designated_agents: boot-time REPORT, never a gate +# --------------------------------------------------------------------------- + + +async def _report_log_events(kernel: object) -> list[Any]: + with structlog.testing.capture_logs() as captured: + await report_designated_agents(kernel) # type: ignore[arg-type] + return list(captured) + + +@pytest.mark.unit +async def test_report_designated_agents_warns_when_designated_agent_not_found() -> None: + """A designated-but-missing Agent logs a warning and moves on; it does + not raise, and it is not the mechanism that skips subscriber work + (the subscriber's own per-apply gate does that).""" + kernel = _kernel( + llm=None, + run_debriefer_agent_id=_DESIGNATED_RUN_DEBRIEFER_ID, + event_store=InMemoryEventStore(), + ) + + events = await _report_log_events(kernel) + + not_found = [ + e for e in events if e.get("event") == "agent_subscriber.designated_agent_not_found" + ] + assert any(e["agent_id"] == str(_DESIGNATED_RUN_DEBRIEFER_ID) for e in not_found) + + +@pytest.mark.unit +async def test_report_designated_agents_no_warning_when_provider_matches() -> None: + """Provider agrees with `settings.llm_provider`: one INFO line, no warning.""" + store = InMemoryEventStore() + await seed_versioned_agent( + store, + agent_id=RUN_DEBRIEFER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + ) + kernel = _kernel(llm=None, llm_provider="anthropic", event_store=store) + + events = await _report_log_events(kernel) + + reports = [e for e in events if e.get("event") == "agent_subscriber.designated_agent"] + assert any( + e["subscriber"] == "run_debriefer" and e["agent_id"] == str(RUN_DEBRIEFER_AGENT_ID) + for e in reports + ) + mismatches = [ + e + for e in events + if e.get("event") == "agent_subscriber.designated_agent_provider_mismatch" + and e.get("subscriber") == "run_debriefer" + ] + assert mismatches == [] + + +@pytest.mark.unit +async def test_report_designated_agents_warns_on_provider_mismatch() -> None: + """Declared provider disagrees with `settings.llm_provider`: a named + warning, but the report still returns normally (never a gate).""" + from cora.agent.aggregates.agent import ModelRef as AgentModelRef + + store = InMemoryEventStore() + await seed_versioned_agent( + store, + agent_id=RUN_DEBRIEFER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + model_ref=AgentModelRef(provider="argo", model="claude-haiku-4-5"), + ) + kernel = _kernel(llm=None, llm_provider="anthropic", event_store=store) + + events = await _report_log_events(kernel) + + mismatches = [ + e for e in events if e.get("event") == "agent_subscriber.designated_agent_provider_mismatch" + ] + assert len(mismatches) == 1 + assert mismatches[0]["subscriber"] == "run_debriefer" + assert mismatches[0]["agent_provider"] == "argo" + assert mismatches[0]["configured_llm_provider"] == "anthropic" 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 be91be2733b..68e9db22117 100644 --- a/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py +++ b/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py @@ -24,6 +24,9 @@ ) from cora.access.aggregates.actor import event_type_name as actor_event_type_name from cora.access.aggregates.actor import to_payload as actor_to_payload +from cora.agent.aggregates.agent import ModelRef as AgentModelRef +from cora.agent.prompts.caution_drafter import DEFAULT_CAUTION_DRAFTER_MODEL +from cora.agent.seed import RUN_DEBRIEFER_AGENT_KIND from cora.agent.seed_caution_drafter import ( CAUTION_DRAFTER_AGENT_ID, CAUTION_DRAFTER_AGENT_KIND, @@ -70,6 +73,7 @@ FakeAllocationLookup, FakeInferenceRecorder, FakeSpendLookup, + seed_defined_agent, seed_suspended_agent, seed_versioned_agent, ) @@ -78,6 +82,7 @@ _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) _PRINCIPAL_ID = UUID("01900000-0000-7000-8000-000000099001") _CORRELATION_ID = UUID("01900000-0000-7000-8000-00000009900a") +_DESIGNATED_AGENT_ID = UUID("01900000-0000-7000-8000-0000cccc0002") # A canned Plan id (Plan must exist for the subscriber to proceed). _PLAN_ID = UUID("01900000-0000-7000-8000-00000000aaaa") @@ -94,9 +99,11 @@ async def _seed_caution_drafter_actor( store: InMemoryEventStore, *, + agent_id: UUID = CAUTION_DRAFTER_AGENT_ID, deactivated: bool = False, ) -> None: - """Write the minimum Actor for the seeded CautionDrafter agent. + """Write the minimum Actor for a CautionDrafter agent (the seeded + singleton by default, or a designated id when `agent_id` is passed). PII vault: V2 payload carries no `name`; display name lives in `actor_profile`. Subscriber tests don't read the display @@ -104,7 +111,7 @@ async def _seed_caution_drafter_actor( """ _ = CAUTION_DRAFTER_AGENT_NAME event = ActorRegistered( - actor_id=CAUTION_DRAFTER_AGENT_ID, + actor_id=agent_id, occurred_at=_NOW, kind=ActorKind.AGENT, ) @@ -120,7 +127,7 @@ async def _seed_caution_drafter_actor( ) await store.append( stream_type="Actor", - stream_id=CAUTION_DRAFTER_AGENT_ID, + stream_id=agent_id, expected_version=0, events=[new_event], ) @@ -128,7 +135,7 @@ async def _seed_caution_drafter_actor( from cora.access.aggregates.actor import ActorDeactivated d_event = ActorDeactivated( - actor_id=CAUTION_DRAFTER_AGENT_ID, + actor_id=agent_id, occurred_at=_NOW, ) d_new_event = to_new_event( @@ -143,7 +150,7 @@ async def _seed_caution_drafter_actor( ) await store.append( stream_type="Actor", - stream_id=CAUTION_DRAFTER_AGENT_ID, + stream_id=agent_id, expected_version=1, events=[d_new_event], ) @@ -249,6 +256,7 @@ async def _build_subscriber( inference_recorder: FakeInferenceRecorder | None = None, spend_lookup: FakeSpendLookup | None = None, allocation_lookup: FakeAllocationLookup | None = None, + agent_id: UUID = CAUTION_DRAFTER_AGENT_ID, ) -> CautionDrafterSubscriber: return CautionDrafterSubscriber( event_store=event_store, @@ -257,6 +265,7 @@ async def _build_subscriber( inference_recorder=inference_recorder, spend_lookup=spend_lookup, allocation_lookup=allocation_lookup, + agent_id=agent_id, ) @@ -1602,3 +1611,198 @@ async def test_apply_coexists_with_run_debriefer_on_same_terminal_event() -> Non assert caution_decision is not None assert caution_decision.choice.value == "NoAction" assert len(caution_llm.received) == 1 + + +# --------------------------------------------------------------------------- +# Subscriber agent designation +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_default_agent_id_is_the_seeded_singleton() -> None: + """Unset means the seeded singleton, so nothing changes on upgrade.""" + subscriber = CautionDrafterSubscriber( + event_store=InMemoryEventStore(), + llm=FakeLLM(), + caution_lookup=AlwaysQuietCautionLookup(), + ) + assert subscriber._agent_id == CAUTION_DRAFTER_AGENT_ID + + +@pytest.mark.unit +async def test_apply_designation_honoured_on_actor_id_lease_and_inference_trace() -> None: + """A designated Agent's id lands on the Decision's actor_id, on the + lease event_id seed, and on the inference trace -- the three places + getting this wrong would be silent.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_PROPOSE_CAUTION]) + recorder = FakeInferenceRecorder() + await _seed_caution_drafter_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=CAUTION_DRAFTER_AGENT_KIND, + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, recorder, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event( + event_type="RunAborted", + run_id=run_id, + reason="rotary stage encoder offline; interlock fired", + ) + + await subscriber.apply(event, conn=None) + + decision = await load_decision(store, _derive_decision_id(event.event_id)) + assert decision is not None + assert decision.decided_by == _DESIGNATED_AGENT_ID + + stored, _version = await store.load("Run", run_id) + leases = [s for s in stored if s.event_type == "DecisionDebriefRequested"] + assert len(leases) == 1 + assert leases[0].payload["debriefer_agent_id"] == str(_DESIGNATED_AGENT_ID) + + assert len(recorder.calls) == 1 + assert recorder.calls[0].trace.agent_id == str(_DESIGNATED_AGENT_ID) + + +@pytest.mark.unit +async def test_apply_designation_serves_the_designated_agents_declared_model() -> None: + """The designated Agent's declared `model_ref` reaches the LLM port, + not the prompt module default. This is the regression `caution_drafter.py` + had (unlike `run_debriefer.py`, which already served the Agent's + declared model): designating a CautionDrafter Agent that declares an + Argo model must actually change what gets served.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=CAUTION_DRAFTER_AGENT_KIND, + model_ref=AgentModelRef(provider="argo", model="claude-sonnet-4-6"), + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + served = llm.received[0].model_ref + assert served.provider == "argo" + assert served.model == "claude-sonnet-4-6" + assert served.provider != DEFAULT_CAUTION_DRAFTER_MODEL.provider + + +@pytest.mark.unit +async def test_apply_falls_back_to_the_default_model_when_no_agent_is_seeded() -> None: + """An Actor-only deployment (seeded default, un-designated) keeps + working: the module default stands in when the Agent stream doesn't + exist.""" + 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 llm.received[0].model_ref == DEFAULT_CAUTION_DRAFTER_MODEL + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_stream_missing() -> None: + """Designated but the Agent stream does not exist (Actor-only): skip. + The seeded default is exempt from this check; an explicitly named + Agent is a deliberate choice and gets checked.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_wrong_kind() -> None: + """Designated Agent's kind isn't CautionDrafter: skip. Attributing a + CautionProposal-context Decision to an agent that doesn't draft + cautions would make the audit trail unreadable.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_not_versioned() -> None: + """Designated but not Versioned: skip via the existing lifecycle + gate, now reading whichever id is threaded in.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_defined_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + occurred_at=_NOW, + kind=CAUTION_DRAFTER_AGENT_KIND, + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None 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 fa11108bb3d..cf1c7c11701 100644 --- a/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py +++ b/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py @@ -31,6 +31,7 @@ RUN_DEBRIEFER_AGENT_KIND, RUN_DEBRIEFER_AGENT_NAME, ) +from cora.agent.seed_caution_drafter import CAUTION_DRAFTER_AGENT_KIND from cora.agent.subscribers._terminal_run_helpers import ( extract_capture_progress as _extract_capture_progress, ) @@ -87,19 +88,22 @@ _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) _PRINCIPAL_ID = UUID("01900000-0000-7000-8000-000000099001") _CORRELATION_ID = UUID("01900000-0000-7000-8000-00000009900a") +_DESIGNATED_AGENT_ID = UUID("01900000-0000-7000-8000-0000cccc0001") async def _seed_run_debrief_actor( store: InMemoryEventStore, *, + agent_id: UUID = RUN_DEBRIEFER_AGENT_ID, deactivated: bool = False, ) -> None: - """Write the bare-minimum Actor for the seeded RunDebriefer agent. + """Write the bare-minimum Actor for a RunDebriefer agent (the seeded + singleton by default, or a designated id when `agent_id` is passed). - The subscriber's `load_actor(event_store, RUN_DEBRIEFER_AGENT_ID)` - needs an Actor row at that id. We only write the Actor (skip the - Agent aggregate write); the subscriber doesn't load the Agent - aggregate at apply()-time. + The subscriber's `load_actor(event_store, self._agent_id)` needs an + Actor row at that id. We only write the Actor (skip the Agent + aggregate write); the subscriber doesn't load the Agent aggregate + at apply()-time. Set `deactivated=True` to also append an `ActorDeactivated` event so the loaded Actor has `active=False` (exercise the security @@ -110,7 +114,7 @@ async def _seed_run_debrief_actor( # surface, so the legacy seed-name constant stays unused here. _ = RUN_DEBRIEFER_AGENT_NAME event = ActorRegistered( - actor_id=RUN_DEBRIEFER_AGENT_ID, + actor_id=agent_id, occurred_at=_NOW, kind=ActorKind.AGENT, ) @@ -126,7 +130,7 @@ async def _seed_run_debrief_actor( ) await store.append( stream_type="Actor", - stream_id=RUN_DEBRIEFER_AGENT_ID, + stream_id=agent_id, expected_version=0, events=[new_event], ) @@ -134,7 +138,7 @@ async def _seed_run_debrief_actor( from cora.access.aggregates.actor import ActorDeactivated deactivated_event = ActorDeactivated( - actor_id=RUN_DEBRIEFER_AGENT_ID, + actor_id=agent_id, occurred_at=_NOW, ) deactivated_new_event = to_new_event( @@ -149,7 +153,7 @@ async def _seed_run_debrief_actor( ) await store.append( stream_type="Actor", - stream_id=RUN_DEBRIEFER_AGENT_ID, + stream_id=agent_id, expected_version=1, events=[deactivated_new_event], ) @@ -240,6 +244,7 @@ async def _build_subscriber( inference_recorder: FakeInferenceRecorder | None = None, spend_lookup: FakeSpendLookup | None = None, allocation_lookup: FakeAllocationLookup | None = None, + agent_id: UUID = RUN_DEBRIEFER_AGENT_ID, ) -> RunDebrieferSubscriber: return RunDebrieferSubscriber( event_store=event_store, @@ -248,6 +253,7 @@ async def _build_subscriber( inference_recorder=inference_recorder, spend_lookup=spend_lookup, allocation_lookup=allocation_lookup, + agent_id=agent_id, ) @@ -1831,3 +1837,166 @@ async def test_apply_falls_back_to_the_default_model_when_no_agent_is_seeded() - await subscriber.apply(event, conn=None) assert llm.received[0].model_ref == DEFAULT_RUN_DEBRIEF_MODEL + + +# --------------------------------------------------------------------------- +# Subscriber agent designation +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_default_agent_id_is_the_seeded_singleton() -> None: + """Unset means the seeded singleton, so nothing changes on upgrade.""" + subscriber = RunDebrieferSubscriber( + event_store=InMemoryEventStore(), + llm=FakeLLM(), + logbook_mirror=None, + ) + assert subscriber._agent_id == RUN_DEBRIEFER_AGENT_ID + + +@pytest.mark.unit +async def test_apply_designation_honoured_on_actor_id_lease_and_inference_trace() -> None: + """A designated Agent's id lands on the Decision's actor_id, on the + lease event_id seed, and on the inference trace -- the three places + getting this wrong would be silent.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + recorder = FakeInferenceRecorder() + await _seed_run_debrief_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + ) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, recorder, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + decision = await load_decision(store, _derive_decision_id(event.event_id)) + assert decision is not None + assert decision.decided_by == _DESIGNATED_AGENT_ID + + stored, _version = await store.load("Run", run_id) + leases = [s for s in stored if s.event_type == "DecisionDebriefRequested"] + assert len(leases) == 1 + assert leases[0].payload["debriefer_agent_id"] == str(_DESIGNATED_AGENT_ID) + + assert len(recorder.calls) == 1 + assert recorder.calls[0].trace.agent_id == str(_DESIGNATED_AGENT_ID) + + +@pytest.mark.unit +async def test_apply_designation_serves_the_designated_agents_declared_model() -> None: + """The designated Agent's declared `model_ref` reaches the LLM port, + not the prompt module default.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + model_ref=AgentModelRef(provider="argo", model="claude-haiku-4-5"), + ) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + served = llm.received[0].model_ref + assert served.provider == "argo" + assert served.model == "claude-haiku-4-5" + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_stream_missing() -> None: + """Designated but the Agent stream does not exist (Actor-only): skip. + The seeded default is exempt from this check; an explicitly named + Agent is a deliberate choice and gets checked.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store, agent_id=_DESIGNATED_AGENT_ID) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_wrong_kind() -> None: + """Designated Agent's kind isn't RunDebriefer: skip. Attributing a + RunDebrief-context Decision to an agent that doesn't debrief would + make the audit trail unreadable.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_versioned_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + kind=CAUTION_DRAFTER_AGENT_KIND, + ) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None + + +@pytest.mark.unit +async def test_apply_skips_when_designated_agent_not_versioned() -> None: + """Designated but not Versioned: skip via the existing lifecycle + gate, now reading whichever id is threaded in.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store, agent_id=_DESIGNATED_AGENT_ID) + await seed_defined_agent( + store, + agent_id=_DESIGNATED_AGENT_ID, + genesis_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + occurred_at=_NOW, + kind=RUN_DEBRIEFER_AGENT_KIND, + ) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, agent_id=_DESIGNATED_AGENT_ID) + event = _terminal_event(event_type="RunCompleted", run_id=run_id) + + await subscriber.apply(event, conn=None) + + assert llm.received == [] + assert await load_decision(store, _derive_decision_id(event.event_id)) is None diff --git a/apps/api/tests/unit/test_settings.py b/apps/api/tests/unit/test_settings.py index 2812f017893..e747b37c30a 100644 --- a/apps/api/tests/unit/test_settings.py +++ b/apps/api/tests/unit/test_settings.py @@ -163,6 +163,19 @@ def test_settings_run_initiator_plan_id_defaults_none_and_parses_uuid( Settings() +@pytest.mark.unit +def test_settings_subscriber_agent_designations_default_to_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unset means the seeded singleton for both LLM subscribers, so + nothing changes on upgrade.""" + monkeypatch.delenv("RUN_DEBRIEFER_AGENT_ID", raising=False) + monkeypatch.delenv("CAUTION_DRAFTER_AGENT_ID", raising=False) + settings = Settings() + assert settings.run_debriefer_agent_id is None + assert settings.caution_drafter_agent_id is None + + @pytest.mark.unit def test_settings_require_authenticated_principal_defaults_to_false( monkeypatch: pytest.MonkeyPatch, diff --git a/docs/deployments/2-bm/llm_debrief.md b/docs/deployments/2-bm/llm_debrief.md index a4089277236..70f4e9da92f 100644 --- a/docs/deployments/2-bm/llm_debrief.md +++ b/docs/deployments/2-bm/llm_debrief.md @@ -128,19 +128,28 @@ once at boot (`cora.agent.build_llm.build_llm`), not hot-reloaded. two separate passes over the corpus with a config change and restart between them, not two routes live at once; `kernel.llm` is a single bound adapter. - !!! warning "The live subscriber defers while an arm is armed" + !!! warning "The live subscriber defers while an arm is armed, unless you designate it" - `LLM_PROVIDER` binds one adapter for the whole process, and the automatic RunDebriefer subscriber uses the - **seeded singleton** Agent, which declares `anthropic`. With `LLM_PROVIDER=argo` or `local`, that - singleton's declared provider no longer matches the bound adapter, the adapter refuses the call, and every - newly completed Run debriefs to `DebriefDeferred` until the setting is put back. + `LLM_PROVIDER` binds one adapter for the whole process, and the automatic RunDebriefer subscriber acts as + the **seeded singleton** Agent by default, which declares `anthropic`. With `LLM_PROVIDER=argo` or + `local`, that singleton's declared provider no longer matches the bound adapter, the adapter refuses the + call, and every newly completed Run debriefs to `DebriefDeferred` until the setting is put back. The refusal is deliberate: cost resolves from the Agent's declared `(provider, model)` while the route comes from configuration, so serving a call through one and pricing it as the other would silently misattribute spend. Failing loudly is the better trade. - Practically: run the comparison in a no-beam window, when no Runs are completing. If beam is live, either - accept deferred automatic debriefs for the duration, or do not arm an alternate provider at all. + `RUN_DEBRIEFER_AGENT_ID` (and `CAUTION_DRAFTER_AGENT_ID` for the sibling subscriber) resolves this without + accepting deferred debriefs: set it to the arm's `agent_id` from [One Agent per + arm](#one-agent-per-arm) and the automatic subscriber acts as that Agent instead of the seeded singleton, + so its declared provider matches the bound adapter. This is exactly 2-BM's situation, not just this + comparison's: `api.anthropic.com` is unreachable from the controls network, so the seeded singleton can + never serve a live call there, and the automatic path stays structurally dead until an Argo- or + in-house-declaring Agent is designated. Restart is still required (settings are read once at boot). + + Without a designation, run the comparison in a no-beam window, when no Runs are completing. If beam is + live, either accept deferred automatic debriefs for the duration, or do not arm an alternate provider at + all. 3. Provider-specific settings, matching the `LLM_PROVIDER` chosen in step 2: **Argo arm:** @@ -166,6 +175,13 @@ once at boot (`cora.agent.build_llm.build_llm`), not hot-reloaded. Allocation envelope; in-house serving is metered-free by design, and what debits the envelope is the catalog entry's token rate (zero, for this entry). - `LOCAL_LLM_DEVICE_ID`: labels the served device in the GPU occupancy meter (default `gpu0`). +4. **`RUN_DEBRIEFER_AGENT_ID`** / **`CAUTION_DRAFTER_AGENT_ID`** (optional): set either to the corresponding arm's + `agent_id` from [One Agent per arm](#one-agent-per-arm) so the automatic subscriber acts as that Agent instead + of the seeded singleton (see the warning under step 2). Unset means the seeded singleton, so this step is + skippable if the comparison only needs the on-demand `regenerate_run_debrief` path used in [Running the + comparison](#running-the-comparison) below. Boot logs one INFO line per subscriber naming the effective Agent + and warns if its declared provider disagrees with `LLM_PROVIDER`; that warning is a report, not a refusal to + boot. ## Grant and activate the envelope