The per-beamline allocation: budget BC, award-window envelope, campaign seal

apps/api/src/cora/agent/_budget_gate.py

@@ -18,6 +18,12 @@
 item gates identically (the non-determinism rule: subscribers decide
 from event facts, not ambient time).
 
+`find_allocation_breach` is the instrument-wide arm above the per-agent
+caps: when the deployment has an Active Allocation envelope (budget
+BC), instance-total spend over the envelope's own lifecycle window
+must stay inside its ceiling. Absent or sealed envelope means no
+constraint (opt-in, like cap declaration).
+
 ## Failure direction
 
 The spend sums undercount (NULL-cost legacy rows, unrecorded cache
@@ -36,9 +42,11 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING
+from uuid import UUID
 
 if TYPE_CHECKING:
     from cora.agent.aggregates.agent import Agent
+    from cora.infrastructure.ports.allocation_lookup import AllocationLookup
     from cora.infrastructure.ports.spend_lookup import SpendLookup
 
 
@@ -132,9 +140,86 @@ async def find_budget_breach(
     return None
 
 
+@dataclass(frozen=True)
+class AllocationBreach:
+    """The Active allocation envelope cannot afford the next call."""
+
+    allocation_id: UUID
+    ceiling_usd: float
+    spent_usd: float
+    window_start: datetime
+
+    def describe(self) -> str:
+        """One human-readable sentence for a deferred Decision's reasoning."""
+        return (
+            f"allocation {self.allocation_id} ceiling of {self.ceiling_usd:g} USD "
+            f"reached (instance-total spend {self.spent_usd:g} since "
+            f"{self.window_start.isoformat()})"
+        )
+
+
+async def find_allocation_breach(
+    *,
+    allocation_lookup: "AllocationLookup",
+    spend_lookup: "SpendLookup",
+    as_of: datetime,
+    pending_usd: float = 0.0,
+) -> AllocationBreach | None:
+    """Return the Active envelope's breach, or None to permit.
+
+    The instrument-wide arm above the per-agent caps: with an Active
+    allocation, instance-total spend is summed over the envelope's
+    own window `[activated_at, as_of)` (the lifecycle IS the window;
+    no calendar arithmetic) across ALL costed rows, agent-attributed
+    and operator-attributed alike. No Active envelope means no
+    constraint: declaring and activating one is what arms this check,
+    so every envelope-less deployment and test stays permissive.
+
+    Two refusal arms because the two caller tiers ask different
+    questions, mirroring the per-agent convention split between
+    `find_budget_breach` (post-hoc, `>=`) and `BudgetSpendGuard`
+    (pre-estimate, strict `>`):
+
+      - `spent + pending > ceiling`: the pre-estimate arm. Callers
+        pass `pending_usd` = the next call's estimated ceiling (plus
+        any in-conduct actuals they carry), and a projection landing
+        exactly ON the ceiling is the last call the envelope affords,
+        so only strictly-over refuses.
+      - `pending == 0 and spent >= ceiling`: the post-hoc arm.
+        Post-hoc callers pass no pending figure, so "exactly
+        exhausted" must refuse the NEXT call (`>` alone would let a
+        balance sitting exactly at the ceiling keep spending forever).
+
+    `as_of` is the triggering event's `occurred_at`, not wall clock,
+    per the subscribers' determinism rule; a replayed event whose
+    `occurred_at` predates `activated_at` sums an empty window and
+    permits. A lookup ERROR propagates (fail closed), same stance as
+    the per-agent gate.
+    """
+    envelope = await allocation_lookup.find_active()
+    if envelope is None:
+        return None
+    spend = await spend_lookup.find_total_spend(
+        window_start=envelope.activated_at,
+        window_end=as_of,
+    )
+    projected_over = spend.usd_spent + pending_usd > envelope.ceiling_usd
+    exhausted_post_hoc = pending_usd == 0.0 and spend.usd_spent >= envelope.ceiling_usd
+    if projected_over or exhausted_post_hoc:
+        return AllocationBreach(
+            allocation_id=envelope.allocation_id,
+            ceiling_usd=envelope.ceiling_usd,
+            spent_usd=spend.usd_spent,
+            window_start=envelope.activated_at,
+        )
+    return None
+
+
 __all__ = [
+    "AllocationBreach",
     "BudgetBreach",
     "calendar_day_window",
     "calendar_month_window",
+    "find_allocation_breach",
     "find_budget_breach",
 ]

apps/api/src/cora/agent/adapters/budget_spend_guard.py

@@ -9,17 +9,25 @@
 
 ## Refusal policy
 
-  - Agent stream missing: permit. Declaration is opt-in; absence of an
-    Agent aggregate must never block (mirrors `find_budget_breach`).
+  - Agent stream missing: the per-agent checks permit. Declaration is
+    opt-in; absence of an Agent aggregate must never block (mirrors
+    `find_budget_breach`). The envelope check below still runs.
   - Agent not Versioned: refuse. Suspend means stop on every path, and
     the pre-call check is the earliest place the steering brain can be
     stopped (the post-hoc record path only notices AFTER spend).
-  - No declared budget: permit.
+  - No declared budget: the per-agent checks permit.
   - Projected breach: refuse when recorded spend plus the estimated
     ceiling would exceed a cap (strictly greater: a call that lands
     exactly on the cap is the last one the envelope affords). Monthly
     USD is checked first, then daily tokens, one lookup per declared
     cap, matching the post-hoc gate's order.
+  - Envelope breach: after the per-agent caps, the instrument-wide
+    allocation envelope is consulted with `pending = estimated_cost_usd`
+    (the caller already folds its in-conduct actuals into that
+    estimate). Instance-total spend plus the estimate strictly over
+    the Active ceiling refuses. Runs even when the agent has no
+    declared budget or no Agent stream: the envelope is armed by
+    declaring an allocation, not by per-agent ceremony.
 
 A lookup ERROR propagates, per the port's fail-closed stance.
 """
@@ -28,23 +36,39 @@
 
 from typing import TYPE_CHECKING
 
-from cora.agent._budget_gate import calendar_day_window, calendar_month_window
+from cora.agent._budget_gate import (
+    calendar_day_window,
+    calendar_month_window,
+    find_allocation_breach,
+)
 from cora.agent.aggregates.agent import AgentStatus, load_agent
+from cora.infrastructure.ports import NoActiveAllocationLookup
 
 if TYPE_CHECKING:
     from datetime import datetime
     from uuid import UUID
 
+    from cora.agent.aggregates.agent import AgentBudget
+    from cora.infrastructure.ports.allocation_lookup import AllocationLookup
     from cora.infrastructure.ports.event_store import EventStore
     from cora.infrastructure.ports.spend_lookup import SpendLookup
 
 
 class BudgetSpendGuard:
     """`SpendGuard` over the Agent aggregate's caps and the spend ledger."""
 
-    def __init__(self, *, event_store: EventStore, spend_lookup: SpendLookup) -> None:
+    def __init__(
+        self,
+        *,
+        event_store: EventStore,
+        spend_lookup: SpendLookup,
+        allocation_lookup: AllocationLookup | None = None,
+    ) -> None:
         self._event_store = event_store
         self._spend_lookup = spend_lookup
+        # Defaults to no Active envelope so direct test construction stays
+        # cap-only; production wiring passes the Kernel's allocation_lookup.
+        self._allocation_lookup = allocation_lookup or NoActiveAllocationLookup()
 
     async def refusal_reason(
         self,
@@ -55,13 +79,49 @@ async def refusal_reason(
         as_of: datetime,
     ) -> str | None:
         agent = await load_agent(self._event_store, agent_id)
-        if agent is None:
-            return None
-        if agent.status is not AgentStatus.VERSIONED:
-            return f"agent is {agent.status.value}, not Versioned; calls are stopped"
-        if agent.budget is None:
+        if agent is not None:
+            if agent.status is not AgentStatus.VERSIONED:
+                return f"agent is {agent.status.value}, not Versioned; calls are stopped"
+            per_agent_reason = await self._per_agent_cap_refusal(
+                agent_budget=agent.budget,
+                agent_id=agent_id,
+                estimated_cost_usd=estimated_cost_usd,
+                estimated_tokens=estimated_tokens,
+                as_of=as_of,
+            )
+            if per_agent_reason is not None:
+                return per_agent_reason
+
+        envelope_breach = await find_allocation_breach(
+            allocation_lookup=self._allocation_lookup,
+            spend_lookup=self._spend_lookup,
+            as_of=as_of,
+            pending_usd=estimated_cost_usd,
+        )
+        if envelope_breach is not None:
+            return (
+                f"allocation envelope {envelope_breach.allocation_id} ceiling of "
+                f"{envelope_breach.ceiling_usd:g} USD would be breached: "
+                f"{envelope_breach.spent_usd:g} instance-total spend since "
+                f"{envelope_breach.window_start.isoformat()} plus an estimated "
+                f"{estimated_cost_usd:g} ceiling for this call"
+            )
+
+        return None
+
+    async def _per_agent_cap_refusal(
+        self,
+        *,
+        agent_budget: AgentBudget | None,
+        agent_id: UUID,
+        estimated_cost_usd: float,
+        estimated_tokens: int,
+        as_of: datetime,
+    ) -> str | None:
+        """Check the declared per-agent caps (monthly USD, then daily tokens)."""
+        if agent_budget is None:
             return None
-        budget = agent.budget
+        budget = agent_budget
 
         if budget.monthly_usd_cap is not None:
             window_start, window_end = calendar_month_window(as_of)

apps/api/src/cora/agent/subscribers/caution_drafter.py

@@ -80,7 +80,7 @@
 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._budget_gate import find_allocation_breach, 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 (
@@ -121,6 +121,7 @@
     AgentInferenceTrace,
     AlwaysZeroSpendLookup,
     ConcurrencyError,
+    NoActiveAllocationLookup,
     NullInferenceRecorder,
 )
 from cora.infrastructure.signing import SIGNED_EVENT_TYPES
@@ -133,6 +134,7 @@
     from cora.infrastructure.kernel import Kernel
     from cora.infrastructure.ports import (
         LLM,
+        AllocationLookup,
         CautionLookup,
         InferenceRecorder,
         LLMChatRequest,
@@ -205,6 +207,7 @@ def __init__(
         signer: Signer | None = None,
         inference_recorder: InferenceRecorder | None = None,
         spend_lookup: SpendLookup | None = None,
+        allocation_lookup: AllocationLookup | None = None,
     ) -> None:
         self.event_store = event_store
         self.llm = llm
@@ -218,6 +221,9 @@ def __init__(
         # don't exercise budget gating; production wiring passes the
         # Kernel's PostgresSpendLookup.
         self.spend_lookup = spend_lookup or AlwaysZeroSpendLookup()
+        # Defaults to no Active envelope so the instrument-wide allocation
+        # check stays disarmed unless a test (or production wiring) arms it.
+        self.allocation_lookup = allocation_lookup or NoActiveAllocationLookup()
 
     async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None:
         """Process one terminal Run event."""
@@ -359,6 +365,43 @@ async def apply(self, event: StoredEvent, conn: ConnectionLike) -> None:
             )
             return
 
+        # Instrument-wide envelope gate (mirrors RunDebriefer): an
+        # exhausted Active allocation stops every LLM caller regardless
+        # of this agent's own headroom (post-hoc arm, pending 0).
+        envelope_breach = await find_allocation_breach(
+            allocation_lookup=self.allocation_lookup,
+            spend_lookup=self.spend_lookup,
+            as_of=event.occurred_at,
+        )
+        if envelope_breach is not None:
+            log.warning(
+                "caution_drafter.allocation_exhausted",
+                allocation_id=str(envelope_breach.allocation_id),
+                ceiling_usd=envelope_breach.ceiling_usd,
+                spent_usd=envelope_breach.spent_usd,
+            )
+            await self._compose_and_append(
+                decision_id=decision_id,
+                actor=actor,
+                run_id=run_id,
+                terminal_event=event,
+                choice="NoAction",
+                confidence=None,
+                reasoning=(
+                    f"Allocation exhausted: {envelope_breach.describe()}; LLM "
+                    "call skipped. Raise the ceiling via "
+                    "amend_allocation_ceiling, or seal or void this envelope "
+                    "and activate a new one."
+                ),
+                extra_inputs={
+                    "failure_error_class": "AllocationExhausted",
+                    "allocation_id": str(envelope_breach.allocation_id),
+                },
+                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
@@ -867,6 +910,7 @@ def make_caution_drafter_subscriber(deps: Kernel) -> CautionDrafterSubscriber:
         signer=deps.signer,
         inference_recorder=deps.inference_recorder,
         spend_lookup=deps.spend_lookup,
+        allocation_lookup=deps.allocation_lookup,
     )