"""Reference solution — the action gateway: the bank's enforcement boundary. Everything before this file is an agent *deciding*. This is where a decision becomes an **effect on the bank**, and it is the layer the whole track has been building toward. The one-line architecture — **the model proposes, the platform disposes** — is implemented here. No agent talks to core banking. It talks to this, which validates the contract, derives retry policy from the side-effect class, enforces idempotency, requires dual control above a threshold, protects the downstream with a breaker, runs multi-step work as a saga with compensations, and writes a hash-chained record that answers "who authorized this?" Deterministic: an injected clock, an injected downstream, derived identifiers, money as integer micro-units divided last. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field, replace from enum import Enum from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple) # ====================================================================================== # 1. Side-effect classes — the field that drives everything else # ====================================================================================== class SideEffect(str, Enum): """The classification every tool must declare. There is no default. A default here would be a default *retry policy*, and the safe default (never retry) makes reads fragile while the convenient default (always retry) double-executes payments. The only correct answer is to refuse to register a tool that has not said. """ READ = "read" WRITE_IDEMPOTENT = "write_idempotent" WRITE_NON_IDEMPOTENT = "write_non_idempotent" IRREVERSIBLE = "irreversible" @dataclass(frozen=True) class EffectPolicy: """What the platform does for a given side-effect class. The point of this table is that retry policy is a PLATFORM decision derived from a declared property — not a per-call-site choice made by whoever wrote the integration at 6pm. Every production double-payment story is a call site that chose. """ max_attempts: int requires_idempotency_key: bool requires_dual_control_above: Optional[int] # micro-units; None = never audit_level: str # "info" | "record" | "evidence" compensable: bool #: Note the asymmetry between the last two rows. A non-idempotent write may be retried #: exactly zero times *without a key* — with one, the store makes it safe. An irreversible #: action is never retried automatically at all, because "did it happen?" is a question #: the gateway cannot answer for a released payment, and guessing costs real money. EFFECT_POLICY: Mapping[SideEffect, EffectPolicy] = { SideEffect.READ: EffectPolicy( max_attempts=3, requires_idempotency_key=False, requires_dual_control_above=None, audit_level="info", compensable=True), SideEffect.WRITE_IDEMPOTENT: EffectPolicy( max_attempts=3, requires_idempotency_key=True, requires_dual_control_above=None, audit_level="record", compensable=True), SideEffect.WRITE_NON_IDEMPOTENT: EffectPolicy( max_attempts=1, requires_idempotency_key=True, requires_dual_control_above=None, audit_level="record", compensable=True), SideEffect.IRREVERSIBLE: EffectPolicy( max_attempts=1, requires_idempotency_key=True, requires_dual_control_above=100_000_000_000, # 100,000 AED in micro-units audit_level="evidence", compensable=False), } # ====================================================================================== # 2. Contract enforcement — schema, then the invariants a schema cannot express # ====================================================================================== class GatewayError(Exception): """Base for every refusal. Each subclass carries an HTTP-shaped code because the distinction between them is what a caller must branch on.""" code = "gateway_error" status = 400 class ContractViolation(GatewayError): code = "contract_violation" status = 422 class IdempotencyConflict(GatewayError): code = "idempotency_conflict" status = 409 class InFlight(GatewayError): code = "in_flight" status = 409 class ApprovalRequired(GatewayError): code = "approval_required" status = 428 class CircuitOpen(GatewayError): code = "circuit_open" status = 503 class DownstreamError(Exception): """Raised by the downstream itself. Deliberately NOT a GatewayError — a downstream failure is a different thing from the gateway refusing, and conflating them is how a retry loop retries a 422 forever.""" def __init__(self, message: str, *, retryable: bool = True) -> None: super().__init__(message) self.retryable = retryable _TYPES: Mapping[str, tuple] = { "string": (str,), "integer": (int,), "number": (int, float), "boolean": (bool,), "object": (dict,), "array": (list,), } def validate_schema(schema: Mapping[str, Any], value: Any, path: str = "$") -> List[str]: """A small JSON-Schema subset. Returns EVERY error, sorted. Every error, not the first: a caller fixing one field at a time is a caller making four round trips to learn about four problems. """ errors: List[str] = [] expected = schema.get("type") if expected: allowed = _TYPES.get(expected, ()) # bool is a subclass of int in Python; a schema saying "integer" does not mean # True. This is the single most common validator bug in the language. if expected in ("integer", "number") and isinstance(value, bool): errors.append(f"{path}: expected {expected}, got boolean") return sorted(errors) if not isinstance(value, allowed): errors.append(f"{path}: expected {expected}, got " f"{type(value).__name__}") return sorted(errors) if expected == "object": props = schema.get("properties", {}) for key in schema.get("required", []): if key not in value: errors.append(f"{path}.{key}: required") for key, sub in props.items(): if key in value: errors.extend(validate_schema(sub, value[key], f"{path}.{key}")) if not schema.get("additionalProperties", True): for key in sorted(set(value) - set(props)): errors.append(f"{path}.{key}: not permitted") elif expected == "array": item = schema.get("items") if item: for i, element in enumerate(value): errors.extend(validate_schema(item, element, f"{path}[{i}]")) elif expected == "string": if "pattern" in schema and not re.fullmatch(schema["pattern"], value): errors.append(f"{path}: does not match {schema['pattern']}") if "enum" in schema and value not in schema["enum"]: errors.append(f"{path}: not one of {sorted(schema['enum'])}") elif expected in ("integer", "number"): if "minimum" in schema and value < schema["minimum"]: errors.append(f"{path}: below minimum {schema['minimum']}") if "maximum" in schema and value > schema["maximum"]: errors.append(f"{path}: above maximum {schema['maximum']}") return sorted(errors) #: An invariant reads the whole call and returns an error string, or None. Invariant = Callable[["ActionRequest"], Optional[str]] @dataclass(frozen=True) class ToolContract: """A tool's declared interface, plus the things a schema cannot say. The split is the lesson. A schema says "amount is an integer ≥ 1". It cannot say "amount is within this agent's limit", "the currency matches the debit account", or "the value date is a business day" — those need the request's context and the bank's state. Both halves are contract enforcement; only one of them is expressible in JSON Schema, and pretending otherwise is how business rules end up in the agent's prompt. """ tool_id: str side_effect: SideEffect schema: Mapping[str, Any] invariants: Tuple[Invariant, ...] = () timeout_ms: int = 5_000 max_value_micros: Optional[int] = None def check(self, request: "ActionRequest") -> List[str]: errors = validate_schema(self.schema, dict(request.arguments)) if errors: # Do NOT run invariants on a structurally invalid payload — they would raise # KeyError and mask the real problem with a stack trace. return errors for invariant in self.invariants: problem = invariant(request) if problem: errors.append(problem) return sorted(errors) # ====================================================================================== # 3. The request and the record # ====================================================================================== @dataclass(frozen=True) class Principal: """Who is acting — the blended subject from Phase 09, plus the chain from Phase 08.""" agent_id: str user_id: Optional[str] tenant: str delegation_chain: Tuple[str, ...] = () def describe(self) -> str: hops = " -> ".join((self.user_id or "system",) + self.delegation_chain + (self.agent_id,)) return hops @dataclass(frozen=True) class ActionRequest: tool_id: str arguments: Mapping[str, Any] principal: Principal idempotency_key: Optional[str] = None approvals: Tuple[str, ...] = () policy_version: str = "" model_version: str = "" trace_id: str = "" value_micros: int = 0 @dataclass(frozen=True) class ActionResult: ok: bool payload: Any = None error: Optional[str] = None attempts: int = 0 replayed: bool = False audit_seq: int = 0 # ====================================================================================== # 4. Idempotency # ====================================================================================== class IdemState(str, Enum): IN_FLIGHT = "in_flight" COMPLETED = "completed" FAILED = "failed" @dataclass(frozen=True) class IdemRecord: key: str state: IdemState request_hash: str response: Any = None created_at: int = 0 completed_at: int = 0 def request_hash(request: ActionRequest) -> str: """The canonical digest of *what was asked*. Excludes the trace id and the model version deliberately: a retry of the same business intent from a different trace is the SAME request, and treating it as different would turn every retry into a 409. """ material = json.dumps({ "tool": request.tool_id, "args": request.arguments, "agent": request.principal.agent_id, "user": request.principal.user_id, "tenant": request.principal.tenant, }, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(material.encode()).hexdigest() class IdempotencyStore: """``key -> (state, request_hash, response)``. Three cases, and they are the phase. 1. **No record** — reserve the key, execute, store the response. 2. **Same key, same hash** — return the stored response. Execute ZERO more times. 3. **Same key, different hash** — 409, and execute NEVER. The caller has a bug: it reused a key for a different request, and the one thing that must not happen is quietly performing the second one. Plus the case people forget: **in flight**. Two concurrent requests with the same key means the first is still running, and the second must not start. 409 with a retry-after, not a queue — a queue turns a duplicate click into a duplicate payment one second later. """ def __init__(self, *, now: Callable[[], int], ttl_ticks: int = 86_400) -> None: self.now = now self.ttl_ticks = ttl_ticks self._records: Dict[str, IdemRecord] = {} def begin(self, key: str, digest: str) -> Optional[IdemRecord]: """Reserve the key, or return the existing record for a replay. Returns None when the caller should proceed to execute. Raises on conflict. """ existing = self._records.get(key) if existing is not None and self._expired(existing): del self._records[key] existing = None if existing is None: self._records[key] = IdemRecord(key, IdemState.IN_FLIGHT, digest, created_at=self.now()) return None if existing.request_hash != digest: raise IdempotencyConflict( f"key {key!r} was used for a different request") if existing.state is IdemState.IN_FLIGHT: raise InFlight(f"key {key!r} is already in flight") return existing def complete(self, key: str, response: Any) -> IdemRecord: record = replace(self._records[key], state=IdemState.COMPLETED, response=response, completed_at=self.now()) self._records[key] = record return record def fail(self, key: str) -> None: """Release the key on failure. A failed non-idempotent write is the genuinely hard case: we do not know whether the downstream applied it. Releasing the key lets the caller retry — which is correct *only* because the downstream is itself keyed. Marking it FAILED and keeping it would be safer and would also strand a caller whose network blipped. This is a judgment call, and it should be a documented one. """ self._records.pop(key, None) def _expired(self, record: IdemRecord) -> bool: return self.now() - record.created_at >= self.ttl_ticks def get(self, key: str) -> Optional[IdemRecord]: return self._records.get(key) # ====================================================================================== # 5. The circuit breaker # ====================================================================================== class BreakerState(str, Enum): CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" class CircuitBreaker: """A rolling-window breaker with a minimum-throughput guard. Two things distinguish a real breaker from the toy version: * **Minimum throughput.** Without it, one failure out of one call is a 100% failure rate and the breaker opens on a single blip at 3 a.m. when traffic is low. * **Half-open is a single probe, not a reopening.** Letting all traffic through to test recovery is how a breaker turns a recovering downstream back into a dead one. And the thing that makes it useful rather than decorative: **open must have a defined behaviour.** A breaker that only fails faster has converted a slow error into a quick one. A breaker whose open state serves a cached value, or queues, or degrades to a narrower answer, has bought something. """ def __init__(self, *, now: Callable[[], int], failure_threshold: float = 0.5, minimum_throughput: int = 5, window_ticks: int = 60, open_ticks: int = 30, half_open_successes: int = 2) -> None: self.now = now self.failure_threshold = failure_threshold self.minimum_throughput = minimum_throughput self.window_ticks = window_ticks self.open_ticks = open_ticks self.half_open_successes = half_open_successes self._events: List[Tuple[int, bool]] = [] # (tick, ok) self._state = BreakerState.CLOSED self._opened_at = 0 self._probe_successes = 0 self.transitions: List[Tuple[int, str]] = [] # -- state ---------------------------------------------------------------------- def state(self) -> BreakerState: if self._state is BreakerState.OPEN and \ self.now() - self._opened_at >= self.open_ticks: self._transition(BreakerState.HALF_OPEN) return self._state def _transition(self, target: BreakerState) -> None: self._state = target self.transitions.append((self.now(), target.value)) if target is BreakerState.OPEN: self._opened_at = self.now() self._probe_successes = 0 if target is BreakerState.HALF_OPEN: self._probe_successes = 0 if target is BreakerState.CLOSED: self._events.clear() # -- the two calls a caller makes ------------------------------------------------ def allow(self) -> bool: state = self.state() if state is BreakerState.OPEN: return False return True # CLOSED, or HALF_OPEN admitting a probe def record(self, ok: bool) -> None: tick = self.now() state = self.state() if state is BreakerState.HALF_OPEN: if not ok: # One failed probe re-opens. Not "some" — the downstream told us. self._transition(BreakerState.OPEN) return self._probe_successes += 1 if self._probe_successes >= self.half_open_successes: self._transition(BreakerState.CLOSED) return self._events.append((tick, ok)) self._prune(tick) if self._should_open(): self._transition(BreakerState.OPEN) def _prune(self, tick: int) -> None: cutoff = tick - self.window_ticks self._events = [e for e in self._events if e[0] > cutoff] def _should_open(self) -> bool: total = len(self._events) if total < self.minimum_throughput: return False failures = sum(1 for _, ok in self._events if not ok) return failures / total >= self.failure_threshold def failure_rate(self) -> float: if not self._events: return 0.0 return sum(1 for _, ok in self._events if not ok) / len(self._events) # ====================================================================================== # 6. Redaction # ====================================================================================== _ACCOUNT = re.compile(r"\b(?:AE\d{2})?\d{9,}\b") _SECRETISH = ("password", "secret", "token", "api_key", "apikey", "credential", "authorization", "pin", "otp") def redact(value: Any, *, key: str = "") -> Any: """Redact before the value reaches a log line — never after. "We'll scrub the logs later" is not a control: the unredacted value already left the process, was buffered, shipped and indexed. Redaction that happens anywhere except before serialization is a report, not a defence. Account numbers keep their last four digits, because an audit record that cannot distinguish two accounts is not much of an audit record. """ if isinstance(value, Mapping): return {k: redact(v, key=k) for k, v in value.items()} if isinstance(value, (list, tuple)): return [redact(v, key=key) for v in value] if any(s in key.lower() for s in _SECRETISH): return "[REDACTED]" if isinstance(value, str): return _ACCOUNT.sub(lambda m: f"****{m.group(0)[-4:]}", value) return value # ====================================================================================== # 7. The audit log — hash-chained # ====================================================================================== GENESIS = "0" * 64 @dataclass(frozen=True) class AuditRecord: """Every field here answers a specific person's question. | field | who asks | |---|---| | actor_chain | the examiner: "who authorized this?" | | policy_version | Internal Audit: "under which rules?" | | model_version | model risk: "which model reasoned about it?" | | idempotency_key | the payments team: "is this the duplicate?" | | approvals | the four-eyes control owner | | outcome + error | the on-call engineer | | prev_hash / this_hash | everyone, implicitly: "has this been edited?" | """ seq: int tick: int trace_id: str tool_id: str side_effect: str actor_chain: str tenant: str arguments: Mapping[str, Any] # redacted outcome: str error: Optional[str] policy_version: str model_version: str idempotency_key: Optional[str] approvals: Tuple[str, ...] value_micros: int prev_hash: str this_hash: str = "" def digest(self) -> str: material = json.dumps({ "seq": self.seq, "tick": self.tick, "trace_id": self.trace_id, "tool_id": self.tool_id, "side_effect": self.side_effect, "actor_chain": self.actor_chain, "tenant": self.tenant, "arguments": self.arguments, "outcome": self.outcome, "error": self.error, "policy_version": self.policy_version, "model_version": self.model_version, "idempotency_key": self.idempotency_key, "approvals": list(self.approvals), "value_micros": self.value_micros, "prev_hash": self.prev_hash, }, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(material.encode()).hexdigest() class AuditLog: """Append-only, hash-chained. Each record's hash covers its own content AND the previous hash, so editing record 7 changes its hash, which breaks record 8's `prev_hash`, and so on to the head. You cannot alter history without rewriting everything after it — and if the head hash is published somewhere you do not control (a WORM store, another team's system, a notary), you cannot rewrite it at all. Note the honest limit: this is tamper-**evident**, not tamper-proof. Someone with write access can rebuild the whole chain. The external anchor is what closes that, and it is a deployment concern rather than a code one. """ def __init__(self, *, now: Callable[[], int]) -> None: self.now = now self.records: List[AuditRecord] = [] def append(self, **fields: Any) -> AuditRecord: prev = self.records[-1].this_hash if self.records else GENESIS record = AuditRecord(seq=len(self.records) + 1, tick=self.now(), prev_hash=prev, **fields) record = replace(record, this_hash=record.digest()) self.records.append(record) return record def verify(self) -> Tuple[bool, Optional[str]]: """Return ``(ok, first_problem)``. Checks both links of every record.""" prev = GENESIS for i, record in enumerate(self.records, 1): if record.seq != i: return False, f"record {i}: sequence is {record.seq}" if record.prev_hash != prev: return False, f"record {i}: prev_hash does not match record {i - 1}" if record.digest() != record.this_hash: return False, f"record {i}: content does not match its hash" prev = record.this_hash return True, None def head(self) -> str: return self.records[-1].this_hash if self.records else GENESIS def for_trace(self, trace_id: str) -> List[AuditRecord]: return [r for r in self.records if r.trace_id == trace_id] # ====================================================================================== # 8. Dual control # ====================================================================================== def check_dual_control(request: ActionRequest, *, threshold: Optional[int], required: int = 2) -> Optional[str]: """Two DISTINCT, authenticated humans — and the agent is never one of them. Three failure modes this exists to prevent, all of which have happened: * one approver clicking twice (hence the set); * the requesting agent counting itself (hence the exclusion); * the initiating *user* approving their own request (hence that exclusion too — four-eyes with one pair of eyes is not four-eyes). Threshold boundaries are INCLUSIVE. "Above 100,000" and "100,000 or more" differ by exactly one transaction, and that transaction is the one the auditor picks. """ if threshold is None or request.value_micros < threshold: return None forbidden = {request.principal.agent_id} if request.principal.user_id: forbidden.add(request.principal.user_id) forbidden.update(request.principal.delegation_chain) approvers = {a for a in request.approvals if a not in forbidden} if len(approvers) < required: return (f"requires {required} distinct approvers, none of whom may be the " f"requester; got {len(approvers)}") return None # ====================================================================================== # 9. The gateway # ====================================================================================== #: A downstream takes (tool_id, arguments) and returns a payload, or raises #: DownstreamError. Injected so the tests are deterministic and the gateway is testable #: without a bank. Downstream = Callable[[str, Mapping[str, Any]], Any] class ActionGateway: """Contract → idempotency → dual control → breaker → execute → audit. The order is not arbitrary. Cheap, purely-local refusals come first, so a malformed request never touches the idempotency store or the downstream; the breaker check sits immediately before execution so an open circuit does not consume an idempotency key. """ def __init__(self, *, now: Callable[[], int], downstream: Downstream, audit: Optional[AuditLog] = None, breakers: Optional[Dict[str, CircuitBreaker]] = None, fallback: Optional[Callable[[ActionRequest], Any]] = None) -> None: self.now = now self.downstream = downstream self.audit = audit or AuditLog(now=now) self.idempotency = IdempotencyStore(now=now) self.breakers: Dict[str, CircuitBreaker] = breakers or {} self.fallback = fallback self.contracts: Dict[str, ToolContract] = {} self.executions: List[Tuple[str, Mapping[str, Any]]] = [] def register(self, contract: ToolContract) -> None: if contract.tool_id in self.contracts: raise ValueError(f"{contract.tool_id} is already registered") self.contracts[contract.tool_id] = contract def breaker_for(self, tool_id: str) -> CircuitBreaker: if tool_id not in self.breakers: self.breakers[tool_id] = CircuitBreaker(now=self.now) return self.breakers[tool_id] # -- the one entry point --------------------------------------------------------- def execute(self, request: ActionRequest) -> ActionResult: contract = self.contracts.get(request.tool_id) if contract is None: return self._refuse(request, None, "unknown_tool", f"{request.tool_id} is not registered") policy = EFFECT_POLICY[contract.side_effect] # 1. contract — schema, then business invariants problems = contract.check(request) if problems: return self._refuse(request, contract, "contract_violation", "; ".join(problems)) # 2. the key must exist when the policy demands one if policy.requires_idempotency_key and not request.idempotency_key: return self._refuse(request, contract, "contract_violation", f"{contract.side_effect.value} requires an " f"idempotency key") # 3. dual control — BEFORE the key is reserved, so a rejected approval does not # burn the key the caller will reuse once they have the approvals. problem = check_dual_control( request, threshold=policy.requires_dual_control_above) if problem: return self._refuse(request, contract, "approval_required", problem) # 4. idempotency digest = request_hash(request) if request.idempotency_key: try: existing = self.idempotency.begin(request.idempotency_key, digest) except GatewayError as exc: return self._refuse(request, contract, exc.code, str(exc)) if existing is not None: self._record(request, contract, "replayed", None) return ActionResult(True, existing.response, attempts=0, replayed=True, audit_seq=self.audit.records[-1].seq) # 5. the breaker, immediately before execution breaker = self.breaker_for(request.tool_id) if not breaker.allow(): if request.idempotency_key: self.idempotency.fail(request.idempotency_key) if self.fallback is not None: payload = self.fallback(request) self._record(request, contract, "fallback", None) return ActionResult(True, payload, audit_seq=self.audit.records[-1].seq) return self._refuse(request, contract, "circuit_open", f"{request.tool_id} circuit is open") # 6. execute, with the attempts the class allows last_error: Optional[str] = None for attempt in range(1, policy.max_attempts + 1): try: payload = self.downstream(request.tool_id, dict(request.arguments)) except DownstreamError as exc: breaker.record(False) last_error = str(exc) if not exc.retryable or attempt >= policy.max_attempts: break continue breaker.record(True) self.executions.append((request.tool_id, dict(request.arguments))) if request.idempotency_key: self.idempotency.complete(request.idempotency_key, payload) self._record(request, contract, "success", None) return ActionResult(True, payload, attempts=attempt, audit_seq=self.audit.records[-1].seq) if request.idempotency_key: self.idempotency.fail(request.idempotency_key) self._record(request, contract, "failure", last_error) return ActionResult(False, None, error=last_error, attempts=policy.max_attempts, audit_seq=self.audit.records[-1].seq) # -- audit helpers ---------------------------------------------------------------- def _record(self, request: ActionRequest, contract: Optional[ToolContract], outcome: str, error: Optional[str]) -> AuditRecord: return self.audit.append( trace_id=request.trace_id, tool_id=request.tool_id, side_effect=contract.side_effect.value if contract else "unknown", actor_chain=request.principal.describe(), tenant=request.principal.tenant, arguments=redact(dict(request.arguments)), outcome=outcome, error=error, policy_version=request.policy_version, model_version=request.model_version, idempotency_key=request.idempotency_key, approvals=tuple(sorted(request.approvals)), value_micros=request.value_micros) def _refuse(self, request: ActionRequest, contract: Optional[ToolContract], outcome: str, message: str) -> ActionResult: """A refusal is audited exactly as carefully as a success. Refusals are the more interesting half of the log: "the platform stopped it" is the sentence that demonstrates the control worked. """ record = self._record(request, contract, outcome, message) return ActionResult(False, None, error=message, audit_seq=record.seq) # ====================================================================================== # 10. Sagas # ====================================================================================== @dataclass(frozen=True) class SagaStep: """A forward action and its **semantic inverse**. Compensation is not rollback. A rollback restores the previous state as if nothing happened; a compensation performs a *new* action that undoes the business effect — and the intermediate state was visible to everyone the whole time. A reversal appears on the customer's statement. The hold was seen by the fraud system. There is no undo button on a bank. """ name: str forward: Callable[[Dict[str, Any]], Any] compensate: Optional[Callable[[Dict[str, Any]], Any]] = None retries: int = 0 @dataclass(frozen=True) class SagaOutcome: ok: bool completed: Tuple[str, ...] compensated: Tuple[str, ...] failed_step: Optional[str] error: Optional[str] orphaned: Tuple[str, ...] = () # compensations that themselves failed class Saga: """Forward steps; on failure, compensate the completed ones in REVERSE order. Reverse order because dependencies run forwards: step 3 may rely on step 2's effect, so undoing 2 before 3 leaves 3's compensation operating on state that no longer exists. Compensations must be **idempotent and retryable**, because they run in exactly the conditions that just broke — a downstream that is flaky, timing out, or half-up. And the case that must not be swallowed: a compensation that *itself* fails. There is no third level of undo. The only correct behaviour is to record it loudly as an orphan and page a human, because the bank is now in a state no code will repair. """ def __init__(self, name: str, steps: Sequence[SagaStep], *, audit: Optional[AuditLog] = None) -> None: self.name = name self.steps = list(steps) self.audit = audit def run(self, context: Optional[Dict[str, Any]] = None) -> SagaOutcome: ctx: Dict[str, Any] = dict(context or {}) completed: List[SagaStep] = [] for step in self.steps: try: result = self._attempt(step, ctx) except Exception as exc: # noqa: BLE001 - boundary compensated, orphaned = self._compensate(completed, ctx) return SagaOutcome(False, tuple(s.name for s in completed), compensated, step.name, str(exc), orphaned) ctx[step.name] = result completed.append(step) return SagaOutcome(True, tuple(s.name for s in completed), (), None, None) def _attempt(self, step: SagaStep, ctx: Dict[str, Any]) -> Any: last: Optional[Exception] = None for _ in range(step.retries + 1): try: return step.forward(ctx) except DownstreamError as exc: last = exc if not exc.retryable: raise raise last # type: ignore[misc] def _compensate(self, completed: Sequence[SagaStep], ctx: Dict[str, Any]) -> Tuple[Tuple[str, ...], Tuple[str, ...]]: done: List[str] = [] orphaned: List[str] = [] for step in reversed(completed): if step.compensate is None: # An uncompensable step among compensable ones is a design error the # saga cannot fix at runtime. Name it. orphaned.append(step.name) continue try: step.compensate(ctx) done.append(step.name) except Exception: # noqa: BLE001 - boundary orphaned.append(step.name) return tuple(done), tuple(orphaned) # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: int = 1000, step: int = 1) -> Callable[[], int]: state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now PAYMENT_SCHEMA = { "type": "object", "required": ["payment_id", "amount_micros", "currency", "debit_account"], "additionalProperties": False, "properties": { "payment_id": {"type": "string", "pattern": r"PMT-\d{3,}"}, "amount_micros": {"type": "integer", "minimum": 1}, "currency": {"type": "string", "enum": ["AED", "USD", "EUR"]}, "debit_account": {"type": "string"}, "value_date": {"type": "string"}, }, } _ACCOUNT_CURRENCY = {"AE070331234567890123456": "AED", "AE070339999999999999999": "USD"} def currency_matches_account(request: ActionRequest) -> Optional[str]: account = request.arguments.get("debit_account", "") expected = _ACCOUNT_CURRENCY.get(account) if expected is None: return f"debit_account {account[-4:]} is not a known account" if request.arguments.get("currency") != expected: return (f"currency {request.arguments.get('currency')} does not match the " f"account currency {expected}") return None def within_agent_limit(request: ActionRequest) -> Optional[str]: limit = 500_000_000_000 # 500,000 AED if request.arguments.get("amount_micros", 0) > limit: return f"amount exceeds the agent's per-action limit of {limit}" return None def main() -> None: # pragma: no cover - narrative output now = _clock() downstream_calls: List[str] = [] failures: Dict[str, int] = {} def downstream(tool_id: str, args: Mapping[str, Any]) -> Any: downstream_calls.append(tool_id) if failures.get(tool_id, 0) > 0: failures[tool_id] -= 1 raise DownstreamError(f"{tool_id} unavailable") return {"status": "APPLIED", "reference": f"REF-{len(downstream_calls):04d}"} gateway = ActionGateway(now=now, downstream=downstream) gateway.register(ToolContract( "payments.release", SideEffect.IRREVERSIBLE, PAYMENT_SCHEMA, invariants=(currency_matches_account, within_agent_limit))) gateway.register(ToolContract( "crm.append_note", SideEffect.WRITE_IDEMPOTENT, {"type": "object", "required": ["case_id", "note"], "properties": {"case_id": {"type": "string"}, "note": {"type": "string"}}})) gateway.register(ToolContract( "payments.lookup", SideEffect.READ, {"type": "object", "required": ["payment_id"], "properties": {"payment_id": {"type": "string"}}})) principal = Principal("payments-investigator", "layla.almansouri", "wholesale", ("orchestrator",)) def payment(**overrides: Any) -> ActionRequest: args = {"payment_id": "PMT-771", "amount_micros": 250_000_000_000, "currency": "AED", "debit_account": "AE070331234567890123456"} args.update(overrides.pop("arguments", {})) base = dict(tool_id="payments.release", arguments=args, principal=principal, idempotency_key="idem-001", approvals=("ahmed", "sara"), policy_version="2026-02-11.3", model_version="gpt-frontier-2026-02-11", trace_id="trace-771", value_micros=args["amount_micros"]) base.update(overrides) return ActionRequest(**base) print("=" * 78) print("1. CONTRACT ENFORCEMENT — SCHEMA, THEN THE INVARIANTS IT CANNOT EXPRESS") print("=" * 78) cases = [ ("valid", {}), ("bad id format", {"payment_id": "771"}), ("negative amount", {"amount_micros": -5}), ("unknown currency", {"currency": "GBP"}), ("extra field", {"memo": "hi"}), ("currency vs account", {"currency": "USD"}), ("over the limit", {"amount_micros": 900_000_000_000}), ] for label, override in cases: result = gateway.execute(payment(arguments=override, idempotency_key=f"idem-{label}")) status = "OK" if result.ok else "REFUSED" print(f" {label:<20} {status:<8} {result.error or ''}") print(" -> the last two are BUSINESS INVARIANTS. No JSON Schema can express") print(" 'the currency matches the account' — and if the gateway does not check") print(" it, the rule lives in the agent's prompt, where it can be argued with.") print() print("=" * 78) print("2. IDEMPOTENCY — THE THREE CASES") print("=" * 78) before = len(downstream_calls) first = gateway.execute(payment(idempotency_key="idem-A")) second = gateway.execute(payment(idempotency_key="idem-A")) third = gateway.execute(payment(idempotency_key="idem-A", arguments={"amount_micros": 1_000_000})) print(f" 1st call, key idem-A -> ok={first.ok} " f"replayed={first.replayed} ref={first.payload['reference']}") print(f" 2nd call, SAME request -> ok={second.ok} " f"replayed={second.replayed} ref={second.payload['reference']}") print(f" 3rd call, DIFFERENT request -> ok={third.ok}: {third.error}") print(f" downstream executions: {len(downstream_calls) - before} — for three calls.") print(" -> the replay returns the STORED response, so the caller sees the same") print(" reference. A fresh execution would have produced a second payment.") print() print("=" * 78) print("3. THE SIDE-EFFECT CLASS DRIVES THE POLICY") print("=" * 78) print(f" {'class':<24} {'attempts':<9} {'key':<6} {'dual control':<14} audit") for effect, policy in EFFECT_POLICY.items(): dual = ("never" if policy.requires_dual_control_above is None else f">= {policy.requires_dual_control_above // 1_000_000:,}") print(f" {effect.value:<24} {policy.max_attempts:<9} " f"{str(policy.requires_idempotency_key):<6} {dual:<14} " f"{policy.audit_level}") result = gateway.execute(ActionRequest( "crm.append_note", {"case_id": "C-1", "note": "checked"}, principal)) print(f" a write with no key -> {result.error}") print(" -> retry policy is a PLATFORM decision derived from a declared property,") print(" not a per-call-site choice made by whoever wrote the integration.") print() print("=" * 78) print("4. DUAL CONTROL") print("=" * 78) variants = [ ("two distinct humans", ("ahmed", "sara")), ("the same human twice", ("ahmed", "ahmed")), ("the agent approving itself", ("ahmed", "payments-investigator")), ("the requesting user approving", ("ahmed", "layla.almansouri")), ("no approvers", ()), ] for label, approvals in variants: result = gateway.execute(payment(approvals=approvals, idempotency_key=f"idem-{label}")) print(f" {label:<32} {'ALLOWED' if result.ok else 'REFUSED'}") small = gateway.execute(payment( arguments={"amount_micros": 50_000_000_000}, approvals=(), idempotency_key="idem-small")) print(f" below the threshold, no approvers {'ALLOWED' if small.ok else 'REFUSED'}") boundary = gateway.execute(payment( arguments={"amount_micros": 100_000_000_000}, approvals=(), idempotency_key="idem-boundary")) print(f" EXACTLY at the threshold, none " f"{'ALLOWED' if boundary.ok else 'REFUSED'}") print(" -> the boundary is inclusive. 'Above 100,000' and '100,000 or more' differ") print(" by exactly one transaction, and that is the one the auditor picks.") print() print("=" * 78) print("5. THE CIRCUIT BREAKER") print("=" * 78) breaker_clock = _clock(start=5000) breaker = CircuitBreaker(now=breaker_clock, failure_threshold=0.5, minimum_throughput=4, window_ticks=100, open_ticks=10, half_open_successes=2) print(f" {'#':<3} {'event':<10} {'observed':<10} {'rate':<6} {'state':<10} why") script = [ (False, "1/1 — but 1 < 4, the minimum throughput"), (False, "2/2 — still below the minimum"), (True, "2/3 = 0.67, but 3 < 4"), (False, "3/4 = 0.75 >= 0.50, and 4 >= 4 -> OPENS"), ] for i, (ok, why) in enumerate(script, 1): breaker.record(ok) observed = f"{sum(1 for _, o in breaker._events if not o)}/{len(breaker._events)}" print(f" {i:<3} {'success' if ok else 'failure':<10} {observed:<10} " f"{breaker.failure_rate():<6.2f} {breaker.state().value:<10} {why}") print(f" allow() while open -> {breaker.allow()}") for _ in range(12): breaker_clock() print(f" after {breaker.open_ticks} ticks -> {breaker.state().value}, and " f"allow() = {breaker.allow()} — ONE probe, not a reopening") breaker.record(True) print(f" probe 1 succeeds -> {breaker.state().value} (needs " f"{breaker.half_open_successes})") breaker.record(True) print(f" probe 2 succeeds -> {breaker.state().value}") print(" -> minimum throughput is why the first two failures did not open it. One") print(" failure out of one call is a 100% failure rate, and without the guard") print(" every low-traffic blip at 3am opens the circuit.") print() print("=" * 78) print("6. OPEN MUST DO SOMETHING") print("=" * 78) failures["payments.lookup"] = 20 fallback_gateway = ActionGateway( now=now, downstream=downstream, fallback=lambda r: {"status": "UNKNOWN", "source": "cache", "note": "stale, breaker open"}) fallback_gateway.register(ToolContract( "payments.lookup", SideEffect.READ, {"type": "object", "required": ["payment_id"], "properties": {"payment_id": {"type": "string"}}})) lookup = ActionRequest("payments.lookup", {"payment_id": "PMT-771"}, principal, trace_id="trace-lookup") for i in range(6): result = fallback_gateway.execute(lookup) state = fallback_gateway.breaker_for("payments.lookup").state().value note = result.payload.get("source") if result.payload else result.error print(f" call {i + 1}: breaker={state:<10} -> {note}") print(" -> a breaker that only fails FASTER has converted a slow error into a") print(" quick one. This one degrades to a stale cached answer, clearly labelled.") print() print("=" * 78) print("7. SAGAS — COMPENSATION IS NOT ROLLBACK") print("=" * 78) ledger: List[str] = [] def step(name: str, fail: bool = False) -> SagaStep: def forward(ctx: Dict[str, Any]) -> str: if fail: raise DownstreamError(f"{name} failed", retryable=False) ledger.append(f"do:{name}") return f"{name}-ok" def compensate(ctx: Dict[str, Any]) -> None: ledger.append(f"undo:{name}") return SagaStep(name, forward, compensate) saga = Saga("investigate-and-refund", [ step("place-hold"), step("open-case"), step("post-refund", fail=True), step("notify-customer")]) outcome = saga.run() print(f" ok={outcome.ok} failed_step={outcome.failed_step}") print(f" completed : {list(outcome.completed)}") print(f" compensated : {list(outcome.compensated)}") print(f" ledger : {ledger}") print(" -> compensations run in REVERSE, because step 2 may depend on step 1.") print(" And note what the ledger shows: the hold WAS placed and the case WAS") print(" opened. Everyone saw them. A compensation is a new, visible business") print(" action — not an undo.") ledger.clear() def orphan_step(name: str) -> SagaStep: def forward(ctx: Dict[str, Any]) -> str: ledger.append(f"do:{name}") return "ok" def compensate(ctx: Dict[str, Any]) -> None: raise DownstreamError("compensation failed too") return SagaStep(name, forward, compensate) broken = Saga("half-undone", [orphan_step("debit"), step("credit", fail=True)]) outcome = broken.run() print(f" a saga whose COMPENSATION fails: orphaned={list(outcome.orphaned)}") print(" -> there is no third level of undo. Record it loudly and page a human;") print(" the bank is now in a state no code will repair.") print() print("=" * 78) print("8. REDACTION AND THE HASH-CHAINED AUDIT LOG") print("=" * 78) dirty = {"debit_account": "AE070331234567890123456", "api_key": "sk-live-abc123", "note": "call 0501234567 about account 1234567890123", "nested": {"password": "hunter2", "amount_micros": 5}} for key, value in redact(dirty).items(): print(f" {key:<16} {value}") print(" -> redaction happens BEFORE the value reaches a log line. 'We'll scrub the") print(" logs later' is not a control: it already left the process.") print() ok, problem = gateway.audit.verify() print(f" audit records: {len(gateway.audit.records)} chain valid: {ok}") print(f" head hash: {gateway.audit.head()[:32]}...") sample = gateway.audit.records[8] print() print(f" record #{sample.seq}:") for key in ("tool_id", "side_effect", "actor_chain", "outcome", "policy_version", "model_version", "idempotency_key", "approvals"): print(f" {key:<18} {getattr(sample, key)}") print(f" {'arguments':<18} {sample.arguments}") print(f" {'prev_hash':<18} {sample.prev_hash[:24]}...") print(f" {'this_hash':<18} {sample.this_hash[:24]}...") print() print(" now edit one field of record 3, the way an attacker would:") tampered = replace(gateway.audit.records[2], arguments={"amount_micros": 1}) gateway.audit.records[2] = tampered ok, problem = gateway.audit.verify() print(f" chain valid: {ok} first problem: {problem}") print(" -> record 3's content no longer matches its own hash. Rewriting that hash") print(" breaks record 4's prev_hash, and so on to the head. Tamper-EVIDENT.") print(" Publish the head hash somewhere you do not control and it becomes") print(" tamper-resistant; nothing in this file can do that for you.") if __name__ == "__main__": # pragma: no cover main()