diff --git a/apps/api/openapi.json b/apps/api/openapi.json index 4c486ee8b3c..0678d9b94ff 100644 --- a/apps/api/openapi.json +++ b/apps/api/openapi.json @@ -681,7 +681,7 @@ "type": "object" }, "AgentStatus": { - "description": "The Agent's lifecycle state.\n\nFour values:\n\n - `Defined` -- registered as config; NOT yet ready for\n invocation (8f-b's subscriber filters on\n Versioned only).\n - `Versioned` -- promoted to ready-for-invocation; rainbow-\n deploy-style signal. Multiple Versioned\n Agents may exist concurrently (different\n `id`s sharing `kind`).\n - `Suspended` -- non-terminal operator-pause from\n `Versioned`. Returns to\n `Versioned` via `resume_agent`. Config\n changes (tools, budget) still permitted so\n the operator can fix permissions while\n paused. Cannot re-Version from Suspended\n (resume is its own dedicated verb).\n - `Deprecated` -- terminal; cannot be re-Defined or re-\n Versioned. Future invocations must pick a\n non-Deprecated Agent of the same kind.\n Reachable from `Defined`, `Versioned`, or\n `Suspended`.", + "description": "The Agent's lifecycle state.\n\nFour values:\n\n - `Defined` -- registered as config; NOT yet ready for\n invocation (the LLM subscribers' lifecycle\n gate fires on Versioned only).\n - `Versioned` -- promoted to ready-for-invocation; rainbow-\n deploy-style signal. Multiple Versioned\n Agents may exist concurrently (different\n `id`s sharing `kind`).\n - `Suspended` -- non-terminal operator-pause from\n `Versioned`. Returns to\n `Versioned` via `resume_agent`. Config\n changes (tools, budget) still permitted so\n the operator can fix permissions while\n paused. Cannot re-Version from Suspended\n (resume is its own dedicated verb).\n - `Deprecated` -- terminal; cannot be re-Defined or re-\n Versioned. Future invocations must pick a\n non-Deprecated Agent of the same kind.\n Reachable from `Defined`, `Versioned`, or\n `Suspended`.", "enum": [ "Defined", "Versioned", @@ -10754,6 +10754,7 @@ "type": "null" } ], + "description": "OTel gen_ai.agent.id. Principal-bound: when set, it MUST equal the calling principal's id (agent-attributed entries are the spend ledger the AgentBudget gate sums; self-reporting only). Mismatches reject the whole batch with 403.", "title": "Agent Id" }, "agent_name": { @@ -10780,6 +10781,19 @@ ], "title": "Conversation Id" }, + "cost_usd": { + "anyOf": [ + { + "minimum": 0.0, + "type": "number" + }, + { + "type": "null" + } + ], + "description": "Actual call cost in USD computed from usage tokens and provider pricing (CORA custom; no OTel attribute exists for call cost).", + "title": "Cost Usd" + }, "duration": { "anyOf": [ { @@ -10811,6 +10825,7 @@ "input_tokens": { "anyOf": [ { + "maximum": 1000000000.0, "minimum": 0.0, "type": "integer" }, @@ -10849,6 +10864,7 @@ "output_tokens": { "anyOf": [ { + "maximum": 1000000000.0, "minimum": 0.0, "type": "integer" }, @@ -28164,7 +28180,7 @@ } } }, - "description": "Authorize port denied the command." + "description": "Authorize port denied the command, or an entry's agent_id does not match the calling principal (agent-attributed entries must be self-reported)." }, "404": { "content": { diff --git a/apps/api/src/cora/agent/_budget_gate.py b/apps/api/src/cora/agent/_budget_gate.py new file mode 100644 index 00000000000..25e6e342f82 --- /dev/null +++ b/apps/api/src/cora/agent/_budget_gate.py @@ -0,0 +1,140 @@ +"""Coarse post-hoc budget gate for LLM-backed agents. + +The enforcement seam that turns a declared `AgentBudget` into a real +control at the coarse tier of the enforcement ladder: debit after each +call (the recorded `cost_usd` on the inference entry) and refuse the +NEXT call once a cap is exhausted. Overspend is bounded to about one +in-flight call per caller, adequate for the cheap, serialized +subscriber agents (RunDebriefer, CautionDrafter). Autonomous or +expensive-long-context callers need the per-call pre-estimate tier +before this gate alone is safe (one call can exhaust a balance before +a post-hoc gate fires). + +## Windows + +`monthly_usd_cap` is summed over the UTC calendar month containing +`as_of`; `daily_token_cap` over the UTC calendar day. `as_of` is the +triggering event's `occurred_at`, NOT wall clock, so a replayed work +item gates identically (the non-determinism rule: subscribers decide +from event facts, not ambient time). + +## Failure direction + +The spend sums undercount (NULL-cost legacy rows, unrecorded cache +tokens; see the SpendLookup port docstring), so the gate errs +PERMISSIVE. A cap of exactly zero refuses every call: the `AgentBudget` +value object already documents zero as recorded no-spend intent, and +the gate is what makes that intent real. + +A SpendLookup ERROR, by contrast, propagates: the subscriber's apply +raises, the projection worker's retry loop with backoff is the failure +handler, and the bookmark does not advance. The gate never fails OPEN +on a lookup error; wrapping it in a permissive except would let a +database outage disable enforcement silently. +""" + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from cora.agent.aggregates.agent import Agent + from cora.infrastructure.ports.spend_lookup import SpendLookup + + +@dataclass(frozen=True) +class BudgetBreach: + """One exhausted cap: which, by how much, over which window.""" + + cap_kind: str # "monthly_usd_cap" | "daily_token_cap" + cap_value: float + spent: float + window_start: datetime + window_end: datetime + + def describe(self) -> str: + """One human-readable sentence for a deferred Decision's reasoning.""" + return ( + f"{self.cap_kind} of {self.cap_value:g} reached " + f"(spent {self.spent:g} in the window starting " + f"{self.window_start.isoformat()})" + ) + + +def calendar_month_window(as_of: datetime) -> tuple[datetime, datetime]: + """Half-open UTC calendar month `[first, first-of-next)` containing `as_of`.""" + at = as_of.astimezone(UTC) + start = datetime(at.year, at.month, 1, tzinfo=UTC) + if at.month == 12: + end = datetime(at.year + 1, 1, 1, tzinfo=UTC) + else: + end = datetime(at.year, at.month + 1, 1, tzinfo=UTC) + return start, end + + +def calendar_day_window(as_of: datetime) -> tuple[datetime, datetime]: + """Half-open UTC calendar day `[midnight, next-midnight)` containing `as_of`.""" + at = as_of.astimezone(UTC) + start = datetime(at.year, at.month, at.day, tzinfo=UTC) + return start, start + timedelta(days=1) + + +async def find_budget_breach( + *, + agent: "Agent | None", + spend_lookup: "SpendLookup", + as_of: datetime, +) -> BudgetBreach | None: + """Return the first exhausted cap for this agent, or None to permit. + + An agent with no declared budget (or no Agent stream at all) is + unlimited: declaration is opt-in, so absence of caps must never + block. Caps are checked in declaration order (monthly USD, then + daily tokens); the first breach wins, one SpendLookup SUM per + declared cap. + """ + if agent is None or agent.budget is None: + return None + budget = agent.budget + + if budget.monthly_usd_cap is not None: + window_start, window_end = calendar_month_window(as_of) + spend = await spend_lookup.find_agent_spend( + agent_id=agent.id, + window_start=window_start, + window_end=window_end, + ) + if spend.usd_spent >= budget.monthly_usd_cap: + return BudgetBreach( + cap_kind="monthly_usd_cap", + cap_value=budget.monthly_usd_cap, + spent=spend.usd_spent, + window_start=window_start, + window_end=window_end, + ) + + if budget.daily_token_cap is not None: + window_start, window_end = calendar_day_window(as_of) + spend = await spend_lookup.find_agent_spend( + agent_id=agent.id, + window_start=window_start, + window_end=window_end, + ) + if spend.tokens_spent >= budget.daily_token_cap: + return BudgetBreach( + cap_kind="daily_token_cap", + cap_value=float(budget.daily_token_cap), + spent=float(spend.tokens_spent), + window_start=window_start, + window_end=window_end, + ) + + return None + + +__all__ = [ + "BudgetBreach", + "calendar_day_window", + "calendar_month_window", + "find_budget_breach", +] diff --git a/apps/api/src/cora/agent/adapters/anthropic_llm.py b/apps/api/src/cora/agent/adapters/anthropic_llm.py index d872e537475..34e99a76616 100644 --- a/apps/api/src/cora/agent/adapters/anthropic_llm.py +++ b/apps/api/src/cora/agent/adapters/anthropic_llm.py @@ -229,15 +229,16 @@ async def chat(self, request: LLMChatRequest) -> LLMResponse: # `record_llm_call` returns the computed USD cost; the # adapter intentionally discards it here. The cost is - # already persisted on the `cora.agent.llm.cost.usd` - # histogram (where dashboards consume it), and surfacing - # it on `LLMResponse` would force every consumer to think - # about pricing semantics that are an observability - # concern, not a domain concern. Tests that want the - # cost value call `compute_cost_usd` directly. + # persisted on the `cora.agent.llm.cost.usd` histogram + # (where dashboards consume it), and durably by the LLM + # producers, which recompute it via `compute_cost_usd` + # (same pure function, same inputs) onto the inference + # entry's cost_usd. Surfacing it on `LLMResponse` would + # force every consumer to think about pricing semantics; + # consumers that want the value compute it from usage. record_llm_call( span, - system="anthropic", + provider_name="anthropic", request_model_ref=request.model_ref, response_model_id=response_model_id, usage=usage, diff --git a/apps/api/src/cora/agent/aggregates/agent/__init__.py b/apps/api/src/cora/agent/aggregates/agent/__init__.py index 68a8c730a1e..22adc8af3a9 100644 --- a/apps/api/src/cora/agent/aggregates/agent/__init__.py +++ b/apps/api/src/cora/agent/aggregates/agent/__init__.py @@ -69,6 +69,7 @@ AgentNotFoundError, AgentNotSeededError, AgentStatus, + AgentSuspendedError, AgentSuspensionReason, AgentVersion, InvalidAgentBudgetError, @@ -129,6 +130,7 @@ "AgentResumed", "AgentStatus", "AgentSuspended", + "AgentSuspendedError", "AgentSuspensionReason", "AgentTargetPlanSet", "AgentToolGranted", diff --git a/apps/api/src/cora/agent/aggregates/agent/state.py b/apps/api/src/cora/agent/aggregates/agent/state.py index 29a83980b35..c71a527267e 100644 --- a/apps/api/src/cora/agent/aggregates/agent/state.py +++ b/apps/api/src/cora/agent/aggregates/agent/state.py @@ -69,8 +69,8 @@ class AgentStatus(StrEnum): Four values: - `Defined` -- registered as config; NOT yet ready for - invocation (8f-b's subscriber filters on - Versioned only). + invocation (the LLM subscribers' lifecycle + gate fires on Versioned only). - `Versioned` -- promoted to ready-for-invocation; rainbow- deploy-style signal. Multiple Versioned Agents may exist concurrently (different @@ -469,6 +469,24 @@ def __init__(self, agent_id: UUID) -> None: self.agent_id = agent_id +class AgentSuspendedError(Exception): + """Cross-aggregate state-gate failure: the Agent is `Suspended`. + + Raised by operator-triggered slices (the regenerate path) when the + target agent is under the reversible operator pause: suspension + means the agent takes no actions, LLM calls included, and driving + it through an on-demand slice would defeat the pause. Recovery is + operator-side: `resume_agent` first. + + Mirrors the subscribers' lifecycle gate (only a Versioned agent + acts). + """ + + def __init__(self, agent_id: UUID) -> None: + super().__init__(f"Agent {agent_id} is suspended; resume via resume_agent before invoking") + self.agent_id = agent_id + + # --------------------------------------------------------------------------- # Bounded-text value objects # --------------------------------------------------------------------------- @@ -660,7 +678,7 @@ class ToolName: @dataclass(frozen=True) class AgentBudget: - """Optional per-agent budget caps (declaration only at this layer). + """Optional per-agent budget caps, enforced at the coarse post-hoc tier. Both `monthly_usd_cap` and `daily_token_cap` are independently nullable so the same VO covers "set both", "set one, clear the @@ -668,17 +686,17 @@ class AgentBudget: one must be non-None at construction (the no-budget shape is `Agent.budget = None` directly). - Enforcement is deferred to a future Budget BC (watch item in - [[project-agent-lifecycle-grants-design]]); these are - declaration-only fields today. Cost telemetry already lands on - `gen_ai.cost.usd` via the `gen_ai` observability helper so the - Budget BC can begin enforcement without further per-agent surface - work. - - Zero caps allowed: recorded as the "no spend" intent, but NOT - enforced today. A 0 cap blocks nothing (same as any other cap) until - the Budget BC lands; the future enforcement layer can treat zero as a - hard stop. Negative caps rejected. + Enforcement: the coarse post-hoc gate (`cora.agent._budget_gate`, + consumed by the LLM subscribers) sums recorded spend from the + inference entries' `cost_usd` and refuses the next call once a cap + is exhausted; per-call telemetry also lands on the + `cora.agent.llm.cost.usd` histogram. Finer tiers (per-call + pre-estimate, reserve-post-void) and a full Budget BC remain + deferred (watch item in [[project-agent-lifecycle-grants-design]]). + + Zero caps allowed and enforced as the recorded "no spend" intent: + the gate treats a 0 cap as a hard stop (zero spent >= zero cap + refuses every call). Negative caps rejected. """ monthly_usd_cap: float | None diff --git a/apps/api/src/cora/agent/features/regenerate_run_debrief/handler.py b/apps/api/src/cora/agent/features/regenerate_run_debrief/handler.py index e83b2b2dd04..0352c66c4e7 100644 --- a/apps/api/src/cora/agent/features/regenerate_run_debrief/handler.py +++ b/apps/api/src/cora/agent/features/regenerate_run_debrief/handler.py @@ -29,7 +29,10 @@ HTTP 404 (already-mapped by Run BC's routes). - RunDebriefer Agent's Actor must exist and be active; raises `AgentNotSeededError` / `AgentDeactivatedError` (both in - `cora.agent.aggregates.agent.state`) -> HTTP 400. + `cora.agent.aggregates.agent.state`) -> HTTP 400. A Suspended + agent raises `AgentSuspendedError` (resume first); the budget + gate is deliberately NOT applied on this operator-triggered + path (conscious, human-accountable spend; still debited). - When `parent_decision_id` is supplied: parent Decision must exist (`DecisionParentNotFoundError` from `cora.decision.aggregates.decision`; HTTP 409 per Decision @@ -58,7 +61,13 @@ from uuid import UUID, uuid5 from cora.access.aggregates.actor import load_actor -from cora.agent.aggregates.agent import AgentDeactivatedError, AgentNotSeededError +from cora.agent.aggregates.agent import ( + AgentDeactivatedError, + AgentNotSeededError, + AgentStatus, + AgentSuspendedError, + load_agent, +) from cora.agent.errors import UnauthorizedError from cora.agent.features.regenerate_run_debrief.command import RegenerateRunDebrief from cora.agent.features.regenerate_run_debrief.context import RegenerateRunDebriefContext @@ -82,6 +91,7 @@ from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.kernel import Kernel from cora.infrastructure.logging import get_logger +from cora.infrastructure.observability.gen_ai import compute_cost_usd from cora.infrastructure.ports import ( AgentInferenceTrace, Deny, @@ -180,6 +190,16 @@ async def handler( if not actor.active: raise AgentDeactivatedError(RUN_DEBRIEFER_AGENT_ID) + # Suspension gate: the reversible operator pause means the agent + # takes no actions, and an on-demand regenerate must not defeat + # it; the operator resumes first. The BUDGET gate is deliberately + # absent here: an operator-triggered regenerate is a conscious, + # human-accountable spend (the coarse post-hoc tier targets the + # autonomous subscribers), and the call still debits the ledger. + agent = await load_agent(deps.event_store, RUN_DEBRIEFER_AGENT_ID) + if agent is not None and agent.status is AgentStatus.SUSPENDED: + raise AgentSuspendedError(RUN_DEBRIEFER_AGENT_ID) + # Pre-load parent Decision when ref set; enforce same-agent + # same-Run scope. if command.parent_decision_id is not None: @@ -339,6 +359,7 @@ async def _record_inference( finish_reasons=(response.stop_reason,) if response.stop_reason else (), input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens, + cost_usd=compute_cost_usd(request.model_ref, response.usage), request_max_tokens=request.max_output_tokens, agent_id=str(RUN_DEBRIEFER_AGENT_ID), agent_name=RUN_DEBRIEFER_AGENT_NAME, diff --git a/apps/api/src/cora/agent/routes.py b/apps/api/src/cora/agent/routes.py index c4611f67541..cac1bf0734b 100644 --- a/apps/api/src/cora/agent/routes.py +++ b/apps/api/src/cora/agent/routes.py @@ -42,6 +42,7 @@ AgentDeactivatedError, AgentNotFoundError, AgentNotSeededError, + AgentSuspendedError, InvalidAgentBudgetError, InvalidAgentCanonicalUriError, InvalidAgentCapabilitiesError, @@ -192,6 +193,7 @@ def register_agent_routes(app: FastAPI) -> None: InvalidModelRefError, AgentNotSeededError, AgentDeactivatedError, + AgentSuspendedError, # promote_caution_proposal validation errors. DecisionNotCautionProposalError, CautionProposalNotActionableError, diff --git a/apps/api/src/cora/agent/subscribers/caution_drafter.py b/apps/api/src/cora/agent/subscribers/caution_drafter.py index b9bb008ec32..c06f743d020 100644 --- a/apps/api/src/cora/agent/subscribers/caution_drafter.py +++ b/apps/api/src/cora/agent/subscribers/caution_drafter.py @@ -80,7 +80,9 @@ from uuid import UUID, uuid5 from cora.access.aggregates.actor import load_actor +from cora.agent._budget_gate import find_budget_breach from cora.agent._subscriber_lease import attempt_debrief_lease +from cora.agent.aggregates.agent import AgentStatus, load_agent from cora.agent.prompts import ( CAUTION_DRAFTER_PROMPT_TEMPLATE_ID, CandidateTarget, @@ -113,8 +115,10 @@ ) from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.logging import get_logger +from cora.infrastructure.observability.gen_ai import compute_cost_usd from cora.infrastructure.ports import ( AgentInferenceTrace, + AlwaysZeroSpendLookup, ConcurrencyError, NullInferenceRecorder, ) @@ -133,6 +137,7 @@ LLMChatRequest, LLMResponse, Signer, + SpendLookup, ) from cora.infrastructure.ports.event_store import EventStore, NewEvent, StoredEvent from cora.infrastructure.projection.handler import ConnectionLike @@ -198,6 +203,7 @@ def __init__( caution_lookup: CautionLookup, signer: Signer | None = None, inference_recorder: InferenceRecorder | None = None, + spend_lookup: SpendLookup | None = None, ) -> None: self.event_store = event_store self.llm = llm @@ -207,6 +213,10 @@ def __init__( # inert; production wiring passes the Kernel's recorder via # `make_caution_drafter_subscriber`. self.inference_recorder = inference_recorder or NullInferenceRecorder() + # Defaults to zero spend so declared caps never block tests that + # don't exercise budget gating; production wiring passes the + # Kernel's PostgresSpendLookup. + self.spend_lookup = spend_lookup or AlwaysZeroSpendLookup() async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: """Process one terminal Run event.""" @@ -266,6 +276,20 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return + # Lifecycle gate (mirrors RunDebriefer): only a Versioned agent + # 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) + 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_status=str(agent.status), + ) + return + # Cross-agent lease (per [[project-run-debriefer-lease-design]]; # mirrors RunDebriefer's gate verbatim). Reserved for future # multi-instance CautionDrafter variants (e.g., different LLM @@ -297,6 +321,43 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return + # Coarse post-hoc budget gate (mirrors RunDebriefer): refuse the + # call once a declared cap is exhausted, recording the refusal as + # this terminal Run's NoAction Decision so operators see WHY no + # caution proposal was drafted. + breach = await find_budget_breach( + agent=agent, + spend_lookup=self.spend_lookup, + as_of=event.occurred_at, + ) + if breach is not None: + log.warning( + "caution_drafter.budget_exhausted", + cap_kind=breach.cap_kind, + cap_value=breach.cap_value, + spent=breach.spent, + ) + await self._compose_and_append( + decision_id=decision_id, + actor=actor, + run_id=run_id, + terminal_event=event, + choice="NoAction", + confidence=None, + reasoning=( + f"Agent budget exhausted: {breach.describe()}; LLM call " + "skipped. Raise the cap via update_agent_budget or wait " + "for the window to reset." + ), + extra_inputs={ + "failure_error_class": "AgentBudgetExhausted", + "budget_cap_kind": breach.cap_kind, + }, + outcome="deferred", + log=log, + ) + return + # Build candidate-target list from Plan.asset_ids. v1 doesn't # include Procedures (Run aggregate doesn't yet carry a # procedure binding); per design memo, watch item for when @@ -431,6 +492,7 @@ async def _record_inference( finish_reasons=(response.stop_reason,) if response.stop_reason else (), input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens, + cost_usd=compute_cost_usd(request.model_ref, response.usage), request_max_tokens=request.max_output_tokens, agent_id=str(CAUTION_DRAFTER_AGENT_ID), agent_name=CAUTION_DRAFTER_AGENT_NAME, @@ -803,6 +865,7 @@ def make_caution_drafter_subscriber(deps: Kernel) -> CautionDrafterSubscriber: caution_lookup=deps.caution_lookup, signer=deps.signer, inference_recorder=deps.inference_recorder, + spend_lookup=deps.spend_lookup, ) diff --git a/apps/api/src/cora/agent/subscribers/run_debriefer.py b/apps/api/src/cora/agent/subscribers/run_debriefer.py index 68a32cf0ba5..9ec53c86c1b 100644 --- a/apps/api/src/cora/agent/subscribers/run_debriefer.py +++ b/apps/api/src/cora/agent/subscribers/run_debriefer.py @@ -139,7 +139,9 @@ from uuid import UUID, uuid5 from cora.access.aggregates.actor import load_actor +from cora.agent._budget_gate import find_budget_breach from cora.agent._subscriber_lease import attempt_debrief_lease +from cora.agent.aggregates.agent import AgentStatus, load_agent from cora.agent.prompts import ( RUN_DEBRIEF_PROMPT_TEMPLATE_ID, RunDebriefPayload, @@ -166,8 +168,10 @@ ) from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.logging import get_logger +from cora.infrastructure.observability.gen_ai import compute_cost_usd from cora.infrastructure.ports import ( AgentInferenceTrace, + AlwaysZeroSpendLookup, ConcurrencyError, NullInferenceRecorder, ) @@ -185,6 +189,7 @@ LLMResponse, LogbookMirror, Signer, + SpendLookup, ) from cora.infrastructure.ports.event_store import EventStore, NewEvent, StoredEvent from cora.infrastructure.projection.handler import ConnectionLike @@ -289,6 +294,7 @@ def __init__( logbook_mirror: LogbookMirror | None, signer: Signer | None = None, inference_recorder: InferenceRecorder | None = None, + spend_lookup: SpendLookup | None = None, ) -> None: self.event_store = event_store self.llm = llm @@ -298,6 +304,10 @@ def __init__( # caller that omits it) stays inert; production wiring passes the # Kernel's recorder via `make_run_debriefer_subscriber`. self.inference_recorder = inference_recorder or NullInferenceRecorder() + # Defaults to zero spend so declared caps never block tests that + # don't exercise budget gating; production wiring passes the + # Kernel's PostgresSpendLookup. + self.spend_lookup = spend_lookup or AlwaysZeroSpendLookup() async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: """Process one terminal Run event. @@ -367,6 +377,25 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return + # Lifecycle gate: only a Versioned agent acts. Suspended (the + # reversible operator pause) and Deprecated (the terminal retire + # verb) both mean no LLM calls and no Decision writes; Defined is + # not yet promoted for invocation per the AgentStatus contract. + # A missing Agent stream stays permissive (legacy deployments + # seeded only the Actor). Per-apply check, so resume restores + # 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) + 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_status=str(agent.status), + ) + return + # Cross-agent 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 @@ -397,6 +426,44 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None: ) return + # Coarse post-hoc budget gate (the enforcement seam): refuse the + # call once a declared cap is exhausted, recording the refusal as + # this terminal Run's Decision so the one-Decision-per-terminal-Run + # invariant holds and operators see WHY the debrief is missing. + # After the lease so only the winning agent pays the spend lookup. + breach = await find_budget_breach( + agent=agent, + spend_lookup=self.spend_lookup, + as_of=event.occurred_at, + ) + if breach is not None: + log.warning( + "run_debriefer.budget_exhausted", + cap_kind=breach.cap_kind, + cap_value=breach.cap_value, + spent=breach.spent, + ) + await self._compose_and_append( + decision_id=decision_id, + actor=actor, + run_id=run_id, + terminal_event=event, + choice="DebriefDeferred", + confidence=None, + reasoning=( + f"Agent budget exhausted: {breach.describe()}; LLM call " + "skipped. Raise the cap via update_agent_budget or wait " + "for the window to reset, then re-trigger the debrief." + ), + extra_inputs={ + "failure_error_class": "AgentBudgetExhausted", + "budget_cap_kind": breach.cap_kind, + }, + outcome="deferred", + log=log, + ) + return + payload = RunDebriefPayload( terminal_event_type=event.event_type, terminal_event_reason=terminal_event_reason, @@ -514,6 +581,7 @@ async def _record_inference( finish_reasons=(response.stop_reason,) if response.stop_reason else (), input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens, + cost_usd=compute_cost_usd(request.model_ref, response.usage), request_max_tokens=request.max_output_tokens, agent_id=str(RUN_DEBRIEFER_AGENT_ID), agent_name=RUN_DEBRIEFER_AGENT_NAME, @@ -797,6 +865,7 @@ def make_run_debriefer_subscriber(deps: Kernel) -> RunDebrieferSubscriber: logbook_mirror=deps.logbook_mirror, signer=deps.signer, inference_recorder=deps.inference_recorder, + spend_lookup=deps.spend_lookup, ) diff --git a/apps/api/src/cora/api/_inference_recorder.py b/apps/api/src/cora/api/_inference_recorder.py index 11e459a44cc..d305e99cd3f 100644 --- a/apps/api/src/cora/api/_inference_recorder.py +++ b/apps/api/src/cora/api/_inference_recorder.py @@ -21,7 +21,7 @@ from uuid import UUID -from cora.decision.errors import UnauthorizedError +from cora.decision.errors import InferenceAgentMismatchError, UnauthorizedError from cora.decision.features.append_inferences.command import ( AppendInferences, ReasoningEntryInput, @@ -61,6 +61,7 @@ async def record( finish_reasons=trace.finish_reasons, input_tokens=trace.input_tokens, output_tokens=trace.output_tokens, + cost_usd=trace.cost_usd, agent_id=trace.agent_id, agent_name=trace.agent_name, ), @@ -73,6 +74,18 @@ async def record( correlation_id=correlation_id, causation_id=causation_id, ) + except InferenceAgentMismatchError as exc: + # Loud like the 403 sibling below: an internal recorder should be + # structurally incapable of this (all callers self-report with the + # agent's own actor id as principal), so hitting it means a new + # agent was wired with a divergent seed/constant and its spend is + # being DROPPED, which starves the budget gate's SUM (fail-open). + _log.warning( + "inference_recorder.agent_mismatch", + decision_id=str(trace.decision_id), + entry_agent_id=exc.entry_agent_id, + principal_id=str(principal_id), + ) except UnauthorizedError as exc: # Loud, actionable: under a real Trust policy the agent principal # must be granted AppendInferences (as RunSupervisor is granted diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index c1c4f743a31..e5557fd3d74 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -124,6 +124,7 @@ register_decision_tools, wire_decision, ) +from cora.decision.adapters import PostgresSpendLookup from cora.enclosure import ( EnclosureHandlers, enclosure_permit_monitor_lifespan, @@ -620,6 +621,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: caution_lookup_factory=PostgresCautionLookup, capability_lookup_factory=PostgresCapabilityLookup, supply_lookup_factory=PostgresSupplyLookup, + spend_lookup_factory=PostgresSpendLookup, run_actor_involvement_lookup_factory=PostgresRunActorInvolvementLookup, consequence_lookup_factory=PostgresConsequenceLookup, dataset_distribution_lookup_factory=PostgresDatasetDistributionLookup, diff --git a/apps/api/src/cora/decision/adapters/__init__.py b/apps/api/src/cora/decision/adapters/__init__.py new file mode 100644 index 00000000000..87adbbbffd5 --- /dev/null +++ b/apps/api/src/cora/decision/adapters/__init__.py @@ -0,0 +1,10 @@ +"""Adapters Decision BC ships for cross-BC ports. + +- `PostgresSpendLookup` implementing `SpendLookup` (consumed by the + budget gate at the LLM producers' seams). Decision BC ships it + because it owns the durable spend fact: `entries_decision_inferences`. +""" + +from cora.decision.adapters.postgres_spend_lookup import PostgresSpendLookup + +__all__ = ["PostgresSpendLookup"] diff --git a/apps/api/src/cora/decision/adapters/postgres_spend_lookup.py b/apps/api/src/cora/decision/adapters/postgres_spend_lookup.py new file mode 100644 index 00000000000..e200255ba20 --- /dev/null +++ b/apps/api/src/cora/decision/adapters/postgres_spend_lookup.py @@ -0,0 +1,89 @@ +"""Postgres adapter implementing `SpendLookup` over `entries_decision_inferences`. + +Consumed by the budget gate at the LLM producers' seams via the +`Kernel.spend_lookup` port. Sums the durable spend facts the inference +entries already carry (`cost_usd`, token counts) for one agent inside +one half-open window. + +## Why query the entries table (not a projection) + +The gate reads one aggregate per call: SUM + COUNT over rows matching +`agent_id` + an `occurred_at` range. No existing index leads with +`agent_id` (the table indexes decision_id/logbook_id/conversation_id +plus a BRIN on recorded_at), so this is a sequential scan of the +append-only entries table; at pilot scale (single-digit LLM calls per +run) that is well inside p95. The escalation ladder when the gate's +read shows up in latency dashboards: first an `(agent_id, occurred_at)` +index, then a dedicated `proj_agent_spend` read model. Neither before +the trigger fires. + +## NULL and type notes + +`cost_usd` is nullable (pre-migration rows) and `COALESCE`d to $0, so +history never inflates a balance. `agent_id` is a TEXT column (OTel +`gen_ai.agent.id` is a string); the adapter stringifies the UUID the +consumer passes. `tokens_spent` counts input + output tokens (cache +tokens are not persisted on the entry; see the port docstring for why +the resulting undercount is the accepted failure direction). +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false + +from datetime import datetime +from uuid import UUID + +import asyncpg + +from cora.infrastructure.ports.spend_lookup import SpendLookupResult + +_SUM_AGENT_SPEND_SQL = """ +SELECT + COALESCE(SUM(cost_usd), 0)::float8 AS usd_spent, + COALESCE(SUM(COALESCE(input_tokens, 0)::numeric + COALESCE(output_tokens, 0)::numeric), 0) + AS tokens_spent, + COUNT(*)::bigint AS call_count +FROM entries_decision_inferences +WHERE agent_id = $1 + AND occurred_at >= $2 + AND occurred_at < $3 +""" + +# Token sums accumulate in NUMERIC (arbitrary precision) so one +# adversarially large row can never raise bigint-out-of-range and wedge +# the subscriber's retry loop; the Python side clamps the Decimal back +# into int range, which still exceeds any cap by construction. +_TOKENS_CLAMP = 2**63 - 1 + + +class PostgresSpendLookup: + """asyncpg-backed `SpendLookup` implementation.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + + async def find_agent_spend( + self, + *, + agent_id: UUID, + window_start: datetime, + window_end: datetime, + ) -> SpendLookupResult: + async with self._pool.acquire() as conn: + row = await conn.fetchrow( + _SUM_AGENT_SPEND_SQL, + str(agent_id), + window_start, + window_end, + ) + assert row is not None # aggregate queries always return one row + return SpendLookupResult( + agent_id=agent_id, + window_start=window_start, + window_end=window_end, + usd_spent=float(row["usd_spent"]), + tokens_spent=min(int(row["tokens_spent"]), _TOKENS_CLAMP), + call_count=int(row["call_count"]), + ) + + +__all__ = ["PostgresSpendLookup"] diff --git a/apps/api/src/cora/decision/aggregates/decision/entries.py b/apps/api/src/cora/decision/aggregates/decision/entries.py index 8626254c255..dcfee4b2b25 100644 --- a/apps/api/src/cora/decision/aggregates/decision/entries.py +++ b/apps/api/src/cora/decision/aggregates/decision/entries.py @@ -124,6 +124,8 @@ class Inference: # --- OTel gen_ai.* token usage (NOT deprecated prompt/completion_tokens) --- input_tokens: int | None # gen_ai.usage.input_tokens output_tokens: int | None # gen_ai.usage.output_tokens + # --- CORA cost (custom; no OTel attribute exists for call cost) --- + cost_usd: float | None # mirrors the cora.agent.llm.cost.usd histogram # --- OTel gen_ai.* agent context --- agent_id: str | None # gen_ai.agent.id agent_name: str | None # gen_ai.agent.name @@ -168,6 +170,15 @@ class Inference: "output_tokens": LogbookFieldSpec( type="int", description="OTel gen_ai.usage.output_tokens" ), + "cost_usd": LogbookFieldSpec( + type="float", + units="USD", + description=( + "Actual call cost computed from usage tokens and provider " + "pricing (CORA custom; mirrors the cora.agent.llm.cost.usd " + "histogram, no OTel attribute exists for call cost)" + ), + ), "agent_id": LogbookFieldSpec(type="string", description="OTel gen_ai.agent.id"), "agent_name": LogbookFieldSpec(type="string", description="OTel gen_ai.agent.name"), "agent_description": LogbookFieldSpec( @@ -223,7 +234,8 @@ async def append(self, rows: list[Inference]) -> None: input_tokens, output_tokens, agent_id, agent_name, agent_description, conversation_id, tool_name, tool_call_id, tool_type, - messages + messages, + cost_usd ) VALUES ( $1, $2, $3, $4, $5, $6, $7, @@ -234,7 +246,8 @@ async def append(self, rows: list[Inference]) -> None: $18, $19, $20, $21, $22, $23, $24, $25, $26, - $27 + $27, + $28 ) ON CONFLICT (event_id) DO NOTHING """ @@ -313,6 +326,7 @@ async def append(self, rows: list[Inference]) -> None: row.tool_call_id, row.tool_type, json.dumps(row.messages) if row.messages is not None else None, + row.cost_usd, ) for row in rows ], diff --git a/apps/api/src/cora/decision/errors.py b/apps/api/src/cora/decision/errors.py index 0130fa042d2..cd13be8e8b7 100644 --- a/apps/api/src/cora/decision/errors.py +++ b/apps/api/src/cora/decision/errors.py @@ -51,7 +51,34 @@ def __init__(self, kind: str) -> None: self.kind = kind +class InferenceAgentMismatchError(Exception): + """An inference entry attributes an agent other than the caller. + + Raised by the `append_inferences` handler when an entry carries an + `agent_id` that differs from the calling principal. Agent-attributed + inference rows are the spend ledger the `AgentBudget` gate sums per + agent, so attribution must be self-reported: a producer may record + its own spend, never another agent's (a mismatched row would let any + authorized producer inflate a victim agent's recorded spend and deny + it service). Rows with `agent_id=None` (operator- or tool-attributed + provenance with no agent claim) are unaffected. + + Maps to HTTP 403 alongside `UnauthorizedError`: the request is + well-formed but the caller may never make it for that agent. + """ + + def __init__(self, *, entry_agent_id: str, principal_id: str) -> None: + super().__init__( + f"entry agent_id={entry_agent_id!r} does not match the calling " + f"principal {principal_id}; agent-attributed inference entries " + "must be self-reported" + ) + self.entry_agent_id = entry_agent_id + self.principal_id = principal_id + + __all__ = [ + "InferenceAgentMismatchError", "InvalidActorKindForDecisionError", "OverrideKindRequiresParentError", "UnauthorizedError", diff --git a/apps/api/src/cora/decision/features/append_inferences/command.py b/apps/api/src/cora/decision/features/append_inferences/command.py index 7da08e1c15d..b63d2842953 100644 --- a/apps/api/src/cora/decision/features/append_inferences/command.py +++ b/apps/api/src/cora/decision/features/append_inferences/command.py @@ -57,6 +57,7 @@ class ReasoningEntryInput: finish_reasons: tuple[str, ...] = field(default_factory=tuple[str, ...]) input_tokens: int | None = None output_tokens: int | None = None + cost_usd: float | None = None agent_id: str | None = None agent_name: str | None = None agent_description: str | None = None diff --git a/apps/api/src/cora/decision/features/append_inferences/handler.py b/apps/api/src/cora/decision/features/append_inferences/handler.py index 47af127b160..e5ce3375649 100644 --- a/apps/api/src/cora/decision/features/append_inferences/handler.py +++ b/apps/api/src/cora/decision/features/append_inferences/handler.py @@ -35,8 +35,11 @@ Decision (own BC) is loaded. The producer's reasoning entries typically reference external concepts (model names, tool names, -agent IDs) but those are stored as opaque strings, no -cross-aggregate validation needed for the entries themselves. +agent IDs) stored as opaque strings without cross-aggregate +validation, with ONE exception: `agent_id` must equal the calling +principal when set (agent-attributed rows are the budget ledger; +see `InferenceAgentMismatchError`). No Agent aggregate is loaded, +the check is a pure string comparison against the envelope. """ from typing import Protocol @@ -53,7 +56,7 @@ load_decision, to_payload, ) -from cora.decision.errors import UnauthorizedError +from cora.decision.errors import InferenceAgentMismatchError, UnauthorizedError from cora.decision.features.append_inferences.command import ( AppendInferences, ReasoningEntryInput, @@ -137,6 +140,28 @@ async def handler( ) raise UnauthorizedError(authz.reason) + # Agent attribution is principal-bound: entry rows keyed by + # agent_id are the spend ledger the AgentBudget gate sums, so a + # producer may only record spend against itself. Reject the + # whole batch BEFORE any write (no logbook open, no partial + # append) so a spoofing request leaves no trace to clean up. + principal_str = str(principal_id) + for entry in command.entries: + if entry.agent_id is not None and entry.agent_id != principal_str: + _log.warning( + "append_inferences.agent_mismatch", + command_name=_COMMAND_NAME, + decision_id=str(command.decision_id), + entry_event_id=str(entry.event_id), + entry_agent_id=entry.agent_id, + principal_id=principal_str, + correlation_id=str(correlation_id), + ) + raise InferenceAgentMismatchError( + entry_agent_id=entry.agent_id, + principal_id=principal_str, + ) + # Resolve the reasoning logbook id, opening it lazily on the # first append. Retries on ConcurrencyError (a parallel writer # incrementing the Decision stream's version between our @@ -259,6 +284,7 @@ def _build_row( finish_reasons=entry.finish_reasons, input_tokens=entry.input_tokens, output_tokens=entry.output_tokens, + cost_usd=entry.cost_usd, agent_id=entry.agent_id, agent_name=entry.agent_name, agent_description=entry.agent_description, diff --git a/apps/api/src/cora/decision/features/append_inferences/route.py b/apps/api/src/cora/decision/features/append_inferences/route.py index 49046894189..4e38b52583a 100644 --- a/apps/api/src/cora/decision/features/append_inferences/route.py +++ b/apps/api/src/cora/decision/features/append_inferences/route.py @@ -109,9 +109,27 @@ class ReasoningEntryRequest(BaseModel): max_length=16, description="OTel gen_ai.response.finish_reasons (multiple stops possible).", ) - input_tokens: int | None = Field(default=None, ge=0) - output_tokens: int | None = Field(default=None, ge=0) - agent_id: str | None = Field(default=None, max_length=200) + input_tokens: int | None = Field(default=None, ge=0, le=1_000_000_000) + output_tokens: int | None = Field(default=None, ge=0, le=1_000_000_000) + cost_usd: float | None = Field( + default=None, + ge=0.0, + allow_inf_nan=False, + description=( + "Actual call cost in USD computed from usage tokens and provider " + "pricing (CORA custom; no OTel attribute exists for call cost)." + ), + ) + agent_id: str | None = Field( + default=None, + max_length=200, + description=( + "OTel gen_ai.agent.id. Principal-bound: when set, it MUST equal " + "the calling principal's id (agent-attributed entries are the " + "spend ledger the AgentBudget gate sums; self-reporting only). " + "Mismatches reject the whole batch with 403." + ), + ) agent_name: str | None = Field(default=None, max_length=200) agent_description: str | None = Field(default=None, max_length=2000) conversation_id: str | None = Field(default=None, max_length=200) @@ -176,7 +194,11 @@ def _get_handler(request: Request) -> Handler: responses={ status.HTTP_403_FORBIDDEN: { "model": ErrorResponse, - "description": "Authorize port denied the command.", + "description": ( + "Authorize port denied the command, or an entry's " + "agent_id does not match the calling principal " + "(agent-attributed entries must be self-reported)." + ), }, status.HTTP_404_NOT_FOUND: { "model": ErrorResponse, @@ -220,6 +242,7 @@ async def post_decisions_reasoning_entries( finish_reasons=tuple(e.finish_reasons), input_tokens=e.input_tokens, output_tokens=e.output_tokens, + cost_usd=e.cost_usd, agent_id=e.agent_id, agent_name=e.agent_name, agent_description=e.agent_description, diff --git a/apps/api/src/cora/decision/features/append_inferences/tool.py b/apps/api/src/cora/decision/features/append_inferences/tool.py index 46d2ceff801..fe7a512d18f 100644 --- a/apps/api/src/cora/decision/features/append_inferences/tool.py +++ b/apps/api/src/cora/decision/features/append_inferences/tool.py @@ -104,9 +104,34 @@ async def append_reasoning_entries_tool( # pyright: ignore[reportUnusedFunction list[str] | None, Field(default=None, max_length=16), ] = None, - input_tokens: Annotated[int | None, Field(default=None, ge=0)] = None, - output_tokens: Annotated[int | None, Field(default=None, ge=0)] = None, - agent_id: Annotated[str | None, Field(default=None, max_length=200)] = None, + input_tokens: Annotated[int | None, Field(default=None, ge=0, le=1_000_000_000)] = None, + output_tokens: Annotated[int | None, Field(default=None, ge=0, le=1_000_000_000)] = None, + cost_usd: Annotated[ + float | None, + Field( + default=None, + ge=0.0, + allow_inf_nan=False, + description=( + "Actual call cost in USD computed from usage tokens and " + "provider pricing (CORA custom; no OTel attribute exists " + "for call cost)." + ), + ), + ] = None, + agent_id: Annotated[ + str | None, + Field( + default=None, + max_length=200, + description=( + "OTel gen_ai.agent.id. Principal-bound: when set, it MUST equal " + "the calling principal's id (agent-attributed entries are the " + "spend ledger the AgentBudget gate sums; self-reporting only). " + "Mismatches reject the whole batch with 403." + ), + ), + ] = None, agent_name: Annotated[str | None, Field(default=None, max_length=200)] = None, agent_description: Annotated[str | None, Field(default=None, max_length=2000)] = None, conversation_id: Annotated[str | None, Field(default=None, max_length=200)] = None, @@ -145,6 +170,7 @@ async def append_reasoning_entries_tool( # pyright: ignore[reportUnusedFunction finish_reasons=tuple(finish_reasons or []), input_tokens=input_tokens, output_tokens=output_tokens, + cost_usd=cost_usd, agent_id=agent_id, agent_name=agent_name, agent_description=agent_description, diff --git a/apps/api/src/cora/decision/routes.py b/apps/api/src/cora/decision/routes.py index ee7b8eb31ae..8d73888a189 100644 --- a/apps/api/src/cora/decision/routes.py +++ b/apps/api/src/cora/decision/routes.py @@ -43,6 +43,7 @@ InvalidReasoningSignatureError, ) from cora.decision.errors import ( + InferenceAgentMismatchError, InvalidActorKindForDecisionError, OverrideKindRequiresParentError, UnauthorizedError, @@ -154,3 +155,4 @@ def register_decision_routes(app: FastAPI) -> None: ): app.add_exception_handler(logbook_state_cls, _handle_logbook_state) app.add_exception_handler(UnauthorizedError, _handle_unauthorized) + app.add_exception_handler(InferenceAgentMismatchError, _handle_unauthorized) diff --git a/apps/api/src/cora/infrastructure/deps.py b/apps/api/src/cora/infrastructure/deps.py index e8823465e73..537f3598393 100644 --- a/apps/api/src/cora/infrastructure/deps.py +++ b/apps/api/src/cora/infrastructure/deps.py @@ -87,6 +87,7 @@ AlwaysPermittedEnclosureLookup, AlwaysQuietCautionLookup, AlwaysRatifiedConsequenceLookup, + AlwaysZeroSpendLookup, AssemblyLookup, AssetLookup, Authorize, @@ -112,6 +113,7 @@ ProfileStore, RoleLookup, RunActorInvolvementLookup, + SpendLookup, SupplyLookup, SystemClock, TokenVerifier, @@ -168,6 +170,7 @@ def make_postgres_kernel( caution_lookup: CautionLookup | None = None, capability_lookup: CapabilityLookup | None = None, supply_lookup: SupplyLookup | None = None, + spend_lookup: SpendLookup | None = None, run_actor_involvement_lookup: RunActorInvolvementLookup | None = None, consequence_lookup: ConsequenceLookup | None = None, dataset_distribution_lookup: DatasetDistributionLookup | None = None, @@ -335,6 +338,7 @@ def make_postgres_kernel( capability_lookup if capability_lookup is not None else AlwaysEmptyCapabilityLookup() ), supply_lookup=(supply_lookup if supply_lookup is not None else AllSatisfiedSupplyLookup()), + spend_lookup=(spend_lookup if spend_lookup is not None else AlwaysZeroSpendLookup()), run_actor_involvement_lookup=( run_actor_involvement_lookup if run_actor_involvement_lookup is not None @@ -397,6 +401,7 @@ def make_inmemory_kernel( caution_lookup: CautionLookup | None = None, capability_lookup: CapabilityLookup | None = None, supply_lookup: SupplyLookup | None = None, + spend_lookup: SpendLookup | None = None, run_actor_involvement_lookup: RunActorInvolvementLookup | None = None, consequence_lookup: ConsequenceLookup | None = None, dataset_distribution_lookup: DatasetDistributionLookup | None = None, @@ -547,6 +552,7 @@ def make_inmemory_kernel( capability_lookup if capability_lookup is not None else AlwaysEmptyCapabilityLookup() ), supply_lookup=(supply_lookup if supply_lookup is not None else AllSatisfiedSupplyLookup()), + spend_lookup=(spend_lookup if spend_lookup is not None else AlwaysZeroSpendLookup()), run_actor_involvement_lookup=( run_actor_involvement_lookup if run_actor_involvement_lookup is not None @@ -681,6 +687,25 @@ def __call__( ) -> SupplyLookup: ... +class SpendLookupFactory(Protocol): + """Builds the production SpendLookup port for the Kernel. + + Decision BC's `cora.decision.adapters.PostgresSpendLookup` is the + production factory; `cora.api.main` binds it. Same factory- + injection shape as the other lookup factories so + `cora.infrastructure.deps` doesn't import from any BC. + + `pool` is `None` only when `app_env=test`; the production factory + requires a real pool. Test mode falls back to + `AlwaysZeroSpendLookup` automatically. + """ + + def __call__( + self, + pool: asyncpg.Pool, + ) -> SpendLookup: ... + + class RunActorInvolvementLookupFactory(Protocol): """Builds the production RunActorInvolvementLookup port for the Kernel. @@ -926,6 +951,7 @@ async def build_kernel( caution_lookup_factory: CautionLookupFactory | None = None, capability_lookup_factory: CapabilityLookupFactory | None = None, supply_lookup_factory: SupplyLookupFactory | None = None, + spend_lookup_factory: SpendLookupFactory | None = None, run_actor_involvement_lookup_factory: RunActorInvolvementLookupFactory | None = None, consequence_lookup_factory: ConsequenceLookupFactory | None = None, dataset_distribution_lookup_factory: DatasetDistributionLookupFactory | None = None, @@ -1036,6 +1062,9 @@ async def build_kernel( if supply_lookup_factory is not None else AllSatisfiedSupplyLookup() ) + spend_lookup: SpendLookup = ( + spend_lookup_factory(pool) if spend_lookup_factory is not None else AlwaysZeroSpendLookup() + ) run_actor_involvement_lookup: RunActorInvolvementLookup = ( run_actor_involvement_lookup_factory(pool) if run_actor_involvement_lookup_factory is not None @@ -1099,6 +1128,7 @@ async def build_kernel( caution_lookup=caution_lookup, capability_lookup=capability_lookup, supply_lookup=supply_lookup, + spend_lookup=spend_lookup, run_actor_involvement_lookup=run_actor_involvement_lookup, consequence_lookup=consequence_lookup, dataset_distribution_lookup=dataset_distribution_lookup, diff --git a/apps/api/src/cora/infrastructure/kernel.py b/apps/api/src/cora/infrastructure/kernel.py index 346fb2b246a..79fa6ec2f8c 100644 --- a/apps/api/src/cora/infrastructure/kernel.py +++ b/apps/api/src/cora/infrastructure/kernel.py @@ -71,6 +71,7 @@ RoleLookup, RunActorInvolvementLookup, Signer, + SpendLookup, SupplyLookup, TokenVerifier, ) @@ -155,6 +156,16 @@ class Kernel: the `ClearanceLookup` / `CautionLookup` test-default pattern. See [[project_supply_preflight_gate_design]]. + `spend_lookup`: cross-BC port consumed by the budget gate at the + LLM subscribers' seams (RunDebriefer / CautionDrafter) to sum an + agent's recorded spend in a cap window before permitting the next + call. Planned consumers, not wired today: the regenerate slice and + the Operation BC steering brain at the per-call pre-estimate tier. + Decision BC ships `PostgresSpendLookup` as the production adapter + (sums `entries_decision_inferences`). Test environments default + to `AlwaysZeroSpendLookup` (nothing spent) so a declared cap + never blocks tests that don't exercise budget gating. + `run_actor_involvement_lookup`: cross-BC port consumed by the authority-revocation holder subscriber (K3) to resolve the in-flight Runs a revoked principal drives and hold each. Run BC @@ -349,6 +360,7 @@ class Kernel: assembly_lookup: AssemblyLookup role_lookup: RoleLookup enclosure_lookup: EnclosureLookup + spend_lookup: SpendLookup profile_store: ProfileStore canonicalization_registry: CanonicalizationRegistry signing_registry: SigningRegistry diff --git a/apps/api/src/cora/infrastructure/observability/gen_ai.py b/apps/api/src/cora/infrastructure/observability/gen_ai.py index add60fd092d..7a2e98c9068 100644 --- a/apps/api/src/cora/infrastructure/observability/gen_ai.py +++ b/apps/api/src/cora/infrastructure/observability/gen_ai.py @@ -11,11 +11,13 @@ Current status: experimental. Per the design memo's watch item, opt in is via `OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental` in production deploy config. Attribute names below match the spec as -of January 2026; if the spec stabilizes with renames, this module -is the single edit point. +of July 2026 (`gen_ai.system` was deprecated in favour of +`gen_ai.provider.name`; the durable Decision-side record was already +on the new name, this module now matches it); if the spec renames +again, this module is the single edit point. Attributes set on the active span: - - `gen_ai.system` ("anthropic") + - `gen_ai.provider.name` ("anthropic") - `gen_ai.operation.name` ("chat") - `gen_ai.request.model` (the model identifier from `ModelRef`) - `gen_ai.request.max_tokens` @@ -76,9 +78,10 @@ class ModelPricing: """Per-million-token USD prices for one LLM model. `cache_write_per_mtok` is the price of bytes WRITTEN to the - Anthropic prompt cache (usually higher than base input because - of the 1-hour TTL overhead). `cache_read_per_mtok` is the price - of bytes READ from cache (usually ~10% of base input). + Anthropic prompt cache at the TTL tier the producers use (2x base + input for the 1-hour TTL; 1.25x for the 5-minute tier). + `cache_read_per_mtok` is the price of bytes READ from cache + (usually ~10% of base input). Providers that don't expose cache pricing (or that don't support caching) set both cache fields equal to `input_per_mtok` so the @@ -92,24 +95,40 @@ class ModelPricing: PRICING: dict[tuple[str, str], ModelPricing] = { - # Anthropic public pricing (Feb 2026; 1h-TTL cache write tier). - # Update when Anthropic publishes a new model or revises prices. + # Anthropic public pricing (Jul 2026). Cache writes are priced at + # the 1-HOUR TTL tier (2x base input), because that is the TTL the + # producers pin on their cache breakpoints; the 5-minute tier would + # be 1.25x. If a producer ever drops to 5m TTL, model per-TTL write + # prices instead of repricing the table. + # Opus dropped to $5/$25 per MTok with the 4.7/4.8 generation. + ("anthropic", "claude-opus-4-8"): ModelPricing( + input_per_mtok=5.00, + output_per_mtok=25.00, + cache_write_per_mtok=10.00, + cache_read_per_mtok=0.50, + ), ("anthropic", "claude-opus-4-7"): ModelPricing( - input_per_mtok=15.00, - output_per_mtok=75.00, - cache_write_per_mtok=18.75, - cache_read_per_mtok=1.50, + input_per_mtok=5.00, + output_per_mtok=25.00, + cache_write_per_mtok=10.00, + cache_read_per_mtok=0.50, + ), + ("anthropic", "claude-sonnet-4-5"): ModelPricing( + input_per_mtok=3.00, + output_per_mtok=15.00, + cache_write_per_mtok=6.00, + cache_read_per_mtok=0.30, ), ("anthropic", "claude-sonnet-4-6"): ModelPricing( input_per_mtok=3.00, output_per_mtok=15.00, - cache_write_per_mtok=3.75, + cache_write_per_mtok=6.00, cache_read_per_mtok=0.30, ), ("anthropic", "claude-haiku-4-5"): ModelPricing( input_per_mtok=1.00, output_per_mtok=5.00, - cache_write_per_mtok=1.25, + cache_write_per_mtok=2.00, cache_read_per_mtok=0.10, ), } @@ -142,7 +161,8 @@ def compute_cost_usd(model_ref: ModelRef, usage: LLMUsage) -> float: `PRICING` entry when they see the warning. Cache-read tokens are billed at ~10% of base input; cache-write - tokens are billed at ~125% of base input (1h tier). Plain input + tokens are billed at 2x base input (the 1-hour TTL tier the + producers pin; the 5-minute tier would be 1.25x). Plain input tokens (`usage.input_tokens` minus cache hits/misses) are billed at base. The Anthropic SDK reports `input_tokens` exclusive of cache tokens, so the three add up to the actual chargeable input. @@ -171,7 +191,7 @@ def compute_cost_usd(model_ref: ModelRef, usage: LLMUsage) -> float: def record_llm_call( span: Span, *, - system: str, + provider_name: str, request_model_ref: ModelRef, response_model_id: str, usage: LLMUsage, @@ -194,7 +214,7 @@ def record_llm_call( when tracing is disabled): set_attribute is a no-op and the histograms are no-op too when no MeterProvider is installed. """ - span.set_attribute("gen_ai.system", system) + span.set_attribute("gen_ai.provider.name", provider_name) span.set_attribute("gen_ai.operation.name", "chat") span.set_attribute("gen_ai.request.model", request_model_ref.model) span.set_attribute("gen_ai.request.max_tokens", max_tokens) @@ -212,7 +232,7 @@ def record_llm_call( ) base_attrs = { - "gen_ai.system": system, + "gen_ai.provider.name": provider_name, "gen_ai.request.model": request_model_ref.model, "gen_ai.response.model": response_model_id, } diff --git a/apps/api/src/cora/infrastructure/ports/__init__.py b/apps/api/src/cora/infrastructure/ports/__init__.py index 451352543a2..b8d39ea54f7 100644 --- a/apps/api/src/cora/infrastructure/ports/__init__.py +++ b/apps/api/src/cora/infrastructure/ports/__init__.py @@ -146,6 +146,11 @@ SignerKeyNotFoundError, SignerUnavailableError, ) +from cora.infrastructure.ports.spend_lookup import ( + AlwaysZeroSpendLookup, + SpendLookup, + SpendLookupResult, +) from cora.infrastructure.ports.supply_lookup import ( AllSatisfiedSupplyLookup, NoSuppliesRegisteredLookup, @@ -172,6 +177,7 @@ "AlwaysPermittedEnclosureLookup", "AlwaysQuietCautionLookup", "AlwaysRatifiedConsequenceLookup", + "AlwaysZeroSpendLookup", "AssemblyLookup", "AssemblyLookupResult", "AssetLookup", @@ -260,6 +266,8 @@ "SignerKeyInactiveError", "SignerKeyNotFoundError", "SignerUnavailableError", + "SpendLookup", + "SpendLookupResult", "StoredEvent", "StreamAppend", "SupplyLookup", diff --git a/apps/api/src/cora/infrastructure/ports/inference_recorder.py b/apps/api/src/cora/infrastructure/ports/inference_recorder.py index 8e4fdcbf8d5..a5bfa6ff410 100644 --- a/apps/api/src/cora/infrastructure/ports/inference_recorder.py +++ b/apps/api/src/cora/infrastructure/ports/inference_recorder.py @@ -57,6 +57,11 @@ class AgentInferenceTrace: `request_model` is what was sent (the agent's configured model); when the provider resolves a dated snapshot, `response_model` carries the exact model that answered (the load-bearing reproducibility field). + + `cost_usd` is the call's actual dollar cost, computed by the producer from + usage tokens and pricing (`compute_cost_usd`), the same math behind the + `cora.agent.llm.cost.usd` histogram. Persisting it here is what turns the + ephemeral meter into a durable ledger fact a spend lookup can sum. """ decision_id: UUID @@ -69,6 +74,7 @@ class AgentInferenceTrace: finish_reasons: tuple[str, ...] = field(default_factory=tuple[str, ...]) input_tokens: int | None = None output_tokens: int | None = None + cost_usd: float | None = None request_max_tokens: int | None = None agent_id: str | None = None agent_name: str | None = None diff --git a/apps/api/src/cora/infrastructure/ports/spend_lookup.py b/apps/api/src/cora/infrastructure/ports/spend_lookup.py new file mode 100644 index 00000000000..d4802abe011 --- /dev/null +++ b/apps/api/src/cora/infrastructure/ports/spend_lookup.py @@ -0,0 +1,131 @@ +"""SpendLookup port: cross-BC query for an agent's recorded LLM spend. + +Consumed by the Agent BC's LLM subscribers (RunDebriefer, +CautionDrafter) to gate an LLM call on the caller's declared +`AgentBudget` caps: the gate sums what the agent has already spent in +the cap's window and refuses the next call once a cap is breached +(coarse post-hoc enforcement per the tier ladder in +[[project_budget_bc_research]]). Planned consumers, not wired today: +the operator-triggered regenerate slice, and the Operation BC steering +brain at the per-call pre-estimate tier. + +## Convention + +Same shape as the `start_run` lookup family (`ClearanceLookup`, +`SupplyLookup`, ...): a consumer-shaped Protocol + frozen result VO + +an always-pass test stub, living in `cora.infrastructure.ports` for +neutral access. The Decision BC ships the production adapter +(`cora.decision.adapters.PostgresSpendLookup`) because it owns the +durable spend fact: `entries_decision_inferences`, where each LLM +call's provenance row carries `agent_id`, `occurred_at`, token counts, +and `cost_usd`. + +## Window semantics + +The port is window-agnostic: the CONSUMER computes `[window_start, +window_end)` (half-open) for the cap it enforces, calendar month (UTC) +for `monthly_usd_cap`, calendar day (UTC) for `daily_token_cap`, and +the adapter just sums inside the bounds. Keeping window policy out of +the adapter means a future award-window allocation (activation and +expiry bound to run lifecycle events rather than the calendar) is a +consumer-side change only. + +## Trust boundary + +Agent-attributed rows are principal-bound at the write path: the +`append_inferences` handler rejects any entry whose `agent_id` +differs from the calling principal (`InferenceAgentMismatchError`, +403), so a producer can record spend against itself only, never +inflate another agent's balance. All internal recorders (both LLM +subscribers and the regenerate slice, via +`DelegatingInferenceRecorder`) self-report with the agent's own +actor id as principal. Residual trust surface: `agent_name` / +`agent_description` remain unverified display strings, and a future +on-behalf-of recorder (the Tier 1.5 steering-spend design) needs an +explicit delegation mechanism rather than a relaxation of the +binding. + +## Known undercount (accepted for the coarse tier) + +Rows with `cost_usd IS NULL` (pre-migration history) sum as $0, and +cache-read/cache-write tokens are not persisted on the entry, so +`tokens_spent` counts input + output only. Both make the gate slightly +PERMISSIVE, never spuriously blocking, which is the right failure +direction for a coarse post-hoc tier. The leak-free reserve-post-void +tier owns exactness (deferred; see [[project_budget_bc_research]]). +""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol +from uuid import UUID + + +@dataclass(frozen=True) +class SpendLookupResult: + """Summed spend for one agent inside one window. + + `usd_spent` sums `cost_usd` (NULLs as $0); `tokens_spent` sums + input + output tokens; `call_count` counts inference rows. The + echoed window bounds make gate log lines self-describing. + """ + + agent_id: UUID + window_start: datetime + window_end: datetime + usd_spent: float + tokens_spent: int + call_count: int + + +class SpendLookup(Protocol): + """Cross-BC port: sum an agent's recorded LLM spend in a window.""" + + async def find_agent_spend( + self, + *, + agent_id: UUID, + window_start: datetime, + window_end: datetime, + ) -> SpendLookupResult: + """Return the agent's summed spend for `[window_start, window_end)`. + + An agent with no recorded calls in the window returns a zero + row (never None): "no spend" and "unknown agent" are the same + answer to a budget gate. + """ + ... + + +class AlwaysZeroSpendLookup: + """Test-default stub: every agent has spent nothing. + + Mirrors `AlwaysCoveredClearanceLookup`'s role: tests that don't + exercise budget gating get this stub from the kernel defaults, so + a declared cap never blocks them. Tests that exercise the gate + override with the real adapter (`PostgresSpendLookup`) or an + in-test fake returning a chosen spend. + """ + + async def find_agent_spend( + self, + *, + agent_id: UUID, + window_start: datetime, + window_end: datetime, + ) -> SpendLookupResult: + return SpendLookupResult( + agent_id=agent_id, + window_start=window_start, + window_end=window_end, + usd_spent=0.0, + tokens_spent=0, + call_count=0, + ) + + +__all__ = [ + "AlwaysZeroSpendLookup", + "SpendLookup", + "SpendLookupResult", +] diff --git a/apps/api/src/cora/operation/adapters/llm_decide_port.py b/apps/api/src/cora/operation/adapters/llm_decide_port.py index b82517d0347..d2881e15ee9 100644 --- a/apps/api/src/cora/operation/adapters/llm_decide_port.py +++ b/apps/api/src/cora/operation/adapters/llm_decide_port.py @@ -1,5 +1,14 @@ """LlmDecidePort: an LLM steering brain behind the `DecidePort` seam. +BUDGET NOTE: this is the one autonomous LLM caller with NO budget gate +and NO durable spend record today. Its calls reach only the OTel cost +histogram (via the adapter), never `entries_decision_inferences`, so +the `SpendLookup` ledger cannot see them and `AgentBudget` caps do not +bind it. The per-call pre-estimate tier plus a durable steering-spend +record are the deferred follow-up (see [[project_budget_bc_research]]); +until then the conduct loop's own iteration and wall-clock bounds are +the only spend limiters on this path. + The DECIDE-axis analogue of the LLM agents in the Agent BC, but homed in the Operation BC because it implements `DecidePort` (which lives here) and consumes only the `LLM` port and `cora.shared` steering value types, never diff --git a/apps/api/tests/contract/test_append_reasoning_entries_endpoint.py b/apps/api/tests/contract/test_append_reasoning_entries_endpoint.py index bdbbf000e44..b1aab644352 100644 --- a/apps/api/tests/contract/test_append_reasoning_entries_endpoint.py +++ b/apps/api/tests/contract/test_append_reasoning_entries_endpoint.py @@ -77,11 +77,17 @@ def test_post_reasoning_entries_handles_dedup_silently_on_retry() -> None: @pytest.mark.contract def test_post_reasoning_entries_accepts_full_otel_field_set() -> None: - """Round-trip every documented OTel gen_ai.* field through the API.""" + """Round-trip every documented OTel gen_ai.* field through the API. + + agent_id is principal-bound (self-reporting only), so the caller + identifies itself via X-Principal-Id and attributes the entry to + that same id.""" + producer_id = str(uuid4()) with TestClient(create_app()) as client: decision_id = _seed_decision(client) response = client.post( f"/decisions/{decision_id}/inferences", + headers={"X-Principal-Id": producer_id}, json={ "entries": [ _good_entry( @@ -98,7 +104,7 @@ def test_post_reasoning_entries_accepts_full_otel_field_set() -> None: finish_reasons=["end_turn"], input_tokens=512, output_tokens=256, - agent_id="agent-7e", + agent_id=producer_id, agent_name="ApprovalAgent", agent_description="Reviews recipes", conversation_id="conv-abc", @@ -195,3 +201,44 @@ def test_post_reasoning_entries_rejects_extra_fields_with_422() -> None: json={"entries": [{**_good_entry(), "rogue_field": "boom"}]}, ) assert response.status_code == 422 + + +# ---------- Agent attribution is principal-bound ---------- + + +@pytest.mark.contract +def test_post_reasoning_entries_rejects_foreign_agent_id_with_403() -> None: + """An entry attributing spend to an agent other than the calling + principal is refused: the ledger only accepts self-reported + agent spend (budget-inflation guard).""" + with TestClient(create_app()) as client: + decision_id = _seed_decision(client) + response = client.post( + f"/decisions/{decision_id}/inferences", + headers={"X-Principal-Id": str(uuid4())}, + json={ + "entries": [ + _good_entry(agent_id=str(uuid4()), cost_usd=250.0), + ] + }, + ) + assert response.status_code == 403 + assert "self-reported" in response.json()["detail"] + + +@pytest.mark.contract +def test_post_reasoning_entries_accepts_self_reported_agent_id() -> None: + producer_id = str(uuid4()) + with TestClient(create_app()) as client: + decision_id = _seed_decision(client) + response = client.post( + f"/decisions/{decision_id}/inferences", + headers={"X-Principal-Id": producer_id}, + json={ + "entries": [ + _good_entry(agent_id=producer_id, cost_usd=0.002), + ] + }, + ) + assert response.status_code == 200 + assert response.json() == {"event_count": 1} diff --git a/apps/api/tests/integration/_helpers.py b/apps/api/tests/integration/_helpers.py index 9dabf82af62..6772204e102 100644 --- a/apps/api/tests/integration/_helpers.py +++ b/apps/api/tests/integration/_helpers.py @@ -29,14 +29,22 @@ from datetime import UTC, datetime from typing import Any -from uuid import UUID +from uuid import UUID, uuid4 import asyncpg +from cora.agent.aggregates.agent import ( + AgentStatus, + AgentVersioned, + load_agent, +) +from cora.agent.aggregates.agent import event_type_name as agent_event_type_name +from cora.agent.aggregates.agent import to_payload as agent_to_payload from cora.infrastructure.adapters.in_memory_facility_lookup import InMemoryFacilityLookup from cora.infrastructure.adapters.postgres_profile_store import PostgresProfileStore from cora.infrastructure.config import Settings from cora.infrastructure.deps import make_postgres_kernel +from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.kernel import Kernel from cora.infrastructure.ports import ( LLM, @@ -383,3 +391,46 @@ def make_pg_profile_store(pool: asyncpg.Pool) -> ProfileStore: "seed_capability_postgres", "seed_run_upstream_chain_postgres", ] + + +async def promote_seeded_agent( + deps: Kernel, + agent_id: UUID, + *, + principal_id: UUID, + correlation_id: UUID, + occurred_at: datetime, +) -> None: + """Promote a seeded Defined fleet Agent to Versioned. + + The subscribers' lifecycle gate fires on Versioned only (Defined + means "registered as config, NOT yet ready for invocation"), so + integration tests that exercise a subscriber end-to-end must mirror + the production bootstrap-then-promote ceremony after seeding. + Direct event append rather than the version_agent handler so pinned + FixedIdGenerator queues in scenario tests stay undisturbed. + Idempotent: an Agent already past Defined is left untouched. + """ + agent = await load_agent(deps.event_store, agent_id) + if agent is None or agent.status is not AgentStatus.DEFINED: + return + versioned = AgentVersioned( + agent_id=agent_id, version=agent.version.value, occurred_at=occurred_at + ) + await deps.event_store.append( + stream_type="Agent", + stream_id=agent_id, + expected_version=1, + events=[ + to_new_event( + event_type=agent_event_type_name(versioned), + payload=agent_to_payload(versioned), + occurred_at=occurred_at, + event_id=uuid4(), + command_name="VersionAgent", + correlation_id=correlation_id, + causation_id=None, + principal_id=principal_id, + ) + ], + ) diff --git a/apps/api/tests/integration/scenarios/test_2bm_agent_cost_overrun_pause.py b/apps/api/tests/integration/scenarios/test_2bm_agent_cost_overrun_pause.py index 33055e4fdc0..e50137a8cbc 100644 --- a/apps/api/tests/integration/scenarios/test_2bm_agent_cost_overrun_pause.py +++ b/apps/api/tests/integration/scenarios/test_2bm_agent_cost_overrun_pause.py @@ -54,18 +54,21 @@ crossed 80% of monthly cap with two weeks of beamtime remaining; runaway projected. 2. Operator suspends the RunDebriefer agent (`suspend_agent`) - citing the cost overrun. The agent's `actor_id` is now - associated with an Actor that the subscriber's revocation - gate (per [[project_run_debrief_design]] security gate-review) - will treat as paused. + citing the cost overrun. The subscriber's suspension gate + (per-apply `Agent.status` check in + `RunDebrieferSubscriber.apply`) now skips every subsequent + terminal Run event: no LLM call, no Decision, until resume. 3. Operator tightens the budget envelope (`update_agent_budget`): drops `monthly_usd_cap` to the remaining-budget amount and adds a per-day token cap to short-circuit any single-day spike. 4. Operator resumes the agent (`resume_agent`) once the new budget is in place. The agent is back in `Versioned` state - and the subscriber will pick up subsequent terminal Run - events. + and the subscriber picks up subsequent terminal Run events, + with the coarse post-hoc budget gate enforcing the tightened + caps against recorded spend (see + `tests/unit/agent/test_budget_gate.py` and the subscriber + budget-gate tests for the enforcement behavior itself). ## Why a separate scenario @@ -193,8 +196,9 @@ async def test_agent_cost_overrun_pause_plays_out_end_to_end( assert versioned.status is AgentStatus.VERSIONED # ----- Initial budget envelope ----- - # Operator declares the initial spend ceiling. Declaration-only; - # enforcement deferred to the Budget BC. + # Operator declares the initial spend ceiling; the subscribers' + # coarse post-hoc gate (cora.agent._budget_gate) enforces it + # against recorded spend at each LLM seam. await bind_update_budget(deps)( UpdateAgentBudget( diff --git a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer.py b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer.py index 8dfbafbffd3..df1482be5ab 100644 --- a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer.py +++ b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer.py @@ -111,7 +111,11 @@ from cora.run.features.start_run import bind as bind_start_run from cora.subject.features.mount_subject import MountSubject from cora.subject.features.mount_subject import bind as bind_mount_subject -from tests.integration._helpers import build_postgres_deps, make_pg_profile_store +from tests.integration._helpers import ( + build_postgres_deps, + make_pg_profile_store, + promote_seeded_agent, +) from tests.integration.scenarios._beamtime_fixture import ( BeamtimeSpec, beamtime_id_prefix, @@ -347,6 +351,13 @@ async def test_run_debrief_agent_fires_on_terminal_run( # Bootstrap the agent (idempotent on re-invocation; uses SYSTEM_PRINCIPAL_ID # internally so the seed events bypass the FixedIdGenerator queue). await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) # Load the terminal RunCompleted event from the Run stream. run_events, _run_version = await deps.event_store.load("Run", _RUN_ID) diff --git a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_aborted.py b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_aborted.py index dc605f356bf..5199c73ebf7 100644 --- a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_aborted.py +++ b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_aborted.py @@ -126,6 +126,7 @@ from tests.integration._helpers import ( build_postgres_deps, make_pg_profile_store, + promote_seeded_agent, seed_capability_postgres, ) from tests.integration.scenarios._beamtime_fixture import ( @@ -423,6 +424,13 @@ async def test_run_debrief_agent_fires_on_equipment_abort( # ----- Agent fires on terminal RunAborted; emits EquipmentAbort ----- await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) run_events, _run_version = await deps.event_store.load("Run", _RUN_ID) terminal_events = [e for e in run_events if e.event_type == "RunAborted"] diff --git a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_degraded.py b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_degraded.py index 58953895b5b..be1a5dcd6b0 100644 --- a/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_degraded.py +++ b/apps/api/tests/integration/scenarios/test_2bm_run_debriefer_degraded.py @@ -105,7 +105,11 @@ from cora.run.features.start_run import bind as bind_start_run from cora.subject.features.mount_subject import MountSubject from cora.subject.features.mount_subject import bind as bind_mount_subject -from tests.integration._helpers import build_postgres_deps, make_pg_profile_store +from tests.integration._helpers import ( + build_postgres_deps, + make_pg_profile_store, + promote_seeded_agent, +) from tests.integration.scenarios._beamtime_fixture import ( BeamtimeSpec, beamtime_id_prefix, @@ -344,6 +348,13 @@ async def test_run_debrief_agent_fires_on_degraded_completion( # ----- Agent fires on terminal RunCompleted; emits DegradedCompletion ----- await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) run_events, _run_version = await deps.event_store.load("Run", _RUN_ID) terminal_events = [e for e in run_events if e.event_type == "RunCompleted"] diff --git a/apps/api/tests/integration/test_append_inferences_handler_postgres.py b/apps/api/tests/integration/test_append_inferences_handler_postgres.py index f74284edcc1..f7d98a96c01 100644 --- a/apps/api/tests/integration/test_append_inferences_handler_postgres.py +++ b/apps/api/tests/integration/test_append_inferences_handler_postgres.py @@ -58,7 +58,7 @@ async def _read_entries_for_decision( response_id, response_model, request_temperature, request_top_p, request_max_tokens, output_type, finish_reasons, - input_tokens, output_tokens, + input_tokens, output_tokens, cost_usd, agent_id, agent_name, agent_description, conversation_id, tool_name, tool_call_id, tool_type, messages @@ -146,7 +146,8 @@ async def test_append_inferences_full_lazy_open_and_jsonb_round_trip( finish_reasons=("end_turn",), input_tokens=512, output_tokens=256, - agent_id="agent-7e", + cost_usd=0.00896, + agent_id=str(_PRINCIPAL_ID), agent_name="ApprovalAgent", conversation_id="conv-abc", messages={ @@ -193,6 +194,8 @@ async def test_append_inferences_full_lazy_open_and_jsonb_round_trip( assert row_a["request_temperature"] == 0.7 assert row_a["input_tokens"] == 512 assert row_a["output_tokens"] == 256 + assert row_a["cost_usd"] == pytest.approx(0.00896) + assert row_b["cost_usd"] is None assert row_a["finish_reasons"] == ["end_turn"] assert row_a["agent_name"] == "ApprovalAgent" assert row_a["conversation_id"] == "conv-abc" @@ -313,6 +316,7 @@ async def test_postgres_reasoning_store_dedups_on_event_id( finish_reasons=(), input_tokens=None, output_tokens=None, + cost_usd=None, agent_id=None, agent_name=None, agent_description=None, @@ -343,6 +347,7 @@ async def test_postgres_reasoning_store_dedups_on_event_id( finish_reasons=(), input_tokens=None, output_tokens=None, + cost_usd=None, agent_id=None, agent_name=None, agent_description=None, diff --git a/apps/api/tests/integration/test_caution_drafter_subscriber_postgres.py b/apps/api/tests/integration/test_caution_drafter_subscriber_postgres.py index 5bdc1806dde..32ca1aaea1b 100644 --- a/apps/api/tests/integration/test_caution_drafter_subscriber_postgres.py +++ b/apps/api/tests/integration/test_caution_drafter_subscriber_postgres.py @@ -63,7 +63,7 @@ from cora.run.aggregates.run import to_payload as run_to_payload from cora.run.aggregates.run.events import RunAborted from cora.shared.identity import ActorId -from tests.integration._helpers import build_postgres_deps +from tests.integration._helpers import build_postgres_deps, promote_seeded_agent _NOW = datetime(2026, 5, 17, 14, 0, 0, tzinfo=UTC) _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) @@ -190,8 +190,22 @@ async def test_subscriber_writes_caution_proposal_decision_end_to_end( """Step 1-5 of the end-to-end walk: subscriber emits the Decision.""" deps = build_postgres_deps(db_pool, now=_NOW, ids=[uuid4() for _ in range(10)]) await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) # Idempotent on real PG. await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) plan_id = uuid4() run_id = uuid4() @@ -235,6 +249,13 @@ async def test_end_to_end_cross_bc_promotion_registers_real_caution( promotes via Agent BC's slice, Caution lands in Caution BC's stream.""" deps = build_postgres_deps(db_pool, now=_NOW, ids=[uuid4() for _ in range(10)]) await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) plan_id = uuid4() run_id = uuid4() @@ -287,6 +308,13 @@ async def test_subscriber_is_at_most_once_on_real_postgres( (ConcurrencyError on the deterministic decision_id is caught).""" deps = build_postgres_deps(db_pool, now=_NOW, ids=[uuid4() for _ in range(10)]) await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) plan_id = uuid4() run_id = uuid4() @@ -363,6 +391,13 @@ async def _seed_supersede_decision( from cora.decision.aggregates.decision import to_payload as decision_to_payload await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) proposed = { "target_kind": "Asset", @@ -608,6 +643,13 @@ async def test_subscriber_records_inference_row_end_to_end( carrying the CautionDrafter LLM call's provenance.""" deps = build_postgres_deps(db_pool, now=_NOW, ids=[uuid4() for _ in range(10)]) await seed_caution_drafter_agent(deps) + await promote_seeded_agent( + deps, + CAUTION_DRAFTER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) plan_id = uuid4() run_id = uuid4() await _seed_plan(deps, plan_id) diff --git a/apps/api/tests/integration/test_run_debriefer_subscriber_postgres.py b/apps/api/tests/integration/test_run_debriefer_subscriber_postgres.py index f48edec45db..301d4da2d9d 100644 --- a/apps/api/tests/integration/test_run_debriefer_subscriber_postgres.py +++ b/apps/api/tests/integration/test_run_debriefer_subscriber_postgres.py @@ -46,7 +46,7 @@ from cora.run.aggregates.run import event_type_name as run_event_type_name from cora.run.aggregates.run import to_payload as run_to_payload from cora.run.aggregates.run.events import RunCompleted -from tests.integration._helpers import build_postgres_deps +from tests.integration._helpers import build_postgres_deps, promote_seeded_agent _NOW = datetime(2026, 5, 17, 14, 0, 0, tzinfo=UTC) _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) @@ -124,8 +124,22 @@ async def test_seed_and_subscriber_write_decision_end_to_end( # Bootstrap: seed RunDebriefer Agent + Actor. await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) # Second call must be a no-op (idempotent on real PG). await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) # Seed a Run. run_id = uuid4() @@ -165,6 +179,13 @@ async def test_subscriber_retry_is_at_most_once_on_real_postgres( treated as no-op).""" deps = build_postgres_deps(db_pool, now=_NOW) await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) run_id = uuid4() plan_id = UUID("01900000-0000-7000-8000-000000000401") @@ -227,6 +248,13 @@ async def test_seed_does_not_collide_with_pre_existing_actor( # Now the seed runs. The Actor stream collides (rolls back the # whole append_streams batch), seed returns cleanly. await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) # Agent stream never got written (the whole batch rolled back). # This is a documented edge case; future iteration can split the @@ -298,6 +326,13 @@ async def test_subscriber_records_inference_row_end_to_end( the inference logbook on the Decision stream.""" deps = build_postgres_deps(db_pool, now=_NOW, ids=_LOGBOOK_IDS) await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) run_id = uuid4() plan_id = UUID("01900000-0000-7000-8000-000000000401") await _seed_run(deps, run_id, plan_id) @@ -336,6 +371,13 @@ async def test_subscriber_inference_write_is_idempotent_on_retry( event_id; the store's ON CONFLICT keeps exactly one row.""" deps = build_postgres_deps(db_pool, now=_NOW, ids=_LOGBOOK_IDS) await seed_run_debriefer_agent(deps) + await promote_seeded_agent( + deps, + RUN_DEBRIEFER_AGENT_ID, + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + occurred_at=_NOW, + ) run_id = uuid4() plan_id = UUID("01900000-0000-7000-8000-000000000401") await _seed_run(deps, run_id, plan_id) diff --git a/apps/api/tests/integration/test_spend_lookup_postgres.py b/apps/api/tests/integration/test_spend_lookup_postgres.py new file mode 100644 index 00000000000..fd08485591b --- /dev/null +++ b/apps/api/tests/integration/test_spend_lookup_postgres.py @@ -0,0 +1,143 @@ +"""Integration tests for `PostgresSpendLookup` over `entries_decision_inferences`. + +Seeds inference rows directly through `PostgresInferenceStore` (the same +write path production uses) and verifies the spend sums respect the +agent filter, the half-open window bounds, and the NULL-cost COALESCE +that keeps pre-migration history from inflating a balance. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import asyncpg +import pytest + +from cora.decision.adapters import PostgresSpendLookup +from cora.decision.aggregates.decision import Inference +from cora.decision.aggregates.decision.entries import PostgresInferenceStore + +_AGENT_ID = UUID("01900000-0000-7000-8000-0000000a0010") +_OTHER_AGENT_ID = UUID("01900000-0000-7000-8000-0000000a0020") + +_WINDOW_START = datetime(2026, 7, 1, tzinfo=UTC) +_WINDOW_END = datetime(2026, 8, 1, tzinfo=UTC) + + +def _row( + *, + agent_id: UUID | None, + occurred_at: datetime, + cost_usd: float | None, + input_tokens: int | None = 100, + output_tokens: int | None = 50, +) -> Inference: + return Inference( + event_id=uuid4(), + decision_id=uuid4(), + logbook_id=uuid4(), + correlation_id=uuid4(), + causation_id=None, + occurred_at=occurred_at, + duration=None, + operation_name="chat", + provider_name="anthropic", + request_model="claude-haiku-4-5", + response_id=None, + response_model=None, + request_temperature=None, + request_top_p=None, + request_max_tokens=None, + output_type=None, + finish_reasons=(), + input_tokens=input_tokens, + output_tokens=output_tokens, + cost_usd=cost_usd, + agent_id=str(agent_id) if agent_id is not None else None, + agent_name=None, + agent_description=None, + conversation_id=None, + tool_name=None, + tool_call_id=None, + tool_type=None, + messages=None, + ) + + +@pytest.mark.integration +async def test_sums_only_the_agents_rows_inside_the_window(db_pool: asyncpg.Pool) -> None: + """usd/tokens/count sum the target agent's in-window rows; another + agent's rows, agentless rows, and out-of-window rows are excluded; + the window is half-open (a row exactly at window_start counts, a + row exactly at window_end does not).""" + store = PostgresInferenceStore(db_pool) + in_window = datetime(2026, 7, 10, tzinfo=UTC) + await store.append( + [ + _row(agent_id=_AGENT_ID, occurred_at=in_window, cost_usd=0.002), + _row(agent_id=_AGENT_ID, occurred_at=in_window, cost_usd=0.003), + _row(agent_id=_AGENT_ID, occurred_at=_WINDOW_START, cost_usd=0.001), + # Excluded: other agent, agentless (operator-attributed row), + # before start, exactly at end (half-open). + _row(agent_id=_OTHER_AGENT_ID, occurred_at=in_window, cost_usd=5.0), + _row(agent_id=None, occurred_at=in_window, cost_usd=3.0), + _row(agent_id=_AGENT_ID, occurred_at=datetime(2026, 6, 30, tzinfo=UTC), cost_usd=7.0), + _row(agent_id=_AGENT_ID, occurred_at=_WINDOW_END, cost_usd=9.0), + ] + ) + + result = await PostgresSpendLookup(db_pool).find_agent_spend( + agent_id=_AGENT_ID, + window_start=_WINDOW_START, + window_end=_WINDOW_END, + ) + + assert result.agent_id == _AGENT_ID + assert result.usd_spent == pytest.approx(0.006) + assert result.tokens_spent == 450 # 3 rows x (100 in + 50 out) + assert result.call_count == 3 + + +@pytest.mark.integration +async def test_null_costs_and_tokens_sum_as_zero_not_poison(db_pool: asyncpg.Pool) -> None: + """Pre-migration rows (cost_usd IS NULL) and rows with NULL token + counts still COUNT as calls but contribute $0 / 0 tokens, keeping + the gate permissive rather than blocking on unknowable history.""" + store = PostgresInferenceStore(db_pool) + agent_id = uuid4() + in_window = datetime(2026, 7, 15, tzinfo=UTC) + await store.append( + [ + _row(agent_id=agent_id, occurred_at=in_window, cost_usd=None), + _row( + agent_id=agent_id, + occurred_at=in_window, + cost_usd=0.01, + input_tokens=None, + output_tokens=None, + ), + ] + ) + + result = await PostgresSpendLookup(db_pool).find_agent_spend( + agent_id=agent_id, + window_start=_WINDOW_START, + window_end=_WINDOW_END, + ) + + assert result.usd_spent == pytest.approx(0.01) + assert result.tokens_spent == 150 # only the NULL-cost row carried tokens + assert result.call_count == 2 + + +@pytest.mark.integration +async def test_agent_with_no_rows_returns_a_zero_row(db_pool: asyncpg.Pool) -> None: + """'No spend' and 'unknown agent' are the same answer to a gate.""" + result = await PostgresSpendLookup(db_pool).find_agent_spend( + agent_id=uuid4(), + window_start=_WINDOW_START, + window_end=_WINDOW_END, + ) + + assert result.usd_spent == 0.0 + assert result.tokens_spent == 0 + assert result.call_count == 0 diff --git a/apps/api/tests/unit/_helpers.py b/apps/api/tests/unit/_helpers.py index dd8df5e8353..31ed32b5693 100644 --- a/apps/api/tests/unit/_helpers.py +++ b/apps/api/tests/unit/_helpers.py @@ -60,6 +60,7 @@ FixedIdGenerator, ProfileStore, RoleLookup, + SpendLookup, ) from cora.recipe.aggregates.capability import ( CapabilityCode, @@ -131,6 +132,7 @@ def build_deps( family_lookup: FamilyLookup | None = None, assembly_lookup: AssemblyLookup | None = None, role_lookup: RoleLookup | None = None, + spend_lookup: SpendLookup | None = None, ) -> Kernel: """Build a Kernel for unit-test handler invocation. @@ -229,6 +231,7 @@ def build_deps( publish_port=InMemoryPublishPort(), signature_port=InMemorySignaturePort(), permit_lookup=InMemoryPermitLookup(), + spend_lookup=spend_lookup, ) diff --git a/apps/api/tests/unit/agent/_helpers.py b/apps/api/tests/unit/agent/_helpers.py index d29a2bd887f..4b3a511a9c3 100644 --- a/apps/api/tests/unit/agent/_helpers.py +++ b/apps/api/tests/unit/agent/_helpers.py @@ -28,6 +28,7 @@ from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.ports import AgentInferenceTrace +from cora.infrastructure.ports.spend_lookup import SpendLookupResult from cora.infrastructure.signing import event_type_to_payload_type from cora.shared.content_hash import canonical_body_bytes, pae_bytes from cora.shared.identity import ActorId @@ -43,6 +44,44 @@ class RecordedInference: causation_id: UUID | None +class FakeSpendLookup: + """SpendLookup stub returning configured sums; records each query. + + `windows` captures every `(agent_id, window_start, window_end)` + asked for, so gate tests can assert the window math without a + real store. + """ + + def __init__( + self, + *, + usd_spent: float = 0.0, + tokens_spent: int = 0, + call_count: int = 0, + ) -> None: + self.usd_spent = usd_spent + self.tokens_spent = tokens_spent + self.call_count = call_count + self.windows: list[tuple[UUID, datetime, datetime]] = [] + + async def find_agent_spend( + self, + *, + agent_id: UUID, + window_start: datetime, + window_end: datetime, + ) -> SpendLookupResult: + self.windows.append((agent_id, window_start, window_end)) + return SpendLookupResult( + agent_id=agent_id, + window_start=window_start, + window_end=window_end, + usd_spent=self.usd_spent, + tokens_spent=self.tokens_spent, + call_count=self.call_count, + ) + + class FakeInferenceRecorder: """Spy `InferenceRecorder` for agent-subscriber tests. @@ -84,8 +123,15 @@ async def seed_defined_agent( correlation_id: UUID, principal_id: UUID, occurred_at: datetime, + monthly_usd_cap: float | None = None, + daily_token_cap: int | None = None, ) -> None: - """Append a single `AgentDefined` event to a fresh Agent stream.""" + """Append a single `AgentDefined` event to a fresh Agent stream. + + Optional caps land on the genesis payload (the additive + `AgentDefined` budget fields) so budget-gate tests can seed a + capped agent without a second `AgentBudgetUpdated` append. + """ genesis = AgentDefined( agent_id=agent_id, kind="RunDebriefer", @@ -97,6 +143,8 @@ async def seed_defined_agent( prompt_template_id=None, capabilities=frozenset(), occurred_at=occurred_at, + monthly_usd_cap=monthly_usd_cap, + daily_token_cap=daily_token_cap, ) await store.append( stream_type="Agent", @@ -127,6 +175,8 @@ async def seed_versioned_agent( principal_id: UUID, defined_at: datetime, versioned_at: datetime, + monthly_usd_cap: float | None = None, + daily_token_cap: int | None = None, ) -> None: """Seed Defined then Versioned, leaving the Agent at stream version 2.""" await seed_defined_agent( @@ -136,6 +186,8 @@ async def seed_versioned_agent( correlation_id=correlation_id, principal_id=principal_id, occurred_at=defined_at, + monthly_usd_cap=monthly_usd_cap, + daily_token_cap=daily_token_cap, ) versioned = AgentVersioned(agent_id=agent_id, version="v1", occurred_at=versioned_at) await store.append( diff --git a/apps/api/tests/unit/agent/test_budget_gate.py b/apps/api/tests/unit/agent/test_budget_gate.py new file mode 100644 index 00000000000..c53e8251f02 --- /dev/null +++ b/apps/api/tests/unit/agent/test_budget_gate.py @@ -0,0 +1,178 @@ +"""Unit tests for the coarse post-hoc budget gate (`cora.agent._budget_gate`).""" + +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest + +from cora.agent._budget_gate import ( + calendar_day_window, + calendar_month_window, + find_budget_breach, +) +from cora.agent.aggregates.agent import load_agent +from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore +from tests.unit.agent._helpers import FakeSpendLookup, seed_defined_agent + +_NOW = datetime(2026, 7, 11, 15, 30, 0, tzinfo=UTC) +_CORRELATION_ID = uuid4() +_PRINCIPAL_ID = uuid4() + + +async def _agent_with_caps( + store: InMemoryEventStore, + *, + monthly_usd_cap: float | None = None, + daily_token_cap: int | None = None, +): + agent_id = uuid4() + await seed_defined_agent( + store, + agent_id=agent_id, + genesis_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + occurred_at=_NOW, + monthly_usd_cap=monthly_usd_cap, + daily_token_cap=daily_token_cap, + ) + return await load_agent(store, agent_id) + + +# ---------- Window helpers ---------- + + +@pytest.mark.unit +def test_calendar_month_window_mid_month() -> None: + start, end = calendar_month_window(_NOW) + assert start == datetime(2026, 7, 1, tzinfo=UTC) + assert end == datetime(2026, 8, 1, tzinfo=UTC) + + +@pytest.mark.unit +def test_calendar_month_window_december_rolls_into_next_year() -> None: + start, end = calendar_month_window(datetime(2026, 12, 31, 23, 59, tzinfo=UTC)) + assert start == datetime(2026, 12, 1, tzinfo=UTC) + assert end == datetime(2027, 1, 1, tzinfo=UTC) + + +@pytest.mark.unit +def test_calendar_day_window_brackets_the_utc_day() -> None: + start, end = calendar_day_window(_NOW) + assert start == datetime(2026, 7, 11, tzinfo=UTC) + assert end == datetime(2026, 7, 12, tzinfo=UTC) + + +# ---------- find_budget_breach ---------- + + +@pytest.mark.unit +async def test_no_agent_stream_permits_without_querying_spend() -> None: + """Declaration is opt-in: a missing Agent stream never blocks.""" + lookup = FakeSpendLookup(usd_spent=1_000_000.0) + + breach = await find_budget_breach(agent=None, spend_lookup=lookup, as_of=_NOW) + + assert breach is None + assert lookup.windows == [] + + +@pytest.mark.unit +async def test_no_declared_budget_permits_without_querying_spend() -> None: + store = InMemoryEventStore() + agent = await _agent_with_caps(store) + lookup = FakeSpendLookup(usd_spent=1_000_000.0) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is None + assert lookup.windows == [] + + +@pytest.mark.unit +async def test_spend_under_both_caps_permits() -> None: + store = InMemoryEventStore() + agent = await _agent_with_caps(store, monthly_usd_cap=500.0, daily_token_cap=2_000_000) + lookup = FakeSpendLookup(usd_spent=499.99, tokens_spent=1_999_999) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is None + assert len(lookup.windows) == 2 # one SUM per declared cap + + +@pytest.mark.unit +async def test_monthly_usd_cap_reached_breaches_over_the_month_window() -> None: + store = InMemoryEventStore() + agent = await _agent_with_caps(store, monthly_usd_cap=120.0) + lookup = FakeSpendLookup(usd_spent=121.3) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is not None + assert breach.cap_kind == "monthly_usd_cap" + assert breach.cap_value == 120.0 + assert breach.spent == pytest.approx(121.3) + assert breach.window_start == datetime(2026, 7, 1, tzinfo=UTC) + assert breach.window_end == datetime(2026, 8, 1, tzinfo=UTC) + assert agent is not None + assert lookup.windows == [(agent.id, breach.window_start, breach.window_end)] + + +@pytest.mark.unit +async def test_daily_token_cap_reached_breaches_over_the_day_window() -> None: + store = InMemoryEventStore() + agent = await _agent_with_caps(store, daily_token_cap=1_000_000) + lookup = FakeSpendLookup(tokens_spent=1_000_000) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is not None + assert breach.cap_kind == "daily_token_cap" + assert breach.window_start == datetime(2026, 7, 11, tzinfo=UTC) + assert breach.window_end == datetime(2026, 7, 12, tzinfo=UTC) + + +@pytest.mark.unit +async def test_monthly_breach_wins_when_both_caps_are_exhausted() -> None: + """Caps are checked in declaration order; the first breach is + reported and the daily SUM is never issued.""" + store = InMemoryEventStore() + agent = await _agent_with_caps(store, monthly_usd_cap=10.0, daily_token_cap=100) + lookup = FakeSpendLookup(usd_spent=11.0, tokens_spent=101) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is not None + assert breach.cap_kind == "monthly_usd_cap" + assert len(lookup.windows) == 1 + + +@pytest.mark.unit +async def test_daily_breach_detected_when_monthly_cap_has_headroom() -> None: + """The daily check must run even after a PASSING monthly check; + an early return after monthly headroom would silently disable + the token cap.""" + store = InMemoryEventStore() + agent = await _agent_with_caps(store, monthly_usd_cap=500.0, daily_token_cap=100) + lookup = FakeSpendLookup(usd_spent=1.0, tokens_spent=101) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is not None + assert breach.cap_kind == "daily_token_cap" + assert len(lookup.windows) == 2 + + +@pytest.mark.unit +async def test_zero_cap_refuses_every_call() -> None: + """AgentBudget documents a zero cap as recorded no-spend intent; + the gate is what makes that intent real (0 spent >= 0 cap).""" + store = InMemoryEventStore() + agent = await _agent_with_caps(store, monthly_usd_cap=0.0) + lookup = FakeSpendLookup(usd_spent=0.0) + + breach = await find_budget_breach(agent=agent, spend_lookup=lookup, as_of=_NOW) + + assert breach is not None + assert breach.cap_kind == "monthly_usd_cap" 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 d3ea16aefb9..2ab2abd5730 100644 --- a/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py +++ b/apps/api/tests/unit/agent/test_caution_drafter_subscriber.py @@ -63,7 +63,13 @@ RunCompleted, ) from tests.unit._helpers import build_deps -from tests.unit.agent._helpers import Ed25519FakeSigner, FakeInferenceRecorder +from tests.unit.agent._helpers import ( + Ed25519FakeSigner, + FakeInferenceRecorder, + FakeSpendLookup, + seed_suspended_agent, + seed_versioned_agent, +) _NOW = datetime(2026, 5, 17, 14, 0, 0, tzinfo=UTC) _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) @@ -238,12 +244,14 @@ async def _build_subscriber( event_store: InMemoryEventStore, llm: FakeLLM, inference_recorder: FakeInferenceRecorder | None = None, + spend_lookup: FakeSpendLookup | None = None, ) -> CautionDrafterSubscriber: return CautionDrafterSubscriber( event_store=event_store, llm=llm, caution_lookup=AlwaysQuietCautionLookup(), inference_recorder=inference_recorder, + spend_lookup=spend_lookup, ) @@ -432,6 +440,112 @@ async def test_apply_writes_no_action_deferred_on_llm_failure() -> None: assert decision.inputs["failure_error_class"] == "LLMServerError" +# ---------- Suspension gate + budget gate (mirror RunDebriefer) ---------- + + +@pytest.mark.unit +async def test_apply_skips_entirely_when_agent_suspended() -> None: + """A Suspended agent takes no actions: no LLM call, no Decision.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await seed_suspended_agent( + store, + agent_id=CAUTION_DRAFTER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + suspend_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + suspended_at=_NOW, + reason="cost overrun", + ) + 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 await load_decision(store, _derive_decision_id(event.event_id)) is None + assert llm.received == [] + + +@pytest.mark.unit +async def test_apply_defers_noaction_when_monthly_usd_cap_exhausted() -> None: + """Cap reached: the LLM call is skipped and the refusal is recorded + as this Run's NoAction Decision (coarse post-hoc tier).""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + recorder = FakeInferenceRecorder() + await _seed_caution_drafter_actor(store) + await seed_versioned_agent( + store, + agent_id=CAUTION_DRAFTER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + monthly_usd_cap=120.0, + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber( + store, llm, recorder, spend_lookup=FakeSpendLookup(usd_spent=121.3) + ) + 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.choice.value == "NoAction" + assert "Agent budget exhausted: monthly_usd_cap" in (decision.reasoning or "") + assert decision.inputs is not None + assert decision.inputs["failure_error_class"] == "AgentBudgetExhausted" + assert decision.inputs["budget_cap_kind"] == "monthly_usd_cap" + assert recorder.calls == [] + assert llm.received == [] + + +@pytest.mark.unit +async def test_apply_proceeds_normally_when_spend_is_under_cap() -> None: + """A declared budget with headroom never interferes with the + draft: the gate is invisible until a cap is exhausted.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_NO_ACTION]) + await _seed_caution_drafter_actor(store) + await seed_versioned_agent( + store, + agent_id=CAUTION_DRAFTER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + monthly_usd_cap=500.0, + ) + await _seed_plan(store) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, spend_lookup=FakeSpendLookup(usd_spent=1.0)) + 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 len(llm.received) == 1 + assert "failure_error_class" not in (decision.inputs or {}) + + class _RaisingLLM: """LLM test-double whose `chat()` always raises a configured exception. @@ -879,6 +993,9 @@ def test_make_caution_drafter_subscriber_constructs_when_llm_is_set() -> None: deps = build_deps(llm=FakeLLM()) subscriber = make_caution_drafter_subscriber(deps) assert isinstance(subscriber, CautionDrafterSubscriber) + # The budget gate silently disables if the factory drops this wiring + # (the constructor default is the permissive AlwaysZeroSpendLookup). + assert subscriber.spend_lookup is deps.spend_lookup # ---------- Signer wiring ---------- @@ -1222,6 +1339,8 @@ async def test_apply_records_inference_on_proposal() -> None: assert call.trace.response_model == "claude-sonnet-4-6-20260201" assert call.trace.input_tokens == 2048 assert call.trace.output_tokens == 320 + # Sonnet 4.6 at $3/$15 per MTok: 2048 in + 320 out. + assert call.trace.cost_usd == pytest.approx(0.006144 + 0.0048) assert call.trace.finish_reasons == ("tool_use",) assert call.trace.agent_id == str(CAUTION_DRAFTER_AGENT_ID) assert call.trace.agent_name == CAUTION_DRAFTER_AGENT_NAME diff --git a/apps/api/tests/unit/agent/test_regenerate_run_debrief_handler.py b/apps/api/tests/unit/agent/test_regenerate_run_debrief_handler.py index b524b12ceb7..8bf445d4d2a 100644 --- a/apps/api/tests/unit/agent/test_regenerate_run_debrief_handler.py +++ b/apps/api/tests/unit/agent/test_regenerate_run_debrief_handler.py @@ -22,7 +22,11 @@ ) 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 AgentDeactivatedError, AgentNotSeededError +from cora.agent.aggregates.agent import ( + AgentDeactivatedError, + AgentNotSeededError, + AgentSuspendedError, +) from cora.agent.errors import UnauthorizedError from cora.agent.features import regenerate_run_debrief from cora.agent.features.regenerate_run_debrief import RegenerateRunDebrief @@ -48,7 +52,12 @@ from cora.run.aggregates.run import to_payload as run_to_payload from cora.shared.identity import ActorId from tests.unit._helpers import build_deps -from tests.unit.agent._helpers import FakeInferenceRecorder +from tests.unit.agent._helpers import ( + FakeInferenceRecorder, + FakeSpendLookup, + seed_suspended_agent, + seed_versioned_agent, +) _NOW = datetime(2026, 5, 17, 16, 0, 0, tzinfo=UTC) _PRINCIPAL_ID = UUID("01900000-0000-7000-8000-000000088001") @@ -353,6 +362,87 @@ async def test_handler_raises_agent_deactivated_when_actor_inactive() -> None: assert llm.received == [] +@pytest.mark.unit +async def test_handler_raises_agent_suspended_when_agent_paused() -> None: + """An operator-suspended agent cannot author on-demand Decisions + either: suspend means STOP, including the manual path. Resume + first. Mirrors the subscriber's Versioned-only lifecycle gate.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + run_id = uuid4() + await _seed_actor(store) + await _seed_run(store, run_id) + await seed_suspended_agent( + store, + agent_id=RUN_DEBRIEFER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + suspend_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + suspended_at=_NOW, + reason="cost overrun", + ) + deps = build_deps( + ids=[_NEW_DECISION_ID], + now=_NOW, + event_store=store, + llm=llm, + ) + handler = bind(deps) + + with pytest.raises(AgentSuspendedError): + await handler( + RegenerateRunDebrief(run_id=run_id), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + assert llm.received == [] + + +@pytest.mark.unit +async def test_handler_proceeds_when_agent_budget_is_exhausted() -> None: + """Pin the deliberate asymmetry with the subscribers: the + operator-triggered regenerate is NOT budget-gated (an accountable + human explicitly asked; the call is still metered and debited), + so an exhausted cap must not block it.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + run_id = uuid4() + await _seed_actor(store) + await _seed_run(store, run_id) + 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, + monthly_usd_cap=100.0, + ) + deps = build_deps( + ids=[_NEW_DECISION_ID], + now=_NOW, + event_store=store, + llm=llm, + spend_lookup=FakeSpendLookup(usd_spent=250.0), + ) + handler = bind(deps) + + decision_id = await handler( + RegenerateRunDebrief(run_id=run_id), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + + assert decision_id == _NEW_DECISION_ID + assert len(llm.received) == 1 + + @pytest.mark.unit async def test_handler_raises_parent_missing_when_parent_absent() -> None: store = InMemoryEventStore() @@ -624,6 +714,8 @@ async def test_handler_records_inference_on_success() -> None: assert call.trace.response_model == "claude-haiku-4-5-20260201" assert call.trace.input_tokens == 1536 assert call.trace.output_tokens == 240 + # Haiku 4.5 at $1/$5 per MTok: 1536 in + 240 out. + assert call.trace.cost_usd == pytest.approx(0.001536 + 0.0012) assert call.trace.finish_reasons == ("tool_use",) # The inference is the agent's, attributed to the agent principal (not # the operator who issued the command). 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 74b296c7570..4238767ca2d 100644 --- a/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py +++ b/apps/api/tests/unit/agent/test_run_debriefer_subscriber.py @@ -21,6 +21,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 AgentDeprecated +from cora.agent.aggregates.agent import event_type_name as agent_event_type_name +from cora.agent.aggregates.agent import to_payload as agent_to_payload from cora.agent.seed import ( RUN_DEBRIEFER_AGENT_ID, RUN_DEBRIEFER_AGENT_NAME, @@ -34,6 +37,7 @@ from cora.agent.subscribers.run_debriefer import ( RunDebrieferSubscriber, _derive_decision_id, + make_run_debriefer_subscriber, redact_secrets, ) from cora.decision.aggregates.decision import load_decision @@ -61,7 +65,15 @@ RunStopped, RunTruncated, ) -from tests.unit.agent._helpers import Ed25519FakeSigner, FakeInferenceRecorder +from tests.unit._helpers import build_deps +from tests.unit.agent._helpers import ( + Ed25519FakeSigner, + FakeInferenceRecorder, + FakeSpendLookup, + seed_defined_agent, + seed_suspended_agent, + seed_versioned_agent, +) _NOW = datetime(2026, 5, 17, 14, 0, 0, tzinfo=UTC) _LATER = datetime(2026, 5, 17, 14, 47, 0, tzinfo=UTC) @@ -218,12 +230,14 @@ async def _build_subscriber( event_store: InMemoryEventStore, llm: FakeLLM, inference_recorder: FakeInferenceRecorder | None = None, + spend_lookup: FakeSpendLookup | None = None, ) -> RunDebrieferSubscriber: return RunDebrieferSubscriber( event_store=event_store, llm=llm, logbook_mirror=None, inference_recorder=inference_recorder, + spend_lookup=spend_lookup, ) @@ -397,6 +411,208 @@ async def test_apply_writes_debrief_deferred_on_llm_failure() -> None: assert decision.inputs["failure_error_class"] == "LLMServerError" +# ---------- Suspension gate + budget gate ---------- + + +@pytest.mark.unit +async def test_apply_skips_entirely_when_agent_suspended() -> None: + """A Suspended agent takes no actions: no LLM call, no Decision. + The per-apply suspension gate is what makes suspend_agent an + actual halt of LLM spend rather than a recorded intention.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store) + await seed_suspended_agent( + store, + agent_id=RUN_DEBRIEFER_AGENT_ID, + genesis_event_id=uuid4(), + version_event_id=uuid4(), + suspend_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + defined_at=_NOW, + versioned_at=_NOW, + suspended_at=_NOW, + reason="cost overrun", + ) + 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 await load_decision(store, _derive_decision_id(event.event_id)) is None + assert llm.received == [] + + +@pytest.mark.unit +async def test_apply_defers_debrief_when_monthly_usd_cap_exhausted() -> None: + """Cap reached: the LLM call is skipped and the refusal is recorded + as this Run's DebriefDeferred Decision (coarse post-hoc tier).""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + recorder = FakeInferenceRecorder() + await _seed_run_debrief_actor(store) + 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, + monthly_usd_cap=120.0, + ) + run_id = uuid4() + await _seed_run(store, run_id) + spend_lookup = FakeSpendLookup(usd_spent=121.3) + subscriber = await _build_subscriber(store, llm, recorder, spend_lookup=spend_lookup) + 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.choice.value == "DebriefDeferred" + assert "Agent budget exhausted: monthly_usd_cap" in (decision.reasoning or "") + assert decision.inputs is not None + assert decision.inputs["failure_error_class"] == "AgentBudgetExhausted" + assert decision.inputs["budget_cap_kind"] == "monthly_usd_cap" + assert recorder.calls == [] + assert llm.received == [] + # The window derives from the terminal event's occurred_at (_LATER, + # 2026-05-17), NOT wall clock: the replay-determinism invariant. + assert spend_lookup.windows == [ + ( + RUN_DEBRIEFER_AGENT_ID, + datetime(2026, 5, 1, tzinfo=UTC), + datetime(2026, 6, 1, tzinfo=UTC), + ) + ] + + +@pytest.mark.unit +async def test_apply_defers_debrief_when_daily_token_cap_exhausted() -> None: + """The daily token cap gates end to end: the deferred Decision + carries the daily cap kind and the window is the event's UTC day.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store) + 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, + daily_token_cap=1_000_000, + ) + run_id = uuid4() + await _seed_run(store, run_id) + spend_lookup = FakeSpendLookup(tokens_spent=1_000_000) + subscriber = await _build_subscriber(store, llm, spend_lookup=spend_lookup) + 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.choice.value == "DebriefDeferred" + assert decision.inputs is not None + assert decision.inputs["budget_cap_kind"] == "daily_token_cap" + assert llm.received == [] + assert spend_lookup.windows == [ + ( + RUN_DEBRIEFER_AGENT_ID, + datetime(2026, 5, 17, tzinfo=UTC), + datetime(2026, 5, 18, tzinfo=UTC), + ) + ] + + +@pytest.mark.unit +async def test_apply_skips_entirely_when_agent_deprecated() -> None: + """The lifecycle gate is Versioned-only: Deprecated is terminal + ("future invocations must pick a non-Deprecated Agent"), so a + retired agent makes no LLM calls and writes no Decisions.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store) + await seed_defined_agent( + store, + agent_id=RUN_DEBRIEFER_AGENT_ID, + genesis_event_id=uuid4(), + correlation_id=_CORRELATION_ID, + principal_id=_PRINCIPAL_ID, + occurred_at=_NOW, + ) + deprecated = AgentDeprecated( + agent_id=RUN_DEBRIEFER_AGENT_ID, + reason="model regression", + occurred_at=_NOW, + ) + await store.append( + stream_type="Agent", + stream_id=RUN_DEBRIEFER_AGENT_ID, + expected_version=1, + events=[ + to_new_event( + event_type=agent_event_type_name(deprecated), + payload=agent_to_payload(deprecated), + occurred_at=deprecated.occurred_at, + event_id=uuid4(), + command_name="DeprecateAgent", + correlation_id=_CORRELATION_ID, + causation_id=None, + principal_id=_PRINCIPAL_ID, + ) + ], + ) + 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 await load_decision(store, _derive_decision_id(event.event_id)) is None + assert llm.received == [] + + +@pytest.mark.unit +async def test_apply_proceeds_normally_when_spend_is_under_cap() -> None: + """A declared budget with headroom never interferes with the + debrief: the gate is invisible until a cap is exhausted.""" + store = InMemoryEventStore() + llm = FakeLLM(responses=[_CANNED_OK]) + await _seed_run_debrief_actor(store) + 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, + monthly_usd_cap=500.0, + ) + run_id = uuid4() + await _seed_run(store, run_id) + subscriber = await _build_subscriber(store, llm, spend_lookup=FakeSpendLookup(usd_spent=1.0)) + 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.choice.value == "NominalCompletion" + + @pytest.mark.unit async def test_debrief_deferred_decision_id_matches_success_path() -> None: """Both success and deferred paths derive the same decision_id @@ -1317,6 +1533,8 @@ async def test_apply_records_inference_on_success() -> None: assert call.trace.response_model == "claude-haiku-4-5-20260201" assert call.trace.input_tokens == 1280 assert call.trace.output_tokens == 214 + # Haiku 4.5 at $1/$5 per MTok: 1280 in + 214 out. + assert call.trace.cost_usd == pytest.approx(0.00128 + 0.00107) assert call.trace.finish_reasons == ("tool_use",) assert call.trace.request_max_tokens == 1024 assert call.trace.agent_id == str(RUN_DEBRIEFER_AGENT_ID) @@ -1386,3 +1604,14 @@ async def test_apply_records_inference_with_stable_event_id_under_retry() -> Non assert len(recorder.calls) == 2 assert recorder.calls[0].trace.event_id == recorder.calls[1].trace.event_id + + +@pytest.mark.unit +def test_make_run_debriefer_subscriber_wires_spend_lookup_from_kernel() -> None: + """Pin the factory wiring: the budget gate silently disables if the + factory drops `deps.spend_lookup` (the constructor default is the + permissive AlwaysZeroSpendLookup).""" + deps = build_deps(llm=FakeLLM()) + subscriber = make_run_debriefer_subscriber(deps) + assert isinstance(subscriber, RunDebrieferSubscriber) + assert subscriber.spend_lookup is deps.spend_lookup diff --git a/apps/api/tests/unit/api/test_delegating_inference_recorder.py b/apps/api/tests/unit/api/test_delegating_inference_recorder.py index 0fd747967df..3e31f397e49 100644 --- a/apps/api/tests/unit/api/test_delegating_inference_recorder.py +++ b/apps/api/tests/unit/api/test_delegating_inference_recorder.py @@ -14,7 +14,7 @@ import structlog from cora.api._inference_recorder import DelegatingInferenceRecorder -from cora.decision.errors import UnauthorizedError +from cora.decision.errors import InferenceAgentMismatchError, UnauthorizedError from cora.decision.features.append_inferences.command import AppendInferences from cora.infrastructure.ports import AgentInferenceTrace from cora.infrastructure.routing import NIL_SENTINEL_ID @@ -67,6 +67,7 @@ def _trace() -> AgentInferenceTrace: finish_reasons=("tool_use",), input_tokens=1280, output_tokens=214, + cost_usd=0.00235, request_max_tokens=1024, agent_id="01900000-0000-7000-8000-0000000a0099", agent_name="RunDebriefer", @@ -99,6 +100,7 @@ async def test_record_maps_trace_to_append_inferences_command() -> None: assert entry.finish_reasons == ("tool_use",) assert entry.input_tokens == 1280 assert entry.output_tokens == 214 + assert entry.cost_usd == 0.00235 assert entry.request_max_tokens == 1024 assert entry.agent_id == "01900000-0000-7000-8000-0000000a0099" assert entry.agent_name == "RunDebriefer" @@ -143,6 +145,32 @@ async def test_record_warns_distinctly_on_authorization_denial() -> None: assert "inference_recorder.failed" not in events +@pytest.mark.unit +async def test_record_warns_distinctly_on_agent_mismatch() -> None: + """An internal recorder is structurally self-reporting, so a + mismatch means a mis-wired agent whose spend is being dropped + (starving the budget gate's SUM). Loud and distinct, never the + generic failed bucket.""" + spy = _SpyAppendInferences( + raises=InferenceAgentMismatchError( + entry_agent_id="someone-else", principal_id=str(_PRINCIPAL_ID) + ) + ) + recorder = DelegatingInferenceRecorder(spy) + + with structlog.testing.capture_logs() as logs: + result = await recorder.record( + _trace(), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + causation_id=_CAUSATION_ID, + ) + assert result is None + events = [entry.get("event") for entry in logs] + assert "inference_recorder.agent_mismatch" in events + assert "inference_recorder.failed" not in events + + @pytest.mark.unit async def test_record_warns_on_unexpected_error() -> None: spy = _SpyAppendInferences(raises=RuntimeError("store down")) diff --git a/apps/api/tests/unit/decision/test_append_inferences_handler.py b/apps/api/tests/unit/decision/test_append_inferences_handler.py index eded179016c..e52a21f5cd2 100644 --- a/apps/api/tests/unit/decision/test_append_inferences_handler.py +++ b/apps/api/tests/unit/decision/test_append_inferences_handler.py @@ -24,6 +24,7 @@ event_type_name, to_payload, ) +from cora.decision.errors import InferenceAgentMismatchError from cora.decision.features import append_inferences from cora.decision.features.append_inferences import ( AppendInferences, @@ -375,6 +376,84 @@ async def test_handler_raises_unauthorized_on_deny() -> None: assert inference_store.all() == [] +# ---------- Agent attribution is principal-bound ---------- + + +@pytest.mark.unit +async def test_handler_rejects_entry_whose_agent_id_is_not_the_principal() -> None: + """Spend rows are summed per agent_id by the budget gate, so a + producer may only attribute entries to itself. A mismatched + agent_id rejects the batch before any write: no logbook open, no + stored rows.""" + event_store = InMemoryEventStore() + await _seed_decision(event_store, _DECISION_ID) + inference_store = InMemoryInferenceStore() + deps = build_deps(ids=[_LOGBOOK_ID, _LOGBOOK_OPEN_EVENT_ID], now=_NOW, event_store=event_store) + victim_agent_id = str(uuid4()) + with pytest.raises(InferenceAgentMismatchError) as exc_info: + await append_inferences.bind(deps, inference_store=inference_store)( + AppendInferences( + decision_id=_DECISION_ID, + entries=(_entry(agent_id=victim_agent_id, cost_usd=999.0),), + ), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + assert exc_info.value.entry_agent_id == victim_agent_id + assert exc_info.value.principal_id == str(_PRINCIPAL_ID) + assert inference_store.all() == [] + state = await load_decision(event_store, _DECISION_ID) + assert state is not None + assert state.logbooks == {} + + +@pytest.mark.unit +async def test_handler_rejects_whole_batch_when_one_entry_mismatches() -> None: + """Atomic rejection: one spoofed entry poisons the batch; the + producer's own well-attributed entries in it are NOT stored.""" + event_store = InMemoryEventStore() + await _seed_decision(event_store, _DECISION_ID) + inference_store = InMemoryInferenceStore() + deps = build_deps(ids=[_LOGBOOK_ID, _LOGBOOK_OPEN_EVENT_ID], now=_NOW, event_store=event_store) + with pytest.raises(InferenceAgentMismatchError): + await append_inferences.bind(deps, inference_store=inference_store)( + AppendInferences( + decision_id=_DECISION_ID, + entries=( + _entry(agent_id=str(_PRINCIPAL_ID)), + _entry(agent_id=str(uuid4())), + ), + ), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + assert inference_store.all() == [] + + +@pytest.mark.unit +async def test_handler_accepts_self_reported_and_agentless_entries() -> None: + """The binding never blocks the legitimate shapes: agent_id equal + to the principal (self-report) and agent_id=None (no agent claim, + operator/tool provenance) both append.""" + event_store = InMemoryEventStore() + await _seed_decision(event_store, _DECISION_ID) + inference_store = InMemoryInferenceStore() + deps = build_deps(ids=[_LOGBOOK_ID, _LOGBOOK_OPEN_EVENT_ID], now=_NOW, event_store=event_store) + count = await append_inferences.bind(deps, inference_store=inference_store)( + AppendInferences( + decision_id=_DECISION_ID, + entries=( + _entry(agent_id=str(_PRINCIPAL_ID), cost_usd=0.01), + _entry(agent_id=None), + ), + ), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + assert count == 2 + assert len(inference_store.all()) == 2 + + # ---------- Decision aggregate state reflects logbook ---------- diff --git a/apps/api/tests/unit/decision/test_decision_inferences.py b/apps/api/tests/unit/decision/test_decision_inferences.py index 18dc591e919..2962107d03a 100644 --- a/apps/api/tests/unit/decision/test_decision_inferences.py +++ b/apps/api/tests/unit/decision/test_decision_inferences.py @@ -46,6 +46,7 @@ def _row(**overrides: object) -> Inference: "finish_reasons": (), "input_tokens": None, "output_tokens": None, + "cost_usd": None, "agent_id": None, "agent_name": None, "agent_description": None, diff --git a/apps/api/tests/unit/infrastructure/test_gen_ai_telemetry.py b/apps/api/tests/unit/infrastructure/test_gen_ai_telemetry.py index 900b7f54299..33660bee58c 100644 --- a/apps/api/tests/unit/infrastructure/test_gen_ai_telemetry.py +++ b/apps/api/tests/unit/infrastructure/test_gen_ai_telemetry.py @@ -27,18 +27,18 @@ def reset_warning_set() -> None: @pytest.mark.unit def test_compute_cost_for_known_model() -> None: - """Opus 4.7 with 1M input tokens at $15/MT = exactly $15.""" + """Opus 4.8 with 1M input tokens at $5/MT = exactly $5.""" cost = compute_cost_usd( - ModelRef(provider="anthropic", model="claude-opus-4-7"), + ModelRef(provider="anthropic", model="claude-opus-4-8"), LLMUsage(input_tokens=1_000_000, output_tokens=0), ) - assert cost == pytest.approx(15.00) + assert cost == pytest.approx(5.00) @pytest.mark.unit def test_compute_cost_sums_all_four_token_types() -> None: """100k input + 50k output + 200k cache_create + 1M cache_read on Haiku 4.5: - 100k*$1 + 50k*$5 + 200k*$1.25 + 1M*$0.10, scaled per MTok.""" + 100k*$1 + 50k*$5 + 200k*$2 (1h-TTL write tier) + 1M*$0.10, per MTok.""" cost = compute_cost_usd( ModelRef(provider="anthropic", model="claude-haiku-4-5"), LLMUsage( @@ -48,7 +48,7 @@ def test_compute_cost_sums_all_four_token_types() -> None: cache_read_input_tokens=1_000_000, ), ) - expected = 0.1 + 0.25 + 0.25 + 0.10 + expected = 0.1 + 0.25 + 0.40 + 0.10 assert cost == pytest.approx(expected) @@ -75,13 +75,35 @@ def test_pricing_table_covers_all_documented_models() -> None: PRICING entry, or compute_cost_usd silently returns $0 and cost dashboards lie. Add to PRICING when adding a model.""" expected = { + ("anthropic", "claude-opus-4-8"), ("anthropic", "claude-opus-4-7"), ("anthropic", "claude-sonnet-4-6"), + ("anthropic", "claude-sonnet-4-5"), ("anthropic", "claude-haiku-4-5"), } assert expected.issubset(set(PRICING)) +@pytest.mark.unit +def test_pricing_table_covers_every_fleet_default_model() -> None: + """The fleet's ACTUAL default ModelRef constants must be priced, + derived from the constants themselves so a default-model bump to + an unpriced model fails here instead of silently metering $0. + cost_usd is enforcement-load-bearing: an unpriced default makes + the monthly USD gate permanently permissive.""" + from cora.agent.prompts.caution_drafter import DEFAULT_CAUTION_DRAFTER_MODEL + from cora.agent.prompts.run_debrief import DEFAULT_RUN_DEBRIEF_MODEL + from cora.operation.adapters._llm_decide_prompt import DEFAULT_LLM_DECIDE_MODEL + + for default in ( + DEFAULT_RUN_DEBRIEF_MODEL, + DEFAULT_CAUTION_DRAFTER_MODEL, + DEFAULT_LLM_DECIDE_MODEL, + ): + key = (default.provider, default.model) + assert key in PRICING, f"fleet default {key} has no PRICING entry" + + @pytest.mark.unit def test_record_llm_call_returns_cost_for_known_model() -> None: """Smoke test: with the no-op tracer (default in tests), every @@ -89,7 +111,7 @@ def test_record_llm_call_returns_cost_for_known_model() -> None: span = trace.get_current_span() # no-op span (no tracer configured) cost = record_llm_call( span, - system="anthropic", + provider_name="anthropic", request_model_ref=ModelRef(provider="anthropic", model="claude-sonnet-4-6"), response_model_id="claude-sonnet-4-6-20260301", usage=LLMUsage(input_tokens=1000, output_tokens=500), @@ -109,7 +131,7 @@ def test_record_llm_call_is_safe_with_noop_span() -> None: # Must not raise: cost = record_llm_call( span, - system="anthropic", + provider_name="anthropic", request_model_ref=ModelRef(provider="anthropic", model="claude-haiku-4-5"), response_model_id="claude-haiku-4-5", usage=LLMUsage( diff --git a/apps/api/tests/unit/infrastructure/test_spend_lookup.py b/apps/api/tests/unit/infrastructure/test_spend_lookup.py new file mode 100644 index 00000000000..017c6232ef7 --- /dev/null +++ b/apps/api/tests/unit/infrastructure/test_spend_lookup.py @@ -0,0 +1,52 @@ +"""Unit tests for the SpendLookup port shapes and the test-default stub.""" + +from dataclasses import FrozenInstanceError +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest + +from cora.infrastructure.ports.spend_lookup import ( + AlwaysZeroSpendLookup, + SpendLookupResult, +) + +_WINDOW_START = datetime(2026, 7, 1, tzinfo=UTC) +_WINDOW_END = datetime(2026, 8, 1, tzinfo=UTC) + + +@pytest.mark.unit +async def test_always_zero_stub_returns_zero_spend_echoing_the_window() -> None: + """The kernel-default stub answers 'nothing spent' for any agent, + echoing the requested window so gate log lines stay self-describing.""" + agent_id = uuid4() + + result = await AlwaysZeroSpendLookup().find_agent_spend( + agent_id=agent_id, + window_start=_WINDOW_START, + window_end=_WINDOW_END, + ) + + assert result == SpendLookupResult( + agent_id=agent_id, + window_start=_WINDOW_START, + window_end=_WINDOW_END, + usd_spent=0.0, + tokens_spent=0, + call_count=0, + ) + + +@pytest.mark.unit +def test_spend_lookup_result_is_frozen() -> None: + """A gate must never mutate the snapshot it decides on.""" + result = SpendLookupResult( + agent_id=uuid4(), + window_start=_WINDOW_START, + window_end=_WINDOW_END, + usd_spent=1.25, + tokens_spent=42, + call_count=3, + ) + with pytest.raises(FrozenInstanceError): + result.usd_spent = 99.0 # type: ignore[misc] diff --git a/infra/atlas/migrations/20260711000000_add_entries_decision_inferences_cost_usd.sql b/infra/atlas/migrations/20260711000000_add_entries_decision_inferences_cost_usd.sql new file mode 100644 index 00000000000..b476a9509ff --- /dev/null +++ b/infra/atlas/migrations/20260711000000_add_entries_decision_inferences_cost_usd.sql @@ -0,0 +1,32 @@ +-- Add cost_usd to entries_decision_inferences: the actual dollar cost of the +-- LLM call each inference entry records. +-- +-- The per-call cost was already computed at the adapter (compute_cost_usd from +-- usage tokens x PRICING) but only emitted to the cora.agent.llm.cost.usd +-- histogram and then discarded: the meter existed, the ledger sink did not. +-- Budget enforcement (the AgentBudget monthly_usd_cap gate, unpaused +-- 2026-07-11) needs a durable per-(agent, window) spend fact to sum, and the +-- inference entry is its natural home: it already carries agent_id, +-- occurred_at, and the token counts the cost derives from, so spend attribution +-- rides the existing provenance row instead of a parallel table. +-- +-- Additive forward-only migration, nullable with NO default: legacy rows keep +-- cost_usd IS NULL ("cost not recorded then", distinct from a true $0 call), +-- and the spend lookup COALESCEs NULL to 0 so pre-migration history never +-- inflates a balance. DOUBLE PRECISION matches the float the producer computes; +-- sub-cent drift is irrelevant at cap granularity (caps are whole dollars). +-- The CHECK rejects negative cost (would under-report spend and let the gate +-- permit what it should refuse), NaN (PG sorts NaN above infinity, so the +-- range check excludes it), and infinity (one +inf row would poison every +-- window's SUM unclearably). NULL passes the CHECK by SQL semantics, which is +-- exactly the legacy-row behavior wanted. +-- No index: the spend lookup filters on agent_id + occurred_at; no existing +-- index leads with agent_id, so the gate's SUM is a sequential scan of the +-- append-only table, fine at pilot scale. Escalation ladder when the read +-- shows up in p95: an (agent_id, occurred_at) index first, a proj_agent_spend +-- read model second (deferred-with-trigger). + +ALTER TABLE entries_decision_inferences + ADD COLUMN cost_usd DOUBLE PRECISION + CONSTRAINT entries_decision_inferences_cost_usd_range + CHECK (cost_usd >= 0 AND cost_usd < 'infinity'::float8); diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index cd8156692f1..3f749a929d2 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:LrBOk59Yp2sE2L8tLKVj8ub9cHMVU+GTR7cEn8QDzeY= +h1:uNiZj/SsGG3NVYsBEsKfUoZkjvCq73yxnx+EtxonZdE= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -156,3 +156,4 @@ h1:LrBOk59Yp2sE2L8tLKVj8ub9cHMVU+GTR7cEn8QDzeY= 20260701020000_init_entries_operation_procedure_outcomes.sql h1:am+Up8MDHuO/yT/pk3rYKAObOVtkvK+Oe1giXczpSu8= 20260705000000_init_proj_run_actor_involvement.sql h1:UmVHB7VBa+uC9xPlxwTPs+BHBK03F5nZITs3SLDJAKQ= 20260706000000_init_proj_trust_ratification_coverage.sql h1:YDYrypgQNOid1GnlfYbXI0j7300NUcGdu5L0Tsev/3I= +20260711000000_add_entries_decision_inferences_cost_usd.sql h1:Ebt1cyOO33cAzACL7Va1e8n7+kuPedHqVMln7mlWqHI=