hermes: bind terminal commit to journal inode

dockerfiles/hermes-execution-safety-regression.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import json
 import os
 import tempfile
+import threading
 import unittest
 from types import SimpleNamespace
 from unittest import mock
@@ -208,6 +209,147 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
         )
         self.assertIn(self._status(task_id), {"blocked", "triage"})
 
+    def test_exact_run_reclaim_succeeds_for_the_authoritative_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="recover exact run")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+
+        self.assertTrue(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="invalid exact journal",
+                expected_run_id=run_id,
+            )
+        )
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "ready")
+        self.assertIsNone(task.current_run_id)
+
+    def test_stale_reclaim_before_transaction_preserves_replacement_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement before txn")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        old_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="old recovery run",
+                expected_run_id=old_run,
+            )
+        )
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        replacement = kanban_db.get_task(self.connection, task_id)
+        replacement_run = replacement.current_run_id
+        self.assertNotEqual(old_run, replacement_run)
+        signals = []
+
+        self.assertFalse(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="stale journal",
+                expected_run_id=old_run,
+                signal_fn=lambda *args: signals.append(args),
+            )
+        )
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(latest.status, "running")
+        self.assertEqual(latest.current_run_id, replacement_run)
+        self.assertEqual(latest.claim_lock, replacement.claim_lock)
+        self.assertEqual(signals, [])
+
+    def test_reclaim_update_guard_preserves_run_changed_inside_transaction(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement in txn")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        old_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        replacement = {}
+
+        def install_replacement(*_args, **_kwargs):
+            cursor = self.connection.execute(
+                "INSERT INTO task_runs (task_id, status, claim_lock, started_at) "
+                "VALUES (?, 'running', ?, strftime('%s','now'))",
+                (task_id, "replacement-lock"),
+            )
+            replacement["run_id"] = int(cursor.lastrowid)
+            self.connection.execute(
+                "UPDATE tasks SET current_run_id = ?, claim_lock = ? WHERE id = ?",
+                (replacement["run_id"], "replacement-lock", task_id),
+            )
+            return {}
+
+        with mock.patch.object(
+            kanban_db,
+            "_terminate_reclaimed_worker",
+            side_effect=install_replacement,
+        ):
+            self.assertFalse(
+                kanban_db.reclaim_task(
+                    self.connection,
+                    task_id,
+                    reason="journal for old run",
+                    expected_run_id=old_run,
+                )
+            )
+
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(latest.status, "running")
+        self.assertEqual(latest.current_run_id, replacement["run_id"])
+        self.assertEqual(latest.claim_lock, "replacement-lock")
+
+    def test_concurrent_exact_reclaim_and_completion_have_one_winner(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="concurrent finalizer")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        barrier = threading.Barrier(2)
+        outcomes = {}
+        errors = []
+
+        def reclaim() -> None:
+            try:
+                with kanban_db.connect_closing() as connection:
+                    barrier.wait()
+                    outcomes["reclaim"] = kanban_db.reclaim_task(
+                        connection,
+                        task_id,
+                        reason="concurrent invalid journal",
+                        expected_run_id=run_id,
+                    )
+            except BaseException as error:  # pragma: no cover - assertion relay
+                errors.append(error)
+
+        def finalize() -> None:
+            try:
+                with kanban_db.connect_closing() as connection:
+                    barrier.wait()
+                    outcomes["complete"] = kanban_db.complete_task(
+                        connection,
+                        task_id,
+                        result="durable winner",
+                        summary="durable winner",
+                        expected_run_id=run_id,
+                    )
+            except BaseException as error:  # pragma: no cover - assertion relay
+                errors.append(error)
+
+        threads = [threading.Thread(target=reclaim), threading.Thread(target=finalize)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=10)
+            self.assertFalse(thread.is_alive())
+
+        self.assertEqual(errors, [])
+        self.assertEqual(set(outcomes), {"reclaim", "complete"})
+        self.assertEqual(sum(bool(value) for value in outcomes.values()), 1)
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertIn(latest.status, {"done", "ready"})
+        self.assertIsNone(latest.current_run_id)
+        runs = kanban_db.list_runs(self.connection, task_id)
+        self.assertEqual(len(runs), 1)
+        self.assertIsNotNone(runs[0].ended_at)
+
     def test_fresh_triage_task_still_auto_promotes(self) -> None:
         task_id = kanban_db.create_task(
             self.connection,

dockerfiles/patch-hermes-execution-safety.py

@@ -124,6 +124,166 @@ db = replace_once(
     "transactional exact ended-run completion",
 )
 
+reclaim_before = '''def reclaim_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    reason: Optional[str] = None,
+    signal_fn=None,
+) -> bool:
+    """Operator-driven reclaim: release the claim and reset to ``ready``.
+
+    Unlike :func:`release_stale_claims` which only acts on tasks whose
+    ``claim_expires`` has passed, this function reclaims immediately
+    regardless of TTL. Intended for the dashboard/CLI recovery flow
+    when an operator wants to abort a running worker without waiting
+    for the TTL to expire (e.g. after seeing a hallucination warning).
+
+    Returns True if a reclaim happened, False if the task isn't in a
+    reclaimable state (not running, or doesn't exist).
+    """
+    row = conn.execute(
+        "SELECT status, claim_lock, worker_pid FROM tasks WHERE id = ?",
+        (task_id,),
+    ).fetchone()
+    if not row:
+        return False
+    if row["status"] != "running" and row["claim_lock"] is None:
+        # Nothing to reclaim — already ready / blocked / done.
+        return False
+    prev_lock = row["claim_lock"]
+    termination = _terminate_reclaimed_worker(
+        row["worker_pid"], prev_lock, signal_fn=signal_fn,
+    )
+    with write_txn(conn):
+        cur = conn.execute(
+            "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+            "claim_expires = NULL, worker_pid = NULL "
+            "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+            "AND claim_lock IS ?",
+            (task_id, prev_lock),
+        )
+        if cur.rowcount != 1:
+            return False
+        run_id = _end_run(
+            conn, task_id,
+            outcome="reclaimed", status="reclaimed",
+            error=(
+                f"manual_reclaim: {reason}" if reason
+                else f"manual_reclaim lock={prev_lock}"
+            ),
+            metadata=termination,
+        )
+        payload = {
+            "manual": True,
+            "reason": reason,
+            "prev_lock": prev_lock,
+        }
+        payload.update(termination)
+        _append_event(
+            conn, task_id, "reclaimed",
+            payload,
+            run_id=run_id,
+        )
+    # Operator intervention — they've looked at the task, so the
+    # consecutive-failures counter is now stale. Give the next retry
+    # a fresh budget. (_clear_failure_counter opens its own write_txn,
+    # so it runs after the enclosing one commits.)
+    _clear_failure_counter(conn, task_id)
+    return True
+'''
+reclaim_after = '''def reclaim_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    reason: Optional[str] = None,
+    signal_fn=None,
+    expected_run_id: Optional[int] = None,
+) -> bool:
+    """Operator-driven reclaim: release the claim and reset to ``ready``.
+
+    Unlike :func:`release_stale_claims` which only acts on tasks whose
+    ``claim_expires`` has passed, this function reclaims immediately
+    regardless of TTL. Intended for the dashboard/CLI recovery flow
+    when an operator wants to abort a running worker without waiting
+    for the TTL to expire (e.g. after seeing a hallucination warning).
+
+    When ``expected_run_id`` is supplied, the run identity is checked after
+    ``BEGIN IMMEDIATE`` and included in the guarded update. A replacement run
+    is therefore neither signalled nor reclaimed by stale recovery evidence.
+
+    Returns True if a reclaim happened, False if the task isn't in a
+    reclaimable state (not running, or doesn't exist), or its run changed.
+    """
+    with write_txn(conn):
+        row = conn.execute(
+            "SELECT status, claim_lock, worker_pid, current_run_id "
+            "FROM tasks WHERE id = ?",
+            (task_id,),
+        ).fetchone()
+        if not row:
+            return False
+        if expected_run_id is not None and row["current_run_id"] != expected_run_id:
+            return False
+        if row["status"] != "running" and row["claim_lock"] is None:
+            # Nothing to reclaim — already ready / blocked / done.
+            return False
+        prev_lock = row["claim_lock"]
+        termination = _terminate_reclaimed_worker(
+            row["worker_pid"], prev_lock, signal_fn=signal_fn,
+        )
+        if expected_run_id is None:
+            cur = conn.execute(
+                "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+                "claim_expires = NULL, worker_pid = NULL "
+                "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+                "AND claim_lock IS ?",
+                (task_id, prev_lock),
+            )
+        else:
+            cur = conn.execute(
+                "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+                "claim_expires = NULL, worker_pid = NULL "
+                "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+                "AND claim_lock IS ? AND current_run_id = ?",
+                (task_id, prev_lock, int(expected_run_id)),
+            )
+        if cur.rowcount != 1:
+            return False
+        run_id = _end_run(
+            conn, task_id,
+            outcome="reclaimed", status="reclaimed",
+            error=(
+                f"manual_reclaim: {reason}" if reason
+                else f"manual_reclaim lock={prev_lock}"
+            ),
+            metadata=termination,
+        )
+        payload = {
+            "manual": True,
+            "reason": reason,
+            "prev_lock": prev_lock,
+        }
+        payload.update(termination)
+        _append_event(
+            conn, task_id, "reclaimed",
+            payload,
+            run_id=run_id,
+        )
+    # Operator intervention — they've looked at the task, so the
+    # consecutive-failures counter is now stale. Give the next retry
+    # a fresh budget. (_clear_failure_counter opens its own write_txn,
+    # so it runs after the enclosing one commits.)
+    _clear_failure_counter(conn, task_id)
+    return True
+'''
+db = replace_once(
+    db,
+    reclaim_before,
+    reclaim_after,
+    "transactional exact-run reclaim",
+)
+
 specify_signature_before = '''def specify_triage_task(
     conn: sqlite3.Connection,
     task_id: str,

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:
@@ -867,6 +893,48 @@ def _completed_result(summary: str = "done") -> dict:
     }
 
 
+def _pending_terminal_record(board: str, task_id: str, run_id: int, summary: str) -> dict:
+    structured = _completed_result(summary)
+    return {
+        "board": board,
+        "task_id": task_id,
+        "expected_run_id": run_id,
+        "result": json.dumps(structured, sort_keys=True),
+        "summary": summary,
+        "metadata": {},
+        "kanban_state": "pending",
+        "recorded_at": lanes.utc_now(),
+    }
+
+
+def _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) -> None:
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        if task.status != "running" or task.current_run_id != kwargs["expected_run_id"]:
+            return False
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.current_run_id = None
+        completions.append(kwargs["result"])
+        return True
+
+    def reclaim_task(_conn, _task_id, **kwargs):
+        reclaims.append(kwargs)
+        return False
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+        reclaim_task=reclaim_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+
 def test_terminal_recovery_can_complete_the_exact_latest_ended_run(
     tmp_path: Path,
     monkeypatch,
@@ -1185,13 +1253,19 @@ def test_malformed_exact_run_journal_is_quarantined_and_reclaimed(
         scoped_current_board=lambda _board: nullcontext(),
         connect=lambda board: Connection(),
         get_task=lambda _conn, _task_id: task,
-        reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True),
+        reclaim_task=lambda *_args, **kwargs: (reclaimed.append(kwargs) or True),
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
 
     assert lanes.recover_pending_finalizations() == 0
 
-    assert reclaimed == [True]
+    assert reclaimed == [{
+        "reason": (
+            "terminal journal recovery failed (malformed-payload); "
+            "exact run may retry"
+        ),
+        "expected_run_id": 12,
+    }]
     assert not path.exists()
     assert lanes._has_pending_finalization("cassandra", "t_partial", 12) is False
     quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
@@ -1417,6 +1491,747 @@ def test_quarantine_preserves_and_replays_an_atomic_replacement(
     assert not path.exists()
 
 
+def test_recovery_quarantines_the_loaded_inode_not_a_before_open_replacement(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_gap"), 51)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"malformed-original")
+    replacement = path.with_name("replacement.valid")
+    lanes.atomic_json(
+        replacement,
+        _pending_terminal_record("cassandra", "t_gap", 51, "preserved replacement"),
+    )
+    task = SimpleNamespace(
+        id="t_gap",
+        status="running",
+        current_run_id=51,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+    real_quarantine = lanes._quarantine_terminal
+    swapped = {"value": False}
+
+    def swap_before_quarantine_reopens(*args, **kwargs):
+        assert kwargs["snapshot"].prefix == b"malformed-original"
+        if not swapped["value"]:
+            os.replace(replacement, path)
+            swapped["value"] = True
+        return real_quarantine(*args, **kwargs)
+
+    monkeypatch.setattr(lanes, "_quarantine_terminal", swap_before_quarantine_reopens)
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert swapped["value"] is True
+    assert task.status == "done"
+    assert len(completions) == 1
+    assert reclaims == []
+    assert not path.exists()
+    committed = next(path.parent.glob("*.terminal.committed.json"))
+    assert json.loads(committed.read_text())["result"] == completions[0]
+
+
+def test_recovery_replays_replacement_installed_after_quarantine_pins_source(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_after_open"), 52)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"malformed-original")
+    replacement = path.with_name("replacement.valid")
+    lanes.atomic_json(
+        replacement,
+        _pending_terminal_record(
+            "cassandra", "t_after_open", 52, "replacement after open"
+        ),
+    )
+    task = SimpleNamespace(
+        id="t_after_open",
+        status="running",
+        current_run_id=52,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+    real_fsync = lanes.os.fsync
+    swapped = {"value": False}
+
+    def swap_after_quarantine_open(descriptor):
+        descriptor_stat = os.fstat(descriptor)
+        if stat.S_ISDIR(descriptor_stat.st_mode) and not swapped["value"]:
+            os.replace(replacement, path)
+            swapped["value"] = True
+        real_fsync(descriptor)
+
+    monkeypatch.setattr(lanes.os, "fsync", swap_after_quarantine_open)
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert swapped["value"] is True
+    assert task.status == "done"
+    assert len(completions) == 1
+    assert reclaims == []
+    assert not path.exists()
+
+
+def test_recovery_survives_repeated_invalid_then_valid_replacements(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_churn"), 53)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"malformed-zero")
+    second = path.with_name("replacement.invalid")
+    second.write_bytes(b"malformed-one")
+    valid = path.with_name("replacement.valid")
+    lanes.atomic_json(
+        valid,
+        _pending_terminal_record("cassandra", "t_churn", 53, "valid after churn"),
+    )
+    task = SimpleNamespace(
+        id="t_churn",
+        status="running",
+        current_run_id=53,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+    real_quarantine = lanes._quarantine_terminal
+    replacements = [second, valid]
+    pinned = []
+
+    def churn_before_quarantine(*args, **kwargs):
+        pinned.append(kwargs["snapshot"].prefix)
+        if replacements:
+            os.replace(replacements.pop(0), path)
+        return real_quarantine(*args, **kwargs)
+
+    monkeypatch.setattr(lanes, "_quarantine_terminal", churn_before_quarantine)
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert pinned == [b"malformed-zero", b"malformed-one"]
+    assert replacements == []
+    assert task.status == "done"
+    assert len(completions) == 1
+    assert reclaims == []
+    assert not path.exists()
+    diagnostics = list((path.parent / "quarantine").glob("*.quarantine"))
+    assert len(diagnostics) == 2
+
+
+def test_recovery_finalizes_the_valid_snapshot_it_classified_before_replacement(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, first = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_valid_gap"),
+        board="cassandra",
+        task_id="t_valid_gap",
+        run_id=54,
+        structured=_completed_result("first valid snapshot"),
+        summary="first valid snapshot",
+        metadata={},
+    )
+    replacement = path.with_name("replacement.valid")
+    lanes.atomic_json(
+        replacement,
+        _pending_terminal_record(
+            "cassandra", "t_valid_gap", 54, "second valid replacement"
+        ),
+    )
+    task = SimpleNamespace(
+        id="t_valid_gap",
+        status="running",
+        current_run_id=54,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+    real_finalize = lanes._finalize_terminal_record
+    swapped = {"value": False}
+
+    def replace_before_finalize(*args, **kwargs):
+        assert kwargs["snapshot"].document == first
+        if not swapped["value"]:
+            os.replace(replacement, path)
+            swapped["value"] = True
+        return real_finalize(*args, **kwargs)
+
+    monkeypatch.setattr(lanes, "_finalize_terminal_record", replace_before_finalize)
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert task.result == first["result"]
+    assert completions == [first["result"]]
+    assert reclaims == []
+    assert not path.exists()
+    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
+    assert len(conflicts) == 1
+    conflict = json.loads(conflicts[0].read_text())
+    assert conflict["summary"] == "second valid replacement"
+
+
+@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 +2466,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 +2475,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 +2489,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(