From 8c71d55853923965bd39b3084f7a92b57035a63c Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 21:32:03 -0300 Subject: [PATCH] hermes: harden journal quarantine races --- services/hermes/scripts/cli_lane_runner.py | 336 ++++++++++++++++----- testing/tests/test_hermes_cli_lanes.py | 271 ++++++++++++++++- 2 files changed, 537 insertions(+), 70 deletions(-) diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 07e90e48..30c84daf 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/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: - descriptor = None - value = json.load(stream) - except (OSError, json.JSONDecodeError): + opened = os.fstat(descriptor) + if ( + not stat.S_ISREG(opened.st_mode) + 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.fileno()) - if ( - opened_stat.st_dev != source_stat.st_dev - or opened_stat.st_ino != source_stat.st_ino - or opened_stat.st_nlink != 1 - ): - raise OSError("terminal journal identity changed while opening") - payload = source.read() - else: - payload = ( - json.dumps( - { - "original_name": path.name, - "reason": reason, - "source_kind": "hardlink" - if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1 - else "non-regular", - }, - sort_keys=True, - ) - + "\n" - ).encode("utf-8") - fingerprint = hashlib.sha256(payload).hexdigest()[:16] + opened_stat = os.fstat(source_descriptor) + if ( + opened_stat.st_dev != source_stat.st_dev + or opened_stat.st_ino != source_stat.st_ino + or opened_stat.st_nlink != 1 + ): + raise OSError("terminal journal identity changed while opening") + # Keep this descriptor open through retirement. The staging entry + # must match the exact fstat-verified inode, not merely a path stat + # captured before a concurrent replacement. + source_stat = opened_stat + prefix = _read_bounded(source_descriptor, QUARANTINE_HASH_BYTES) + hashed_bytes = len(prefix) + hash_complete = opened_stat.st_size <= QUARANTINE_HASH_BYTES + digest = hashlib.sha256(prefix).hexdigest() + source_kind = ( + "regular" + if safe_source + else "hardlink" + if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1 + else "non-regular" + ) + diagnostic = { + "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( - path.name, - dir_fd=board_descriptor, - follow_symlinks=False, + retirement = _retire_terminal_entry( + path, + source_stat, + 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; - # never chmod or write through the source inode. - if board_descriptor is not None and not retired: + # path eligible for replay forever. The same identity-safe staging + # protocol works in the board directory when quarantine is unusable. + if board_descriptor is not None and source_stat is not None: try: - os.unlink(path.name, dir_fd=board_descriptor) - retired = True - os.fsync(board_descriptor) + retirement = _retire_terminal_entry( + path, + 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() - removed += 1 - board_removed += 1 + if _unlink_artifact_if_same(path, file_stat): + removed += 1 + board_removed += 1 continue if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds: - path.unlink() - removed += 1 - board_removed += 1 + if _unlink_artifact_if_same(path, file_stat): + 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() - removed += 1 - board_removed += 1 + if _unlink_artifact_if_same(path, file_stat): + removed += 1 + board_removed += 1 continue retained_count += 1 retained_bytes += file_stat.st_size diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 2db9b1ae..94d9cb58 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/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,