R2: run manifest / provenance writer for reproducible, auditable runs

write_manifest() freezes at circuit-experiment start everything needed to
reproduce and audit a SOR run and writes an immutable
output/sor-runs/<ts>/manifest.json: the R1 --sor-seed, topology/selector/churn
schedule id, one persona fingerprint per participating node, worktree git SHA,
isolated engine kind + image digest, pip/cargo dependency freeze, and
start/stop timestamps. The events.jsonl SHA-256 field is reserved for R3.

Provenance only: no forwarder, no engine spawned. Defense-in-depth to keep
containment load-bearing, the schema refuses to record a non-isolated
("local") engine at run or node level; the binding assertion still lands with
the R4 forwarder.

- cmd_chat/sor/provenance.py: Node/RunManifest, node_fingerprint (bit-for-bit
  mirror of persona.rs::fingerprint_of), git/deps capture, explicit
  dependency-free schema validator.
- hh/src/persona.rs: fingerprint_of known-vector parity test.
- tests: R2 acceptance predicate (manifest present + schema-valid, non-empty
  seed, >=1 fingerprint/node), immutability, containment rejection of local
  engines, and cross-language fingerprint parity.

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

cmd_chat/sor/provenance.py

@@ -0,0 +1,251 @@
+"""R2 — Run manifest / provenance writer for the SOR measurement instrument.
+
+`write_manifest()` freezes, at circuit-experiment start, everything needed to
+reproduce and audit a run: the R1 `--sor-seed`, the topology / selector / churn
+schedule id, one persona fingerprint per participating node (mirrors
+`hh/src/persona.rs::fingerprint_of`), the worktree git SHA, the isolated engine
+kind + image digest, a pip/cargo dependency freeze, and start/stop timestamps.
+It writes an immutable `output/sor-runs/<ts>/manifest.json`.
+
+Provenance only — this module forwards no traffic and spawns no engine. As
+defense-in-depth it refuses to record a non-isolated (`local`) engine, but the
+load-bearing containment assertion lives with the R4 forwarder. The event-log
+SHA-256 field is reserved here and filled by R3 when the log is closed.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import subprocess
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+SCHEMA_VERSION = "sor-manifest/1"
+
+# Engines that satisfy the containment law (isolated-engine-only). "local" is
+# never a valid engine for a SOR run — recording it is refused.
+ISOLATED_ENGINES = ("docker", "multipass")
+
+
+def node_fingerprint(pub_b64: str) -> str:
+    """Short fingerprint of a base64 Ed25519 pubkey: sha256(raw)[:4] hex.
+
+    Bit-for-bit mirror of hh/src/persona.rs::fingerprint_of. Raises ValueError
+    on undecodable input (a node with no valid persona cannot be recorded)."""
+    try:
+        raw = base64.b64decode(pub_b64, validate=True)
+    except Exception as exc:  # noqa: BLE001 - normalize to ValueError
+        raise ValueError(f"invalid persona pubkey: {exc}") from exc
+    return hashlib.sha256(raw).digest()[:4].hex()
+
+
+@dataclass
+class Node:
+    """A participating node in a SOR circuit run."""
+
+    role: str  # "host" | "hop" | "bridge"
+    persona_pub_b64: str
+    engine: str = "docker"
+    image_digest: Optional[str] = None
+
+    def to_entry(self) -> Dict[str, Any]:
+        if self.engine not in ISOLATED_ENGINES:
+            raise ValueError(
+                f"containment: node engine {self.engine!r} is not isolated "
+                f"(allowed: {ISOLATED_ENGINES})"
+            )
+        return {
+            "role": self.role,
+            "persona_pub_b64": self.persona_pub_b64,
+            "fingerprint": node_fingerprint(self.persona_pub_b64),
+            "engine": self.engine,
+            "image_digest": self.image_digest,
+        }
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def capture_git(worktree_root: Path) -> Dict[str, Any]:
+    """Best-effort worktree git provenance. Never raises; records nulls if the
+    worktree is not a git repo (a manifest must still be schema-valid)."""
+
+    def _run(args: List[str]) -> Optional[str]:
+        try:
+            out = subprocess.run(
+                ["git", *args],
+                cwd=str(worktree_root),
+                capture_output=True,
+                text=True,
+                timeout=10,
+            )
+            return out.stdout.strip() if out.returncode == 0 else None
+        except Exception:  # noqa: BLE001
+            return None
+
+    sha = _run(["rev-parse", "HEAD"])
+    branch = _run(["rev-parse", "--abbrev-ref", "HEAD"])
+    status = _run(["status", "--porcelain"])
+    return {
+        "worktree_sha": sha,
+        "branch": branch,
+        "dirty": bool(status) if status is not None else None,
+    }
+
+
+def capture_deps(worktree_root: Path) -> Dict[str, Any]:
+    """Freeze dependency provenance: sha256 of the pip freeze list and of the
+    Cargo.lock (if present). Records the freeze list itself for auditability."""
+    try:
+        freeze = subprocess.run(
+            [sys.executable, "-m", "pip", "freeze"],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        ).stdout
+    except Exception:  # noqa: BLE001
+        freeze = ""
+    pip_lines = sorted(l for l in freeze.splitlines() if l.strip())
+    pip_blob = "\n".join(pip_lines).encode()
+
+    cargo_lock = worktree_root / "hh" / "Cargo.lock"
+    cargo_sha: Optional[str] = None
+    if cargo_lock.is_file():
+        cargo_sha = hashlib.sha256(cargo_lock.read_bytes()).hexdigest()
+
+    return {
+        "pip_freeze_sha256": hashlib.sha256(pip_blob).hexdigest(),
+        "pip_freeze": pip_lines,
+        "cargo_lock_sha256": cargo_sha,
+    }
+
+
+@dataclass
+class RunManifest:
+    """In-memory manifest; `write()` renders it immutably to disk."""
+
+    run_id: str
+    sor_seed: int
+    topology: str
+    selector: str
+    churn_schedule_id: str
+    nodes: List[Node]
+    engine_kind: str = "docker"
+    engine_image_digest: Optional[str] = None
+    worktree_root: Path = field(default_factory=lambda: Path.cwd())
+    start_utc: Optional[str] = None
+    stop_utc: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        if self.engine_kind not in ISOLATED_ENGINES:
+            raise ValueError(
+                f"containment: run engine {self.engine_kind!r} is not isolated "
+                f"(allowed: {ISOLATED_ENGINES})"
+            )
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "run_id": self.run_id,
+            "created_utc": _utc_now_iso(),
+            "sor_seed": self.sor_seed,
+            "topology": self.topology,
+            "selector": self.selector,
+            "churn_schedule_id": self.churn_schedule_id,
+            "nodes": [n.to_entry() for n in self.nodes],
+            "engine": {
+                "kind": self.engine_kind,
+                "image_digest": self.engine_image_digest,
+            },
+            "git": capture_git(self.worktree_root),
+            "deps": capture_deps(self.worktree_root),
+            "timestamps": {
+                "start_utc": self.start_utc or _utc_now_iso(),
+                "stop_utc": self.stop_utc,
+            },
+            # Reserved for R3: sha256 of the closed events.jsonl.
+            "events": {"path": "events.jsonl", "sha256": None},
+        }
+
+
+def write_manifest(run_dir: Path, manifest: RunManifest) -> Dict[str, Any]:
+    """Render `manifest` to `<run_dir>/manifest.json` and return the dict.
+
+    Refuses to overwrite an existing manifest (artifacts are immutable). The
+    written document is validated before return; an invalid manifest raises and
+    leaves nothing partial on disk."""
+    run_dir = Path(run_dir)
+    run_dir.mkdir(parents=True, exist_ok=True)
+    path = run_dir / "manifest.json"
+    if path.exists():
+        raise FileExistsError(f"manifest already exists (immutable): {path}")
+
+    doc = manifest.to_dict()
+    validate_manifest(doc)
+    path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n")
+    return doc
+
+
+# --- schema -----------------------------------------------------------------
+
+def validate_manifest(doc: Dict[str, Any]) -> None:
+    """Explicit, dependency-free schema check. Raises ValueError on any
+    violation. This IS the R2 acceptance predicate (schema-valid + non-empty
+    seed + >=1 fingerprint per participating node)."""
+
+    def req(cond: bool, msg: str) -> None:
+        if not cond:
+            raise ValueError(f"manifest schema: {msg}")
+
+    req(isinstance(doc, dict), "root must be an object")
+    req(doc.get("schema_version") == SCHEMA_VERSION, "wrong/missing schema_version")
+    for key in (
+        "run_id",
+        "created_utc",
+        "sor_seed",
+        "topology",
+        "selector",
+        "churn_schedule_id",
+        "nodes",
+        "engine",
+        "git",
+        "deps",
+        "timestamps",
+        "events",
+    ):
+        req(key in doc, f"missing top-level key {key!r}")
+
+    req(isinstance(doc["run_id"], str) and doc["run_id"], "run_id must be non-empty str")
+    req(isinstance(doc["sor_seed"], int), "sor_seed must be an int")
+    req(0 <= doc["sor_seed"] <= (1 << 64) - 1, "sor_seed out of u64 range")
+
+    for f in ("topology", "selector", "churn_schedule_id"):
+        req(isinstance(doc[f], str) and doc[f], f"{f} must be non-empty str")
+
+    nodes = doc["nodes"]
+    req(isinstance(nodes, list) and len(nodes) >= 1, "nodes must be a non-empty list")
+    for i, n in enumerate(nodes):
+        req(isinstance(n, dict), f"node[{i}] must be an object")
+        req(isinstance(n.get("role"), str) and n["role"], f"node[{i}].role required")
+        fp = n.get("fingerprint")
+        req(
+            isinstance(fp, str) and len(fp) == 8 and all(c in "0123456789abcdef" for c in fp),
+            f"node[{i}] must carry a persona fingerprint (8 hex chars)",
+        )
+        eng = n.get("engine")
+        req(eng in ISOLATED_ENGINES, f"node[{i}].engine {eng!r} not isolated (containment)")
+
+    engine = doc["engine"]
+    req(isinstance(engine, dict), "engine must be an object")
+    req(engine.get("kind") in ISOLATED_ENGINES, "engine.kind not isolated (containment)")
+
+    ts = doc["timestamps"]
+    req(isinstance(ts, dict) and isinstance(ts.get("start_utc"), str) and ts["start_utc"],
+        "timestamps.start_utc required")
+
+    ev = doc["events"]
+    req(isinstance(ev, dict) and "sha256" in ev and "path" in ev, "events block malformed")

hh/src/persona.rs

@@ -155,4 +155,14 @@ mod tests {
         assert_eq!(fp.len(), 8, "4 bytes → 8 hex chars");
         assert_eq!(Some(fp), fingerprint_of(&p.pub_b64()));
     }
+
+    // Cross-language parity anchor for the R2 manifest writer: the same known
+    // pubkey (raw bytes 0..32) maps to this fingerprint in
+    // cmd_chat/sor/provenance.py (tests/test_sor_provenance.py). If either side
+    // changes the fingerprint formula, one of the two tests fails.
+    #[test]
+    fn fingerprint_of_known_vector() {
+        let pub_b64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
+        assert_eq!(fingerprint_of(pub_b64).as_deref(), Some("630dcd29"));
+    }
 }

tests/test_sor_provenance.py

@@ -0,0 +1,111 @@
+"""R2 — provenance manifest acceptance checks.
+
+Acceptance predicate (roadmap R2): a manifest is present + schema-valid after a
+run, carries a non-empty seed, and >=1 persona fingerprint per participating
+node. The fingerprint parity constant below is locked identically in
+hh/src/persona.rs::tests::fingerprint_of_known_vector.
+"""
+
+import json
+
+import pytest
+
+from cmd_chat.sor.provenance import (
+    Node,
+    RunManifest,
+    node_fingerprint,
+    validate_manifest,
+    write_manifest,
+)
+
+# Known 32-byte pubkey (bytes 0..32) -> base64 -> fingerprint. Same vector as
+# the Rust persona test.
+KNOWN_PUB_B64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+KNOWN_FINGERPRINT = "630dcd29"
+
+
+def _fixture_manifest(root):
+    return RunManifest(
+        run_id="20260719T000000Z",
+        sor_seed=123456789,
+        topology="1-house",
+        selector="static",
+        churn_schedule_id="fixture-churn-0",
+        nodes=[
+            Node(role="host", persona_pub_b64=KNOWN_PUB_B64, engine="docker"),
+            Node(role="hop", persona_pub_b64="QUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVphYmNk", engine="docker"),
+        ],
+        engine_kind="docker",
+        engine_image_digest="sha256:deadbeef",
+        worktree_root=root,
+    )
+
+
+def test_fingerprint_parity_with_rust():
+    assert node_fingerprint(KNOWN_PUB_B64) == KNOWN_FINGERPRINT
+
+
+def test_manifest_written_and_schema_valid(tmp_path):
+    run_dir = tmp_path / "sor-runs" / "20260719T000000Z"
+    doc = write_manifest(run_dir, _fixture_manifest(tmp_path))
+
+    # present on disk
+    path = run_dir / "manifest.json"
+    assert path.is_file()
+    on_disk = json.loads(path.read_text())
+    assert on_disk == doc
+
+    # schema-valid, non-empty seed, >=1 fingerprint per node (the R2 predicate)
+    validate_manifest(on_disk)
+    assert on_disk["sor_seed"] == 123456789
+    assert len(on_disk["nodes"]) >= 1
+    for n in on_disk["nodes"]:
+        assert n["fingerprint"]
+    assert on_disk["nodes"][0]["fingerprint"] == KNOWN_FINGERPRINT
+
+
+def test_manifest_is_immutable(tmp_path):
+    run_dir = tmp_path / "run"
+    write_manifest(run_dir, _fixture_manifest(tmp_path))
+    with pytest.raises(FileExistsError):
+        write_manifest(run_dir, _fixture_manifest(tmp_path))
+
+
+def test_containment_rejects_local_engine(tmp_path):
+    m = _fixture_manifest(tmp_path)
+    m.engine_kind = "local"
+    with pytest.raises(ValueError):
+        write_manifest(tmp_path / "run2", m)
+
+
+def test_containment_rejects_local_node_engine(tmp_path):
+    m = _fixture_manifest(tmp_path)
+    m.nodes[1].engine = "local"
+    with pytest.raises(ValueError):
+        write_manifest(tmp_path / "run3", m)
+
+
+def test_invalid_pubkey_rejected():
+    with pytest.raises(ValueError):
+        node_fingerprint("!!! not base64 !!!")
+
+
+def test_validator_catches_empty_nodes():
+    with pytest.raises(ValueError):
+        validate_manifest(
+            {
+                "schema_version": "sor-manifest/1",
+                "run_id": "x",
+                "created_utc": "t",
+                "sor_seed": 1,
+                "topology": "1-house",
+                "selector": "static",
+                "churn_schedule_id": "c",
+                "nodes": [],
+                "engine": {"kind": "docker", "image_digest": None},
+                "git": {},
+                "deps": {},
+                "timestamps": {"start_utc": "t", "stop_utc": None},
+                "events": {"path": "events.jsonl", "sha256": None},
+            }
+        )