"""Lab 01 — The action gateway: the bank's enforcement boundary. This is where a model's *suggestion* becomes an *effect on the bank*. Work top to bottom; each section's tests pass once that section is done: 1. side-effect classes and the policy table 2. schema validation 3. contracts and business invariants 4. idempotency 5. the circuit breaker 6. redaction 7. the hash-chained audit log 8. dual control 9. the gateway 10. sagas pytest # against your work LAB_MODULE=solution pytest # against the reference Determinism rules: the clock and the downstream are injected, money is an integer number of micro-units divided last, and identifiers are derived rather than random. """ 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.""" 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 #: TODO: fill in the table. The tests pin these properties: #: #: * every SideEffect has an entry; #: * READ may be retried (> 1 attempt) and needs no idempotency key; #: * WRITE_NON_IDEMPOTENT and IRREVERSIBLE are attempted exactly ONCE; #: * every write class requires an idempotency key; #: * only IRREVERSIBLE requires dual control (use 100_000_000_000 micro-units); #: * IRREVERSIBLE is not compensable. #: #: 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. EFFECT_POLICY: Mapping[SideEffect, EffectPolicy] = {} # ====================================================================================== # 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]: """TODO: a small JSON-Schema subset. Return EVERY error, SORTED. Support: ``type``, ``required``, ``properties``, ``additionalProperties: false``, ``items``, ``pattern`` (whole-string), ``enum``, ``minimum``, ``maximum``. Two things the tests pin: * **A boolean is not an integer.** ``bool`` is a subclass of ``int`` in Python, so the naive ``isinstance(value, int)`` accepts ``True`` for an amount. This is the single most common validator bug in the language. * **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. Error format: ``"$.path.to.field: what is wrong"``; array elements as ``$.xs[0]``. """ raise NotImplementedError #: 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 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]: """TODO: schema first; if it fails, return immediately. Do NOT run invariants on a structurally invalid payload — they would raise KeyError and mask the real problem with a stack trace. Otherwise run every invariant and return all their complaints, sorted. """ raise NotImplementedError # ====================================================================================== # 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: """TODO: ``user -> hop -> hop -> agent``, with ``"system"`` when there is no user. This string is the answer to "who authorized this?", so it must show the whole path, not just the last hop.""" raise NotImplementedError @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: """TODO: a canonical sha256 digest of *what was asked*. Cover the tool, the arguments, and the principal (agent, user, tenant). EXCLUDE the trace id and the model version: a retry of the same business intent from a different trace is the SAME request, and treating it as different turns every retry into a 409. Serialize with ``sort_keys=True`` so argument order does not change the digest. """ raise NotImplementedError 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** — conflict, 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. Conflict 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: raise NotImplementedError def begin(self, key: str, digest: str) -> Optional[IdemRecord]: """TODO: reserve the key, or return the existing record for a replay. * expired record -> treat as fresh; * no record -> store IN_FLIGHT, return None (the caller should execute); * different hash -> raise ``IdempotencyConflict`` — check this BEFORE the in-flight check, because a conflicting request must never be told "try again shortly", which is what would make it eventually execute; * IN_FLIGHT -> raise ``InFlight``; * COMPLETED -> return the record so the caller can replay its response. """ raise NotImplementedError def complete(self, key: str, response: Any) -> IdemRecord: raise NotImplementedError def fail(self, key: str) -> None: """TODO: release the key so the caller may retry. A failed non-idempotent write is the genuinely hard case: we do not know whether the downstream applied it. Releasing lets the caller retry — 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. Whichever you choose, document it; the tests here expect release. """ raise NotImplementedError def get(self, key: str) -> Optional[IdemRecord]: raise NotImplementedError # ====================================================================================== # 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. """ 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: # TODO: store the config, an event list of (tick, ok), the state, the time it # opened, a probe counter, and a transition log for the incident timeline. raise NotImplementedError def state(self) -> BreakerState: """TODO: current state — and this is where OPEN ages into HALF_OPEN once ``open_ticks`` have passed. Doing it here rather than on a timer keeps the whole class deterministic under an injected clock.""" raise NotImplementedError def allow(self) -> bool: """TODO: False only when OPEN. HALF_OPEN admits the probe.""" raise NotImplementedError def record(self, ok: bool) -> None: """TODO: the outcome of one call. * **HALF_OPEN**: one failure re-opens immediately (the downstream told us); successes accumulate, and reaching ``half_open_successes`` closes it. * **CLOSED**: append the event, drop events older than ``window_ticks``, and open when BOTH ``len(events) >= minimum_throughput`` AND ``failures / total >= failure_threshold``. Closing must CLEAR the window — otherwise the failures that opened it are still in the window and it re-opens on the first new failure. """ raise NotImplementedError def failure_rate(self) -> float: """TODO: 0.0 for an empty window, not a ZeroDivisionError.""" raise NotImplementedError # ====================================================================================== # 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: """TODO: redact recursively, 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. * a key containing any of ``_SECRETISH`` (case-insensitively) -> ``"[REDACTED]"``; * a long digit run in a string -> ``****`` plus the last four, because an audit record that cannot distinguish two accounts is not much of an audit record; * recurse into mappings and sequences; leave everything else alone. """ raise NotImplementedError # ====================================================================================== # 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: """TODO: a canonical sha256 over every field EXCEPT ``this_hash`` — including ``prev_hash``, which is what makes it a chain rather than a list of hashes.""" raise NotImplementedError 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. Honest limit: this is tamper-**evident**, not tamper-proof. Someone with write access can rebuild the whole chain. Publishing the head hash somewhere you do not control is what closes that, and it is a deployment concern rather than a code one. """ def __init__(self, *, now: Callable[[], int]) -> None: raise NotImplementedError def append(self, **fields: Any) -> AuditRecord: """TODO: seq is ``len(records) + 1``; ``prev_hash`` is the previous record's ``this_hash`` or ``GENESIS``; compute ``this_hash`` last.""" raise NotImplementedError def verify(self) -> Tuple[bool, Optional[str]]: """TODO: return ``(ok, first_problem)``. Check THREE things per record: * the sequence number is right (catches a deletion); * ``prev_hash`` matches the previous record's ``this_hash`` (catches a re-hashed edit); * ``digest()`` matches ``this_hash`` (catches a raw content edit). Only checking one of the three leaves an attack open; the tests cover all three. """ raise NotImplementedError def head(self) -> str: raise NotImplementedError def for_trace(self, trace_id: str) -> List[AuditRecord]: raise NotImplementedError # ====================================================================================== # 8. Dual control # ====================================================================================== def check_dual_control(request: ActionRequest, *, threshold: Optional[int], required: int = 2) -> Optional[str]: """TODO: return None when satisfied, or an explanation. 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 a SET, not a list); * the requesting agent counting itself; * the initiating *user* approving their own request — four-eyes with one pair of eyes is not four-eyes. Exclude everyone in the delegation chain too. Threshold boundaries are **inclusive**: ``value_micros >= threshold`` requires approval. "Above 100,000" and "100,000 or more" differ by exactly one transaction, and that transaction is the one the auditor picks. ``threshold=None`` means dual control never applies. """ raise NotImplementedError # ====================================================================================== # 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.""" 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: raise NotImplementedError def register(self, contract: ToolContract) -> None: """TODO: reject a duplicate tool id.""" raise NotImplementedError def breaker_for(self, tool_id: str) -> CircuitBreaker: """TODO: one breaker per tool, created on demand. Per tool, not per gateway — a breaker shared across tools lets one sick downstream shut off healthy ones.""" raise NotImplementedError def execute(self, request: ActionRequest) -> ActionResult: """TODO: the one entry point. The ORDER is the design: 1. **unknown tool** -> refuse, audited; 2. **contract** — schema, then invariants; 3. **the key must exist** when the class demands one; 4. **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; 5. **idempotency** — conflict and in-flight refuse; a completed record replays its stored response and executes ZERO more times; 6. **the breaker**, immediately before execution — so an open circuit does not consume an idempotency key. Open + a fallback = a degraded answer; open with no fallback = refuse; 7. **execute**, up to ``policy.max_attempts``, stopping early on a non-retryable error, recording each outcome in the breaker; 8. **audit** every path — success, replay, failure and every refusal. Cheap local refusals come first so a malformed request never touches the idempotency store or the downstream. """ raise NotImplementedError # ====================================================================================== # 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. 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.""" def __init__(self, name: str, steps: Sequence[SagaStep], *, audit: Optional[AuditLog] = None) -> None: raise NotImplementedError def run(self, context: Optional[Dict[str, Any]] = None) -> SagaOutcome: """TODO: run each step forward, storing its result in the context under the step's name so later steps can read it. On a failure, compensate the COMPLETED steps — not the failing one, which by definition did not take effect.""" raise NotImplementedError def _attempt(self, step: SagaStep, ctx: Dict[str, Any]) -> Any: """TODO: up to ``retries + 1`` attempts; a non-retryable ``DownstreamError`` stops immediately.""" raise NotImplementedError def _compensate(self, completed: Sequence[SagaStep], ctx: Dict[str, Any]) -> Tuple[Tuple[str, ...], Tuple[str, ...]]: """TODO: compensate in REVERSE order — step 3 may rely on step 2's effect, so undoing 2 first leaves 3's compensation operating on state that no longer exists. Return ``(compensated, orphaned)``. A step lands in ``orphaned`` when it has no compensation, or when its compensation itself raised. Do NOT let one failed compensation stop the others, and do NOT swallow it: there is no third level of undo, so an orphan must be recorded loudly for a human, because the bank is now in a state no code will repair. """ raise NotImplementedError # ====================================================================================== # Worked example # ====================================================================================== def main() -> None: # pragma: no cover """TODO (optional): once the tests pass, build the eight-section demo. Compare against ``python solution.py`` — but only after your own runs. """ raise NotImplementedError if __name__ == "__main__": # pragma: no cover main()