R3: immutable append-only event log sealed into the run manifest

Adds the tamper-evident event stream that makes every DV auditable. Each
measurable moment is appended as one JSON object per line to an append-only
output/sor-runs/<ts>/events.jsonl over the closed vocabulary {consent_request,
consent_accept, consent_reject, circuit_build, hop_add, bridge_forward,
churn_kill, churn_spawn, rebuild_start, rebuild_done}; on close the file is
SHA-256'd and the digest is sealed exactly once into manifest.json (the single
sanctioned None->hash completion of the manifest).

Records carry only metadata (fingerprints, byte counts, latencies, decisions),
never message plaintext, so the zero-knowledge relay property is preserved. The
deterministic replay_fixture_circuit emits a seed-driven event stream with no
engine, socket, or traffic (containment-safe) and is the basis of
instrument-validation gate item 6.

- cmd_chat/sor/events.py: EventLog (append-only, per-record schema),
  replay_fixture_circuit, replay_and_seal.
- cmd_chat/sor/provenance.py: seal_manifest (seal-once events SHA + stop ts).
- tests: replay -> schema-valid JSONL whose SHA matches the manifest;
  append-only (original prefix byte-identical after further appends);
  seed-determinism modulo timestamps; seal-once immutability.

Live emit wiring into the _sor handlers lands with R4/R5.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd_chat/sor/events.py

@@ -0,0 +1,211 @@
+"""R3 — Immutable structured event log for the SOR measurement instrument.
+
+Every measurable moment of a run is appended as one JSON object per line to an
+append-only ``output/sor-runs/<ts>/events.jsonl``. On close the file is SHA-256'd
+and the digest is sealed into ``manifest.json`` (R2), so the event stream is
+tamper-evident and reproducible: a replayed fixture circuit yields a
+schema-valid log whose hash matches the manifest (instrument-validation gate
+item 6).
+
+Records carry only *metadata* — fingerprints, byte counts, latencies, decisions
+— never message plaintext, keeping the zero-knowledge relay property intact. The
+live emit points are wired by the R4 forwarder and R5 consent handlers; this
+module is the logging primitive plus a deterministic fixture replay used by the
+acceptance check. Replay spawns no engine and moves no traffic — it is pure
+seeded event emission (containment-safe).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from cmd_chat.sor import config as sor_config
+from cmd_chat.sor import provenance
+
+# The closed vocabulary of loggable events (roadmap R3). Emitting anything else
+# is a programming error and is refused.
+EVENT_TYPES = frozenset(
+    {
+        "consent_request",
+        "consent_accept",
+        "consent_reject",
+        "circuit_build",
+        "hop_add",
+        "bridge_forward",
+        "churn_kill",
+        "churn_spawn",
+        "rebuild_start",
+        "rebuild_done",
+    }
+)
+
+# Every record carries exactly these keys (null where not applicable), so the
+# JSONL is uniform and machine-checkable.
+RECORD_KEYS = (
+    "ts",
+    "run_id",
+    "event",
+    "node_fp",
+    "circuit_id",
+    "hop_index",
+    "bytes",
+    "latency_ms",
+    "decision",
+    "seed",
+)
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def validate_record(rec: Dict[str, Any]) -> None:
+    """Raise ValueError unless `rec` is a schema-valid event record."""
+
+    def req(cond: bool, msg: str) -> None:
+        if not cond:
+            raise ValueError(f"event schema: {msg}")
+
+    req(isinstance(rec, dict), "record must be an object")
+    req(set(rec.keys()) == set(RECORD_KEYS), f"record keys must be exactly {RECORD_KEYS}")
+    req(rec["event"] in EVENT_TYPES, f"unknown event {rec['event']!r}")
+    req(isinstance(rec["ts"], str) and rec["ts"], "ts must be non-empty str")
+    req(isinstance(rec["run_id"], str) and rec["run_id"], "run_id must be non-empty str")
+    req(isinstance(rec["seed"], int) and 0 <= rec["seed"] <= (1 << 64) - 1, "seed u64")
+    # Optional-but-typed fields.
+    req(rec["node_fp"] is None or isinstance(rec["node_fp"], str), "node_fp str|null")
+    req(rec["circuit_id"] is None or isinstance(rec["circuit_id"], str), "circuit_id str|null")
+    req(rec["hop_index"] is None or isinstance(rec["hop_index"], int), "hop_index int|null")
+    req(rec["bytes"] is None or (isinstance(rec["bytes"], int) and rec["bytes"] >= 0), "bytes>=0|null")
+    req(
+        rec["latency_ms"] is None or isinstance(rec["latency_ms"], (int, float)),
+        "latency_ms number|null",
+    )
+    req(rec["decision"] is None or isinstance(rec["decision"], str), "decision str|null")
+
+
+class EventLog:
+    """Append-only JSONL writer. Never edits a byte already written; the only
+    file operation is append. `close()` returns the SHA-256 of the whole file."""
+
+    def __init__(self, run_dir: Path, run_id: str, seed: int) -> None:
+        self.run_dir = Path(run_dir)
+        self.run_dir.mkdir(parents=True, exist_ok=True)
+        self.path = self.run_dir / "events.jsonl"
+        self.run_id = run_id
+        self.seed = seed
+        self._closed = False
+        # Append mode: existing content is never truncated or rewritten.
+        self._fh = self.path.open("a", encoding="utf-8")
+
+    def emit(
+        self,
+        event: str,
+        *,
+        node_fp: Optional[str] = None,
+        circuit_id: Optional[str] = None,
+        hop_index: Optional[int] = None,
+        bytes_: Optional[int] = None,
+        latency_ms: Optional[float] = None,
+        decision: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        if self._closed:
+            raise RuntimeError("event log is closed")
+        rec = {
+            "ts": _utc_now_iso(),
+            "run_id": self.run_id,
+            "event": event,
+            "node_fp": node_fp,
+            "circuit_id": circuit_id,
+            "hop_index": hop_index,
+            "bytes": bytes_,
+            "latency_ms": latency_ms,
+            "decision": decision,
+            "seed": self.seed,
+        }
+        validate_record(rec)
+        # Deterministic key order so the line bytes are stable.
+        self._fh.write(json.dumps(rec, sort_keys=True, separators=(",", ":")) + "\n")
+        self._fh.flush()
+        return rec
+
+    def close(self) -> str:
+        """Close the file and return its SHA-256 hex digest (over exact bytes)."""
+        if not self._closed:
+            self._fh.close()
+            self._closed = True
+        return hashlib.sha256(self.path.read_bytes()).hexdigest()
+
+    def __enter__(self) -> "EventLog":
+        return self
+
+    def __exit__(self, *exc: Any) -> None:
+        if not self._closed:
+            self._fh.close()
+            self._closed = True
+
+
+def replay_fixture_circuit(
+    run_dir: Path,
+    run_id: str,
+    seed: int,
+    pool: int = 5,
+    hops: int = 3,
+    rebuilds: int = 1,
+) -> str:
+    """Emit a deterministic fixture circuit's event stream from `seed` alone and
+    return the SHA-256 of the closed log. Same seed -> identical circuit-build
+    sequence (R1) -> identical event bodies (modulo wall-clock `ts`).
+
+    Pure bookkeeping: no engine, no socket, no traffic. Used by the R3 acceptance
+    check and instrument-validation gate item 6 (provenance integrity)."""
+    circuits = sor_config.bringup(seed, pool, hops, rebuilds)
+    log = EventLog(run_dir, run_id, seed)
+    with log:
+        for c_idx, hops_seq in enumerate(circuits):
+            circuit_id = f"c{c_idx}"
+            if c_idx > 0:
+                log.emit("rebuild_start", circuit_id=circuit_id)
+            # Consent handshake per hop, then build.
+            for h_idx, node in enumerate(hops_seq):
+                node_fp = f"{node:08x}"
+                log.emit("consent_request", node_fp=node_fp, circuit_id=circuit_id, hop_index=h_idx)
+                log.emit(
+                    "consent_accept",
+                    node_fp=node_fp,
+                    circuit_id=circuit_id,
+                    hop_index=h_idx,
+                    decision="accept",
+                )
+            log.emit("circuit_build", circuit_id=circuit_id, hop_index=len(hops_seq))
+            for h_idx, node in enumerate(hops_seq):
+                log.emit(
+                    "hop_add",
+                    node_fp=f"{node:08x}",
+                    circuit_id=circuit_id,
+                    hop_index=h_idx,
+                    bytes_=1024,
+                    latency_ms=1.0 + h_idx,
+                )
+            if c_idx > 0:
+                log.emit("rebuild_done", circuit_id=circuit_id)
+    return log.close()
+
+
+def replay_and_seal(
+    run_dir: Path,
+    manifest: provenance.RunManifest,
+    pool: int = 5,
+    hops: int = 3,
+    rebuilds: int = 1,
+) -> Dict[str, Any]:
+    """End-to-end fixture: write the R2 manifest, replay the fixture event
+    stream, seal its SHA-256 into the manifest, and return the sealed manifest.
+    This is exactly the gate item 6 flow (schema-valid events + hash match)."""
+    provenance.write_manifest(run_dir, manifest)
+    sha = replay_fixture_circuit(run_dir, manifest.run_id, manifest.sor_seed, pool, hops, rebuilds)
+    return provenance.seal_manifest(run_dir, sha)

cmd_chat/sor/provenance.py

@@ -172,6 +172,27 @@ class RunManifest:
         }
 
 
+def seal_manifest(
+    run_dir: Path, events_sha256: str, stop_utc: Optional[str] = None
+) -> Dict[str, Any]:
+    """R3 close step: seal the closed events.jsonl SHA-256 (and stop timestamp)
+    into an existing manifest.json exactly once.
+
+    This is the one sanctioned mutation of a manifest — a single None -> hash
+    transition completing the document. Re-sealing an already-sealed manifest is
+    refused, preserving artifact immutability."""
+    run_dir = Path(run_dir)
+    path = run_dir / "manifest.json"
+    doc = json.loads(path.read_text())
+    if doc.get("events", {}).get("sha256") is not None:
+        raise ValueError("events already sealed (manifest is immutable)")
+    doc["events"]["sha256"] = events_sha256
+    doc["timestamps"]["stop_utc"] = stop_utc or _utc_now_iso()
+    validate_manifest(doc)
+    path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n")
+    return doc
+
+
 def write_manifest(run_dir: Path, manifest: RunManifest) -> Dict[str, Any]:
     """Render `manifest` to `<run_dir>/manifest.json` and return the dict.
 

tests/test_sor_events.py

@@ -0,0 +1,123 @@
+"""R3 — immutable event log acceptance checks.
+
+Acceptance predicate (roadmap R3): replaying a fixture circuit produces a
+schema-valid JSONL whose SHA-256 matches the manifest; the file is append-only
+(no in-place edits). This is also instrument-validation gate item 6.
+"""
+
+import hashlib
+import json
+
+import pytest
+
+from cmd_chat.sor.events import (
+    EVENT_TYPES,
+    EventLog,
+    replay_and_seal,
+    replay_fixture_circuit,
+    validate_record,
+)
+from cmd_chat.sor.provenance import Node, RunManifest
+
+KNOWN_PUB_B64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+
+def _manifest(root, run_id="20260719T010203Z", seed=123456789):
+    return RunManifest(
+        run_id=run_id,
+        sor_seed=seed,
+        topology="1-house",
+        selector="static",
+        churn_schedule_id="fixture-churn-0",
+        nodes=[Node(role="host", persona_pub_b64=KNOWN_PUB_B64, engine="docker")],
+        engine_kind="docker",
+        worktree_root=root,
+    )
+
+
+def _read_lines(path):
+    return [json.loads(l) for l in path.read_text().splitlines() if l.strip()]
+
+
+def test_replay_schema_valid_and_hash_matches_manifest(tmp_path):
+    run_dir = tmp_path / "run"
+    sealed = replay_and_seal(run_dir, _manifest(tmp_path), pool=5, hops=3, rebuilds=2)
+
+    events_path = run_dir / "events.jsonl"
+    assert events_path.is_file()
+
+    # Every line schema-valid, uses a known event type, and carries the seed.
+    records = _read_lines(events_path)
+    assert len(records) > 0
+    for rec in records:
+        validate_record(rec)
+        assert rec["event"] in EVENT_TYPES
+        assert rec["seed"] == 123456789
+
+    # SHA-256 of the closed file matches what was sealed into the manifest.
+    disk_sha = hashlib.sha256(events_path.read_bytes()).hexdigest()
+    assert sealed["events"]["sha256"] == disk_sha
+    manifest_doc = json.loads((run_dir / "manifest.json").read_text())
+    assert manifest_doc["events"]["sha256"] == disk_sha
+    assert manifest_doc["timestamps"]["stop_utc"] is not None
+
+
+def test_replay_is_deterministic_modulo_timestamps(tmp_path):
+    # Same seed -> identical circuit-build sequence -> identical event bodies
+    # (drop the wall-clock ts). Two independent runs must match field-for-field.
+    a_dir = tmp_path / "a"
+    b_dir = tmp_path / "b"
+    replay_fixture_circuit(a_dir, "runA", 42, pool=6, hops=3, rebuilds=3)
+    replay_fixture_circuit(b_dir, "runB", 42, pool=6, hops=3, rebuilds=3)
+
+    def _strip(path, run_id):
+        out = []
+        for r in _read_lines(path):
+            r = dict(r)
+            r.pop("ts")
+            r["run_id"] = "X"  # normalize the only other per-run field
+            out.append(r)
+        return out
+
+    assert _strip(a_dir / "events.jsonl", "runA") == _strip(b_dir / "events.jsonl", "runB")
+
+
+def test_append_only_no_in_place_edits(tmp_path):
+    run_dir = tmp_path / "run"
+    log = EventLog(run_dir, "runX", 7)
+    log.emit("circuit_build", circuit_id="c0", hop_index=0)
+    first_bytes = (run_dir / "events.jsonl").read_bytes()
+    first_len = len(first_bytes)
+
+    log.emit("hop_add", node_fp="00000001", circuit_id="c0", hop_index=0, bytes_=10, latency_ms=1.0)
+    log.close()
+
+    after = (run_dir / "events.jsonl").read_bytes()
+    # The file only grew; the original prefix is byte-identical (append-only).
+    assert len(after) > first_len
+    assert after[:first_len] == first_bytes
+
+
+def test_seal_is_once_only(tmp_path):
+    run_dir = tmp_path / "run"
+    replay_and_seal(run_dir, _manifest(tmp_path), rebuilds=1)
+    # A second seal attempt on the same manifest is refused (immutability).
+    from cmd_chat.sor.provenance import seal_manifest
+
+    with pytest.raises(ValueError):
+        seal_manifest(run_dir, "0" * 64)
+
+
+def test_unknown_event_rejected(tmp_path):
+    log = EventLog(tmp_path / "run", "runX", 1)
+    with pytest.raises(ValueError):
+        log.emit("not_a_real_event")
+    log.close()
+
+
+def test_closed_log_refuses_emit(tmp_path):
+    log = EventLog(tmp_path / "run", "runX", 1)
+    log.emit("circuit_build", circuit_id="c0")
+    log.close()
+    with pytest.raises(RuntimeError):
+        log.emit("circuit_build", circuit_id="c1")