diff --git a/.github/workflows/repository-integrity.yml b/.github/workflows/repository-integrity.yml index 9004e1d..3cecc2a 100644 --- a/.github/workflows/repository-integrity.yml +++ b/.github/workflows/repository-integrity.yml @@ -30,3 +30,8 @@ jobs: - name: Validate repository structure and internal links run: python tools/check_repository.py + + - name: Run E001 invariant tests + env: + PYTHONPATH: src + run: python -m unittest discover -s tests -p "test_*.py" -v diff --git a/experiments/E001/config/CANONICAL_MATRIX__E001__v0.1__2026-07-13.json b/experiments/E001/config/CANONICAL_MATRIX__E001__v0.1__2026-07-13.json new file mode 100644 index 0000000..ebc21b3 --- /dev/null +++ b/experiments/E001/config/CANONICAL_MATRIX__E001__v0.1__2026-07-13.json @@ -0,0 +1,35 @@ +{ + "configurations": [ + "central", + "global_barrier", + "local_cbf" + ], + "parameters": { + "capacity_per_loop": 8, + "credit_lease_ticks": 3, + "injection_end_tick": 119, + "interlock_transfer_limit": 1, + "maximum_drain_horizon": 360, + "normal_service_period": 1, + "primary_horizon": 240, + "work_deadline_ticks": 60 + }, + "scenarios": [ + "balanced", + "burst", + "saturated_receiver", + "slow_neighbor", + "failed_neighbor", + "circular_wait", + "stale_feedback", + "malformed_duplicate" + ], + "schema_version": "e001.matrix.v1", + "seeds": [ + 17, + 29, + 43 + ], + "specification": "EXPERIMENT__E001__THREE_RING_BOUNDED_FLOW__v0.1__2026-07-13.md" +} + diff --git a/src/README.md b/src/README.md index f0c910e..58f487e 100644 --- a/src/README.md +++ b/src/README.md @@ -13,3 +13,19 @@ The first implementation should be a deterministic discrete-event simulator with - interchangeable centralized, global-barrier, and local-interlock schedulers. The implementation must conform to the [frozen E001 specification](../experiments/E001/EXPERIMENT__E001__THREE_RING_BOUNDED_FLOW__v0.1__2026-07-13.md). A later runtime may replace Python if it preserves canonical inputs, traces, and declared semantics. + +## E001 Stage A + +The dependency-free simulator lives in `src/superloop_e001/` and can run one configuration or the complete canonical matrix: + +```bash +PYTHONPATH=src python -m superloop_e001 run \ + --configuration local_cbf \ + --scenario balanced \ + --seed 17 \ + --print-summary + +PYTHONPATH=src python -m superloop_e001 matrix --output /tmp/e001-results +``` + +The matrix command writes generated evidence only to an explicit output directory. Canonical repository evidence should be promoted separately after review of the source commit and complete manifest. diff --git a/src/superloop_e001/__init__.py b/src/superloop_e001/__init__.py new file mode 100644 index 0000000..6c29d49 --- /dev/null +++ b/src/superloop_e001/__init__.py @@ -0,0 +1,6 @@ +"""Deterministic Stage A simulator for Superloop experiment E001.""" + +from .simulator import RunResult, run_matrix, run_simulation + +__all__ = ["RunResult", "run_matrix", "run_simulation"] + diff --git a/src/superloop_e001/__main__.py b/src/superloop_e001/__main__.py new file mode 100644 index 0000000..72f2cfb --- /dev/null +++ b/src/superloop_e001/__main__.py @@ -0,0 +1,5 @@ +from .cli import main + + +raise SystemExit(main()) + diff --git a/src/superloop_e001/cli.py b/src/superloop_e001/cli.py new file mode 100644 index 0000000..746f7f7 --- /dev/null +++ b/src/superloop_e001/cli.py @@ -0,0 +1,53 @@ +"""Command-line interface for E001 Stage A.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .simulator import matrix_manifest, run_matrix, run_simulation, write_result +from .workloads import CONFIGURATIONS, SCENARIOS, SEEDS + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Run the deterministic E001 observation chamber.") + subparsers = parser.add_subparsers(dest="command", required=True) + + run_parser = subparsers.add_parser("run", help="Run one canonical configuration.") + run_parser.add_argument("--configuration", choices=CONFIGURATIONS, required=True) + run_parser.add_argument("--scenario", choices=SCENARIOS, required=True) + run_parser.add_argument("--seed", choices=SEEDS, type=int, required=True) + run_parser.add_argument("--output", type=Path) + run_parser.add_argument("--print-summary", action="store_true") + + matrix_parser = subparsers.add_parser("matrix", help="Run all 72 canonical configurations.") + matrix_parser.add_argument("--output", type=Path, required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + if args.command == "run": + result = run_simulation(args.configuration, args.scenario, args.seed) + if args.output: + write_result(result, args.output) + if args.print_summary or not args.output: + print(json.dumps(result.summary, ensure_ascii=False, indent=2, sort_keys=True)) + return int(result.summary["invariant_violation_count"] > 0) + + results = run_matrix() + for result in results: + write_result(result, args.output) + manifest = matrix_manifest(results) + (args.output / "matrix_manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True)) + return int(bool(manifest["invalid_run_ids"] or manifest["workload_equivalence_mismatches"])) + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/src/superloop_e001/model.py b/src/superloop_e001/model.py new file mode 100644 index 0000000..470bd05 --- /dev/null +++ b/src/superloop_e001/model.py @@ -0,0 +1,179 @@ +"""Core E001 state and canonical trace representations.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from typing import Any + + +LOOP_IDS = ("A", "B", "C") +NEXT_LOOP = {"A": "B", "B": "C", "C": "A"} +ROUTES = { + "A": ("A", "B", "C"), + "B": ("B", "C", "A"), + "C": ("C", "A", "B"), +} + +TRACE_FIELDS = ( + "schema_version", + "run_id", + "configuration", + "scenario", + "seed", + "workload_digest", + "event_id", + "simulation_tick", + "event_type", + "loop_id", + "interlock_id", + "work_id", + "parent_event_ids", + "validation_state", + "credit_before", + "credit_after", + "occupancy_before", + "occupancy_after", + "lease_fresh_until", + "obligation_created", + "obligation_resolved", + "provenance_complete", + "invariant_checks", + "terminal_state", + "terminal_reason", +) + + +def canonical_json(value: Any) -> str: + """Serialize a JSON-compatible value deterministically.""" + + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def digest(value: Any) -> str: + """Return a SHA-256 digest of canonical JSON.""" + + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +def raw_digest(value: str) -> str: + """Return a SHA-256 digest of exact UTF-8 text bytes.""" + + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class Offer: + """One externally offered work event.""" + + tick: int + work_id: str + source_loop: str + schema_version: str = "e001.work.v1" + duplicate_replay: bool = False + + def canonical(self) -> dict[str, Any]: + return { + "duplicate_replay": self.duplicate_replay, + "schema_version": self.schema_version, + "source_loop": self.source_loop, + "tick": self.tick, + "work_id": self.work_id, + } + + +@dataclass +class WorkItem: + """Authoritative work state while an item is active.""" + + work_id: str + source_loop: str + route: tuple[str, str, str] + route_index: int + created_at: int + admitted_at: int + deadline: int + schema_version: str + payload_digest: str + priority: str = "normal" + provenance: list[str] = field(default_factory=list) + processed: bool = False + + @property + def location(self) -> str: + return self.route[self.route_index] + + @property + def final_stage(self) -> bool: + return self.route_index == len(self.route) - 1 + + +@dataclass +class LoopState: + """Bounded local reservoir and FIFO work ordering.""" + + loop_id: str + capacity: int + queue: list[str] = field(default_factory=list) + + @property + def occupancy(self) -> int: + return len(self.queue) + + @property + def free_capacity(self) -> int: + return self.capacity - self.occupancy + + +class TraceRecorder: + """Construct canonical, complete transition records.""" + + def __init__( + self, + *, + configuration: str, + scenario: str, + seed: int, + workload_digest: str, + ) -> None: + self.configuration = configuration + self.scenario = scenario + self.seed = seed + self.workload_digest = workload_digest + self.run_id = f"{configuration}__{scenario}__seed-{seed}" + self.events: list[dict[str, Any]] = [] + self._sequence = 0 + + def emit(self, *, simulation_tick: int, event_type: str, **values: Any) -> str: + self._sequence += 1 + event_id = f"{self.run_id}:{self._sequence:07d}" + record = {field_name: None for field_name in TRACE_FIELDS} + record.update( + { + "schema_version": "e001.trace.v1", + "run_id": self.run_id, + "configuration": self.configuration, + "scenario": self.scenario, + "seed": self.seed, + "workload_digest": self.workload_digest, + "event_id": event_id, + "simulation_tick": simulation_tick, + "event_type": event_type, + "parent_event_ids": [], + "obligation_created": False, + "obligation_resolved": False, + "invariant_checks": [], + } + ) + unknown = set(values) - set(TRACE_FIELDS) + if unknown: + raise ValueError(f"unknown trace fields: {sorted(unknown)}") + record.update(values) + self.events.append(record) + return event_id + + def jsonl(self) -> str: + if not self.events: + return "" + return "\n".join(canonical_json(event) for event in self.events) + "\n" diff --git a/src/superloop_e001/simulator.py b/src/superloop_e001/simulator.py new file mode 100644 index 0000000..d327cf8 --- /dev/null +++ b/src/superloop_e001/simulator.py @@ -0,0 +1,1052 @@ +"""Deterministic discrete-event implementation of E001 Stage A.""" + +from __future__ import annotations + +import json +import math +import platform +import time +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable + +from .model import ( + LOOP_IDS, + ROUTES, + LoopState, + Offer, + TraceRecorder, + WorkItem, + digest, + raw_digest, +) +from .workloads import ( + CONFIGURATIONS, + SCENARIOS, + SEEDS, + build_offers, + build_circular_preload, + offers_by_tick, + workload_digest, +) + + +CAPACITY = 8 +CREDIT_LEASE_TICKS = 3 +WORK_DEADLINE_TICKS = 60 +PRIMARY_HORIZON = 240 +MAXIMUM_DRAIN_HORIZON = 360 + + +@dataclass +class RunResult: + """Canonical trace plus machine-readable summary.""" + + trace: list[dict[str, Any]] + trace_jsonl: str + summary: dict[str, Any] + execution: dict[str, Any] + + +class Simulation: + """One configuration, scenario, and seed.""" + + def __init__(self, configuration: str, scenario: str, seed: int) -> None: + if configuration not in CONFIGURATIONS: + raise ValueError(f"unsupported configuration: {configuration}") + if scenario not in SCENARIOS: + raise ValueError(f"unsupported scenario: {scenario}") + if seed not in SEEDS: + raise ValueError(f"seed is outside the canonical set: {seed}") + + self.configuration = configuration + self.scenario = scenario + self.seed = seed + self.workload_digest = workload_digest(scenario, seed) + self.trace = TraceRecorder( + configuration=configuration, + scenario=scenario, + seed=seed, + workload_digest=self.workload_digest, + ) + self.loops = {loop_id: LoopState(loop_id, CAPACITY) for loop_id in LOOP_IDS} + self.items: dict[str, WorkItem] = {} + self.terminal_items: dict[str, tuple[str, str | None, int]] = {} + self.seen_work_ids: set[str] = set() + self.last_event_by_work: dict[str, str] = {} + self.offers = build_offers(scenario) + self.offers_by_tick = offers_by_tick(self.offers) + self.offered_attempts = 0 + self.admitted_count = 0 + self.rejected_counts: Counter[str] = Counter() + self.outstanding_obligations = 0 + self.max_outstanding_obligations = 0 + self.stale_rejection_count = 0 + self.stale_authorization_count = 0 + self.duplicate_promotion_count = 0 + self.invariant_violations: list[str] = [] + self.max_occupancy = {loop_id: 0 for loop_id in LOOP_IDS} + self.occupancy_time = {loop_id: 0 for loop_id in LOOP_IDS} + self.occupancy_history = {loop_id: [] for loop_id in LOOP_IDS} + self.progress_ticks = {loop_id: set() for loop_id in LOOP_IDS} + self.completion_ticks: list[int] = [] + self.elapsed_ticks = 0 + + def _emit( + self, + *, + tick: int, + event_type: str, + item: WorkItem | None = None, + work_id: str | None = None, + loop_id: str | None = None, + parents: Iterable[str] = (), + update_work_parent: bool = True, + **values: Any, + ) -> str: + effective_work_id = work_id if work_id is not None else (item.work_id if item else None) + if item is not None: + values.setdefault("provenance_complete", bool(item.provenance)) + loop_id = loop_id or item.location + event_id = self.trace.emit( + simulation_tick=tick, + event_type=event_type, + loop_id=loop_id, + work_id=effective_work_id, + parent_event_ids=list(parents), + **values, + ) + if effective_work_id is not None and update_work_parent: + self.last_event_by_work[effective_work_id] = event_id + if loop_id in self.progress_ticks and event_type in { + "process", + "transfer_commit", + "complete", + }: + self.progress_ticks[loop_id].add(tick) + return event_id + + def _parent(self, work_id: str) -> tuple[str, ...]: + parent = self.last_event_by_work.get(work_id) + return (parent,) if parent else () + + def _loop_available(self, loop_id: str, tick: int) -> bool: + return not (self.scenario == "failed_neighbor" and loop_id == "B" and 40 <= tick <= 79) + + def _transfer_admission_open(self, loop_id: str, tick: int) -> bool: + if not self._loop_available(loop_id, tick): + return False + return not ( + self.scenario == "saturated_receiver" + and loop_id == "B" + and 35 <= tick <= 54 + ) + + def _service_due(self, loop_id: str, tick: int) -> bool: + if not self._loop_available(loop_id, tick): + return False + if self.scenario == "slow_neighbor" and loop_id == "B" and 40 <= tick <= 99: + return (tick - 40) % 4 == 0 + return True + + def _admit_offer(self, offer: Offer, tick: int) -> None: + self.offered_attempts += 1 + duplicate = offer.duplicate_replay or offer.work_id in self.seen_work_ids + offer_event = self._emit( + tick=tick, + event_type="offer", + work_id=offer.work_id, + loop_id=offer.source_loop, + validation_state="proposed", + update_work_parent=not duplicate, + ) + if duplicate: + self.rejected_counts["duplicate_ignored"] += 1 + self._emit( + tick=tick, + event_type="reject", + work_id=offer.work_id, + loop_id=offer.source_loop, + parents=(offer_event,), + validation_state="rejected_duplicate", + terminal_state="duplicate_ignored", + terminal_reason="work_id already exists", + update_work_parent=False, + ) + return + self.seen_work_ids.add(offer.work_id) + if offer.schema_version != "e001.work.v1": + self.rejected_counts["rejected_invalid"] += 1 + self._emit( + tick=tick, + event_type="reject", + work_id=offer.work_id, + loop_id=offer.source_loop, + parents=(offer_event,), + validation_state="rejected_schema", + terminal_state="rejected_invalid", + terminal_reason=f"unsupported schema: {offer.schema_version}", + ) + return + + loop = self.loops[offer.source_loop] + if loop.free_capacity <= 0 or not self._loop_available(loop.loop_id, tick): + self.rejected_counts["rejected_capacity"] += 1 + self._emit( + tick=tick, + event_type="reject", + work_id=offer.work_id, + loop_id=offer.source_loop, + parents=(offer_event,), + validation_state="rejected_capacity", + occupancy_before=loop.occupancy, + occupancy_after=loop.occupancy, + terminal_state="rejected_capacity", + terminal_reason="source loop has no admissible capacity", + ) + return + + item = WorkItem( + work_id=offer.work_id, + source_loop=offer.source_loop, + route=ROUTES[offer.source_loop], + route_index=0, + created_at=offer.tick, + admitted_at=tick, + deadline=tick + WORK_DEADLINE_TICKS, + schema_version=offer.schema_version, + payload_digest=digest({"work_id": offer.work_id, "source": offer.source_loop}), + provenance=[offer_event], + ) + before = loop.occupancy + self.items[item.work_id] = item + loop.queue.append(item.work_id) + self.admitted_count += 1 + event_id = self._emit( + tick=tick, + event_type="admit", + item=item, + parents=(offer_event,), + validation_state="accepted", + occupancy_before=before, + occupancy_after=loop.occupancy, + ) + item.provenance.append(event_id) + + def _preload_circular_wait(self) -> None: + for record in build_circular_preload(): + loop_id = str(record["source_loop"]) + work_id = str(record["work_id"]) + loop = self.loops[loop_id] + self.offered_attempts += 1 + self.seen_work_ids.add(work_id) + offer_event = self._emit( + tick=0, + event_type="offer", + work_id=work_id, + loop_id=loop_id, + validation_state="proposed", + ) + item = WorkItem( + work_id=work_id, + source_loop=loop_id, + route=tuple(record["route"]), + route_index=int(record["route_index"]), + created_at=0, + admitted_at=int(record["admitted_at"]), + deadline=int(record["deadline"]), + schema_version="e001.work.v1", + payload_digest=digest({"work_id": work_id, "source": loop_id}), + provenance=[offer_event], + processed=bool(record["processed"]), + ) + before = loop.occupancy + self.items[work_id] = item + loop.queue.append(work_id) + self.admitted_count += 1 + admit_event = self._emit( + tick=0, + event_type="admit", + item=item, + parents=(offer_event,), + validation_state="accepted", + occupancy_before=before, + occupancy_after=loop.occupancy, + ) + item.provenance.append(admit_event) + process_event = self._emit( + tick=0, + event_type="process", + item=item, + parents=(admit_event,), + occupancy_before=loop.occupancy, + occupancy_after=loop.occupancy, + ) + item.provenance.append(process_event) + + def _expire_due_items(self, tick: int) -> None: + for work_id in sorted(list(self.items)): + item = self.items[work_id] + if tick < item.deadline: + continue + loop = self.loops[item.location] + before = loop.occupancy + loop.queue.remove(work_id) + del self.items[work_id] + self.terminal_items[work_id] = ("expired_deadline", "deadline reached", tick) + self._emit( + tick=tick, + event_type="expire", + item=item, + loop_id=loop.loop_id, + parents=self._parent(work_id), + occupancy_before=before, + occupancy_after=loop.occupancy, + terminal_state="expired_deadline", + terminal_reason="deadline reached", + ) + + def _process(self, tick: int) -> None: + for loop_id in LOOP_IDS: + if not self._service_due(loop_id, tick): + continue + loop = self.loops[loop_id] + candidate = next( + (self.items[work_id] for work_id in loop.queue if not self.items[work_id].processed), + None, + ) + if candidate is None: + continue + candidate.processed = True + process_event = self._emit( + tick=tick, + event_type="process", + item=candidate, + loop_id=loop_id, + parents=self._parent(candidate.work_id), + occupancy_before=loop.occupancy, + occupancy_after=loop.occupancy, + ) + candidate.provenance.append(process_event) + if candidate.final_stage: + before = loop.occupancy + loop.queue.remove(candidate.work_id) + del self.items[candidate.work_id] + self.terminal_items[candidate.work_id] = ("completed", None, tick) + self.completion_ticks.append(tick) + self._emit( + tick=tick, + event_type="complete", + item=candidate, + loop_id=loop_id, + parents=(process_event,), + occupancy_before=before, + occupancy_after=loop.occupancy, + terminal_state="completed", + ) + + def _ready_candidate(self, loop_id: str) -> WorkItem | None: + for work_id in self.loops[loop_id].queue: + item = self.items[work_id] + if item.processed and not item.final_stage: + return item + return None + + def _can_transfer(self, item: WorkItem, tick: int) -> bool: + destination = item.route[item.route_index + 1] + return ( + self._loop_available(item.location, tick) + and self._transfer_admission_open(destination, tick) + and self.loops[destination].free_capacity > 0 + ) + + def _commit_transfer( + self, + item: WorkItem, + tick: int, + *, + parent: str, + validation_state: str, + lease_fresh_until: int | None, + credit_before: int | None, + ) -> None: + source_id = item.location + destination_id = item.route[item.route_index + 1] + source = self.loops[source_id] + destination = self.loops[destination_id] + source_before = source.occupancy + destination_before = destination.occupancy + source.queue.remove(item.work_id) + destination.queue.append(item.work_id) + item.route_index += 1 + item.processed = False + interlock_id = f"{source_id}{destination_id}" + self.outstanding_obligations += 1 + self.max_outstanding_obligations = max( + self.max_outstanding_obligations, + self.outstanding_obligations, + ) + commit_event = self._emit( + tick=tick, + event_type="transfer_commit", + item=item, + loop_id=source_id, + parents=(parent,), + interlock_id=interlock_id, + validation_state=validation_state, + credit_before=credit_before, + credit_after=0 if credit_before is not None else None, + occupancy_before=source_before, + occupancy_after=source.occupancy, + lease_fresh_until=lease_fresh_until, + obligation_created=True, + ) + item.provenance.append(commit_event) + self.outstanding_obligations -= 1 + receipt_event = self._emit( + tick=tick, + event_type="receipt", + item=item, + loop_id=destination_id, + parents=(commit_event,), + interlock_id=interlock_id, + validation_state="acknowledged", + occupancy_before=destination_before, + occupancy_after=destination.occupancy, + obligation_resolved=True, + ) + item.provenance.append(receipt_event) + + def _commit_transfer_batch( + self, + candidates: list[WorkItem], + tick: int, + *, + validation_state: str, + ) -> None: + """Atomically rotate a globally coordinated transfer batch.""" + + before = {loop_id: self.loops[loop_id].occupancy for loop_id in LOOP_IDS} + validation_events: dict[str, str] = {} + movements: list[tuple[WorkItem, str, str]] = [] + for item in candidates: + source_id = item.location + destination_id = item.route[item.route_index + 1] + validation_events[item.work_id] = self._emit( + tick=tick, + event_type="validate", + item=item, + loop_id=source_id, + parents=self._parent(item.work_id), + interlock_id=f"{source_id}{destination_id}", + validation_state=validation_state, + ) + movements.append((item, source_id, destination_id)) + + for item, source_id, _ in movements: + self.loops[source_id].queue.remove(item.work_id) + for item, _, destination_id in movements: + self.loops[destination_id].queue.append(item.work_id) + item.route_index += 1 + item.processed = False + + after = {loop_id: self.loops[loop_id].occupancy for loop_id in LOOP_IDS} + for item, source_id, destination_id in movements: + interlock_id = f"{source_id}{destination_id}" + self.outstanding_obligations += 1 + self.max_outstanding_obligations = max( + self.max_outstanding_obligations, + self.outstanding_obligations, + ) + commit_event = self._emit( + tick=tick, + event_type="transfer_commit", + item=item, + loop_id=source_id, + parents=(validation_events[item.work_id],), + interlock_id=interlock_id, + validation_state=validation_state, + occupancy_before=before[source_id], + occupancy_after=after[source_id], + obligation_created=True, + ) + item.provenance.append(commit_event) + self.outstanding_obligations -= 1 + receipt_event = self._emit( + tick=tick, + event_type="receipt", + item=item, + loop_id=destination_id, + parents=(commit_event,), + interlock_id=interlock_id, + validation_state="acknowledged", + occupancy_before=before[destination_id], + occupancy_after=after[destination_id], + obligation_resolved=True, + ) + item.provenance.append(receipt_event) + + def _batch_capacity_available(self, candidates: list[WorkItem], tick: int) -> bool: + incoming = Counter(item.route[item.route_index + 1] for item in candidates) + outgoing = Counter(item.location for item in candidates) + return all( + self._transfer_admission_open(destination, tick) + and incoming_count + <= self.loops[destination].free_capacity + outgoing[destination] + for destination, incoming_count in incoming.items() + ) + + def _central_transfers(self, tick: int) -> None: + offset = (tick + self.seed) % len(LOOP_IDS) + order = LOOP_IDS[offset:] + LOOP_IDS[:offset] + ready = [ + item + for loop_id in order + if (item := self._ready_candidate(loop_id)) is not None + ] + candidates = [ + item + for item in ready + if self._loop_available(item.location, tick) + and self._transfer_admission_open(item.route[item.route_index + 1], tick) + ] + if candidates and self._batch_capacity_available(candidates, tick): + self._commit_transfer_batch( + candidates, + tick, + validation_state="accepted_central", + ) + committed = {item.work_id for item in candidates} + for item in ready: + if item.work_id in committed: + continue + source = item.location + destination = item.route[item.route_index + 1] + self._emit( + tick=tick, + event_type="wait", + item=item, + loop_id=source, + parents=self._parent(item.work_id), + interlock_id=f"{source}{destination}", + validation_state="wait_capacity", + ) + return + for loop_id in order: + item = self._ready_candidate(loop_id) + if item is None: + continue + destination = item.route[item.route_index + 1] + if not self._can_transfer(item, tick): + self._emit( + tick=tick, + event_type="wait", + item=item, + loop_id=loop_id, + parents=self._parent(item.work_id), + interlock_id=f"{loop_id}{destination}", + validation_state="wait_capacity", + ) + continue + validation_event = self._emit( + tick=tick, + event_type="validate", + item=item, + loop_id=loop_id, + parents=self._parent(item.work_id), + interlock_id=f"{loop_id}{destination}", + validation_state="accepted_central", + ) + self._commit_transfer( + item, + tick, + parent=validation_event, + validation_state="accepted_central", + lease_fresh_until=None, + credit_before=None, + ) + + def _global_barrier_transfers(self, tick: int) -> None: + candidates = [ + item + for loop_id in LOOP_IDS + if (item := self._ready_candidate(loop_id)) is not None + ] + if not candidates: + return + all_available = all(self._loop_available(loop_id, tick) for loop_id in LOOP_IDS) + capacity_available = self._batch_capacity_available(candidates, tick) + if not all_available or not capacity_available: + for item in candidates: + source = item.location + destination = item.route[item.route_index + 1] + self._emit( + tick=tick, + event_type="wait", + item=item, + loop_id=source, + parents=self._parent(item.work_id), + interlock_id=f"{source}{destination}", + validation_state="wait_global_barrier", + ) + return + self._commit_transfer_batch( + candidates, + tick, + validation_state="accepted_global_barrier", + ) + + def _local_cbf_transfers(self, tick: int) -> None: + offset = (tick + self.seed) % len(LOOP_IDS) + order = LOOP_IDS[offset:] + LOOP_IDS[:offset] + for loop_id in order: + item = self._ready_candidate(loop_id) + if item is None: + continue + destination_id = item.route[item.route_index + 1] + interlock_id = f"{loop_id}{destination_id}" + proposal = self._emit( + tick=tick, + event_type="propose", + item=item, + loop_id=loop_id, + parents=self._parent(item.work_id), + interlock_id=interlock_id, + validation_state="proposed", + ) + destination = self.loops[destination_id] + credit_before = destination.free_capacity + if not self._can_transfer(item, tick): + self._emit( + tick=tick, + event_type="credit_deny", + item=item, + loop_id=loop_id, + parents=(proposal,), + interlock_id=interlock_id, + validation_state="wait_capacity", + credit_before=max(credit_before, 0), + credit_after=max(credit_before, 0), + ) + self._emit( + tick=tick, + event_type="wait", + item=item, + loop_id=loop_id, + parents=self._parent(item.work_id), + interlock_id=interlock_id, + validation_state="wait_capacity", + ) + continue + lease_fresh_until = tick + CREDIT_LEASE_TICKS + credit = self._emit( + tick=tick, + event_type="credit_grant", + item=item, + loop_id=loop_id, + parents=(proposal,), + interlock_id=interlock_id, + validation_state="credit_granted", + credit_before=credit_before, + credit_after=credit_before - 1, + lease_fresh_until=lease_fresh_until, + ) + validation = self._emit( + tick=tick, + event_type="validate", + item=item, + loop_id=loop_id, + parents=(credit,), + interlock_id=interlock_id, + validation_state="accepted", + lease_fresh_until=lease_fresh_until, + ) + self._commit_transfer( + item, + tick, + parent=validation, + validation_state="accepted", + lease_fresh_until=lease_fresh_until, + credit_before=1, + ) + + def _inject_stale_feedback(self, tick: int) -> None: + if self.scenario != "stale_feedback" or tick != 30: + return + self.stale_rejection_count += 1 + self._emit( + tick=tick, + event_type="validate", + work_id="stale-replay-30", + loop_id="A", + interlock_id="AB", + validation_state="rejected_stale", + lease_fresh_until=27, + terminal_state="rejected_invalid", + terminal_reason="acceptance lease expired", + ) + + def _fault_boundaries(self, tick: int) -> None: + if self.scenario != "failed_neighbor": + return + if tick == 40: + self._emit( + tick=tick, + event_type="isolate", + loop_id="B", + validation_state="isolated_fault", + terminal_reason="scheduled failure begins", + ) + elif tick == 80: + self._emit( + tick=tick, + event_type="recover", + loop_id="B", + validation_state="recovered", + terminal_reason="scheduled failure ends", + ) + + def _check_invariants(self, tick: int) -> None: + violations: list[str] = [] + queued_ids: list[str] = [] + for loop_id, loop in self.loops.items(): + if not 0 <= loop.occupancy <= loop.capacity: + violations.append(f"capacity:{loop_id}:{loop.occupancy}") + queued_ids.extend(loop.queue) + for work_id in loop.queue: + item = self.items.get(work_id) + if item is None: + violations.append(f"queue_without_item:{loop_id}:{work_id}") + elif item.location != loop_id: + violations.append(f"location_mismatch:{work_id}:{loop_id}:{item.location}") + elif item.payload_digest != digest( + {"work_id": item.work_id, "source": item.source_loop} + ): + violations.append(f"payload_digest_mismatch:{work_id}") + if len(queued_ids) != len(set(queued_ids)): + violations.append("duplicate_authoritative_queue_entry") + if set(queued_ids) != set(self.items): + violations.append("active_ledger_partition_mismatch") + if set(self.items) & set(self.terminal_items): + violations.append("active_and_terminal_overlap") + if self.outstanding_obligations < 0: + violations.append("negative_obligation_count") + for event in self.trace.events: + if event["simulation_tick"] != tick: + continue + for field_name in ("credit_before", "credit_after"): + credit = event[field_name] + if credit is not None and not 0 <= credit <= CAPACITY: + violations.append(f"credit_range:{event['event_id']}:{field_name}:{credit}") + if self.offered_attempts != self.admitted_count + sum(self.rejected_counts.values()): + violations.append("offer_accounting_mismatch") + if self.admitted_count != len(self.items) + len(self.terminal_items): + violations.append("admission_accounting_mismatch") + if violations: + self.invariant_violations.extend(f"tick-{tick}:{value}" for value in violations) + self._emit( + tick=tick, + event_type="invariant_check", + invariant_checks=violations, + terminal_reason="hard invariant violation", + ) + + def _sample_occupancy(self) -> None: + for loop_id, loop in self.loops.items(): + occupancy = loop.occupancy + self.max_occupancy[loop_id] = max(self.max_occupancy[loop_id], occupancy) + self.occupancy_time[loop_id] += occupancy + self.occupancy_history[loop_id].append(occupancy) + + def run(self) -> RunResult: + started = time.perf_counter() + if self.scenario == "circular_wait": + self._preload_circular_wait() + + for tick in range(MAXIMUM_DRAIN_HORIZON): + self._fault_boundaries(tick) + self._expire_due_items(tick) + for offer in self.offers_by_tick.get(tick, []): + self._admit_offer(offer, tick) + self._inject_stale_feedback(tick) + self._process(tick) + if self.configuration == "central": + self._central_transfers(tick) + elif self.configuration == "global_barrier": + self._global_barrier_transfers(tick) + else: + self._local_cbf_transfers(tick) + self._check_invariants(tick) + self._sample_occupancy() + self.elapsed_ticks = tick + 1 + if tick + 1 >= PRIMARY_HORIZON and not self.items: + break + + if self.items: + self.invariant_violations.append("active_work_after_maximum_drain_horizon") + + trace_jsonl = self.trace.jsonl() + summary = self._build_summary(trace_jsonl=trace_jsonl) + execution = { + "schema_version": "e001.execution.v1", + "run_id": self.trace.run_id, + "duration_seconds": round(time.perf_counter() - started, 9), + "python": platform.python_version(), + "platform": platform.platform(), + } + return RunResult(self.trace.events, trace_jsonl, summary, execution) + + def _nearest_rank(self, values: list[int], percentile: float) -> int | None: + if not values: + return None + ordered = sorted(values) + rank = max(1, math.ceil(percentile * len(ordered))) + return ordered[rank - 1] + + def _fault_metrics(self) -> tuple[int, int]: + if self.scenario == "slow_neighbor": + start, end = 40, 99 + elif self.scenario == "failed_neighbor": + start, end = 40, 79 + else: + return 0, 0 + healthy = ("A", "C") + healthy_progress_ticks = len( + set().union(*(self.progress_ticks[loop_id] for loop_id in healthy)) + & set(range(start, end + 1)) + ) + affected = 0 + for loop_id in healthy: + stalled_with_demand = 0 + loop_affected = False + for tick in range(start, min(end + 1, self.elapsed_ticks)): + occupancy = self.occupancy_history[loop_id][tick] + if occupancy > 0 and tick not in self.progress_ticks[loop_id]: + stalled_with_demand += 1 + if stalled_with_demand >= 5: + loop_affected = True + else: + stalled_with_demand = 0 + affected += int(loop_affected) + return healthy_progress_ticks, affected + + def _build_summary(self, *, trace_jsonl: str) -> dict[str, Any]: + terminal_counts = Counter(state for state, _, _ in self.terminal_items.values()) + terminal_counts.update(self.rejected_counts) + latencies = [ + terminal_tick + 1 - item_admitted + for work_id, (state, _, terminal_tick) in self.terminal_items.items() + if state == "completed" + for item_admitted in [self._admitted_at_from_trace(work_id)] + if item_admitted is not None + ] + coordination_types = { + "propose", + "credit_grant", + "credit_deny", + "validate", + "receipt", + "wait", + } + coordination_counts = Counter( + event["event_type"] + for event in self.trace.events + if event["event_type"] in coordination_types + ) + healthy_progress_ticks, fault_radius = self._fault_metrics() + promoted_items = { + event["work_id"] + for event in self.trace.events + if event["event_type"] == "transfer_commit" and event["work_id"] is not None + } + provenance_complete = { + event["work_id"] + for event in self.trace.events + if event["event_type"] == "transfer_commit" + and event["work_id"] is not None + and event["provenance_complete"] + } + summary: dict[str, Any] = { + "schema_version": "e001.summary.v1", + "run_id": self.trace.run_id, + "configuration": self.configuration, + "scenario": self.scenario, + "seed": self.seed, + "workload_digest": self.workload_digest, + "elapsed_simulation_ticks": self.elapsed_ticks, + "offered": self.offered_attempts, + "admitted": self.admitted_count, + "active_at_end": len(self.items), + "terminal_counts": dict(sorted(terminal_counts.items())), + "completed": terminal_counts["completed"], + "maximum_occupancy": self.max_occupancy, + "mean_occupancy": { + loop_id: round(self.occupancy_time[loop_id] / self.elapsed_ticks, 6) + for loop_id in LOOP_IDS + }, + "occupancy_time": self.occupancy_time, + "throughput_per_tick": round( + terminal_counts["completed"] / self.elapsed_ticks, + 9, + ), + "latency": { + "p50": self._nearest_rank(latencies, 0.50), + "p95": self._nearest_rank(latencies, 0.95), + "maximum": max(latencies) if latencies else None, + }, + "maximum_wait": self._maximum_wait(), + "healthy_progress_ticks": healthy_progress_ticks, + "fault_radius": fault_radius, + "recovery_time": None, + "coordination_events": dict(sorted(coordination_counts.items())), + "coordination_event_total": sum(coordination_counts.values()), + "maximum_unresolved_obligations": self.max_outstanding_obligations, + "final_unresolved_obligations": self.outstanding_obligations, + "stale_rejection_count": self.stale_rejection_count, + "stale_authorization_count": self.stale_authorization_count, + "duplicate_promotion_count": self.duplicate_promotion_count, + "provenance_completeness_percent": ( + 100.0 + if not promoted_items + else round(100 * len(provenance_complete) / len(promoted_items), 6) + ), + "invariant_violation_count": len(self.invariant_violations), + "invariant_violations": self.invariant_violations, + "trace_digest": raw_digest(trace_jsonl), + "summary_digest": None, + } + canonical_for_digest = dict(summary) + canonical_for_digest["summary_digest"] = None + summary["summary_digest"] = digest(canonical_for_digest) + return summary + + def _admitted_at_from_trace(self, work_id: str) -> int | None: + for event in self.trace.events: + if event["work_id"] == work_id and event["event_type"] == "admit": + return int(event["simulation_tick"]) + return None + + def _maximum_wait(self) -> int: + wait_started: dict[str, int] = {} + maximum = 0 + for event in self.trace.events: + work_id = event["work_id"] + if work_id is None: + continue + if event["event_type"] == "wait": + wait_started.setdefault(work_id, event["simulation_tick"]) + elif event["event_type"] in {"transfer_commit", "complete", "expire"}: + started = wait_started.pop(work_id, None) + if started is not None: + maximum = max(maximum, event["simulation_tick"] - started) + for started in wait_started.values(): + maximum = max(maximum, self.elapsed_ticks - started) + return maximum + + +def _recovery_time(result: RunResult, baseline_rate: float) -> int | None: + scenario = result.summary["scenario"] + if scenario == "slow_neighbor": + disturbance_end = 99 + elif scenario == "failed_neighbor": + disturbance_end = 79 + else: + return None + target = baseline_rate * 0.9 + completion_ticks = [ + event["simulation_tick"] + for event in result.trace + if event["event_type"] == "complete" + ] + horizon = result.summary["elapsed_simulation_ticks"] + for start in range(disturbance_end + 1, max(disturbance_end + 1, horizon - 9)): + completions = sum(start <= tick < start + 10 for tick in completion_ticks) + if completions / 10 >= target: + return start - disturbance_end - 1 + return None + + +def _finalize_summary_digest(summary: dict[str, Any]) -> None: + canonical_for_digest = dict(summary) + canonical_for_digest["summary_digest"] = None + summary["summary_digest"] = digest(canonical_for_digest) + + +def run_simulation( + configuration: str, + scenario: str, + seed: int, + *, + calculate_recovery: bool = True, +) -> RunResult: + result = Simulation(configuration, scenario, seed).run() + if calculate_recovery and scenario in {"slow_neighbor", "failed_neighbor"}: + baseline = run_simulation( + configuration, + "balanced", + seed, + calculate_recovery=False, + ) + result.summary["recovery_time"] = _recovery_time( + result, + baseline.summary["throughput_per_tick"], + ) + _finalize_summary_digest(result.summary) + return result + + +def run_matrix() -> list[RunResult]: + """Execute the frozen 72-run canonical matrix in stable order.""" + + return [ + run_simulation(configuration, scenario, seed) + for configuration in CONFIGURATIONS + for scenario in SCENARIOS + for seed in SEEDS + ] + + +def write_result(result: RunResult, output_root: Path) -> None: + """Write one trace and summary beneath an explicit output root.""" + + raw_dir = output_root / "raw" + summary_dir = output_root / "summary" + execution_dir = output_root / "execution" + raw_dir.mkdir(parents=True, exist_ok=True) + summary_dir.mkdir(parents=True, exist_ok=True) + execution_dir.mkdir(parents=True, exist_ok=True) + stem = result.summary["run_id"] + (raw_dir / f"{stem}.jsonl").write_text(result.trace_jsonl, encoding="utf-8") + (summary_dir / f"{stem}.json").write_text( + json.dumps(result.summary, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + (execution_dir / f"{stem}.json").write_text( + json.dumps(result.execution, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def matrix_manifest(results: list[RunResult]) -> dict[str, Any]: + workloads: dict[str, set[str]] = defaultdict(set) + for result in results: + key = f"{result.summary['scenario']}__seed-{result.summary['seed']}" + workloads[key].add(result.summary["workload_digest"]) + mismatches = {key: sorted(values) for key, values in workloads.items() if len(values) != 1} + return { + "schema_version": "e001.matrix-manifest.v1", + "run_count": len(results), + "expected_run_count": 72, + "workload_equivalence_mismatches": mismatches, + "invalid_run_ids": [ + result.summary["run_id"] + for result in results + if result.summary["invariant_violation_count"] > 0 + or result.summary["active_at_end"] > 0 + ], + "run_summary_digests": { + result.summary["run_id"]: result.summary["summary_digest"] + for result in results + }, + } diff --git a/src/superloop_e001/workloads.py b/src/superloop_e001/workloads.py new file mode 100644 index 0000000..882d854 --- /dev/null +++ b/src/superloop_e001/workloads.py @@ -0,0 +1,125 @@ +"""Frozen workload schedules for E001 v0.1.""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Iterable + +from .model import LOOP_IDS, ROUTES, Offer, digest + + +SCENARIOS = ( + "balanced", + "burst", + "saturated_receiver", + "slow_neighbor", + "failed_neighbor", + "circular_wait", + "stale_feedback", + "malformed_duplicate", +) +CONFIGURATIONS = ("central", "global_barrier", "local_cbf") +SEEDS = (17, 29, 43) + + +def _balanced_offers() -> list[Offer]: + offers: list[Offer] = [] + for index, tick in enumerate(range(0, 120, 3)): + source = LOOP_IDS[index % len(LOOP_IDS)] + offers.append(Offer(tick=tick, work_id=f"regular-{index:03d}", source_loop=source)) + return offers + + +def build_offers(scenario: str) -> list[Offer]: + """Return the immutable offered event stream for one scenario.""" + + if scenario not in SCENARIOS: + raise ValueError(f"unsupported scenario: {scenario}") + if scenario == "circular_wait": + return [] + + offers = _balanced_offers() + if scenario == "burst": + for tick in (30, 60): + for source in LOOP_IDS: + for offset in range(4): + offers.append( + Offer( + tick=tick, + work_id=f"burst-{tick}-{source}-{offset}", + source_loop=source, + ) + ) + elif scenario == "saturated_receiver": + for offset in range(8): + offers.append( + Offer( + tick=30, + work_id=f"saturation-30-A-{offset}", + source_loop="A", + ) + ) + elif scenario == "malformed_duplicate": + offers.extend( + ( + Offer( + tick=30, + work_id="malformed-30", + source_loop="A", + schema_version="unsupported.work.v0", + ), + Offer( + tick=30, + work_id="regular-000", + source_loop="A", + duplicate_replay=True, + ), + ) + ) + + return sorted( + offers, + key=lambda offer: ( + offer.tick, + offer.source_loop, + offer.work_id, + offer.schema_version, + ), + ) + + +def offers_by_tick(offers: Iterable[Offer]) -> dict[int, list[Offer]]: + indexed: dict[int, list[Offer]] = defaultdict(list) + for offer in offers: + indexed[offer.tick].append(offer) + return dict(indexed) + + +def build_circular_preload() -> list[dict[str, object]]: + """Return the canonical full-ring preload used by S6.""" + + return [ + { + "admitted_at": 0, + "deadline": 60, + "processed": True, + "route": list(ROUTES[loop_id]), + "route_index": 0, + "source_loop": loop_id, + "work_id": f"circular-{loop_id}-{index:02d}", + } + for loop_id in LOOP_IDS + for index in range(8) + ] + + +def workload_digest(scenario: str, seed: int) -> str: + """Digest the canonical offered stream and seeded ordering context.""" + + payload = { + "scenario": scenario, + "seed": seed, + "offers": [offer.canonical() for offer in build_offers(scenario)], + "circular_preload": build_circular_preload() if scenario == "circular_wait" else [], + } + return digest(payload) diff --git a/tests/README.md b/tests/README.md index 520f2b2..24dd948 100644 --- a/tests/README.md +++ b/tests/README.md @@ -17,3 +17,9 @@ Tests will be organized around architectural properties rather than implementati Every implementation test should identify the concept invariant it protects. E001 tests must map explicitly to the ten hard invariants in the [E001 specification](../experiments/E001/EXPERIMENT__E001__THREE_RING_BOUNDED_FLOW__v0.1__2026-07-13.md). + +Run the Stage A suite with: + +```bash +PYTHONPATH=src python -m unittest discover -s tests -p "test_*.py" -v +``` diff --git a/tests/test_e001_simulator.py b/tests/test_e001_simulator.py new file mode 100644 index 0000000..49fb41c --- /dev/null +++ b/tests/test_e001_simulator.py @@ -0,0 +1,146 @@ +"""Invariant, determinism, and comparison tests for E001 Stage A.""" + +from __future__ import annotations + +import json +import unittest +from pathlib import Path + +from superloop_e001.model import TRACE_FIELDS +from superloop_e001.simulator import matrix_manifest, run_matrix, run_simulation +from superloop_e001.workloads import CONFIGURATIONS, SCENARIOS, SEEDS, build_offers + + +class E001SimulatorTests(unittest.TestCase): + def test_frozen_matrix_dimensions(self) -> None: + self.assertEqual(3, len(CONFIGURATIONS)) + self.assertEqual(8, len(SCENARIOS)) + self.assertEqual(3, len(SEEDS)) + self.assertEqual(40, len(build_offers("balanced"))) + self.assertEqual(64, len(build_offers("burst"))) + self.assertEqual(48, len(build_offers("saturated_receiver"))) + self.assertEqual(42, len(build_offers("malformed_duplicate"))) + matrix_path = ( + Path(__file__).resolve().parents[1] + / "experiments" + / "E001" + / "config" + / "CANONICAL_MATRIX__E001__v0.1__2026-07-13.json" + ) + matrix = json.loads(matrix_path.read_text(encoding="utf-8")) + self.assertEqual(list(CONFIGURATIONS), matrix["configurations"]) + self.assertEqual(list(SCENARIOS), matrix["scenarios"]) + self.assertEqual(list(SEEDS), matrix["seeds"]) + self.assertEqual(8, matrix["parameters"]["capacity_per_loop"]) + self.assertEqual(3, matrix["parameters"]["credit_lease_ticks"]) + self.assertEqual(60, matrix["parameters"]["work_deadline_ticks"]) + self.assertEqual(240, matrix["parameters"]["primary_horizon"]) + self.assertEqual(360, matrix["parameters"]["maximum_drain_horizon"]) + + def test_trace_is_byte_deterministic(self) -> None: + first = run_simulation("local_cbf", "burst", 17) + second = run_simulation("local_cbf", "burst", 17) + self.assertEqual(first.trace_jsonl, second.trace_jsonl) + self.assertEqual(first.summary["trace_digest"], second.summary["trace_digest"]) + self.assertEqual(first.summary["summary_digest"], second.summary["summary_digest"]) + + def test_trace_records_have_exact_required_keys(self) -> None: + result = run_simulation("local_cbf", "balanced", 17) + self.assertTrue(result.trace) + for event in result.trace: + self.assertEqual(set(TRACE_FIELDS), set(event)) + json.dumps(event, sort_keys=True) + + def test_full_matrix_preserves_hard_invariants_and_equivalent_work(self) -> None: + results = run_matrix() + manifest = matrix_manifest(results) + self.assertEqual(72, len(results)) + self.assertEqual(72, manifest["run_count"]) + self.assertEqual({}, manifest["workload_equivalence_mismatches"]) + self.assertEqual([], manifest["invalid_run_ids"]) + for result in results: + summary = result.summary + self.assertEqual(0, summary["invariant_violation_count"], summary["run_id"]) + self.assertEqual(0, summary["active_at_end"], summary["run_id"]) + self.assertLessEqual(max(summary["maximum_occupancy"].values()), 8) + self.assertEqual(0, summary["final_unresolved_obligations"]) + self.assertEqual(0, summary["stale_authorization_count"]) + self.assertEqual(0, summary["duplicate_promotion_count"]) + self.assertEqual(100.0, summary["provenance_completeness_percent"]) + + def test_stale_feedback_is_rejected(self) -> None: + result = run_simulation("local_cbf", "stale_feedback", 29) + self.assertEqual(1, result.summary["stale_rejection_count"]) + self.assertEqual(0, result.summary["stale_authorization_count"]) + stale_events = [ + event for event in result.trace if event["validation_state"] == "rejected_stale" + ] + self.assertEqual(1, len(stale_events)) + + def test_malformed_and_duplicate_offers_do_not_promote(self) -> None: + result = run_simulation("local_cbf", "malformed_duplicate", 43) + terminals = result.summary["terminal_counts"] + self.assertEqual(1, terminals["rejected_invalid"]) + self.assertEqual(1, terminals["duplicate_ignored"]) + self.assertEqual(0, result.summary["duplicate_promotion_count"]) + + def test_circular_wait_terminates_accountably(self) -> None: + central = run_simulation("central", "circular_wait", 17) + barrier = run_simulation("global_barrier", "circular_wait", 17) + local = run_simulation("local_cbf", "circular_wait", 17) + for result in (central, barrier, local): + self.assertEqual(24, result.summary["admitted"]) + self.assertEqual(0, result.summary["active_at_end"]) + self.assertEqual(24, central.summary["completed"]) + self.assertEqual(24, barrier.summary["completed"]) + self.assertEqual(24, local.summary["terminal_counts"]["expired_deadline"]) + self.assertEqual(0, local.summary["completed"]) + + def test_offer_and_admission_ledgers_balance(self) -> None: + for result in run_matrix(): + summary = result.summary + rejected = sum( + summary["terminal_counts"].get(state, 0) + for state in ("rejected_invalid", "rejected_capacity", "duplicate_ignored") + ) + self.assertEqual(summary["offered"], summary["admitted"] + rejected) + + def test_execution_metadata_is_separate_from_canonical_summary(self) -> None: + result = run_simulation("local_cbf", "balanced", 17) + self.assertNotIn("execution", result.summary) + self.assertEqual(result.summary["run_id"], result.execution["run_id"]) + + def test_every_transfer_commit_has_one_receipt(self) -> None: + result = run_simulation("local_cbf", "burst", 17) + commits = [event for event in result.trace if event["event_type"] == "transfer_commit"] + receipts = [event for event in result.trace if event["event_type"] == "receipt"] + self.assertEqual(len(commits), len(receipts)) + receipt_parents = [event["parent_event_ids"] for event in receipts] + self.assertCountEqual([[event["event_id"]] for event in commits], receipt_parents) + + def test_local_interlocks_preserve_fault_locality_signal(self) -> None: + global_barrier = run_simulation("global_barrier", "failed_neighbor", 17) + local_cbf = run_simulation("local_cbf", "failed_neighbor", 17) + self.assertGreaterEqual( + local_cbf.summary["healthy_progress_ticks"], + global_barrier.summary["healthy_progress_ticks"], + ) + self.assertLessEqual( + local_cbf.summary["fault_radius"], + global_barrier.summary["fault_radius"], + ) + + def test_central_trace_accounts_for_work_blocked_by_failed_neighbor(self) -> None: + result = run_simulation("central", "failed_neighbor", 17) + blocked_transfer_waits = [ + event + for event in result.trace + if event["event_type"] == "wait" + and event["interlock_id"] == "AB" + and 40 <= event["simulation_tick"] <= 79 + ] + self.assertTrue(blocked_transfer_waits) + + +if __name__ == "__main__": + unittest.main()