diff --git a/cmd_chat/sor/events.py b/cmd_chat/sor/events.py new file mode 100644 index 0000000..09c7021 --- /dev/null +++ b/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//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) diff --git a/cmd_chat/sor/provenance.py b/cmd_chat/sor/provenance.py index afd4666..431dcbd 100644 --- a/cmd_chat/sor/provenance.py +++ b/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 `/manifest.json` and return the dict. diff --git a/tests/test_sor_events.py b/tests/test_sor_events.py new file mode 100644 index 0000000..3706e3c --- /dev/null +++ b/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")