fix(evidence): hold flock + derive offset from fd to close chain-corruption

evidence/append_event_stub.py

@@ -19,6 +19,7 @@
 from __future__ import annotations
 
 import argparse
+import fcntl
 import hashlib
 import json
 import os
@@ -39,6 +40,16 @@ def canonical_bytes(value: Any) -> bytes:
 
 
 def _strip_prefix(digest: str) -> str:
+    """Strip a scheme prefix ('sha256:') from a digest string.
+
+    Silently accepting a bare hex string is deliberate for BACKWARDS-COMPATIBILITY
+    with a small number of legacy records — but new writers must always emit the
+    `sha256:` form, and `append_event` below constructs its writes that way. If a
+    future writer starts emitting bare hex, `verify_journal` will still succeed on
+    the chain check but the reader loses the scheme labelling used elsewhere; that
+    is a separate constraint tightening (fail-closed on bare-hex on read) and NOT
+    what this file's durability fix set out to change.
+    """
     return digest.split(":", 1)[1] if ":" in digest else digest
 
 
@@ -57,28 +68,69 @@ def tip_digest(journal_path: Path) -> str:
 
 
 def append_event(payload: dict[str, Any], journal_path: Path = DEFAULT_JOURNAL) -> dict[str, Any]:
-    """Append one evidence record and return its real offset and digests."""
+    """Append one evidence record and return its real offset and digests.
+
+    CONCURRENCY. The read-tip → compute-chain → write → publish-offset window
+    is guarded by fcntl.flock(LOCK_EX) on the journal file descriptor for the
+    duration of the append. Without it, two concurrent writers observe the SAME
+    `previous`, compute two records whose `previousDigest` both link to the same
+    point, and both write — the chain silently branches and `verify_journal`
+    fails at the second record. Same class as the ledger.py fix in
+    SocioProphet/evidence-intake-kernel: durability is not asserted, it is held
+    by the lock.
+
+    Offset is derived from the file descriptor INSIDE the lock via `tell()`
+    after `seek(0, io.SEEK_END)`, NOT via a separate `Path.stat()` before the
+    write. A pre-write stat + a separate open leaves a TOCTOU window: another
+    writer can extend the file between the two syscalls, and the returned offset
+    points into the middle of another record — every downstream reader
+    (`cairn://<task>/<offset>` handles) dereferences garbage.
+    """
     event = payload.get("event", payload)
     digest = hashlib.sha256(canonical_bytes(event)).hexdigest()
 
     journal_path.parent.mkdir(parents=True, exist_ok=True)
-    previous = tip_digest(journal_path)
-    chain = hashlib.sha256(f"{previous}{digest}".encode("utf-8")).hexdigest()
 
-    record = {
-        "appendedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
-        "chainDigest": f"sha256:{chain}",
-        "evidenceDigest": f"sha256:{digest}",
-        "event": event,
-        "previousDigest": f"sha256:{previous}",
-    }
-
-    # Offset is taken before the write so it marks where this record starts.
-    journal_offset = journal_path.stat().st_size if journal_path.exists() else 0
-    with journal_path.open("a", encoding="utf-8") as handle:
-        handle.write(json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n")
-        handle.flush()
-        os.fsync(handle.fileno())
+    line = ""  # populated inside the lock; hoisted for post-lock return
+    journal_offset = 0
+    chain = ""
+    # Open once in "a+" so we can read the tip and write, both under the same
+    # advisory lock. `O_APPEND` semantics still guarantee the write goes to the
+    # end even if some other writer sneaks past a broken flock; the lock is the
+    # primary guarantee, and O_APPEND is the belt.
+    with journal_path.open("a+", encoding="utf-8") as handle:
+        try:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+            # Read the tip UNDER THE LOCK. tip_digest() opens its own file
+            # handle and would race — read directly from the locked fd.
+            handle.seek(0)
+            last = None
+            for read_line in handle:
+                if read_line.strip():
+                    last = read_line
+            previous = _strip_prefix(json.loads(last)["chainDigest"]) if last else GENESIS
+
+            chain = hashlib.sha256(f"{previous}{digest}".encode("utf-8")).hexdigest()
+            record = {
+                "appendedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+                "chainDigest": f"sha256:{chain}",
+                "evidenceDigest": f"sha256:{digest}",
+                "event": event,
+                "previousDigest": f"sha256:{previous}",
+            }
+            line = json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n"
+
+            # Offset derived from the fd (not stat) so nothing can extend the
+            # file between measurement and write.
+            handle.seek(0, os.SEEK_END)
+            journal_offset = handle.tell()
+            handle.write(line)
+            handle.flush()
+            os.fsync(handle.fileno())
+        finally:
+            # Release the lock explicitly; the close would release too, but
+            # being explicit keeps the intent legible next to the acquire.
+            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
 
     return {
         "appended": True,

tests/test_evidence_journal_concurrency.py

@@ -0,0 +1,110 @@
+"""Concurrency tests for evidence/append_event_stub.append_event.
+
+The defect these tests pin: before the fcntl.flock fix, two concurrent writers
+observed the same `previous` tip, both computed `chainDigest` from it, and both
+wrote — the chain silently BRANCHED at that point and verify_journal broke at
+the second record. Same class as the ledger fix in evidence-intake-kernel.
+
+A separate defect: `journalOffset` was taken via `Path.stat().st_size` BEFORE
+opening the file for the append — a TOCTOU where another writer extended the
+file between the stat and the write meant the returned offset pointed into the
+middle of another record.
+
+Both defects are latent unless the same journal is written by concurrent
+processes/threads — which is the scenario every real emitter (agent_runtime +
+a concurrent CLI append) creates.
+"""
+from __future__ import annotations
+
+import json
+import threading
+from pathlib import Path
+
+import pytest
+
+from evidence.append_event_stub import append_event, verify_journal
+
+
+def _worker(journal: Path, thread_id: int, events_per_thread: int) -> list[dict]:
+    """Append N events, capture each append's returned offset + digest."""
+    results = []
+    for i in range(events_per_thread):
+        results.append(
+            append_event(
+                {"event": {"kind": "test", "thread": thread_id, "idx": i}},
+                journal_path=journal,
+            )
+        )
+    return results
+
+
+def _run_concurrent_appends(journal: Path, threads: int, per_thread: int) -> list[dict]:
+    """Run `threads` threads each appending `per_thread` events; join and return."""
+    out: list[list[dict]] = [[] for _ in range(threads)]
+
+    def _entry(tid: int) -> None:
+        out[tid] = _worker(journal, tid, per_thread)
+
+    ts = [threading.Thread(target=_entry, args=(t,)) for t in range(threads)]
+    for t in ts:
+        t.start()
+    for t in ts:
+        t.join()
+    return [r for group in out for r in group]
+
+
+def test_concurrent_appends_verify_clean(tmp_path: Path) -> None:
+    """20 threads × 5 events each = 100 records; the chain must verify clean and
+    record count must equal thread_count × events_per_thread. Any lost or
+    branched record blows up here."""
+    journal = tmp_path / "j.jsonl"
+    _run_concurrent_appends(journal, threads=20, per_thread=5)
+    result = verify_journal(journal)
+    assert result["verified"], f"chain broken under concurrency: {result}"
+    assert result["records"] == 100, f"lost {100 - result['records']} events"
+
+
+def test_returned_offsets_are_distinct_and_point_at_starts(tmp_path: Path) -> None:
+    """Every appended record's `journalOffset` must be the true byte offset at
+    which its record starts. Under the old stat-before-write pattern, concurrent
+    writers observed the same offset and every downstream reader dereferenced
+    garbage. Pin this by asserting each returned offset maps to a line whose
+    chainDigest matches what the append returned."""
+    journal = tmp_path / "j.jsonl"
+    results = _run_concurrent_appends(journal, threads=10, per_thread=10)
+    assert len(results) == 100
+
+    offsets = [r["journalOffset"] for r in results]
+    assert len(set(offsets)) == len(offsets), "duplicate offsets — TOCTOU regression"
+
+    # Every returned offset must sit at the start of the line whose chainDigest
+    # it claims. This is the invariant `cairn://<task>/<offset>` handles depend on.
+    raw = journal.read_bytes()
+    for r in results:
+        chunk = raw[r["journalOffset"]:]
+        first_newline = chunk.index(b"\n")
+        record = json.loads(chunk[:first_newline].decode("utf-8"))
+        assert record["chainDigest"] == r["chainDigest"], (
+            f"offset {r['journalOffset']} does not point at record {r['chainDigest']}"
+        )
+
+
+def test_single_writer_appends_still_work(tmp_path: Path) -> None:
+    """Sanity — the concurrency fix must not regress the single-writer path."""
+    journal = tmp_path / "j.jsonl"
+    r1 = append_event({"event": {"kind": "x", "n": 1}}, journal_path=journal)
+    r2 = append_event({"event": {"kind": "x", "n": 2}}, journal_path=journal)
+    assert r1["appended"] and r2["appended"]
+    assert r1["journalOffset"] == 0
+    assert r2["journalOffset"] > 0
+    result = verify_journal(journal)
+    assert result["verified"] and result["records"] == 2
+
+
+def test_empty_journal_bootstraps_to_genesis(tmp_path: Path) -> None:
+    """The first record's previousDigest must be the genesis constant."""
+    journal = tmp_path / "j.jsonl"
+    append_event({"event": {"kind": "bootstrap"}}, journal_path=journal)
+    with journal.open("r", encoding="utf-8") as h:
+        record = json.loads(h.readline())
+    assert record["previousDigest"] == "sha256:" + "0" * 64