Trajectory step 5: HMAC attestation of the chain head at run end

CHANGELOG.md

@@ -6,6 +6,7 @@ Versions follow [SemVer](https://semver.org).
 
 ## [Unreleased]
 
+- Run endings append one `attest` trajectory event: an HMAC-SHA256 over the run id and chain head with a deployment key created owner-only beside the operator settings file, and `outerloop trajectory --verify` reports `attested` or `attestation mismatch`. Older kernels ignore the event; no ledger or PR-body change.
 - The PR body gains a report-only Number check that lists values in the author's report matching no `measure` event or Experiments result when rounded to the report's decimals; nothing is blocked and no trajectory event is added.
 - Session end records each tool call from the retained Claude or Codex log as a `tool.call` trajectory event, once per call even when a session resumes. `outerloop trajectory` and a new Process section in the PR body show counts by tool. Older kernels ignore these events.
 - `outerloop status` shows a run's priced session cost as a lower bound when some legs are unpriced (a wake killed mid-turn leaves no usage event), for example `session-cost>=$4.2000 (1 unpriced)`, instead of hiding the priced part behind `unknown`; the status row and JSON gain the priced-leg count (#516).

docs/design/trajectory.md

@@ -57,4 +57,11 @@ report-only number check: numbers on report lines that mention the metric,
 baseline or candidate are compared with every `measure` event's values and the
 Experiments result cells, after rounding to the report's decimals, and the
 unmatched ones are listed under "## Number check". Nothing is blocked and no
-event is written for it. Attestation is a later step.
+event is written for it. At the end of a run the kernel appends one `attest`
+event carrying an HMAC-SHA256 over the run id and the chain head. Its 32-byte
+key is created owner-only beside the operator settings file
+(`trajectory_attest.key`). `--verify` reports `attested` when the last attest
+event matches that key and `attestation mismatch` otherwise, which shows the
+deployment's own chain was not altered after ending but is not a public proof:
+anyone holding the key can recompute it, and a missing event or key proves
+nothing.

src/outerloop/attempt.py

@@ -152,7 +152,7 @@
 from outerloop.syscall import install_tool as syscall_install_tool
 from outerloop.syscall import refresh_tool as syscall_refresh_tool
 from outerloop.syscall import write_siblings as syscall_write_siblings
-from outerloop.trajectory import file_refs
+from outerloop.trajectory import attest, file_refs
 from outerloop.trajectory import read as read_trajectory
 from outerloop.trajectory import record as record_event
 from outerloop.trajectory import summarize as summarize_trajectory
@@ -6410,6 +6410,7 @@ def finish_run(
         {"state": ending, "reason": note},
         refs=file_refs(directory, report),
     )
+    attest(directory, record.run_id)
     ref = str(record.stage.get("candidate_ref") or "")
     if snapshot_attempted and not ref:
         ws = None

src/outerloop/cli.py

@@ -1319,7 +1319,7 @@ def main(argv: list[str] | None = None) -> int:
         return pr(args)
     if args.command == "trajectory":
         from outerloop.runstate import run_dir
-        from outerloop.trajectory import read, summarize, verify
+        from outerloop.trajectory import check_attest, read, summarize, unattested_tail, verify
 
         values = env_file_values(keys=("OUTERLOOP_ROOT",)) if not args.root else {}
         root = Path(
@@ -1337,6 +1337,11 @@ def main(argv: list[str] | None = None) -> int:
             print(f"cannot read trajectory: {exc}", file=sys.stderr)
             return 1
         problems = verify(directory, args.run_id) if args.verify else []
+        attested = None
+        if args.verify:
+            attested = check_attest(directory, args.run_id)
+            if attested is False:
+                problems.append("attestation mismatch")
         if args.json:
             print(
                 json.dumps(
@@ -1354,6 +1359,13 @@ def main(argv: list[str] | None = None) -> int:
                 print("no trajectory")
             if args.verify:
                 print("; ".join(problems) if problems else "chain intact")
+                if attested and not problems:
+                    tail = unattested_tail(directory)
+                    print(
+                        "attested"
+                        if not tail
+                        else f"attested; {tail} later event(s) not covered by the attestation"
+                    )
             if events:
                 summary = summarize(events)
                 verbs = ", ".join(

src/outerloop/trajectory.py

@@ -5,19 +5,25 @@
 import contextlib
 import fcntl
 import hashlib
+import hmac
 import json
 import logging
+import os
 import time
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from outerloop import paths
+
 if TYPE_CHECKING:
     from outerloop.harness import SessionResult
 
 log = logging.getLogger(__name__)
 LEDGER = "trajectory.jsonl"
 MAX_LEDGER_BYTES = 4_000_000
+ATTEST_KEY = "trajectory_attest.key"
+ATTEST_ALG = "hmac-sha256"
 _cap_warned: set[Path] = set()
 
 
@@ -45,6 +51,20 @@ def record(
     The directory name is the run id. The lock covers reading and appending;
     malformed lines remain visible to verify, including after recovery.
     """
+    _append(run_dir, type, data, actor=actor, refs=refs, at=at)
+
+
+def _append(
+    run_dir: Path,
+    type: str,
+    data: dict[str, Any] | Callable[[str], dict[str, Any]],
+    *,
+    actor: str = "kernel",
+    refs: Iterable[dict[str, str]] = (),
+    at: float | None = None,
+) -> None:
+    """Append one event under the exclusive lock, so `prev` and any data built
+    from it describe the chain as it is at the write."""
     try:
         path = run_dir / LEDGER
         run_dir.mkdir(parents=True, exist_ok=True)
@@ -69,7 +89,7 @@ def record(
                     ts=time.time() if at is None else float(at),
                     type=type,
                     actor=actor,
-                    data=data,
+                    data=data(prev) if callable(data) else data,
                     refs=list(refs),
                     prev=prev,
                 )
@@ -83,6 +103,94 @@ def record(
         log.warning("could not write trajectory in %s: %s", run_dir, exc)
 
 
+def _attestation_key(*, create: bool) -> bytes | None:
+    """The deployment's 32-byte key beside the operator settings file, created
+    owner-only on first use. Only `attest` creates it."""
+    try:
+        path = paths.env_file().parent / ATTEST_KEY
+        if create and not path.exists():
+            path.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            except FileExistsError:
+                pass  # another process created it first
+            else:
+                with os.fdopen(fd, "wb") as stream:
+                    stream.write(os.urandom(32))
+        key = path.read_bytes()
+    except (OSError, ValueError) as exc:
+        log.warning("attestation key unavailable: %s", exc)
+        return None
+    if len(key) != 32:
+        log.warning("attestation key is not 32 bytes")
+        return None
+    return key
+
+
+def attest(run_dir: Path, run_id: str) -> None:
+    """Append one HMAC over the run id and the chain head; best effort."""
+    key = _attestation_key(create=True)
+    if key is None:
+        return
+
+    def data(head: str) -> dict[str, Any]:
+        mac = hmac.new(key, f"{run_id}\n{head}".encode(), hashlib.sha256).hexdigest()
+        return {"head": head, "alg": ATTEST_ALG, "mac": mac}
+
+    _append(run_dir, "attest", data)
+
+
+def check_attest(run_dir: Path, run_id: str) -> bool | None:
+    """Does the last attest event verify with this deployment's key?
+
+    None when the chain has no attest event or no key exists; True only when the
+    recorded head is the actual preceding line and the MAC matches it.
+    """
+    key = _attestation_key(create=False)
+    if key is None:
+        return None
+    try:
+        with (run_dir / LEDGER).open("rb") as stream:
+            raw = stream.read(MAX_LEDGER_BYTES + 1)
+    except OSError:
+        return None
+    found: tuple[dict[str, Any], str] | None = None
+    prev = hashlib.sha256(run_id.encode()).hexdigest()
+    for line in raw.splitlines(keepends=True):
+        if not line.endswith(b"\n"):
+            break
+        if (row := _event(line[:-1])) is None:
+            continue
+        if row.get("type") == "attest":
+            found = (row, prev)
+        prev = hashlib.sha256(line[:-1]).hexdigest()
+    if found is None:
+        return None
+    row, head = found
+    data = row.get("data")
+    if not isinstance(data, dict) or data.get("alg") != ATTEST_ALG:
+        return False
+    mac = hmac.new(key, f"{run_id}\n{head}".encode(), hashlib.sha256).hexdigest()
+    recorded = data.get("mac")
+    return (
+        data.get("head") == head
+        and isinstance(recorded, str)
+        and hmac.compare_digest(recorded, mac)
+    )
+
+
+def unattested_tail(run_dir: Path) -> int:
+    """Events after the last attest event: the attestation does not cover them.
+    The kernel may legitimately record a few after an ending (e.g. cancelled
+    launches), so this is reported, not treated as a mismatch."""
+    try:
+        events = read(run_dir)
+    except OSError:
+        return 0
+    last = max((i for i, e in enumerate(events) if e.get("type") == "attest"), default=None)
+    return 0 if last is None else len(events) - last - 1
+
+
 def _warn_cap(path: Path) -> None:
     if path not in _cap_warned:
         _cap_warned.add(path)

tests/conftest.py

@@ -69,6 +69,17 @@ def _configured_claude_model(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("OUTERLOOP_CLAUDE_MODEL", "claude-test-model")
 
 
+@pytest.fixture(autouse=True)
+def _test_attestation_key():
+    """Run endings must not create a deployment key in the runner's home; the
+    trajectory tests put the real key in a temp config dir themselves. Its own
+    MonkeyPatch, so a test's `monkeypatch.undo()` cannot remove the guard."""
+    guard = pytest.MonkeyPatch()
+    guard.setattr("outerloop.trajectory._attestation_key", lambda *, create: b"k" * 32)
+    yield
+    guard.undo()
+
+
 @pytest.fixture(autouse=True)
 def _no_tick_lease_settle(monkeypatch):
     """The settle read-back after a fresh tick lease is a one-second sleep in

tests/test_attempt_review.py

@@ -1526,11 +1526,12 @@ def fail(*args, **kwargs):
     assert released == ["refs/test/snapshot"]
     assert load_record(root, "tsp-r1").state == ENDED
     assert "notebook unavailable" in caplog.text
-    from outerloop.trajectory import read, verify
+    from outerloop.trajectory import check_attest, read, verify
 
     directory = run_dir(root, record.run_id)
     events = read(directory)
-    assert [event["type"] for event in events] == ["ending"]
+    assert [event["type"] for event in events] == ["ending", "attest"]
+    assert check_attest(directory, record.run_id) is True
     assert events[0]["data"] == {"state": "merged", "reason": "PR merged"}
     assert events[0]["ts"] > NOW  # wall-clock time, not the caller's tick time
     assert events[0]["refs"][0]["path"] == "report.md"

tests/test_trajectory.py

@@ -6,6 +6,10 @@
 
 from outerloop import trajectory
 
+# conftest stubs the key so unrelated run-ending tests touch no real config;
+# these tests exercise the file itself.
+REAL_ATTESTATION_KEY = trajectory._attestation_key
+
 
 def test_summarize() -> None:
     start = {"type": "session.start", "data": {"backend": "claude", "model": "model-x"}}
@@ -146,3 +150,73 @@ def test_non_serializable_data_is_best_effort(tmp_path: Path, caplog) -> None:
     assert trajectory.read(tmp_path) == []
     trajectory.record(tmp_path, "wake", {})
     assert trajectory.verify(tmp_path, tmp_path.name) == []
+
+
+def test_attest_round_trip(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setattr(trajectory, "_attestation_key", REAL_ATTESTATION_KEY)
+    config = tmp_path / "config"
+    monkeypatch.setenv("OUTERLOOP_ENV_FILE", str(config / ".env"))
+    directory = tmp_path / "runs" / "sample"
+    trajectory.record(directory, "measure", {"value": 1}, at=1)
+
+    trajectory.attest(directory, "sample")
+
+    events = trajectory.read(directory)
+    assert [event["type"] for event in events] == ["measure", "attest"]
+    data = events[-1]["data"]
+    assert data["alg"] == "hmac-sha256"
+    assert data["head"] == events[-1]["prev"]
+    assert len(data["mac"]) == 64
+    assert trajectory.check_attest(directory, "sample") is True
+    assert trajectory.verify(directory, "sample") == []
+    key = config / trajectory.ATTEST_KEY
+    assert key.stat().st_mode & 0o777 == 0o600
+    assert key.read_bytes() != b""
+    assert not (directory / trajectory.ATTEST_KEY).exists()
+
+
+def test_verify_reports_attested_then_mismatch(tmp_path: Path, monkeypatch, capsys) -> None:
+    from outerloop.cli import main
+
+    monkeypatch.setattr(trajectory, "_attestation_key", REAL_ATTESTATION_KEY)
+    monkeypatch.setenv("OUTERLOOP_ENV_FILE", str(tmp_path / "config" / ".env"))
+    directory = tmp_path / "runs" / "sample"
+    trajectory.record(directory, "measure", {"value": 1}, at=1)
+    trajectory.record(directory, "ending", {}, at=2)
+    trajectory.attest(directory, "sample")
+    args = ["trajectory", "sample", "--root", str(tmp_path), "--verify"]
+
+    assert main(args) == 0
+    out = capsys.readouterr().out
+    assert "chain intact" in out
+    assert "attested\n" in out
+
+    # an event appended after the attestation is reported as not covered
+    trajectory.record(directory, "launch.result", {}, at=3)
+    assert main(args) == 0
+    assert "attested; 1 later event(s) not covered" in capsys.readouterr().out
+
+    path = directory / trajectory.LEDGER
+    lines = path.read_bytes().splitlines(keepends=True)
+    lines[1] = lines[1].replace(b'"data": {}', b'"data": {"x": 1}')
+    path.write_bytes(b"".join(lines))
+
+    assert main(args) == 1
+    out = capsys.readouterr().out
+    assert "attestation mismatch" in out
+
+
+def test_attest_without_a_key_skips_the_event(tmp_path: Path, monkeypatch, caplog) -> None:
+    monkeypatch.setattr(trajectory, "_attestation_key", REAL_ATTESTATION_KEY)
+    config = tmp_path / "config"
+    config.write_text("")  # a file where the config dir should be
+    monkeypatch.setenv("OUTERLOOP_ENV_FILE", str(config / ".env"))
+    directory = tmp_path / "runs" / "sample"
+    trajectory.record(directory, "measure", {"value": 1}, at=1)
+
+    with caplog.at_level("WARNING"):
+        trajectory.attest(directory, "sample")
+
+    assert [event["type"] for event in trajectory.read(directory)] == ["measure"]
+    assert trajectory.check_attest(directory, "sample") is None
+    assert "attestation key unavailable" in caplog.text