hermes: harden journal quarantine races

services/hermes/scripts/cli_lane_runner.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import concurrent.futures
+import ctypes
+import errno
 import hashlib
 import json
 import os
@@ -49,6 +51,8 @@ ARTIFACT_GC_INTERVAL_SECONDS = 5 * 60
 ARTIFACT_RETENTION_AGE_SECONDS = 30 * 24 * 60 * 60
 ARTIFACT_RETENTION_COUNT = 256
 ARTIFACT_RETENTION_BYTES = 128 * 1024 * 1024
+MAX_TERMINAL_RECORD_BYTES = 1024 * 1024
+QUARANTINE_HASH_BYTES = 64 * 1024
 PROVIDER_HEALTH_PATHS = {
     "codex": DATA_ROOT / "provider-health/codex.json",
     "claude": DATA_ROOT / "provider-health/claude.json",
@@ -193,7 +197,7 @@ def load_json(path: Path) -> dict[str, Any]:
         descriptor = os.open(path, flags)
         with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
             value = json.load(stream)
-    except (OSError, json.JSONDecodeError):
+    except (OSError, UnicodeError, json.JSONDecodeError):
         return {}
     return value if isinstance(value, dict) else {}
 
@@ -414,8 +418,21 @@ def _terminal_identity(path: Path) -> TerminalIdentity | None:
     )
 
 
+def _read_bounded(descriptor: int, limit: int) -> bytes:
+    """Read at most ``limit`` bytes from a regular file descriptor."""
+    chunks: list[bytes] = []
+    remaining = max(0, limit)
+    while remaining:
+        chunk = os.read(descriptor, min(64 * 1024, remaining))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
 def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]:
-    """Read through the already-validated board directory without symlink hops."""
+    """Read one small, singly-linked journal through its validated directory."""
     if _terminal_identity(path) != identity:
         return {}
     directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
@@ -426,10 +443,30 @@ def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any
     try:
         directory = os.open(path.parent, directory_flags)
         descriptor = os.open(path.name, file_flags, dir_fd=directory)
-        with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
+        opened = os.fstat(descriptor)
-            descriptor = None
+        if (
-            value = json.load(stream)
+            not stat.S_ISREG(opened.st_mode)
-    except (OSError, json.JSONDecodeError):
+            or opened.st_nlink != 1
+            or opened.st_size > MAX_TERMINAL_RECORD_BYTES
+        ):
+            return {}
+        payload = _read_bounded(descriptor, MAX_TERMINAL_RECORD_BYTES + 1)
+        if len(payload) > MAX_TERMINAL_RECORD_BYTES:
+            return {}
+        current = os.stat(
+            path.name,
+            dir_fd=directory,
+            follow_symlinks=False,
+        )
+        if (
+            current.st_dev != opened.st_dev
+            or current.st_ino != opened.st_ino
+            or current.st_size != opened.st_size
+            or len(payload) != opened.st_size
+        ):
+            return {}
+        value = json.loads(payload.decode("utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError):
         return {}
     finally:
         if descriptor is not None:
@@ -1093,8 +1130,100 @@ def _board_call(
     raise last_storage_error
 
 
+def _rename_noreplace(
+    source: str,
+    destination: str,
+    *,
+    source_dir: int,
+    destination_dir: int,
+) -> None:
+    """Atomically rename a directory entry without replacing a collision."""
+    renameat2 = getattr(ctypes.CDLL(None, use_errno=True), "renameat2", None)
+    if renameat2 is None:
+        raise OSError(errno.ENOSYS, "renameat2 is unavailable")
+    renameat2.argtypes = (
+        ctypes.c_int,
+        ctypes.c_char_p,
+        ctypes.c_int,
+        ctypes.c_char_p,
+        ctypes.c_uint,
+    )
+    renameat2.restype = ctypes.c_int
+    result = renameat2(
+        source_dir,
+        ctypes.c_char_p(os.fsencode(source)),
+        destination_dir,
+        ctypes.c_char_p(os.fsencode(destination)),
+        1,  # RENAME_NOREPLACE
+    )
+    if result != 0:
+        error = ctypes.get_errno()
+        raise OSError(error, os.strerror(error), source, destination)
+
+
+def _retire_terminal_entry(
+    path: Path,
+    source_stat: os.stat_result,
+    *,
+    board_descriptor: int,
+    quarantine_descriptor: int,
+) -> str:
+    """Retire only the inode previously inspected; preserve any replacement."""
+    try:
+        current = os.stat(
+            path.name,
+            dir_fd=board_descriptor,
+            follow_symlinks=False,
+        )
+    except FileNotFoundError:
+        return "missing"
+    if current.st_dev != source_stat.st_dev or current.st_ino != source_stat.st_ino:
+        return "replacement"
+    identity = hashlib.sha256(
+        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}"
+        try:
+            _rename_noreplace(
+                path.name,
+                staging,
+                source_dir=board_descriptor,
+                destination_dir=quarantine_descriptor,
+            )
+        except FileExistsError:
+            continue
+        break
+    else:
+        return "collision"
+    staged = os.stat(
+        staging,
+        dir_fd=quarantine_descriptor,
+        follow_symlinks=False,
+    )
+    if staged.st_dev == source_stat.st_dev and staged.st_ino == source_stat.st_ino:
+        os.unlink(staging, dir_fd=quarantine_descriptor)
+        os.fsync(quarantine_descriptor)
+        os.fsync(board_descriptor)
+        return "retired"
+    # The path changed after the pre-check but before the rename. Put the
+    # replacement back without overwriting an even newer entry.
+    try:
+        _rename_noreplace(
+            staging,
+            path.name,
+            source_dir=quarantine_descriptor,
+            destination_dir=board_descriptor,
+        )
+    except FileExistsError:
+        return "replacement-staged"
+    os.fsync(quarantine_descriptor)
+    os.fsync(board_descriptor)
+    return "replacement"
+
+
 def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path:
-    """Copy an unusable journal to a private file, then retire its exact name."""
+    """Record bounded metadata, then retire only the inspected journal inode."""
     board_dir = path.parent
     try:
         board_stat = board_dir.stat(follow_symlinks=False)
@@ -1114,9 +1243,10 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason:
     create_flags |= getattr(os, "O_NOFOLLOW", 0)
     board_descriptor = None
     quarantine_descriptor = None
+    source_descriptor = None
     source_stat = None
     destination = path
-    retired = False
+    retirement = "not-attempted"
     try:
         board_descriptor = os.open(board_dir, board_flags)
         source_stat = os.stat(
@@ -1125,37 +1255,50 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason:
             follow_symlinks=False,
         )
         safe_source = stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink == 1
-        payload: bytes
+        digest = None
+        hashed_bytes = 0
+        hash_complete = False
         if safe_source:
             source_descriptor = os.open(
                 path.name,
                 file_flags,
                 dir_fd=board_descriptor,
             )
-            with os.fdopen(source_descriptor, "rb") as source:
+            opened_stat = os.fstat(source_descriptor)
-                opened_stat = os.fstat(source.fileno())
+            if (
-                if (
+                opened_stat.st_dev != source_stat.st_dev
-                    opened_stat.st_dev != source_stat.st_dev
+                or opened_stat.st_ino != source_stat.st_ino
-                    or opened_stat.st_ino != source_stat.st_ino
+                or opened_stat.st_nlink != 1
-                    or opened_stat.st_nlink != 1
+            ):
-                ):
+                raise OSError("terminal journal identity changed while opening")
-                    raise OSError("terminal journal identity changed while opening")
+            # Keep this descriptor open through retirement. The staging entry
-                payload = source.read()
+            # must match the exact fstat-verified inode, not merely a path stat
-        else:
+            # captured before a concurrent replacement.
-            payload = (
+            source_stat = opened_stat
-                json.dumps(
+            prefix = _read_bounded(source_descriptor, QUARANTINE_HASH_BYTES)
-                    {
+            hashed_bytes = len(prefix)
-                        "original_name": path.name,
+            hash_complete = opened_stat.st_size <= QUARANTINE_HASH_BYTES
-                        "reason": reason,
+            digest = hashlib.sha256(prefix).hexdigest()
-                        "source_kind": "hardlink"
+        source_kind = (
-                        if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1
+            "regular"
-                        else "non-regular",
+            if safe_source
-                    },
+            else "hardlink"
-                    sort_keys=True,
+            if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1
-                )
+            else "non-regular"
-                + "\n"
+        )
-            ).encode("utf-8")
+        diagnostic = {
-        fingerprint = hashlib.sha256(payload).hexdigest()[:16]
+            "hash_complete": hash_complete,
+            "hashed_bytes": hashed_bytes,
+            "original_name": path.name,
+            "reason": reason,
+            "sha256": digest,
+            "size": source_stat.st_size,
+            "source_kind": source_kind,
+        }
+        payload = (json.dumps(diagnostic, sort_keys=True) + "\n").encode("utf-8")
+        fingerprint = (
+            (digest or hashlib.sha256(path.name.encode("utf-8")).hexdigest())[:16]
+        )
         safe_reason = (
             re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
         )
@@ -1198,47 +1341,46 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason:
         else:
             raise OSError("could not reserve a unique quarantine destination")
         os.fsync(quarantine_descriptor)
-        current = os.stat(
+        retirement = _retire_terminal_entry(
-            path.name,
+            path,
-            dir_fd=board_descriptor,
+            source_stat,
-            follow_symlinks=False,
+            board_descriptor=board_descriptor,
+            quarantine_descriptor=quarantine_descriptor,
         )
-        if (
-            current.st_dev != source_stat.st_dev
-            or current.st_ino != source_stat.st_ino
-        ):
-            print(
-                f"terminal journal changed before retirement; removing new entry: {path}",
-                file=sys.stderr,
-                flush=True,
-            )
-        os.unlink(path.name, dir_fd=board_descriptor)
-        retired = True
-        os.fsync(board_descriptor)
     except OSError as error:
         # A bad quarantine target must not make an attacker-controlled pending
-        # path eligible for replay forever. Retire only the directory entry;
+        # path eligible for replay forever. The same identity-safe staging
-        # never chmod or write through the source inode.
+        # protocol works in the board directory when quarantine is unusable.
-        if board_descriptor is not None and not retired:
+        if board_descriptor is not None and source_stat is not None:
             try:
-                os.unlink(path.name, dir_fd=board_descriptor)
+                retirement = _retire_terminal_entry(
-                retired = True
+                    path,
-                os.fsync(board_descriptor)
+                    source_stat,
+                    board_descriptor=board_descriptor,
+                    quarantine_descriptor=(
+                        quarantine_descriptor
+                        if quarantine_descriptor is not None
+                        else board_descriptor
+                    ),
+                )
             except OSError:
-                pass
+                retirement = "deferred"
         print(
             f"terminal journal quarantine degraded safely: {type(error).__name__}: {error}",
             file=sys.stderr,
             flush=True,
         )
     finally:
+        if source_descriptor is not None:
+            os.close(source_descriptor)
         if quarantine_descriptor is not None:
             os.close(quarantine_descriptor)
         if board_descriptor is not None:
             os.close(board_descriptor)
     print(
         "quarantined terminal journal "
-        f"{path.name}: {reason}; identity={identity or 'unparseable'}",
+        f"{path.name}: {reason}; identity={identity or 'unparseable'}; "
+        f"retirement={retirement}",
         file=sys.stderr,
         flush=True,
     )
@@ -1356,6 +1498,23 @@ def _finalize_terminal_record(
     return outcome
 
 
+def _replay_replacement_terminal(
+    kanban_db: Any,
+    path: Path,
+    identity: TerminalIdentity,
+) -> tuple[bool, bool]:
+    """Replay a valid journal atomically substituted during quarantine."""
+    replacement = _load_terminal_json(path, identity)
+    if not _terminal_record_valid(replacement, identity):
+        return False, False
+    try:
+        outcome = _finalize_terminal_record(kanban_db, path, replacement)
+    except Exception as error:
+        _record_board_access_error(identity.board, error)
+        return True, False
+    return True, outcome == "committed"
+
+
 def recover_pending_finalizations() -> int:
     """Replay accepted exact-run results before scheduling more provider work."""
     from hermes_cli import kanban_db
@@ -1369,11 +1528,23 @@ def recover_pending_finalizations() -> int:
             continue
         if not _terminal_record_valid(record):
             _quarantine_terminal(path, identity, "malformed-payload")
-            _recover_exact_run(kanban_db, identity, "malformed-payload")
+            replacement, committed = _replay_replacement_terminal(
+                kanban_db, path, identity
+            )
+            if committed:
+                recovered += 1
+            elif not replacement:
+                _recover_exact_run(kanban_db, identity, "malformed-payload")
             continue
         if not _terminal_record_valid(record, identity):
             _quarantine_terminal(path, identity, "foreign-identity")
-            _recover_exact_run(kanban_db, identity, "foreign-identity")
+            replacement, committed = _replay_replacement_terminal(
+                kanban_db, path, identity
+            )
+            if committed:
+                recovered += 1
+            elif not replacement:
+                _recover_exact_run(kanban_db, identity, "foreign-identity")
             continue
         try:
             outcome = _finalize_terminal_record(kanban_db, path, record)
@@ -1384,7 +1555,13 @@ def recover_pending_finalizations() -> int:
             recovered += 1
         elif outcome in {"invalid", "foreign", "stale"}:
             _quarantine_terminal(path, identity, outcome)
-            _recover_exact_run(kanban_db, identity, outcome)
+            replacement, committed = _replay_replacement_terminal(
+                kanban_db, path, identity
+            )
+            if committed:
+                recovered += 1
+            elif not replacement:
+                _recover_exact_run(kanban_db, identity, outcome)
     return recovered
 
 
@@ -1422,6 +1599,27 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
     return candidates
 
 
+def _unlink_artifact_if_same(path: Path, observed: os.stat_result) -> bool:
+    """Remove only the exact artifact inode selected by retention."""
+    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    directory_flags |= getattr(os, "O_NOFOLLOW", 0)
+    descriptor = None
+    try:
+        descriptor = os.open(path.parent, directory_flags)
+        outcome = _retire_terminal_entry(
+            path,
+            observed,
+            board_descriptor=descriptor,
+            quarantine_descriptor=descriptor,
+        )
+        return outcome == "retired"
+    except OSError:
+        return False
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+
+
 def gc_lane_artifacts(
     *,
     now: float | None = None,
@@ -1443,14 +1641,14 @@ def gc_lane_artifacts(
             except OSError:
                 continue
             if not stat.S_ISREG(file_stat.st_mode):
-                path.unlink()
+                if _unlink_artifact_if_same(path, file_stat):
-                removed += 1
+                    removed += 1
-                board_removed += 1
+                    board_removed += 1
                 continue
             if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds:
-                path.unlink()
+                if _unlink_artifact_if_same(path, file_stat):
-                removed += 1
+                    removed += 1
-                board_removed += 1
+                    board_removed += 1
             else:
                 entries.append((path, file_stat))
         retained_count = 0
@@ -1463,9 +1661,9 @@ def gc_lane_artifacts(
                 max_bytes >= 0 and retained_bytes + file_stat.st_size > max_bytes
             )
             if exceeds_count or exceeds_bytes:
-                path.unlink()
+                if _unlink_artifact_if_same(path, file_stat):
-                removed += 1
+                    removed += 1
-                board_removed += 1
+                    board_removed += 1
                 continue
             retained_count += 1
             retained_bytes += file_stat.st_size

testing/tests/test_hermes_cli_lanes.py

@@ -8,6 +8,7 @@ import hashlib
 import json
 import os
 import signal
+import stat
 import sys
 from contextlib import nullcontext
 from pathlib import Path
@@ -1300,8 +1301,11 @@ def test_quarantine_avoids_symlink_and_mode_collision_destinations(
 
     assert not path.exists()
     assert destination.name == f"{base}.2.quarantine"
-    assert destination.read_bytes() == b"invalid"
     assert destination.stat().st_mode & 0o777 == 0o600
+    diagnostic = json.loads(destination.read_text(encoding="utf-8"))
+    assert diagnostic["size"] == len(b"invalid")
+    assert diagnostic["sha256"] == hashlib.sha256(b"invalid").hexdigest()
+    assert diagnostic["source_kind"] == "regular"
     assert victim.read_text(encoding="utf-8") == "unchanged"
     assert collision.read_text(encoding="utf-8") == "attacker collision"
     assert collision.stat().st_mode & 0o777 == 0o644
@@ -1339,6 +1343,241 @@ def test_hardlinked_terminal_source_is_never_chmodded_or_copied_as_data(
     assert "foreign inode contents" not in destination.read_text(encoding="utf-8")
 
 
+def test_quarantine_preserves_and_replays_an_atomic_replacement(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_swap"), 21)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"malformed")
+    identity = lanes._terminal_identity(path)
+    assert identity is not None
+    structured = _completed_result("replacement result")
+    replacement_record = {
+        "board": "cassandra",
+        "task_id": "t_swap",
+        "expected_run_id": 21,
+        "result": json.dumps(structured, sort_keys=True),
+        "summary": structured["summary"],
+        "metadata": {},
+        "kanban_state": "pending",
+        "recorded_at": lanes.utc_now(),
+    }
+    replacement = path.with_name("replacement.tmp")
+    replacement.write_text(
+        json.dumps(replacement_record, sort_keys=True),
+        encoding="utf-8",
+    )
+    replacement.chmod(0o600)
+    real_fsync = lanes.os.fsync
+    swapped = {"value": False}
+
+    def swap_on_quarantine_directory_fsync(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_on_quarantine_directory_fsync)
+
+    lanes._quarantine_terminal(path, identity, "malformed-payload")
+
+    assert swapped["value"] is True
+    assert path.exists()
+    assert lanes._terminal_record_valid(
+        lanes._load_terminal_json(path, identity), identity
+    )
+
+    task = SimpleNamespace(
+        id="t_swap", status="running", current_run_id=21, result=None, 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,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert task.status == "done"
+    assert not path.exists()
+
+
+def test_invalid_utf8_journal_quarantines_and_does_not_stop_later_replay(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    bad = lanes._terminal_path(lanes.state_path("cassandra", "a_bad_utf8"), 31)
+    bad.parent.mkdir(parents=True)
+    bad.write_bytes(b"\xff\xfe\x80not-json")
+    good, _record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "z_good"),
+        board="cassandra",
+        task_id="z_good",
+        run_id=32,
+        structured=_completed_result("valid after invalid UTF-8"),
+        summary="valid after invalid UTF-8",
+        metadata={},
+    )
+    tasks = {
+        "a_bad_utf8": SimpleNamespace(
+            id="a_bad_utf8",
+            status="running",
+            current_run_id=31,
+            result=None,
+            assignee="cli-auto",
+        ),
+        "z_good": SimpleNamespace(
+            id="z_good",
+            status="running",
+            current_run_id=32,
+            result=None,
+            assignee="cli-auto",
+        ),
+    }
+
+    class Connection:
+        def close(self):
+            return None
+
+    def reclaim_task(_conn, task_id, **_kwargs):
+        tasks[task_id].status = "ready"
+        tasks[task_id].current_run_id = None
+        return True
+
+    def complete_task(_conn, task_id, **kwargs):
+        tasks[task_id].status = "done"
+        tasks[task_id].result = kwargs["result"]
+        tasks[task_id].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: tasks[task_id],
+        reclaim_task=reclaim_task,
+        complete_task=complete_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert not bad.exists()
+    assert not good.exists()
+    assert tasks["a_bad_utf8"].status == "ready"
+    assert tasks["z_good"].status == "done"
+    diagnostic_path = next((bad.parent / "quarantine").glob("*.quarantine"))
+    diagnostic = json.loads(diagnostic_path.read_text(encoding="utf-8"))
+    assert diagnostic["size"] == len(b"\xff\xfe\x80not-json")
+    assert diagnostic_path.stat().st_size < 4096
+
+
+@pytest.mark.parametrize(
+    "size",
+    [lanes.MAX_TERMINAL_RECORD_BYTES + 1, 16 * 1024 * 1024],
+)
+def test_oversized_sparse_journal_has_bounded_read_and_quarantine(
+    tmp_path: Path,
+    monkeypatch,
+    size: int,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    huge = lanes._terminal_path(lanes.state_path("cassandra", "a_huge"), 41)
+    huge.parent.mkdir(parents=True)
+    with huge.open("wb") as stream:
+        stream.seek(size - 1)
+        stream.write(b"\0")
+    good, _record = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "z_after_huge"),
+        board="cassandra",
+        task_id="z_after_huge",
+        run_id=42,
+        structured=_completed_result("valid after oversized journal"),
+        summary="valid after oversized journal",
+        metadata={},
+    )
+    tasks = {
+        "a_huge": SimpleNamespace(
+            id="a_huge",
+            status="running",
+            current_run_id=41,
+            result=None,
+            assignee="cli-auto",
+        ),
+        "z_after_huge": SimpleNamespace(
+            id="z_after_huge",
+            status="running",
+            current_run_id=42,
+            result=None,
+            assignee="cli-auto",
+        ),
+    }
+    reads = []
+    real_read_bounded = lanes._read_bounded
+
+    def observe_read_limit(descriptor, limit):
+        reads.append((os.fstat(descriptor).st_size, limit))
+        return real_read_bounded(descriptor, limit)
+
+    monkeypatch.setattr(lanes, "_read_bounded", observe_read_limit)
+
+    class Connection:
+        def close(self):
+            return None
+
+    def reclaim_task(_conn, task_id, **_kwargs):
+        tasks[task_id].status = "ready"
+        tasks[task_id].current_run_id = None
+        return True
+
+    def complete_task(_conn, task_id, **kwargs):
+        tasks[task_id].status = "done"
+        tasks[task_id].result = kwargs["result"]
+        tasks[task_id].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: tasks[task_id],
+        reclaim_task=reclaim_task,
+        complete_task=complete_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert not huge.exists()
+    assert not good.exists()
+    assert tasks["a_huge"].status == "ready"
+    assert tasks["z_after_huge"].status == "done"
+    diagnostics = list((huge.parent / "quarantine").glob("*.quarantine"))
+    assert len(diagnostics) == 1
+    assert diagnostics[0].stat().st_size < 4096
+    diagnostic = json.loads(diagnostics[0].read_text(encoding="utf-8"))
+    assert diagnostic["size"] == size
+    assert diagnostic["hashed_bytes"] == lanes.QUARANTINE_HASH_BYTES
+    assert diagnostic["hash_complete"] is False
+    assert (size, lanes.QUARANTINE_HASH_BYTES) in reads
+    assert (size, lanes.MAX_TERMINAL_RECORD_BYTES + 1) not in reads
+    assert max(limit for _file_size, limit in reads) <= (
+        lanes.MAX_TERMINAL_RECORD_BYTES + 1
+    )
+
+
 def test_recovery_does_not_parse_committed_journals_every_tick(
     tmp_path: Path,
     monkeypatch,
@@ -1453,6 +1692,36 @@ def test_artifact_gc_prunes_oldest_by_count_and_total_bytes(
     assert [path.exists() for path in files] == [False, False, True]
 
 
+def test_artifact_gc_preserves_a_concurrent_replacement(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    board = lanes.STATE_ROOT / "cassandra"
+    board.mkdir(parents=True)
+    path = board / "t.run-1.provider-1.result.json"
+    path.write_text("old", encoding="utf-8")
+    os.utime(path, (100, 100))
+    replacement = board / "replacement.tmp"
+    replacement.write_text("new", encoding="utf-8")
+    real_unlink = lanes._unlink_artifact_if_same
+    swapped = {"value": False}
+
+    def swap_before_identity_checked(candidate, observed):
+        if not swapped["value"]:
+            os.replace(replacement, candidate)
+            swapped["value"] = True
+        return real_unlink(candidate, observed)
+
+    monkeypatch.setattr(lanes, "_unlink_artifact_if_same", swap_before_identity_checked)
+
+    assert lanes.gc_lane_artifacts(
+        now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
+    ) == 0
+    assert swapped["value"] is True
+    assert path.read_text(encoding="utf-8") == "new"
+
+
 def test_artifact_gc_is_interval_gated(
     tmp_path: Path,
     monkeypatch,