Model Gateways, Routing, and Fallbacks

from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
import json
import math

COST_RELEASE_ID = "support-release-2026-09-cost-v1"
GATEWAY_POLICY_ID = "gateway-policy-v1"
REQUIRED_ANSWER_SCHEMA = "cited-support-answer-v3"
MAX_GENERATED_ANSWER_USD = Decimal("0.005000")
MAX_GENERATION_ATTEMPTS = 2
REQUEST_DEADLINE_MS = 2_500
MAX_REQUEST_SPEND_USD = Decimal("0.010000")
ATTEMPT_RESERVATION_USD = Decimal("0.005000")

def owned_text(value: object) -> bool:
    return type(value) is str and bool(value.strip())

def integer_count(value: object) -> bool:
    return type(value) is int and value >= 0

def money(value: object) -> bool:
    return type(value) is Decimal and value.is_finite() and value >= 0

def seconds(value: object) -> bool:
    if type(value) not in (int, float):
        return False
    try:
        return math.isfinite(value) and value >= 0
    except OverflowError:
        return False

class DataClass(str, Enum):
    PUBLIC = "public"
    TENANT_PRIVATE = "tenant_private"

@dataclass(frozen=True)
class GatewayRequest:
    request_id: str
    task: str
    data_class: DataClass
    context_tokens: int
    risk_amount_cents: int = 0
    requires_citations: bool = False
    requires_schema: bool = True

    def __post_init__(self) -> None:
        if not owned_text(self.request_id) or not owned_text(self.task) or type(self.data_class) is not DataClass:
            raise ValueError("need owned request/task identifiers and a known data class")
        if not integer_count(self.context_tokens) or self.context_tokens == 0 or not integer_count(self.risk_amount_cents):
            raise ValueError("need positive context and nonnegative integer risk")
        if type(self.requires_citations) is not bool or type(self.requires_schema) is not bool:
            raise ValueError("requirement flags must be known Booleans")

@dataclass(frozen=True)
class RouteContract:
    request_id: str
    data_class: DataClass
    context_tokens: int
    needs_citations: bool
    needs_schema: bool
    needs_human_review: bool
    max_answer_cost_usd: Decimal

    def __post_init__(self) -> None:
        if not owned_text(self.request_id) or type(self.data_class) is not DataClass or not integer_count(self.context_tokens) or self.context_tokens == 0:
            raise ValueError("need a named contract, known data class, and positive context")
        if any(type(value) is not bool for value in (self.needs_citations, self.needs_schema, self.needs_human_review)) or self.needs_schema is not True:
            raise ValueError("known requirements and the inherited schema are required")
        if not money(self.max_answer_cost_usd) or self.max_answer_cost_usd > MAX_GENERATED_ANSWER_USD:
            raise ValueError("cost ceiling must fit the inherited policy")

@dataclass(frozen=True)
class Lane:
    name: str
    provider: str
    allowed_data_classes: frozenset[DataClass]
    max_context_tokens: int
    supports_schema: bool
    supports_citations: bool
    supports_human_review: bool
    evaluated_answer_cost_usd: Decimal
    expected_latency_ms: int

    def __post_init__(self) -> None:
        if not owned_text(self.name) or not owned_text(self.provider):
            raise ValueError("lane and provider must be named")
        if type(self.allowed_data_classes) is not frozenset or not self.allowed_data_classes or any(type(value) is not DataClass for value in self.allowed_data_classes):
            raise ValueError("need a nonempty set of known data classes")
        if not integer_count(self.max_context_tokens) or self.max_context_tokens == 0 or not integer_count(self.expected_latency_ms):
            raise ValueError("invalid context capacity or latency")
        if any(type(value) is not bool for value in (self.supports_schema, self.supports_citations, self.supports_human_review)):
            raise ValueError("capability flags must be known Booleans")
        if not money(self.evaluated_answer_cost_usd):
            raise ValueError("evaluated cost must be a finite nonnegative Decimal")

LANES = [
    Lane(
        "fast-public-json", "hosted-fast", frozenset({DataClass.PUBLIC}),
        16_000, True, False, False, Decimal("0.001100"), 260,
    ),
    Lane(
        "public-cited-review", "hosted-cited", frozenset({DataClass.PUBLIC}),
        64_000, True, True, True, Decimal("0.003800"), 860,
    ),
    Lane(
        "primary-private-cited-review", "hosted-private", frozenset({DataClass.TENANT_PRIVATE}),
        64_000, True, True, True, Decimal("0.004200"), 940,
    ),
    Lane(
        "local-private-cited-review", "local-private", frozenset({DataClass.TENANT_PRIVATE}),
        32_000, True, True, True, Decimal("0.004500"), 1_100,
    ),
    Lane(
        "regional-private-cited-review", "regional-private", frozenset({DataClass.TENANT_PRIVATE}),
        64_000, True, True, True, Decimal("0.004560"), 1_300,
    ),
    Lane(
        "cheap-text-fallback", "hosted-cheap", frozenset({DataClass.PUBLIC, DataClass.TENANT_PRIVATE}),
        32_000, False, False, False, Decimal("0.001500"), 420,
    ),
]

print(f"cost_release={COST_RELEASE_ID}")
print(f"gateway_policy={GATEWAY_POLICY_ID}")
print(f"required_answer_schema={REQUIRED_ANSWER_SCHEMA}")
print(f"max_generated_answer_usd={MAX_GENERATED_ANSWER_USD}")
print(f"max_generation_attempts={MAX_GENERATION_ATTEMPTS}")
print(f"request_deadline_ms={REQUEST_DEADLINE_MS}")
print(f"registered_lanes={len(LANES)}")

if request.data_class == "tenant_private":
    return "local-private"
if request.risk_amount_cents >= 50_000:
    return "human-review"

HIGH_RISK_ACCESS_CENTS = 50_000

def compile_contract(request: GatewayRequest) -> RouteContract:
    if type(request) is not GatewayRequest:
        raise ValueError("need a validated gateway request")
    if request.task not in {"prod_access_decision", "deploy_policy"}:
        raise ValueError("unknown task")
    access_task = request.task == "prod_access_decision"
    return RouteContract(
        request_id=request.request_id,
        data_class=DataClass.TENANT_PRIVATE if access_task else request.data_class,
        context_tokens=request.context_tokens,
        needs_citations=access_task or request.requires_citations,
        # The inherited release contract requires its schema on every route.
        # A caller can't opt out by setting requires_schema=False.
        needs_schema=True,
        needs_human_review=request.risk_amount_cents >= HIGH_RISK_ACCESS_CENTS,
        max_answer_cost_usd=MAX_GENERATED_ANSWER_USD,
    )

private_access = GatewayRequest(
    request_id="access-R900",
    task="prod_access_decision",
    data_class=DataClass.TENANT_PRIVATE,
    context_tokens=24_000,
    risk_amount_cents=90_000,
    requires_citations=True,
)
private_contract = compile_contract(private_access)
schema_downgrade = GatewayRequest(
    "schema-bypass",
    "prod_access_decision",
    DataClass.TENANT_PRIVATE,
    2_000,
    requires_schema=False,
)
assert compile_contract(schema_downgrade).needs_schema is True

print(f"request={private_contract.request_id}")
print(f"data_class={private_contract.data_class.value}")
print(f"needs_citations={private_contract.needs_citations}")
print(f"needs_human_review={private_contract.needs_human_review}")
print(f"max_answer_cost_usd={private_contract.max_answer_cost_usd}")

def contract_violations(lane: Lane, contract: RouteContract) -> list[str]:
    if type(lane) is not Lane or type(contract) is not RouteContract:
        raise ValueError("need a validated lane and contract")
    failures: list[str] = []
    if contract.data_class not in lane.allowed_data_classes:
        failures.append("data_boundary")
    if lane.max_context_tokens < contract.context_tokens:
        failures.append("context_length")
    if contract.needs_schema and not lane.supports_schema:
        failures.append("schema")
    if contract.needs_citations and not lane.supports_citations:
        failures.append("citations")
    if contract.needs_human_review and not lane.supports_human_review:
        failures.append("human_review")
    if lane.evaluated_answer_cost_usd > contract.max_answer_cost_usd:
        failures.append("budget")
    return failures

def compatible_lanes(contract: RouteContract) -> list[Lane]:
    return [lane for lane in LANES if not contract_violations(lane, contract)]

for lane in LANES:
    failures = contract_violations(lane, private_contract)
    status = "compatible" if not failures else "reject=" + ",".join(failures)
    print(f"{lane.name}: {status}")

@dataclass(frozen=True)
class RouteDecision:
    request_id: str
    lane: str | None
    action: str
    reasons: tuple[str, ...]

def choose_primary(contract: RouteContract) -> RouteDecision:
    feasible = compatible_lanes(contract)
    if not feasible:
        return RouteDecision(contract.request_id, None, "escalate", ("no_compatible_lane",))
    lane = min(feasible, key=lambda candidate: (candidate.evaluated_answer_cost_usd, candidate.expected_latency_ms, candidate.name))
    reasons = (
        f"data={contract.data_class.value}",
        f"review={str(contract.needs_human_review).lower()}",
        f"citations={str(contract.needs_citations).lower()}",
        f"budget<={contract.max_answer_cost_usd}",
    )
    return RouteDecision(contract.request_id, lane.name, "generate", reasons)

status_request = GatewayRequest(
    "docs-Q102", "deploy_policy", DataClass.PUBLIC, 2_000,
    requires_citations=False,
)

for request in (status_request, private_access):
    decision = choose_primary(compile_contract(request))
    print(f"{decision.request_id} -> {decision.lane} action={decision.action}")
    print("  " + " ".join(decision.reasons))

class FailureKind(str, Enum):
    RATE_LIMIT_BEFORE_OUTPUT = "rate_limit_before_output"
    TIMEOUT_BEFORE_OUTPUT = "timeout_before_output"
    QUOTA_EXHAUSTED = "quota_exhausted"
    CONTEXT_REJECTED = "context_rejected"
    MID_STREAM_DROP = "mid_stream_drop"
    SCHEMA_INVALID = "schema_invalid"
    AUTH_DENIED = "auth_denied"
    POLICY_REFUSAL = "policy_refusal"

def may_retry_transparently(failure: FailureKind) -> bool:
    return type(failure) is FailureKind and failure in {
        FailureKind.RATE_LIMIT_BEFORE_OUTPUT,
        FailureKind.TIMEOUT_BEFORE_OUTPUT,
    }

for failure in FailureKind:
    action = "try_compatible_fallback" if may_retry_transparently(failure) else "stop_or_escalate"
    print(f"{failure.value}: {action}")

def choose_fallback(contract: RouteContract, failed_lane: str) -> RouteDecision:
    if not owned_text(failed_lane) or failed_lane not in {lane.name for lane in LANES}:
        raise ValueError("need a known failed lane")
    candidates = [
        lane for lane in compatible_lanes(contract)
        if lane.name != failed_lane
    ]
    if not candidates:
        return RouteDecision(contract.request_id, None, "escalate", ("fallback_contract_unmet",))
    lane = min(candidates, key=lambda candidate: (candidate.evaluated_answer_cost_usd, candidate.expected_latency_ms, candidate.name))
    return RouteDecision(
        contract.request_id,
        lane.name,
        "fallback_generate",
        (f"primary_failed={failed_lane}", "contract_preserved=true"),
    )

primary = choose_primary(private_contract)
fallback = choose_fallback(private_contract, primary.lane or "")
print(f"primary={primary.lane}")
print(f"fallback={fallback.lane} action={fallback.action}")
print("cheap_text_rejected=" + ",".join(contract_violations(LANES[-1], private_contract)))

class CircuitStatus(str, Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitState:
    status: CircuitStatus = CircuitStatus.CLOSED
    failures: int = 0
    opened_until: float = 0.0
    probe_until: float = 0.0

class CircuitBreaker:
    def __init__(self, threshold: int = 2, cooldown_seconds: float = 10.0,
                 probe_timeout_seconds: float = 2.5) -> None:
        if not integer_count(threshold) or threshold == 0 or not seconds(cooldown_seconds) or cooldown_seconds == 0 or not seconds(probe_timeout_seconds) or probe_timeout_seconds == 0:
            raise ValueError("positive finite threshold, cooldown, and probe lifetime required")
        self.threshold = threshold
        self.cooldown_seconds = cooldown_seconds
        self.probe_timeout_seconds = probe_timeout_seconds
        self.states: dict[str, CircuitState] = {}

    def permit(self, provider: str, now: float) -> bool:
        if not owned_text(provider) or not seconds(now) or not seconds(now + self.cooldown_seconds) or not seconds(now + self.probe_timeout_seconds):
            return False
        state = self.states.setdefault(provider, CircuitState())
        if state.status == CircuitStatus.HALF_OPEN:
            if now >= state.probe_until:
                state.status = CircuitStatus.OPEN
                state.opened_until = now + self.cooldown_seconds
            return False
        if state.status == CircuitStatus.OPEN:
            if now < state.opened_until:
                return False
            state.status = CircuitStatus.HALF_OPEN
            state.probe_until = now + self.probe_timeout_seconds
            return True
        return state.status == CircuitStatus.CLOSED

    def failure(self, provider: str, now: float) -> None:
        if not owned_text(provider) or not seconds(now) or not seconds(now + self.cooldown_seconds):
            raise ValueError("need a named target and finite nonnegative clock")
        state = self.states.setdefault(provider, CircuitState())
        state.failures += 1
        if state.status == CircuitStatus.HALF_OPEN or state.failures >= self.threshold:
            state.status = CircuitStatus.OPEN
            state.opened_until = now + self.cooldown_seconds

    def success(self, provider: str) -> None:
        if not owned_text(provider):
            raise ValueError("need a named target")
        self.states[provider] = CircuitState()

breaker = CircuitBreaker()
breaker.failure("hosted-private", 100.0)
breaker.failure("hosted-private", 101.0)
print(f"during_cooldown={breaker.permit('hosted-private', 105.0)}")
print(f"probe_after_cooldown={breaker.permit('hosted-private', 112.0)}")
breaker.success("hosted-private")
print(f"after_success={breaker.states['hosted-private'].status.value}")

unanswered_probe = CircuitBreaker(threshold=1)
unanswered_probe.failure("probe-target", 100.0)
assert unanswered_probe.permit("probe-target", 110.0)
assert not unanswered_probe.permit("probe-target", 110.1)
assert not unanswered_probe.permit("probe-target", 112.5)
assert unanswered_probe.states["probe-target"].status is CircuitStatus.OPEN
print(f"expired_probe={unanswered_probe.states['probe-target'].status.value}")

@dataclass(frozen=True)
class RouteEvent:
    request_id: str
    policy_id: str
    cost_release_id: str
    action: str
    lane: str | None
    contract_summary: str
    reason: str
    evaluated_cost_usd: Decimal
    attempted_lanes: tuple[str, ...] = ()
    elapsed_ms: int = 0
    fixture_spend_usd: Decimal = Decimal("0")

def lane_by_name(name: str) -> Lane:
    return next(lane for lane in LANES if lane.name == name)

def summarize_contract(contract: RouteContract) -> str:
    return (
        f"data={contract.data_class.value};"
        f"context_tokens={contract.context_tokens};"
        f"schema={str(contract.needs_schema).lower()};"
        f"schema_id={REQUIRED_ANSWER_SCHEMA};"
        f"citations={str(contract.needs_citations).lower()};"
        f"review={str(contract.needs_human_review).lower()};"
        f"budget<={contract.max_answer_cost_usd}"
    )

@dataclass(frozen=True)
class Attempt:
    duration_ms: int
    charge_usd: Decimal
    failure: FailureKind | None = None

    def __post_init__(self) -> None:
        if not integer_count(self.duration_ms) or not money(self.charge_usd):
            raise ValueError("need a nonnegative integer duration and finite Decimal charge")
        if self.failure is not None and type(self.failure) is not FailureKind:
            raise ValueError("need a known failure or explicit completion")

def execute_with_failure(
    request: GatewayRequest,
    failure: FailureKind | None,
    now: float = 200.0,
    scripted: dict[str, Attempt] | None = None,
    retry_delay_ms: int = 0,
    request_cap: Decimal = MAX_REQUEST_SPEND_USD,
) -> RouteEvent:
    if failure is not None and type(failure) is not FailureKind:
        raise ValueError("need a classified failure or explicit completion")
    if not seconds(now):
        raise ValueError("need a finite nonnegative fixture clock")
    if not money(request_cap):
        raise ValueError("request cap must be a finite nonnegative Decimal")
    if scripted is None:
        scripted = {}
    if type(scripted) is not dict or any(
        not owned_text(name) or name not in {lane.name for lane in LANES}
        or type(attempt) is not Attempt or attempt.charge_usd > ATTEMPT_RESERVATION_USD
        for name, attempt in scripted.items()
    ):
        raise ValueError("need known scripted lanes and charges within the reserved bound")
    contract = compile_contract(request)
    contract_summary = summarize_contract(contract)
    candidates = sorted(
        compatible_lanes(contract),
        key=lambda lane: (lane.evaluated_answer_cost_usd, lane.expected_latency_ms, lane.name),
    )
    if type(retry_delay_ms) is not int or retry_delay_ms < 0:
        raise ValueError("retry delay must be nonnegative integer milliseconds")
    # Synthetic known charges, distinct from registry cost estimates.
    charges = dict(zip((lane.name for lane in LANES), map(Decimal, (
        "0.001100", "0.003800", "0.004200", "0.004500", "0.004560", "0.001500",
    ))))
    attempted: list[str] = []
    elapsed = 0
    spend = Decimal("0")
    reason = "no_primary_lane" if not candidates else "no_permitted_compatible_lane"

    def event(action: str, lane: Lane | None = None) -> RouteEvent:
        return RouteEvent(
            request.request_id, GATEWAY_POLICY_ID, COST_RELEASE_ID, action,
            lane.name if lane else None, contract_summary, reason,
            lane.evaluated_answer_cost_usd if lane else Decimal("0"),
            tuple(attempted), elapsed, spend,
        )

    for index, lane in enumerate(candidates):
        if elapsed >= REQUEST_DEADLINE_MS:
            reason = "deadline_exhausted"
            break
        if len(attempted) >= MAX_GENERATION_ATTEMPTS:
            reason = "attempts_exhausted"
            break
        # Don't wait or spend for a fallback that can't finish in the time left.
        delay = retry_delay_ms if attempted else 0
        if attempted and delay + lane.expected_latency_ms >= REQUEST_DEADLINE_MS - elapsed:
            reason = "deadline_exhausted"
            continue
        if spend + ATTEMPT_RESERVATION_USD > request_cap:
            reason = "request_spend_reservation_denied"
            break
        if not breaker.permit(lane.provider, now + elapsed / 1000):
            if index == 0:
                reason = "primary_circuit_open"
            continue
        elapsed += delay
        attempt = scripted.get(lane.name, Attempt(
            lane.expected_latency_ms, charges[lane.name], failure if index == 0 else None,
        ))
        attempted.append(lane.name)
        remaining = REQUEST_DEADLINE_MS - elapsed
        elapsed += min(attempt.duration_ms, remaining)
        spend += attempt.charge_usd
        outcome = attempt.failure
        if attempt.duration_ms >= remaining and outcome is not FailureKind.MID_STREAM_DROP:
            outcome = FailureKind.TIMEOUT_BEFORE_OUTPUT
        if outcome is None:
            breaker.success(lane.provider)
            reason = "primary_contract_match" if index == 0 else reason + ";contract_preserved"
            return event("served" if index == 0 else "served_fallback", lane)
        # Auth, refusal, and schema failures aren't provider-health evidence here.
        if outcome in {FailureKind.TIMEOUT_BEFORE_OUTPUT, FailureKind.MID_STREAM_DROP}:
            breaker.failure(lane.provider, now + elapsed / 1000)
        elif breaker.states[lane.provider].status == CircuitStatus.HALF_OPEN:
            breaker.success(lane.provider)  # reachable, even if response was unusable
        reason = ("primary_" if index == 0 else "fallback_") + outcome.value
        if not may_retry_transparently(outcome):
            return event("escalate")
    return event("escalate")

breaker = CircuitBreaker()
before_output = execute_with_failure(private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT, now=200.0)
mid_stream = execute_with_failure(private_access, FailureKind.MID_STREAM_DROP, now=203.0)
breaker.failure("local-private", 204.0)
breaker.failure("local-private", 205.0)
skip_open_local = execute_with_failure(private_access, None, now=206.0)

# These are scripted charges, not provider-reported usage.
primary_timeout_usage = Decimal("0.004200")
fallback_usage = Decimal("0.004500")
request_attempt_costs = [primary_timeout_usage, fallback_usage]
request_total_spend = sum(request_attempt_costs, Decimal("0"))
assert before_output.evaluated_cost_usd == Decimal("0.004500")
assert request_total_spend == Decimal("0.008700")
assert request_total_spend > MAX_GENERATED_ANSWER_USD
assert before_output.fixture_spend_usd == request_total_spend
assert before_output.elapsed_ms == 2_040

print(f"before_output={before_output.action} lane={before_output.lane} reason={before_output.reason}")
print(f"audit_policy={before_output.policy_id} cost_release={before_output.cost_release_id}")
print(f"audit_contract={before_output.contract_summary}")
print(f"mid_stream={mid_stream.action} lane={mid_stream.lane} reason={mid_stream.reason}")
print(f"circuit_after_failures={breaker.states['hosted-private'].status.value}")
print(f"open_local_skipped={skip_open_local.action} lane={skip_open_local.lane} reason={skip_open_local.reason}")
print(f"attempt_costs={','.join(str(c) for c in request_attempt_costs)}")
print(f"request_total_spend={request_total_spend} admission_ceiling={MAX_GENERATED_ANSWER_USD}")

too_large_private_request = GatewayRequest(
    "access-long-context", "prod_access_decision", DataClass.TENANT_PRIVATE, 70_000,
    risk_amount_cents=90_000, requires_citations=True,
)

breaker = CircuitBreaker(threshold=1, cooldown_seconds=10.0)
replay_cases = [
    (status_request, None),
    (private_access, None),
    (private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT),
    (private_access, None),
    (too_large_private_request, None),
]
events = [
    execute_with_failure(request, failure, now=300.0 + 3 * index)
    for index, (request, failure) in enumerate(replay_cases)
]

served = []
expected_actions = ["served", "served", "served_fallback", "served_fallback", "escalate"]
for (request, _), event, expected in zip(replay_cases, events, expected_actions):
    assert event.action == expected
    assert event.elapsed_ms <= REQUEST_DEADLINE_MS
    assert len(event.attempted_lanes) <= MAX_GENERATION_ATTEMPTS
    assert event.fixture_spend_usd <= MAX_REQUEST_SPEND_USD
    if event.action.startswith("served"):
        lane = lane_by_name(event.lane or "")
        assert not contract_violations(lane, compile_contract(request))
        assert event.evaluated_cost_usd <= MAX_GENERATED_ANSWER_USD
        served.append(event)

for event in events:
    print(f"{event.request_id}: {event.action} lane={event.lane}")
route_field_violations = sum(
    len(contract_violations(lane_by_name(event.lane), compile_contract(request)))
    for (request, _), event in zip(replay_cases, events)
    if event.action.startswith("served")
)
assert route_field_violations == 0
print(f"compatible_transport_completions={len(served)}/{len(events)}")
print(f"route_field_violations={route_field_violations}")

primary_name = "primary-private-cited-review"
local_name = "local-private-cited-review"
cases = {
    "fallback_also_fails": ({
        local_name: Attempt(500, Decimal("0.002000"), FailureKind.TIMEOUT_BEFORE_OUTPUT),
    }, 0, MAX_REQUEST_SPEND_USD, "attempts_exhausted", 2),
    "shared_deadline": ({
        primary_name: Attempt(1_800, Decimal("0.004200"), FailureKind.TIMEOUT_BEFORE_OUTPUT),
    }, 0, MAX_REQUEST_SPEND_USD, "deadline_exhausted", 1),
    "retry_delay_uses_deadline": ({}, 2_000, MAX_REQUEST_SPEND_USD, "deadline_exhausted", 1),
    "request_cap": ({}, 0, Decimal("0.008000"), "request_spend_reservation_denied", 1),
    "auth_failure": ({
        primary_name: Attempt(20, Decimal("0"), FailureKind.AUTH_DENIED),
    }, 0, MAX_REQUEST_SPEND_USD, "primary_auth_denied", 1),
    "visible_output_at_deadline": ({
        primary_name: Attempt(3_000, Decimal("0.004200"), FailureKind.MID_STREAM_DROP),
    }, 0, MAX_REQUEST_SPEND_USD, "primary_mid_stream_drop", 1),
}
for name, (scripted, delay, cap, reason, attempts) in cases.items():
    breaker = CircuitBreaker()
    result = execute_with_failure(
        private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT,
        scripted=scripted, retry_delay_ms=delay, request_cap=cap,
    )
    assert result.action == "escalate" and result.reason == reason
    assert len(result.attempted_lanes) == attempts
    assert result.elapsed_ms <= REQUEST_DEADLINE_MS
    assert result.fixture_spend_usd <= cap
    print(f"{name}: {reason}, attempts={attempts}, elapsed_ms={result.elapsed_ms}")

def validate_reply(payload: object, contract: RouteContract, evidence_ids: frozenset[str]) -> dict:
    if type(contract) is not RouteContract or type(evidence_ids) is not frozenset or any(not owned_text(item) for item in evidence_ids):
        raise ValueError("need a validated contract and owned evidence identifiers")
    fields = {"schema", "status", "answer", "citations", "review"}
    if type(payload) is not dict or set(payload) != fields:
        raise ValueError("unexpected reply shape")
    if payload["schema"] != REQUIRED_ANSWER_SCHEMA or payload["status"] != "complete":
        raise ValueError("wrong schema, refusal, or incomplete reply")
    if type(payload["answer"]) is not str or not payload["answer"].strip():
        raise ValueError("missing answer")
    citations = payload["citations"]
    if type(citations) is not list or any(type(item) is not str for item in citations):
        raise ValueError("invalid citations")
    if len(set(citations)) != len(citations) or not set(citations) <= evidence_ids:
        raise ValueError("duplicate or unknown citation")
    if contract.needs_citations and not citations:
        raise ValueError("citations required")
    # Require the marker; a real workflow must separately create/verify handoff state.
    expected_review = "pending" if contract.needs_human_review else "not_required"
    if payload["review"] != expected_review:
        raise ValueError("invalid review state")
    return payload | {"citations": citations.copy()}

good_reply = {
    "schema": REQUIRED_ANSWER_SCHEMA, "status": "complete",
    "answer": "Send this access request for human review.",
    "citations": ["E1"], "review": "pending",
}
assert validate_reply(good_reply, private_contract, frozenset({"E1"})) == good_reply
for name, patch in {
    "refused": {"status": "refused"},
    "truncated": {"status": "length"},
    "wrong_schema": {"schema": "plain-json-v1"},
    "invented_citation": {"citations": ["E99"]},
    "self_approved": {"review": "approved"},
}.items():
    try:
        validate_reply(good_reply | patch, private_contract, frozenset({"E1"}))
    except ValueError:
        print(f"{name}: blocked before publication")
    else:
        raise AssertionError(name)

@dataclass(frozen=True)
class Caller:
    principal: str
    tenant: str
    permissions: frozenset[str]

    def __post_init__(self) -> None:
        if not owned_text(self.principal) or not owned_text(self.tenant) or type(self.permissions) is not frozenset or any(not owned_text(value) for value in self.permissions):
            raise ValueError("need owned caller identity and explicit permission names")

ledger: dict[tuple[str, str, str], tuple[tuple[str, int], str]] = {}
write_count = 0

def grant_fixture(caller: Caller, tenant: str, operation_key: str,
                  target: str, minutes: int, human_approved: bool) -> str:
    global write_count
    if type(caller) is not Caller or not owned_text(tenant):
        raise PermissionError("need a validated caller and named tenant")
    if caller.tenant != tenant or "access:grant" not in caller.permissions:
        raise PermissionError("grant not authorized")
    if human_approved is not True:
        raise PermissionError("trusted human approval required")
    if not owned_text(operation_key) or not owned_text(target) or not integer_count(minutes) or minutes == 0:
        raise ValueError("invalid operation")
    key = (tenant, caller.principal, operation_key)
    parameters = (target, minutes)
    if key in ledger:
        previous, receipt = ledger[key]
        if previous != parameters:
            raise ValueError("idempotency key reused with different parameters")
        return receipt
    write_count += 1
    receipt = f"grant-receipt-{write_count}"
    ledger[key] = (parameters, receipt)
    return receipt

caller = Caller("reviewer-7", "tenant-a", frozenset({"access:grant"}))
receipt = grant_fixture(caller, "tenant-a", "access-R900-grant", "prod-console", 15, True)
# Retrying after a lost write response returns the same receipt.
assert grant_fixture(caller, "tenant-a", "access-R900-grant", "prod-console", 15, True) == receipt
breaker = CircuitBreaker()
generation = execute_with_failure(private_access, FailureKind.TIMEOUT_BEFORE_OUTPUT)
assert generation.action == "served_fallback" and write_count == 1
for tenant, minutes, approved in [("tenant-b", 15, True), ("tenant-a", 30, True), ("tenant-a", 15, False)]:
    try:
        grant_fixture(caller, tenant, "access-R900-grant", "prod-console", minutes, approved)
    except (PermissionError, ValueError):
        pass
    else:
        raise AssertionError("unsafe write accepted")
assert write_count == 1
print(f"receipt={receipt} writes={write_count} generation={generation.action}")
print("cross-tenant, changed-parameters, and unapproved writes blocked")

policy_artifact = {
    "policy_id": GATEWAY_POLICY_ID,
    "cost_release_id": COST_RELEASE_ID,
    "required_answer_schema": REQUIRED_ANSWER_SCHEMA,
    "max_generated_answer_usd": str(MAX_GENERATED_ANSWER_USD),
    "retry_limits": {
        "max_generation_attempts": MAX_GENERATION_ATTEMPTS,
        "request_deadline_ms": REQUEST_DEADLINE_MS,
        "request_spend_cap_usd": str(MAX_REQUEST_SPEND_USD),
        "max_attempt_reservation_usd": str(ATTEMPT_RESERVATION_USD),
    },
    "fixture_examples": {
        "public_deploy_policy": "fast-public-json",
        "private_high_risk_access": "primary-private-cited-review",
        "private_high_risk_access_fallback": "local-private-cited-review",
    },
    "escalate_when": [
        "no lane preserves all contract fields",
        "failure occurs after visible output begins",
        "approved private context capacity is exceeded",
        "retry attempts or request deadline are exhausted",
        "request spend reservation is denied",
        "failure is not eligible for transparent generation retry",
        "normalized reply fails the versioned application contract",
    ],
}

print(json.dumps(policy_artifact, indent=2))