hermes: bind terminal commit to journal inode

services/hermes/scripts/cli_lane_runner.py

@@ -148,6 +148,21 @@ class TerminalIdentity:
     state: str
 
 
+@dataclass
+class TerminalSnapshot:
+    """One inode-bound, bounded journal read held open across finalization."""
+
+    document: dict[str, Any]
+    file_stat: os.stat_result
+    descriptor: int
+    directory_descriptor: int
+
+    def close(self) -> None:
+        """Release the pinned file and directory descriptors."""
+        os.close(self.descriptor)
+        os.close(self.directory_descriptor)
+
+
 def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -385,6 +400,10 @@ _TERMINAL_NAME = re.compile(
     r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
     r"terminal\.(?P<state>pending|committed)\.json$"
 )
+_TERMINAL_EVIDENCE_NAME = re.compile(
+    r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
+    r"terminal\.(?P<state>prepared|conflict)-(?P<digest>[a-f0-9]{32})\.json$"
+)
 _SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$")
 
 
@@ -418,6 +437,36 @@ def _terminal_identity(path: Path) -> TerminalIdentity | None:
     )
 
 
+def _terminal_evidence_identity(path: Path) -> TerminalIdentity | None:
+    """Parse immutable prepared/conflict evidence authority from its path."""
+    try:
+        relative = path.relative_to(STATE_ROOT)
+    except ValueError:
+        return None
+    if len(relative.parts) != 2:
+        return None
+    board, filename = relative.parts
+    match = _TERMINAL_EVIDENCE_NAME.fullmatch(filename)
+    try:
+        board_stat = path.parent.stat(follow_symlinks=False)
+    except OSError:
+        return None
+    if (
+        not match
+        or not stat.S_ISDIR(board_stat.st_mode)
+        or not _SAFE_BOARD.fullmatch(board)
+        or board in {".", ".."}
+        or match.group("task") in {".", ".."}
+    ):
+        return None
+    return TerminalIdentity(
+        board=board,
+        task_id=match.group("task"),
+        run_id=int(match.group("run")),
+        state=match.group("state"),
+    )
+
+
 def _read_bounded(descriptor: int, limit: int) -> bytes:
     """Read at most ``limit`` bytes from a regular file descriptor."""
     chunks: list[bytes] = []
@@ -431,10 +480,8 @@ def _read_bounded(descriptor: int, limit: int) -> bytes:
     return b"".join(chunks)
 
 
-def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]:
+def _open_small_json_snapshot(path: Path) -> TerminalSnapshot | None:
-    """Read one small, singly-linked journal through its validated directory."""
+    """Open and pin one bounded, singly-linked JSON document."""
-    if _terminal_identity(path) != identity:
-        return {}
     directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
     directory_flags |= getattr(os, "O_NOFOLLOW", 0)
     file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
@@ -449,10 +496,10 @@ def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any
             or opened.st_nlink != 1
             or opened.st_size > MAX_TERMINAL_RECORD_BYTES
         ):
-            return {}
+            return None
         payload = _read_bounded(descriptor, MAX_TERMINAL_RECORD_BYTES + 1)
         if len(payload) > MAX_TERMINAL_RECORD_BYTES:
-            return {}
+            return None
         current = os.stat(
             path.name,
             dir_fd=directory,
@@ -464,16 +511,53 @@ def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any
             or current.st_size != opened.st_size
             or len(payload) != opened.st_size
         ):
-            return {}
+            return None
         value = json.loads(payload.decode("utf-8"))
+        if not isinstance(value, dict):
+            return None
+        snapshot = TerminalSnapshot(value, opened, descriptor, directory)
+        descriptor = None
+        directory = None
+        return snapshot
     except (OSError, UnicodeError, json.JSONDecodeError):
-        return {}
+        return None
     finally:
         if descriptor is not None:
             os.close(descriptor)
         if directory is not None:
             os.close(directory)
-    return value if isinstance(value, dict) else {}
+
+
+def _open_terminal_snapshot(
+    path: Path,
+    identity: TerminalIdentity,
+) -> TerminalSnapshot | None:
+    """Open a terminal journal only after its lexical identity is verified."""
+    if _terminal_identity(path) != identity:
+        return None
+    return _open_small_json_snapshot(path)
+
+
+def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]:
+    """Read one small, singly-linked journal through its validated directory."""
+    snapshot = _open_terminal_snapshot(path, identity)
+    if snapshot is None:
+        return {}
+    try:
+        return snapshot.document
+    finally:
+        snapshot.close()
+
+
+def _load_small_json(path: Path) -> dict[str, Any]:
+    """Read one bounded immutable evidence document without following links."""
+    snapshot = _open_small_json_snapshot(path)
+    if snapshot is None:
+        return {}
+    try:
+        return snapshot.document
+    finally:
+        snapshot.close()
 
 
 def _persist_candidate(
@@ -580,7 +664,7 @@ def _terminal_record_valid(
         and structured["blockers"] == []
         and cli_lane_goal.unfinished_result_reason(structured) is None
         and record["summary"] == structured["summary"]
-        and record.get("kanban_state") in {"pending", "committed"}
+        and record.get("kanban_state") in {"pending", "prepared", "committed"}
     )
     if not valid or identity is None:
         return valid
@@ -588,6 +672,7 @@ def _terminal_record_valid(
         record.get("board") == identity.board
         and record.get("task_id") == identity.task_id
         and record.get("expected_run_id") == identity.run_id
+        and record.get("kanban_state") == identity.state
     )
 
 
@@ -1183,7 +1268,7 @@ def _retire_terminal_entry(
         f"{path.name}\0{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8")
     ).hexdigest()[:32]
     for sequence in range(32):
-        staging = f".retire.{path.name}.{identity}.{sequence}"
+        staging = f".retire.{identity}.{sequence}"
         try:
             _rename_noreplace(
                 path.name,
@@ -1222,6 +1307,182 @@ def _retire_terminal_entry(
     return "replacement"
 
 
+def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool:
+    """Durably create one immutable JSON artifact without replacing a peer."""
+    payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
+    if len(payload) > MAX_TERMINAL_RECORD_BYTES + 4096:
+        raise ValueError("terminal evidence exceeds the bounded artifact limit")
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    os.chmod(path.parent, 0o700, follow_symlinks=False)
+    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    directory_flags |= getattr(os, "O_NOFOLLOW", 0)
+    create_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    create_flags |= getattr(os, "O_NOFOLLOW", 0)
+    directory = os.open(path.parent, directory_flags)
+    temporary = f".{path.name}.{uuid.uuid4().hex}.tmp"
+    temporary_stat = None
+    descriptor = None
+    try:
+        descriptor = os.open(temporary, create_flags, 0o600, dir_fd=directory)
+        temporary_stat = os.fstat(descriptor)
+        with os.fdopen(descriptor, "wb") as stream:
+            descriptor = None
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+            written = os.fstat(stream.fileno())
+            if stat.S_IMODE(written.st_mode) != 0o600 or written.st_nlink != 1:
+                raise OSError("terminal evidence is not private and singly linked")
+        try:
+            _rename_noreplace(
+                temporary,
+                path.name,
+                source_dir=directory,
+                destination_dir=directory,
+            )
+        except FileExistsError:
+            return False
+        os.fsync(directory)
+        return True
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+        if temporary_stat is not None:
+            try:
+                _retire_terminal_entry(
+                    Path(temporary),
+                    temporary_stat,
+                    board_descriptor=directory,
+                    quarantine_descriptor=directory,
+                )
+            except OSError:
+                pass
+        os.close(directory)
+
+
+def _terminal_document_digest(document: dict[str, Any]) -> str:
+    """Return the stable identity for one exact accepted result document."""
+    immutable = {
+        key: document.get(key)
+        for key in (
+            "board",
+            "task_id",
+            "expected_run_id",
+            "result",
+            "summary",
+            "metadata",
+        )
+    }
+    canonical = json.dumps(immutable, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:32]
+
+
+def _terminal_evidence_path(
+    identity: TerminalIdentity,
+    state: str,
+    document: dict[str, Any],
+) -> Path:
+    """Return a deterministic immutable prepared or conflict evidence path."""
+    if state not in {"prepared", "conflict"}:
+        raise ValueError(f"invalid terminal evidence state: {state}")
+    base = state_path(identity.board, identity.task_id)
+    digest = _terminal_document_digest(document)
+    return base.with_name(
+        f"{base.stem}.run-{identity.run_id}.terminal.{state}-{digest}.json"
+    )
+
+
+def _same_terminal_document(left: dict[str, Any], right: dict[str, Any]) -> bool:
+    """Compare the immutable result-bearing fields of terminal evidence."""
+    keys = ("board", "task_id", "expected_run_id", "result", "summary", "metadata")
+    return all(left.get(key) == right.get(key) for key in keys)
+
+
+def _terminal_evidence_valid(
+    document: dict[str, Any],
+    identity: TerminalIdentity,
+    state: str,
+) -> bool:
+    """Validate semantic terminal content plus its immutable evidence state."""
+    if document.get("kanban_state") != state:
+        return False
+    semantic = dict(document)
+    semantic_identity = identity
+    if state == "conflict":
+        semantic["kanban_state"] = "prepared"
+        semantic_identity = TerminalIdentity(
+            identity.board,
+            identity.task_id,
+            identity.run_id,
+            "prepared",
+        )
+    return _terminal_record_valid(semantic, semantic_identity)
+
+
+def _persist_prepared_evidence(
+    identity: TerminalIdentity,
+    document: dict[str, Any],
+) -> Path:
+    """Fsync an immutable accepted result before entering the DB boundary."""
+    prepared = dict(document)
+    prepared["kanban_state"] = "prepared"
+    prepared["prepared_at"] = utc_now()
+    path = _terminal_evidence_path(identity, "prepared", document)
+    if _write_json_noreplace(path, prepared):
+        return path
+    if _terminal_evidence_identity(path) != TerminalIdentity(
+        identity.board, identity.task_id, identity.run_id, "prepared"
+    ):
+        raise OSError("prepared terminal evidence identity is invalid")
+    existing = _load_small_json(path)
+    prepared_identity = TerminalIdentity(
+        identity.board,
+        identity.task_id,
+        identity.run_id,
+        "prepared",
+    )
+    if not _terminal_evidence_valid(
+        existing,
+        prepared_identity,
+        "prepared",
+    ) or not _same_terminal_document(existing, prepared):
+        raise OSError("prepared terminal evidence path contains a conflicting result")
+    return path
+
+
+def _persist_conflict_evidence(
+    identity: TerminalIdentity,
+    document: dict[str, Any],
+    reason: str,
+) -> Path:
+    """Preserve a losing valid result in full under deterministic retention."""
+    conflict = dict(document)
+    conflict["kanban_state"] = "conflict"
+    conflict["conflict_reason"] = reason
+    conflict["conflicted_at"] = utc_now()
+    path = _terminal_evidence_path(identity, "conflict", document)
+    if _write_json_noreplace(path, conflict):
+        return path
+    if _terminal_evidence_identity(path) != TerminalIdentity(
+        identity.board, identity.task_id, identity.run_id, "conflict"
+    ):
+        raise OSError("terminal conflict evidence identity is invalid")
+    existing = _load_small_json(path)
+    conflict_identity = TerminalIdentity(
+        identity.board,
+        identity.task_id,
+        identity.run_id,
+        "conflict",
+    )
+    if not _terminal_evidence_valid(
+        existing,
+        conflict_identity,
+        "conflict",
+    ) or not _same_terminal_document(existing, conflict):
+        raise OSError("terminal conflict evidence path contains a different result")
+    return path
+
+
 def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path:
     """Record bounded metadata, then retire only the inspected journal inode."""
     board_dir = path.parent
@@ -1429,36 +1690,85 @@ def _recover_exact_run(
     return recovered
 
 
-def _commit_terminal_file(path: Path, identity: TerminalIdentity, document: dict[str, Any]) -> Path:
+def _retire_snapshot(path: Path, snapshot: TerminalSnapshot) -> str:
-    """Persist committed state, then atomically retire a pending journal."""
+    """Retire only the directory entry still naming an opened snapshot inode."""
+    return _retire_terminal_entry(
+        path,
+        snapshot.file_stat,
+        board_descriptor=snapshot.directory_descriptor,
+        quarantine_descriptor=snapshot.directory_descriptor,
+    )
+
+
+def _retire_snapshot_after_db(path: Path, snapshot: TerminalSnapshot) -> str:
+    """Keep a committed DB result recoverable across retirement fsync errors."""
+    try:
+        return _retire_snapshot(path, snapshot)
+    except OSError as error:
+        print(
+            f"terminal journal retirement deferred after DB commit: "
+            f"{type(error).__name__}: {error}",
+            file=sys.stderr,
+            flush=True,
+        )
+        return "deferred"
+
+
+def _discard_evidence(path: Path) -> str:
+    """Identity-safely remove one immutable prepared evidence artifact."""
+    snapshot = _open_small_json_snapshot(path)
+    if snapshot is None:
+        return "missing"
+    try:
+        return _retire_snapshot(path, snapshot)
+    finally:
+        snapshot.close()
+
+
+def _promote_prepared_evidence(
+    identity: TerminalIdentity,
+    document: dict[str, Any],
+    prepared: Path,
+) -> tuple[Path, str]:
+    """Publish the DB-winning result without overwriting first-writer evidence."""
     committed = _terminal_path(
         state_path(identity.board, identity.task_id),
         identity.run_id,
         "committed",
     )
-    document["kanban_state"] = "committed"
+    committed_document = dict(document)
-    document["committed_at"] = utc_now()
+    committed_document["kanban_state"] = "committed"
-    atomic_json(path, document)
+    committed_document["committed_at"] = utc_now()
-    os.replace(path, committed)
+    state = "committed"
-    os.chmod(committed, 0o600, follow_symlinks=False)
+    if not _write_json_noreplace(committed, committed_document):
-    _fsync_directory(committed.parent)
+        committed_identity = TerminalIdentity(
-    return committed
+            identity.board,
+            identity.task_id,
+            identity.run_id,
+            "committed",
+        )
+        existing = _load_terminal_json(committed, committed_identity)
+        if (
+            not _terminal_record_valid(existing, committed_identity)
+            or existing.get("kanban_state") != "committed"
+            or not _same_terminal_document(existing, committed_document)
+        ):
+            _persist_conflict_evidence(
+                identity,
+                document,
+                "canonical committed evidence already contains a different result",
+            )
+            state = "conflict"
+    _discard_evidence(prepared)
+    return committed, state
 
 
-def _finalize_terminal_record(
+def _finalize_document_db(
     kanban_db: Any,
-    path: Path,
+    identity: TerminalIdentity,
-    _record: dict[str, Any] | None = None,
+    document: dict[str, Any],
 ) -> str:
-    """Commit one exact-run terminal journal, or leave it safely pending."""
+    """Apply one validated exact-run result to Kanban under DB run guards."""
-    identity = _terminal_identity(path)
-    if identity is None or identity.state != "pending":
-        return "invalid"
-    document = _load_terminal_json(path, identity)
-    if not _terminal_record_valid(document):
-        return "invalid"
-    if not _terminal_record_valid(document, identity):
-        return "foreign"
 
     def operation(conn: Any) -> str:
         task = kanban_db.get_task(conn, identity.task_id)
@@ -1490,11 +1800,111 @@ def _finalize_terminal_record(
             metadata=document["metadata"],
             **completion_guard,
         )
-        return "committed" if completed else "pending"
+        if completed:
+            return "committed"
+        # A duplicate finalizer can lose the guarded UPDATE to an identical
+        # first writer. Re-read rather than reporting a false pending state.
+        latest = kanban_db.get_task(conn, identity.task_id)
+        if latest is not None and str(_task_value(latest, "status", "")) == "done":
+            return (
+                "committed"
+                if str(_task_value(latest, "result", "") or "") == document["result"]
+                else "stale"
+            )
+        return "pending"
 
-    outcome = str(_board_call(kanban_db, identity.board, operation))
+    return str(_board_call(kanban_db, identity.board, operation))
-    if outcome == "committed":
+
-        _commit_terminal_file(path, identity, document)
+
+def _resolve_pending_after_winner(
+    path: Path,
+    identity: TerminalIdentity,
+    winning_document: dict[str, Any],
+) -> bool:
+    """Retire duplicates and preserve differing replacements as conflicts."""
+    for _attempt in range(8):
+        snapshot = _open_terminal_snapshot(path, identity)
+        if snapshot is None:
+            return False
+        document = snapshot.document
+        if not _terminal_record_valid(document, identity):
+            snapshot.close()
+            return False
+        differing = not _same_terminal_document(document, winning_document)
+        try:
+            if differing:
+                _persist_conflict_evidence(
+                    identity,
+                    document,
+                    "different valid result lost the exact-run first-writer race",
+                )
+            retirement = _retire_snapshot_after_db(path, snapshot)
+        finally:
+            snapshot.close()
+        if retirement in {"retired", "missing"}:
+            if differing:
+                print(
+                    f"preserved terminal result conflict for {identity.board}/"
+                    f"{identity.task_id} run {identity.run_id}",
+                    file=sys.stderr,
+                    flush=True,
+                )
+            return differing
+        if retirement not in {"replacement", "replacement-staged", "collision"}:
+            return differing
+    return False
+
+
+def _finalize_terminal_record(
+    kanban_db: Any,
+    path: Path,
+    _record: dict[str, Any] | None = None,
+) -> str:
+    """Commit one exact-run terminal journal, or leave it safely pending."""
+    identity = _terminal_identity(path)
+    if identity is None or identity.state != "pending":
+        return "invalid"
+    snapshot = _open_terminal_snapshot(path, identity)
+    if snapshot is None:
+        return "invalid"
+    document = snapshot.document
+    try:
+        if not _terminal_record_valid(document):
+            return "invalid"
+        if not _terminal_record_valid(document, identity):
+            return "foreign"
+        prepared = _persist_prepared_evidence(identity, document)
+        outcome = _finalize_document_db(kanban_db, identity, document)
+        if outcome == "committed":
+            _committed, evidence_state = _promote_prepared_evidence(
+                identity,
+                document,
+                prepared,
+            )
+            retirement = _retire_snapshot_after_db(path, snapshot)
+        elif outcome == "stale":
+            _persist_conflict_evidence(
+                identity,
+                document,
+                "valid terminal result no longer matches the authoritative task run",
+            )
+            _discard_evidence(prepared)
+            retirement = _retire_snapshot_after_db(path, snapshot)
+            outcome = "conflict"
+            evidence_state = "conflict"
+        else:
+            return outcome
+    finally:
+        snapshot.close()
+    if outcome == "committed" and retirement not in {"retired", "missing"}:
+        _resolve_pending_after_winner(path, identity, document)
+    if evidence_state == "conflict":
+        print(
+            f"terminal first-writer conflict recorded for {identity.board}/"
+            f"{identity.task_id} run {identity.run_id}",
+            file=sys.stderr,
+            flush=True,
+        )
     return outcome
 
 
@@ -1515,11 +1925,81 @@ def _replay_replacement_terminal(
     return True, outcome == "committed"
 
 
+def _recover_prepared_finalizations(kanban_db: Any) -> int:
+    """Recover an accepted result staged before an interrupted DB boundary."""
+    recovered = 0
+    prepared_paths = list(STATE_ROOT.glob("*/*.terminal.prepared-*.json"))
+
+    def durable_order(path: Path) -> tuple[int, str]:
+        try:
+            return path.stat(follow_symlinks=False).st_mtime_ns, str(path)
+        except OSError:
+            return 2**63 - 1, str(path)
+
+    # When no DB winner exists yet, the first durably prepared result owns the
+    # run. The DB transaction remains the final arbiter across runner processes.
+    for path in sorted(prepared_paths, key=durable_order):
+        identity = _terminal_evidence_identity(path)
+        snapshot = _open_small_json_snapshot(path)
+        if identity is None or identity.state != "prepared" or snapshot is None:
+            if snapshot is not None:
+                snapshot.close()
+            _quarantine_terminal(path, identity, "malformed-prepared-evidence")
+            continue
+        document = snapshot.document
+        try:
+            expected = _terminal_evidence_path(identity, "prepared", document)
+            if (
+                expected.name != path.name
+                or not _terminal_evidence_valid(document, identity, "prepared")
+            ):
+                _quarantine_terminal(path, identity, "foreign-prepared-evidence")
+                continue
+            try:
+                outcome = _finalize_document_db(kanban_db, identity, document)
+            except Exception as error:
+                _record_board_access_error(identity.board, error)
+                continue
+            try:
+                if outcome == "committed":
+                    _promote_prepared_evidence(identity, document, path)
+                    pending = _terminal_path(
+                        state_path(identity.board, identity.task_id),
+                        identity.run_id,
+                        "pending",
+                    )
+                    pending_identity = TerminalIdentity(
+                        identity.board,
+                        identity.task_id,
+                        identity.run_id,
+                        "pending",
+                    )
+                    _resolve_pending_after_winner(
+                        pending,
+                        pending_identity,
+                        document,
+                    )
+                    recovered += 1
+                elif outcome == "stale":
+                    _persist_conflict_evidence(
+                        identity,
+                        document,
+                        "prepared result no longer matches the authoritative task run",
+                    )
+                    _discard_evidence(path)
+            except Exception as error:
+                _record_board_access_error(identity.board, error)
+                continue
+        finally:
+            snapshot.close()
+    return recovered
+
+
 def recover_pending_finalizations() -> int:
     """Replay accepted exact-run results before scheduling more provider work."""
     from hermes_cli import kanban_db
 
-    recovered = 0
+    recovered = _recover_prepared_finalizations(kanban_db)
     for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")):
         identity = _terminal_identity(path)
         record = _load_terminal_json(path, identity) if identity is not None else {}
@@ -1573,14 +2053,26 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
     try:
         path.stat()
     except FileNotFoundError:
-        return False
+        pass
     except OSError:
         return False
-    identity = _terminal_identity(path)
+    else:
-    if identity is None:
+        identity = _terminal_identity(path)
-        return False
+        if identity is not None:
-    record = _load_terminal_json(path, identity)
+            record = _load_terminal_json(path, identity)
-    return _terminal_record_valid(record, identity)
+            if _terminal_record_valid(record, identity):
+                return True
+    base = state_path(board, task_id)
+    for prepared in base.parent.glob(
+        f"{base.stem}.run-{run_id}.terminal.prepared-*.json"
+    ):
+        identity = _terminal_evidence_identity(prepared)
+        if identity is None or identity.state != "prepared":
+            continue
+        record = _load_small_json(prepared)
+        if _terminal_evidence_valid(record, identity, "prepared"):
+            return True
+    return False
 
 
 def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
@@ -1590,6 +2082,7 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
         "*.provider-*.result.json",
         "*.candidate-*.json",
         "*.terminal.committed.json",
+        "*.terminal.conflict-*.json",
     )
     for pattern in patterns:
         candidates.extend(board_dir.glob(pattern))
@@ -1969,9 +2462,10 @@ def execute_claim(board: str, task_id: str) -> None:
                         identity = _terminal_identity(terminal_file)
                         replayable = bool(
                             identity is not None
-                            and _terminal_record_valid(
+                            and _has_pending_finalization(
-                                _load_terminal_json(terminal_file, identity),
+                                identity.board,
-                                identity,
+                                identity.task_id,
+                                identity.run_id,
                             )
                         )
                         if not replayable:

testing/tests/test_hermes_cli_lanes.py

@@ -10,6 +10,7 @@ import os
 import signal
 import stat
 import sys
+import threading
 from contextlib import nullcontext
 from pathlib import Path
 from types import SimpleNamespace
@@ -804,6 +805,28 @@ def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization(
     assert reclaimed == []
 
 
+def test_prepared_evidence_without_pending_still_pins_the_exact_run(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    pending, record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_prepared_pin"),
+        board="cassandra",
+        task_id="t_prepared_pin",
+        run_id=14,
+        structured=_completed_result("accepted and prepared"),
+        summary="accepted and prepared",
+        metadata={},
+    )
+    identity = lanes._terminal_identity(pending)
+    assert identity is not None
+    lanes._persist_prepared_evidence(identity, record)
+    pending.unlink()
+
+    assert lanes._has_pending_finalization("cassandra", "t_prepared_pin", 14) is True
+
+
 def test_terminal_replay_never_crosses_into_a_replacement_run(
     tmp_path: Path,
     monkeypatch,
@@ -851,8 +874,11 @@ def test_terminal_replay_never_crosses_into_a_replacement_run(
     assert lanes.recover_pending_finalizations() == 0
     assert completions == []
     assert not terminal_path.exists()
-    quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
+    conflicts = list((state_root / "cassandra").glob("*.terminal.conflict-*.json"))
-    assert len(quarantined) == 1
+    assert len(conflicts) == 1
+    conflict = json.loads(conflicts[0].read_text(encoding="utf-8"))
+    assert conflict["result"] == _record["result"]
+    assert conflict["kanban_state"] == "conflict"
 
 
 def _completed_result(summary: str = "done") -> dict:
@@ -1417,6 +1443,555 @@ def test_quarantine_preserves_and_replays_an_atomic_replacement(
     assert not path.exists()
 
 
+@pytest.mark.parametrize(
+    "swap_point",
+    [
+        "during-complete",
+        "after-db-before-promote",
+        "before-committed-create",
+        "before-pending-retire",
+    ],
+)
+def test_terminal_first_writer_preserves_valid_replacement_conflicts(
+    tmp_path: Path,
+    monkeypatch,
+    swap_point: str,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, old_record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_first_writer"),
+        board="cassandra",
+        task_id="t_first_writer",
+        run_id=27,
+        structured=_completed_result("first result"),
+        summary="first result",
+        metadata={"writer": "first"},
+    )
+    replacement_structured = _completed_result("replacement result")
+    replacement_record = {
+        "board": "cassandra",
+        "task_id": "t_first_writer",
+        "expected_run_id": 27,
+        "result": json.dumps(replacement_structured, sort_keys=True),
+        "summary": "replacement result",
+        "metadata": {"writer": "replacement"},
+        "kanban_state": "pending",
+        "recorded_at": lanes.utc_now(),
+    }
+    replacement = path.with_name(f"replacement-{swap_point}.tmp")
+    lanes.atomic_json(replacement, replacement_record)
+    swapped = {"value": False}
+
+    def swap_pending():
+        if not swapped["value"]:
+            os.replace(replacement, path)
+            swapped["value"] = True
+
+    task = SimpleNamespace(
+        id="t_first_writer",
+        status="running",
+        result=None,
+        current_run_id=27,
+        assignee="cli-auto",
+    )
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        if swap_point == "during-complete":
+            swap_pending()
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+    if swap_point == "after-db-before-promote":
+        real_promote = lanes._promote_prepared_evidence
+
+        def swap_then_promote(*args, **kwargs):
+            swap_pending()
+            return real_promote(*args, **kwargs)
+
+        monkeypatch.setattr(lanes, "_promote_prepared_evidence", swap_then_promote)
+    elif swap_point == "before-committed-create":
+        real_create = lanes._write_json_noreplace
+
+        def swap_before_committed(path_arg, value):
+            if path_arg.name.endswith(".terminal.committed.json"):
+                swap_pending()
+            return real_create(path_arg, value)
+
+        monkeypatch.setattr(lanes, "_write_json_noreplace", swap_before_committed)
+    elif swap_point == "before-pending-retire":
+        real_retire = lanes._retire_snapshot
+
+        def swap_before_retire(path_arg, snapshot):
+            if path_arg == path:
+                swap_pending()
+            return real_retire(path_arg, snapshot)
+
+        monkeypatch.setattr(lanes, "_retire_snapshot", swap_before_retire)
+
+    assert lanes._finalize_terminal_record(fake_db, path, old_record) == "committed"
+
+    assert swapped["value"] is True
+    assert task.result == old_record["result"]
+    assert not path.exists()
+    committed = list(path.parent.glob("*.terminal.committed.json"))
+    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
+    prepared = list(path.parent.glob("*.terminal.prepared-*.json"))
+    assert len(committed) == 1
+    assert len(conflicts) == 1
+    assert prepared == []
+    assert committed[0].stat().st_mode & 0o777 == 0o600
+    assert conflicts[0].stat().st_mode & 0o777 == 0o600
+    committed_document = json.loads(committed[0].read_text(encoding="utf-8"))
+    conflict_document = json.loads(conflicts[0].read_text(encoding="utf-8"))
+    assert committed_document["result"] == old_record["result"]
+    assert committed_document["kanban_state"] == "committed"
+    assert conflict_document["result"] == replacement_record["result"]
+    assert conflict_document["metadata"] == {"writer": "replacement"}
+    assert conflict_document["kanban_state"] == "conflict"
+
+
+def test_terminal_commit_directory_fsync_failure_recovers_from_prepared_evidence(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_post_db_fsync"),
+        board="cassandra",
+        task_id="t_post_db_fsync",
+        run_id=28,
+        structured=_completed_result("durable DB winner"),
+        summary="durable DB winner",
+        metadata={},
+    )
+    task = SimpleNamespace(
+        id="t_post_db_fsync",
+        status="running",
+        result=None,
+        current_run_id=28,
+        assignee="cli-auto",
+    )
+    replacement_structured = _completed_result("replacement after DB commit")
+    replacement_record = {
+        "board": "cassandra",
+        "task_id": "t_post_db_fsync",
+        "expected_run_id": 28,
+        "result": json.dumps(replacement_structured, sort_keys=True),
+        "summary": "replacement after DB commit",
+        "metadata": {"writer": "replacement"},
+        "kanban_state": "pending",
+        "recorded_at": lanes.utc_now(),
+    }
+    replacement = path.with_name("post-db-fsync-replacement.tmp")
+    lanes.atomic_json(replacement, replacement_record)
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        os.replace(replacement, path)
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+    real_fsync = lanes.os.fsync
+    failed = {"value": False}
+
+    def fail_committed_directory_fsync(descriptor):
+        if (
+            stat.S_ISDIR(os.fstat(descriptor).st_mode)
+            and list(path.parent.glob("*.terminal.committed.json"))
+            and not failed["value"]
+        ):
+            failed["value"] = True
+            raise OSError(errno.ENOSPC, "post-DB committed directory fsync failed")
+        real_fsync(descriptor)
+
+    monkeypatch.setattr(lanes.os, "fsync", fail_committed_directory_fsync)
+
+    with pytest.raises(OSError, match="post-DB committed"):
+        lanes._finalize_terminal_record(fake_db, path, record)
+
+    assert task.status == "done"
+    assert task.result == record["result"]
+    assert path.exists()
+    assert json.loads(path.read_text(encoding="utf-8"))["result"] == (
+        replacement_record["result"]
+    )
+    assert len(list(path.parent.glob("*.terminal.prepared-*.json"))) == 1
+
+    monkeypatch.setattr(lanes.os, "fsync", real_fsync)
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    assert lanes.recover_pending_finalizations() == 1
+    assert not path.exists()
+    assert list(path.parent.glob("*.terminal.prepared-*.json")) == []
+    committed = list(path.parent.glob("*.terminal.committed.json"))
+    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
+    assert len(committed) == 1
+    assert len(conflicts) == 1
+    assert json.loads(committed[0].read_text())["result"] == record["result"]
+    assert json.loads(conflicts[0].read_text())["result"] == replacement_record["result"]
+
+
+def test_unknown_db_completion_outcome_replays_from_prepared_evidence(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_unknown_db_outcome"),
+        board="cassandra",
+        task_id="t_unknown_db_outcome",
+        run_id=35,
+        structured=_completed_result("DB committed before transport error"),
+        summary="DB committed before transport error",
+        metadata={},
+    )
+    task = SimpleNamespace(
+        id="t_unknown_db_outcome",
+        status="running",
+        result=None,
+        current_run_id=35,
+        assignee="cli-auto",
+    )
+    fail_once = {"value": True}
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        if fail_once["value"]:
+            fail_once["value"] = False
+            raise RuntimeError("transport failed after DB commit")
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+
+    with pytest.raises(RuntimeError, match="after DB commit"):
+        lanes._finalize_terminal_record(fake_db, path, record)
+
+    assert task.status == "done"
+    assert path.exists()
+    assert len(list(path.parent.glob("*.terminal.prepared-*.json"))) == 1
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    assert lanes.recover_pending_finalizations() == 1
+    assert not path.exists()
+    assert list(path.parent.glob("*.terminal.prepared-*.json")) == []
+    committed = list(path.parent.glob("*.terminal.committed.json"))
+    assert len(committed) == 1
+    assert json.loads(committed[0].read_text())["result"] == record["result"]
+
+
+def test_pending_retirement_fsync_failure_keeps_db_winner_committed(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_retire_fsync"),
+        board="cassandra",
+        task_id="t_retire_fsync",
+        run_id=33,
+        structured=_completed_result("retirement fsync winner"),
+        summary="retirement fsync winner",
+        metadata={},
+    )
+    task = SimpleNamespace(
+        id="t_retire_fsync",
+        status="running",
+        result=None,
+        current_run_id=33,
+        assignee="cli-auto",
+    )
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+    real_fsync = lanes.os.fsync
+    failed = {"value": False}
+
+    def fail_pending_retirement_fsync(descriptor):
+        if (
+            stat.S_ISDIR(os.fstat(descriptor).st_mode)
+            and not path.exists()
+            and list(path.parent.glob("*.terminal.committed.json"))
+            and not failed["value"]
+        ):
+            failed["value"] = True
+            raise OSError(errno.ENOSPC, "pending retirement fsync failed")
+        real_fsync(descriptor)
+
+    monkeypatch.setattr(lanes.os, "fsync", fail_pending_retirement_fsync)
+
+    assert lanes._finalize_terminal_record(fake_db, path, record) == "committed"
+    assert failed["value"] is True
+    assert task.status == "done"
+    assert task.result == record["result"]
+    assert not path.exists()
+    assert len(list(path.parent.glob("*.terminal.committed.json"))) == 1
+    assert list(path.parent.glob("*.terminal.prepared-*.json")) == []
+
+
+def test_committed_first_writer_is_never_overwritten_by_a_db_winner(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    state_file = lanes.state_path("cassandra", "t_committed_collision")
+    pending, winner = lanes._write_terminal_record(
+        state_file,
+        board="cassandra",
+        task_id="t_committed_collision",
+        run_id=34,
+        structured=_completed_result("DB winner"),
+        summary="DB winner",
+        metadata={"writer": "db"},
+    )
+    committed_path = lanes._terminal_path(state_file, 34, "committed")
+    first_writer = {
+        "board": "cassandra",
+        "task_id": "t_committed_collision",
+        "expected_run_id": 34,
+        "result": json.dumps(_completed_result("evidence first writer"), sort_keys=True),
+        "summary": "evidence first writer",
+        "metadata": {"writer": "evidence"},
+        "kanban_state": "committed",
+        "recorded_at": lanes.utc_now(),
+    }
+    lanes.atomic_json(committed_path, first_writer)
+    task = SimpleNamespace(
+        id="t_committed_collision",
+        status="running",
+        result=None,
+        current_run_id=34,
+        assignee="cli-auto",
+    )
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+
+    assert lanes._finalize_terminal_record(fake_db, pending, winner) == "committed"
+    assert task.result == winner["result"]
+    assert json.loads(committed_path.read_text())["result"] == first_writer["result"]
+    conflicts = list(pending.parent.glob("*.terminal.conflict-*.json"))
+    assert len(conflicts) == 1
+    assert json.loads(conflicts[0].read_text())["result"] == winner["result"]
+    assert not pending.exists()
+
+
+def test_concurrent_duplicate_terminal_finalizers_are_idempotent(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_duplicate"),
+        board="cassandra",
+        task_id="t_duplicate",
+        run_id=29,
+        structured=_completed_result("same accepted result"),
+        summary="same accepted result",
+        metadata={},
+    )
+    task_state = {
+        "status": "running",
+        "result": None,
+        "current_run_id": 29,
+    }
+    state_lock = threading.Lock()
+    readers = threading.Barrier(2)
+    completions = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def get_task(_conn, _task_id):
+        with state_lock:
+            snapshot = SimpleNamespace(
+                id="t_duplicate",
+                assignee="cli-auto",
+                **task_state,
+            )
+        if snapshot.status == "running":
+            readers.wait(timeout=5)
+        return snapshot
+
+    def complete_task(_conn, _task_id, **kwargs):
+        with state_lock:
+            if task_state["status"] != "running":
+                return False
+            task_state.update(
+                status="done",
+                result=kwargs["result"],
+                current_run_id=None,
+            )
+            completions.append(kwargs["result"])
+            return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=get_task,
+        complete_task=complete_task,
+    )
+    outcomes = []
+    errors = []
+
+    def finalize():
+        try:
+            outcomes.append(lanes._finalize_terminal_record(fake_db, path, record))
+        except Exception as error:  # pragma: no cover - assertion reports detail
+            errors.append(error)
+
+    workers = [threading.Thread(target=finalize) for _index in range(2)]
+    for worker in workers:
+        worker.start()
+    for worker in workers:
+        worker.join(timeout=10)
+
+    assert all(not worker.is_alive() for worker in workers)
+    assert errors == []
+    assert outcomes == ["committed", "committed"]
+    assert len(completions) == 1
+    assert not path.exists()
+    assert len(list(path.parent.glob("*.terminal.committed.json"))) == 1
+    assert list(path.parent.glob("*.terminal.conflict-*.json")) == []
+    assert list(path.parent.glob("*.terminal.prepared-*.json")) == []
+
+
+def test_recovery_uses_first_durable_prepared_result_as_db_writer(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    state_file = lanes.state_path("cassandra", "t_prepared_order")
+    pending, first = lanes._write_terminal_record(
+        state_file,
+        board="cassandra",
+        task_id="t_prepared_order",
+        run_id=30,
+        structured=_completed_result("first durable result"),
+        summary="first durable result",
+        metadata={"writer": "first"},
+    )
+    identity = lanes._terminal_identity(pending)
+    assert identity is not None
+    first_prepared = lanes._persist_prepared_evidence(identity, first)
+    os.utime(first_prepared, ns=(100, 100))
+    second_structured = _completed_result("second durable result")
+    second = {
+        "board": "cassandra",
+        "task_id": "t_prepared_order",
+        "expected_run_id": 30,
+        "result": json.dumps(second_structured, sort_keys=True),
+        "summary": "second durable result",
+        "metadata": {"writer": "second"},
+        "kanban_state": "pending",
+        "recorded_at": lanes.utc_now(),
+    }
+    second_prepared = lanes._persist_prepared_evidence(identity, second)
+    os.utime(second_prepared, ns=(200, 200))
+    lanes.atomic_json(pending, second)
+    task = SimpleNamespace(
+        id="t_prepared_order",
+        status="running",
+        result=None,
+        current_run_id=30,
+        assignee="cli-auto",
+    )
+    completions = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        if task.status != "running":
+            return False
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        completions.append(kwargs["result"])
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert completions == [first["result"]]
+    assert task.result == first["result"]
+    assert not pending.exists()
+    assert list(pending.parent.glob("*.terminal.prepared-*.json")) == []
+    committed = list(pending.parent.glob("*.terminal.committed.json"))
+    conflicts = list(pending.parent.glob("*.terminal.conflict-*.json"))
+    assert len(committed) == 1
+    assert len(conflicts) == 1
+    assert json.loads(committed[0].read_text())["result"] == first["result"]
+    assert json.loads(conflicts[0].read_text())["result"] == second["result"]
+
+
 def test_invalid_utf8_journal_quarantines_and_does_not_stop_later_replay(
     tmp_path: Path,
     monkeypatch,
@@ -1651,6 +2226,7 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
         board / "t.run-1.candidate-1.json",
         board / "t.run-1.provider-1.result.json",
         board / "t.run-1.terminal.committed.json",
+        board / f"t.run-1.terminal.conflict-{'a' * 32}.json",
         quarantine / "t.invalid.1234.quarantine",
     ]
     for artifact in old_artifacts:
@@ -1659,10 +2235,13 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
     quarantine_symlink.symlink_to(tmp_path / "missing-target")
     pending = board / "t.run-1.terminal.pending.json"
     pending.write_text("{}", encoding="utf-8")
+    prepared = board / f"t.run-1.terminal.prepared-{'b' * 32}.json"
+    prepared.write_text("{}", encoding="utf-8")
     old_time = 100.0
     for artifact in old_artifacts:
         os.utime(artifact, (old_time, old_time))
     os.utime(pending, (old_time, old_time))
+    os.utime(prepared, (old_time, old_time))
 
     assert lanes.gc_lane_artifacts(
         now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
@@ -1670,6 +2249,7 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
     assert not any(artifact.exists() for artifact in old_artifacts)
     assert not quarantine_symlink.is_symlink()
     assert pending.exists()
+    assert prepared.exists()
 
 
 def test_artifact_gc_prunes_oldest_by_count_and_total_bytes(