#!/usr/bin/env python3 """Typed failure injector for Paper 188. The programme runs a synthetic refund workflow under controlled faults in five operational domains: identity, context, tool, state, and evidence. It is a teaching and design-assurance artefact, not a production certification tool. Examples: python paper-188-typed-failure-injector.py --profile baseline --trials 2000 python paper-188-typed-failure-injector.py --profile typed --trials 2000 python paper-188-typed-failure-injector.py --profile typed --fault tool --json """ from __future__ import annotations import argparse import json from dataclasses import asdict, dataclass from enum import Enum from typing import Generic, Iterable, Literal, Sequence, TypeVar T = TypeVar("T") class LCG: """Small deterministic generator shared with the browser lab. It is deliberately simple so the article and command-line artefact produce identical synthetic runs. It is not suitable for security or simulation work that depends on high-quality randomness. """ def __init__(self, seed: int) -> None: self.state = seed & 0xFFFFFFFF def random(self) -> float: self.state = (1664525 * self.state + 1013904223) & 0xFFFFFFFF return self.state / 4294967296.0 def choice(self, values: Sequence[T]) -> T: if not values: raise IndexError("cannot choose from an empty sequence") return values[min(len(values) - 1, int(self.random() * len(values)))] class FaultDomain(str, Enum): IDENTITY = "identity" CONTEXT = "context" TOOL = "tool" STATE = "state" EVIDENCE = "evidence" @dataclass(frozen=True) class Controls: bind_identity: bool pre_effect_authority: bool refresh_context: bool state_precondition: bool idempotency: bool effect_readback: bool evidence_gate: bool @classmethod def baseline(cls) -> "Controls": return cls(False, False, False, False, False, False, False) @classmethod def typed(cls) -> "Controls": return cls(True, True, True, True, True, True, True) @dataclass(frozen=True) class FaultSpec: domain: FaultDomain mode: str target: str timing: str intensity: float = 1.0 scope: str = "one synthetic order" @dataclass(frozen=True) class TrialOutcome: domain: str mode: str detected: bool contained: bool recovered: bool verified: bool useful_service: bool unsafe_escape: bool duplicate_effect: bool unverifiable_release: bool service_loss: bool effects: int verdict: Literal["release", "hold", "deny", "unsafe"] @dataclass class Aggregate: trials: int = 0 detected: int = 0 contained: int = 0 recovered: int = 0 verified: int = 0 useful_service: int = 0 unsafe_escape: int = 0 duplicate_effect: int = 0 unverifiable_release: int = 0 service_loss: int = 0 def add(self, outcome: TrialOutcome) -> None: self.trials += 1 for field in ( "detected", "contained", "recovered", "verified", "useful_service", "unsafe_escape", "duplicate_effect", "unverifiable_release", "service_loss", ): setattr(self, field, getattr(self, field) + int(getattr(outcome, field))) def rates(self) -> dict[str, float | int]: n = max(self.trials, 1) return { "trials": self.trials, "detected": self.detected / n, "contained": self.contained / n, "recovered": self.recovered / n, "verified": self.verified / n, "useful_service": self.useful_service / n, "unsafe_escape": self.unsafe_escape / n, "duplicate_effect": self.duplicate_effect / n, "unverifiable_release": self.unverifiable_release / n, "service_loss": self.service_loss / n, } def choose_fault(domain: FaultDomain, rng: LCG) -> FaultSpec: modes: dict[FaultDomain, Sequence[tuple[str, str, str]]] = { FaultDomain.IDENTITY: ( ("subject_swap", "action contract", "after proposal, before authority check"), ("revoked_capability", "capability grant", "after approval, before effect"), ), FaultDomain.CONTEXT: ( ("stale_policy", "policy assertion", "after retrieval, before effect"), ("cross_source_conflict", "eligibility assertion", "during context assembly"), ), FaultDomain.TOOL: ( ("timeout_after_commit", "refund API", "after commit, before response"), ("partial_success_response", "tool response", "after effect"), ), FaultDomain.STATE: ( ("checkpoint_rollback", "workflow checkpoint", "after effect, before resume"), ("concurrent_version_change", "order state", "after read, before write"), ), FaultDomain.EVIDENCE: ( ("drop_effect_receipt", "evidence ledger", "after effect"), ("break_trace_lineage", "trace context", "between authority and tool"), ), } mode, target, timing = rng.choice(modes[domain]) return FaultSpec(domain=domain, mode=mode, target=target, timing=timing) def _safe_denial(domain: FaultDomain, mode: str) -> TrialOutcome: return TrialOutcome( domain=domain.value, mode=mode, detected=True, contained=True, recovered=True, verified=True, useful_service=True, unsafe_escape=False, duplicate_effect=False, unverifiable_release=False, service_loss=False, effects=0, verdict="deny", ) def _unsafe(domain: FaultDomain, mode: str, effects: int = 1, duplicate: bool = False) -> TrialOutcome: return TrialOutcome( domain=domain.value, mode=mode, detected=False, contained=False, recovered=False, verified=False, useful_service=False, unsafe_escape=True, duplicate_effect=duplicate, unverifiable_release=False, service_loss=False, effects=effects, verdict="unsafe", ) def simulate_trial(spec: FaultSpec, controls: Controls, rng: LCG) -> TrialOutcome: """Run one synthetic intervention. The intended service is domain-dependent. Identity and context faults should be denied. Tool and state faults should complete exactly once or stop in a provably safe hold. Evidence faults should never be released as verified unless a correlated receipt or independent readback exists. """ d, mode = spec.domain, spec.mode if d is FaultDomain.IDENTITY: if mode == "subject_swap": if controls.bind_identity: return _safe_denial(d, mode) return _unsafe(d, mode) if controls.pre_effect_authority: return _safe_denial(d, mode) return _unsafe(d, mode) if d is FaultDomain.CONTEXT: if controls.refresh_context: return _safe_denial(d, mode) return _unsafe(d, mode) if d is FaultDomain.TOOL: if mode == "timeout_after_commit": effects = 1 detected = True if controls.effect_readback: return TrialOutcome(d.value, mode, detected, True, True, True, True, False, False, False, False, effects, "release") if controls.idempotency: # A retry with the same key returns the original logical result. return TrialOutcome(d.value, mode, detected, True, True, True, True, False, False, False, False, effects, "release") # A blind retry uses a new request identity and commits again. effects = 2 return TrialOutcome(d.value, mode, detected, False, False, False, False, True, True, False, False, effects, "unsafe") # Effect occurred, but the response is missing the authoritative receipt. effects = 1 if controls.effect_readback: return TrialOutcome(d.value, mode, True, True, True, True, True, False, False, False, False, effects, "release") if controls.evidence_gate: return TrialOutcome(d.value, mode, True, True, False, False, False, False, False, False, True, effects, "hold") return TrialOutcome(d.value, mode, False, True, False, False, False, False, False, True, False, effects, "release") if d is FaultDomain.STATE: if mode == "concurrent_version_change": if controls.state_precondition: return _safe_denial(d, mode) return _unsafe(d, mode) # The external effect already happened, but the restored checkpoint says it did not. effects = 1 if controls.effect_readback or controls.idempotency: return TrialOutcome(d.value, mode, True, True, True, True, True, False, False, False, False, effects, "release") effects = 2 return TrialOutcome(d.value, mode, False, False, False, False, False, True, True, False, False, effects, "unsafe") if d is FaultDomain.EVIDENCE: effects = 1 if mode == "drop_effect_receipt" and controls.effect_readback: return TrialOutcome(d.value, mode, True, True, True, True, True, False, False, False, False, effects, "release") if mode == "break_trace_lineage" and controls.evidence_gate: # The effect can be found, but the injector refuses to claim causal proof. return TrialOutcome(d.value, mode, True, True, False, False, False, False, False, False, True, effects, "hold") if controls.evidence_gate: return TrialOutcome(d.value, mode, True, True, False, False, False, False, False, False, True, effects, "hold") return TrialOutcome(d.value, mode, False, True, False, False, False, False, False, True, False, effects, "release") raise AssertionError(f"Unhandled domain: {d}") def run_experiment( controls: Controls, domains: Sequence[FaultDomain], trials: int, seed: int, ) -> tuple[dict[str, Aggregate], list[TrialOutcome]]: rng = LCG(seed) aggregates = {domain.value: Aggregate() for domain in domains} outcomes: list[TrialOutcome] = [] for i in range(trials): domain = domains[i % len(domains)] spec = choose_fault(domain, rng) outcome = simulate_trial(spec, controls, rng) aggregates[domain.value].add(outcome) outcomes.append(outcome) return aggregates, outcomes def pct(value: float) -> str: return f"{100.0 * value:6.1f}%" def release_assessment(overall: dict[str, float | int]) -> tuple[str, list[str]]: reasons: list[str] = [] if float(overall["unsafe_escape"]) > 0.0: reasons.append("at least one represented fault escaped into an unsafe effect") if float(overall["unverifiable_release"]) > 0.0: reasons.append("at least one success was released without outcome proof") if float(overall["contained"]) < 0.99: reasons.append("containment was below the 99% synthetic target") if float(overall["useful_service"]) < 0.80: reasons.append("useful service was below the 80% synthetic target") return ("PASS ON THIS SYNTHETIC ASSAY" if not reasons else "BLOCK OR REDESIGN", reasons) def aggregate_all(aggregates: Iterable[Aggregate]) -> Aggregate: total = Aggregate() for agg in aggregates: for field in asdict(total): setattr(total, field, getattr(total, field) + getattr(agg, field)) return total def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--profile", choices=("baseline", "typed"), default="typed") parser.add_argument( "--fault", choices=("all",) + tuple(domain.value for domain in FaultDomain), default="all", help="Fault family to inject.", ) parser.add_argument("--trials", type=int, default=2000) parser.add_argument("--seed", type=int, default=188) parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") return parser.parse_args() def main() -> int: args = parse_args() if args.trials < 1: raise SystemExit("--trials must be at least 1") controls = Controls.typed() if args.profile == "typed" else Controls.baseline() domains = list(FaultDomain) if args.fault == "all" else [FaultDomain(args.fault)] aggregates, outcomes = run_experiment(controls, domains, args.trials, args.seed) overall_agg = aggregate_all(aggregates.values()) overall = overall_agg.rates() assessment, reasons = release_assessment(overall) if args.json: result = { "profile": args.profile, "controls": asdict(controls), "seed": args.seed, "fault_domains": [d.value for d in domains], "by_domain": {name: agg.rates() for name, agg in aggregates.items()}, "overall": overall, "assessment": assessment, "reasons": reasons, "interpretation": { "positive": "The configured controls contained the represented synthetic faults under this harness.", "negative": "A negative result identifies a control or recovery gap in the represented path.", "cannot_establish": "The harness does not establish production reliability, unseen-fault coverage, or a causal role for any language model component.", }, } print(json.dumps(result, indent=2, sort_keys=True)) return 0 print("Typed failure-injection assay (synthetic)") print(f"Profile: {args.profile} | trials: {args.trials} | seed: {args.seed}") print("Controls:") for name, enabled in asdict(controls).items(): print(f" {name:24} {'ON' if enabled else 'off'}") print() header = ( f"{'domain':10} {'n':>6} {'detect':>8} {'contain':>8} {'recover':>8} " f"{'verify':>8} {'service':>8} {'unsafe':>8} {'dupe':>8} {'unproved':>9}" ) print(header) print("-" * len(header)) for domain in domains: r = aggregates[domain.value].rates() print( f"{domain.value:10} {int(r['trials']):6d} {pct(float(r['detected'])):>8} " f"{pct(float(r['contained'])):>8} {pct(float(r['recovered'])):>8} " f"{pct(float(r['verified'])):>8} {pct(float(r['useful_service'])):>8} " f"{pct(float(r['unsafe_escape'])):>8} {pct(float(r['duplicate_effect'])):>8} " f"{pct(float(r['unverifiable_release'])):>9}" ) print("-" * len(header)) print( f"{'overall':10} {int(overall['trials']):6d} {pct(float(overall['detected'])):>8} " f"{pct(float(overall['contained'])):>8} {pct(float(overall['recovered'])):>8} " f"{pct(float(overall['verified'])):>8} {pct(float(overall['useful_service'])):>8} " f"{pct(float(overall['unsafe_escape'])):>8} {pct(float(overall['duplicate_effect'])):>8} " f"{pct(float(overall['unverifiable_release'])):>9}" ) print(f"\nAssessment: {assessment}") for reason in reasons: print(f" - {reason}") print("\nInterpretation boundary: this is evidence about represented faults in a synthetic configured system, not proof of production safety.") return 0 if __name__ == "__main__": raise SystemExit(main())