Implement the E001 Stage A observation chamber

.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

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"
+}
+

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.

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"]
+

src/superloop_e001/__main__.py

@@ -0,0 +1,5 @@
+from .cli import main
+
+
+raise SystemExit(main())
+

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())
+

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"