From 8e0a13e9dbaa9399286e78dd76ae1f241c01d019 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 20:55:46 -0300 Subject: [PATCH] hermes: close terminal replay edge cases --- .../hermes-execution-safety-regression.py | 62 ++++ dockerfiles/patch-hermes-execution-safety.py | 108 +++++- services/hermes/scripts/cli_lane_runner.py | 259 +++++++++++---- testing/tests/test_hermes_cli_lanes.py | 311 +++++++++++++++++- 4 files changed, 669 insertions(+), 71 deletions(-) diff --git a/dockerfiles/hermes-execution-safety-regression.py b/dockerfiles/hermes-execution-safety-regression.py index a632e51b..a77ca92d 100644 --- a/dockerfiles/hermes-execution-safety-regression.py +++ b/dockerfiles/hermes-execution-safety-regression.py @@ -146,6 +146,68 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): self.assertEqual(completions.calls, 1) self.assertEqual(self._status(task_id), "ready") + def test_latest_ended_run_can_replay_a_durable_completion(self) -> None: + task_id = kanban_db.create_task(self.connection, title="journaled result") + 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.block_task( + self.connection, + task_id, + reason="legacy post-journal failure", + kind="capability", + expected_run_id=run_id, + ) + ) + + self.assertTrue( + kanban_db.complete_task( + self.connection, + task_id, + result="durable terminal result", + summary="durable terminal result", + replay_ended_run_id=run_id, + ) + ) + self.assertEqual(self._status(task_id), "done") + + def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None: + task_id = kanban_db.create_task(self.connection, title="replacement guard") + 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="first run", + kind="capability", + 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_run = kanban_db.get_task(self.connection, task_id).current_run_id + self.assertNotEqual(old_run, replacement_run) + self.assertTrue( + kanban_db.block_task( + self.connection, + task_id, + reason="replacement run", + kind="capability", + expected_run_id=replacement_run, + ) + ) + + self.assertFalse( + kanban_db.complete_task( + self.connection, + task_id, + result="stale result", + replay_ended_run_id=old_run, + ) + ) + self.assertIn(self._status(task_id), {"blocked", "triage"}) + def test_fresh_triage_task_still_auto_promotes(self) -> None: task_id = kanban_db.create_task( self.connection, diff --git a/dockerfiles/patch-hermes-execution-safety.py b/dockerfiles/patch-hermes-execution-safety.py index 621d508d..6abd6f26 100644 --- a/dockerfiles/patch-hermes-execution-safety.py +++ b/dockerfiles/patch-hermes-execution-safety.py @@ -1,4 +1,4 @@ -"""Patch upstream Hermes auto-decomposition to respect execution history.""" +"""Patch upstream Hermes exact-run completion and decomposition safety.""" import os from pathlib import Path @@ -18,6 +18,112 @@ source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes")) db_path = source_root / "hermes_cli/kanban_db.py" db = db_path.read_text(encoding="utf-8") +complete_signature_before = '''def complete_task( + conn: sqlite3.Connection, + task_id: str, + *, + result: Optional[str] = None, + summary: Optional[str] = None, + metadata: Optional[dict] = None, + created_cards: Optional[Iterable[str]] = None, + expected_run_id: Optional[int] = None, +) -> bool: +''' +complete_signature_after = '''def complete_task( + conn: sqlite3.Connection, + task_id: str, + *, + result: Optional[str] = None, + summary: Optional[str] = None, + metadata: Optional[dict] = None, + created_cards: Optional[Iterable[str]] = None, + expected_run_id: Optional[int] = None, + replay_ended_run_id: Optional[int] = None, +) -> bool: +''' +db = replace_once( + db, + complete_signature_before, + complete_signature_after, + "exact ended-run completion signature", +) + +complete_branch_before = ''' with write_txn(conn): + if expected_run_id is None: + cur = conn.execute( + """ + UPDATE tasks + SET status = 'done', + result = ?, + completed_at = ?, + claim_lock = NULL, + claim_expires= NULL, + worker_pid = NULL, + block_kind = NULL, + block_recurrences = 0 + WHERE id = ? + AND status IN ('running', 'ready', 'blocked', 'scheduled') + """, + (result, now, task_id), + ) + else: +''' +complete_branch_after = ''' # Journal recovery may arrive after a legacy error path ended the + # accepted run. Verify that it is still the latest run in this same + # transaction; a replacement attempt must make the replay fail closed. + with write_txn(conn): + if expected_run_id is not None and replay_ended_run_id is not None: + raise ValueError("expected_run_id and replay_ended_run_id are mutually exclusive") + if replay_ended_run_id is not None: + cur = conn.execute( + """ + UPDATE tasks + SET status = 'done', + result = ?, + completed_at = ?, + claim_lock = NULL, + claim_expires= NULL, + worker_pid = NULL, + block_kind = NULL, + block_recurrences = 0 + WHERE id = ? + AND status IN ('ready', 'blocked') + AND current_run_id IS NULL + AND ? = ( + SELECT id FROM task_runs + WHERE task_id = ? + ORDER BY id DESC + LIMIT 1 + ) + """, + (result, now, task_id, int(replay_ended_run_id), task_id), + ) + elif expected_run_id is None: + cur = conn.execute( + """ + UPDATE tasks + SET status = 'done', + result = ?, + completed_at = ?, + claim_lock = NULL, + claim_expires= NULL, + worker_pid = NULL, + block_kind = NULL, + block_recurrences = 0 + WHERE id = ? + AND status IN ('running', 'ready', 'blocked', 'scheduled') + """, + (result, now, task_id), + ) + else: +''' +db = replace_once( + db, + complete_branch_before, + complete_branch_after, + "transactional exact ended-run completion", +) + specify_signature_before = '''def specify_triage_task( conn: sqlite3.Connection, task_id: str, diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 7fa30e34..07e90e48 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -522,10 +522,27 @@ def _terminal_record_valid( structured = json.loads(record["result"]) except (TypeError, json.JSONDecodeError): return False + if not isinstance(structured, dict): + return False + required = set(RESULT_SCHEMA["required"]) + if set(structured) != required: + return False + if ( + type(structured["status"]) is not str + or structured["status"] not in cli_lane_goal.RESULT_STATUSES + or type(structured["summary"]) is not str + or not structured["summary"].strip() + ): + return False + for key in ("changed_files", "tests_run", "artifacts", "findings", "blockers"): + value = structured[key] + if type(value) is not list or any(type(item) is not str for item in value): + return False valid = ( - isinstance(structured, dict) - and structured.get("status") == "completed" - and not structured.get("blockers") + structured["status"] == "completed" + and structured["blockers"] == [] + and cli_lane_goal.unfinished_result_reason(structured) is None + and record["summary"] == structured["summary"] and record.get("kanban_state") in {"pending", "committed"} ) if not valid or identity is None: @@ -1077,7 +1094,7 @@ def _board_call( def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path: - """Move an unusable journal aside with a deterministic, private name.""" + """Copy an unusable journal to a private file, then retire its exact name.""" board_dir = path.parent try: board_stat = board_dir.stat(follow_symlinks=False) @@ -1090,44 +1107,135 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: flush=True, ) return path - quarantine = board_dir / "quarantine" - quarantine.mkdir(parents=True, exist_ok=True, mode=0o700) - os.chmod(quarantine, 0o700, follow_symlinks=False) + board_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + board_flags |= getattr(os, "O_NOFOLLOW", 0) + file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + create_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + create_flags |= getattr(os, "O_NOFOLLOW", 0) + board_descriptor = None + quarantine_descriptor = None + source_stat = None + destination = path + retired = False try: - source_stat = path.stat(follow_symlinks=False) - regular = stat.S_ISREG(source_stat.st_mode) - except OSError: - regular = False - try: - if not regular: - raise OSError("journal is not a regular file") - flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) - descriptor = os.open(path, flags) - with os.fdopen(descriptor, "rb") as stream: - fingerprint = hashlib.sha256(stream.read()).hexdigest()[:16] - except OSError: - fingerprint = hashlib.sha256(path.name.encode("utf-8")).hexdigest()[:16] - safe_reason = re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid" - destination = quarantine / f"{path.name}.{safe_reason}.{fingerprint}.quarantine" - source_exists = path.exists() or path.is_symlink() - if source_exists and regular and not destination.exists(): - os.chmod(path, 0o600, follow_symlinks=False) - os.replace(path, destination) - os.chmod(destination, 0o600, follow_symlinks=False) - _fsync_directory(board_dir) - _fsync_directory(quarantine) - elif source_exists: - path.unlink() - _fsync_directory(board_dir) - if not destination.exists(): - atomic_json( - destination, - { - "original_name": path.name, - "reason": reason, - "recorded_at": utc_now(), - }, + board_descriptor = os.open(board_dir, board_flags) + source_stat = os.stat( + path.name, + dir_fd=board_descriptor, + follow_symlinks=False, + ) + safe_source = stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink == 1 + payload: bytes + 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] + safe_reason = ( + re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid" + ) + try: + os.mkdir("quarantine", 0o700, dir_fd=board_descriptor) + except FileExistsError: + pass + quarantine_descriptor = os.open( + "quarantine", + board_flags, + dir_fd=board_descriptor, + ) + for sequence in range(32): + name = f"{path.name}.{safe_reason}.{fingerprint}.{sequence}.quarantine" + try: + destination_descriptor = os.open( + name, + create_flags, + 0o600, + dir_fd=quarantine_descriptor, + ) + except FileExistsError: + continue + try: + with os.fdopen(destination_descriptor, "wb") as target: + target.write(payload) + target.flush() + os.fsync(target.fileno()) + target_stat = os.fstat(target.fileno()) + if ( + stat.S_IMODE(target_stat.st_mode) != 0o600 + or target_stat.st_nlink != 1 + ): + raise OSError("quarantine destination is not private and singly linked") + except Exception: + os.unlink(name, dir_fd=quarantine_descriptor) + raise + destination = board_dir / "quarantine" / name + break + 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, + ) + 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: + try: + os.unlink(path.name, dir_fd=board_descriptor) + retired = True + os.fsync(board_descriptor) + except OSError: + pass + print( + f"terminal journal quarantine degraded safely: {type(error).__name__}: {error}", + file=sys.stderr, + flush=True, + ) + finally: + 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'}", @@ -1137,12 +1245,12 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: return destination -def _recover_quarantined_run( +def _recover_exact_run( kanban_db: Any, identity: TerminalIdentity | None, reason: str, ) -> bool: - """Make an exact external run retryable after its journal is quarantined.""" + """Make an exact external run retryable after journal recovery fails.""" if identity is None: return False @@ -1160,7 +1268,7 @@ def _recover_quarantined_run( kanban_db.reclaim_task( conn, identity.task_id, - reason=f"terminal journal quarantined ({reason}); exact run may retry", + reason=f"terminal journal recovery failed ({reason}); exact run may retry", ) ) @@ -1172,7 +1280,7 @@ def _recover_quarantined_run( if recovered: print( f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} " - f"after terminal journal quarantine: {reason}", + f"after terminal journal recovery: {reason}", file=sys.stderr, flush=True, ) @@ -1221,10 +1329,16 @@ def _finalize_terminal_record( if str(_task_value(task, "result", "") or "") == document["result"] else "stale" ) - if ( - status != "running" - or _task_value(task, "current_run_id", None) != identity.run_id - ): + current_run_id = _task_value(task, "current_run_id", None) + completion_guard: dict[str, int] + if status == "running" and current_run_id == identity.run_id: + completion_guard = {"expected_run_id": identity.run_id} + elif status in {"ready", "blocked"} and current_run_id is None: + # The journal may have survived an older post-persistence error + # path that ended its run. The patched DB verifies atomically that + # this is still the latest ended run before allowing completion. + completion_guard = {"replay_ended_run_id": identity.run_id} + else: return "stale" completed = kanban_db.complete_task( conn, @@ -1232,7 +1346,7 @@ def _finalize_terminal_record( result=document["result"], summary=document["summary"], metadata=document["metadata"], - expected_run_id=identity.run_id, + **completion_guard, ) return "committed" if completed else "pending" @@ -1255,11 +1369,11 @@ def recover_pending_finalizations() -> int: continue if not _terminal_record_valid(record): _quarantine_terminal(path, identity, "malformed-payload") - _recover_quarantined_run(kanban_db, identity, "malformed-payload") + _recover_exact_run(kanban_db, identity, "malformed-payload") continue if not _terminal_record_valid(record, identity): _quarantine_terminal(path, identity, "foreign-identity") - _recover_quarantined_run(kanban_db, identity, "foreign-identity") + _recover_exact_run(kanban_db, identity, "foreign-identity") continue try: outcome = _finalize_terminal_record(kanban_db, path, record) @@ -1270,7 +1384,7 @@ def recover_pending_finalizations() -> int: recovered += 1 elif outcome in {"invalid", "foreign", "stale"}: _quarantine_terminal(path, identity, outcome) - _recover_quarantined_run(kanban_db, identity, outcome) + _recover_exact_run(kanban_db, identity, outcome) return recovered @@ -1329,6 +1443,9 @@ def gc_lane_artifacts( except OSError: continue if not stat.S_ISREG(file_stat.st_mode): + path.unlink() + removed += 1 + board_removed += 1 continue if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds: path.unlink() @@ -1631,28 +1748,44 @@ def execute_claim(board: str, task_id: str) -> None: and result.returncode == 0 and completion_problem is None ): + terminal_file = _terminal_path(state_file, run_id, "pending") metadata["terminal_record"] = str( _terminal_path(state_file, run_id, "committed") ) - terminal_file, terminal_record = _write_terminal_record( - state_file, - board=board, - task_id=task_id, - run_id=run_id, - structured=structured, - summary=str(structured.get("summary") or "Completed"), - metadata=metadata, - ) try: + terminal_file, terminal_record = _write_terminal_record( + state_file, + board=board, + task_id=task_id, + run_id=run_id, + structured=structured, + summary=str(structured.get("summary") or "Completed"), + metadata=metadata, + ) outcome = _finalize_terminal_record( kanban_db, terminal_file, terminal_record, ) except Exception as error: + identity = _terminal_identity(terminal_file) + replayable = bool( + identity is not None + and _terminal_record_valid( + _load_terminal_json(terminal_file, identity), + identity, + ) + ) + if not replayable: + _recover_exact_run( + kanban_db, + identity, + f"persistence raised {type(error).__name__}", + ) raise TerminalFinalizationPending( - "accepted worker result is durably journaled; " - f"Kanban finalization raised {type(error).__name__}" + "accepted worker result remains outside the generic block path; " + f"terminal replayable={replayable}; persistence/finalization " + f"raised {type(error).__name__}" ) from error if outcome != "committed": raise TerminalFinalizationPending( diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index c4ae715a..23c54027 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -2,7 +2,9 @@ from __future__ import annotations +import errno import importlib.util +import hashlib import json import os import signal @@ -864,6 +866,55 @@ def _completed_result(summary: str = "done") -> dict: } +def test_terminal_recovery_can_complete_the_exact_latest_ended_run( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + state_file = lanes.state_path("cassandra", "t_ended") + pending, _record = lanes._write_terminal_record( + state_file, + board="cassandra", + task_id="t_ended", + run_id=17, + structured=_completed_result("accepted before legacy block"), + summary="accepted before legacy block", + metadata={}, + ) + task = SimpleNamespace( + id="t_ended", + status="blocked", + result=None, + current_run_id=None, + assignee="cli-auto", + ) + guards = [] + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + guards.append(kwargs) + task.status = "done" + task.result = 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 guards[0]["replay_ended_run_id"] == 17 + assert "expected_run_id" not in guards[0] + assert task.status == "done" + assert not pending.exists() + + def test_complete_exception_after_journal_is_replayable_and_never_blocks( tmp_path: Path, monkeypatch, @@ -937,6 +988,103 @@ def test_complete_exception_after_journal_is_replayable_and_never_blocks( assert not pending[0].exists() +def test_terminal_replace_then_directory_fsync_enospc_replays_after_restart( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + task = SimpleNamespace( + id="t_enospc", + status="running", + result=None, + current_run_id=52, + assignee="cli-auto", + max_runtime_seconds=60, + ) + blocks = [] + comments = [] + completions = [] + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + completions.append(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, + worker_log_path=lambda _task_id, board: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_enospc"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "Finish without losing the result.", + heartbeat_worker=lambda *_args, **_kwargs: True, + add_comment=lambda _conn, _task_id, _author, body: comments.append(body), + complete_task=complete_task, + block_task=lambda *_args, **kwargs: blocks.append(kwargs), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr( + lanes, + "select_route", + lambda *_args, **_kwargs: lanes.Route( + "codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, () + ), + ) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult( + 0, "", _completed_result("persisted before ENOSPC"), False + ), + ) + real_replace = lanes.os.replace + real_fsync_directory = lanes._fsync_directory + terminal_replaced = {"value": False} + fail_once = {"value": True} + + def replace_then_mark(source, destination): + real_replace(source, destination) + if str(destination).endswith(".terminal.pending.json"): + terminal_replaced["value"] = True + + def fail_after_terminal_replace(directory): + if terminal_replaced["value"] and fail_once["value"]: + fail_once["value"] = False + raise OSError(errno.ENOSPC, "no space after terminal rename") + real_fsync_directory(directory) + + monkeypatch.setattr(lanes.os, "replace", replace_then_mark) + monkeypatch.setattr(lanes, "_fsync_directory", fail_after_terminal_replace) + + lanes.execute_claim("cassandra", "t_enospc") + + pending = list((state_root / "cassandra").glob("*.terminal.pending.json")) + assert terminal_replaced["value"] is True + assert fail_once["value"] is False + assert len(pending) == 1 + assert blocks == [] + assert completions == [] + assert task.status == "running" + assert any("terminal replayable=True" in body for body in comments) + + # A restarted runner sees the exact pending journal and completes the run. + assert lanes.recover_pending_finalizations() == 1 + assert len(completions) == 1 + assert task.status == "done" + assert not pending[0].exists() + committed = list((state_root / "cassandra").glob("*.terminal.committed.json")) + assert len(committed) == 1 + + def test_terminal_payload_cannot_select_a_different_board( tmp_path: Path, monkeypatch, @@ -1050,6 +1198,147 @@ def test_malformed_exact_run_journal_is_quarantined_and_reclaimed( assert quarantined[0].stat().st_mode & 0o777 == 0o600 +@pytest.mark.parametrize( + "structured", + [ + {"status": "completed", "blockers": {}}, + {**_completed_result(), "changed_files": "src/a.py"}, + {**_completed_result(), "unexpected": True}, + {**_completed_result(), "blockers": ["work remains"]}, + {**_completed_result("tests are still running")}, + ], +) +def test_terminal_record_requires_exact_completed_result_contract(structured): + record = { + "board": "cassandra", + "task_id": "t_schema", + "expected_run_id": 1, + "result": json.dumps(structured), + "summary": str(structured.get("summary") or "done"), + "metadata": {}, + "kanban_state": "pending", + } + + assert lanes._terminal_record_valid(record) is False + + +def test_minimal_completed_result_with_mapping_blockers_quarantines_without_completion( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_schema"), 6) + lanes.atomic_json( + path, + { + "board": "cassandra", + "task_id": "t_schema", + "expected_run_id": 6, + "result": json.dumps({"status": "completed", "blockers": {}}), + "summary": "done", + "metadata": {}, + "kanban_state": "pending", + }, + ) + task = SimpleNamespace( + id="t_schema", status="running", current_run_id=6, assignee="cli-auto" + ) + completed = [] + reclaimed = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=lambda *_args, **_kwargs: completed.append(True), + reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 0 + assert completed == [] + assert reclaimed == [True] + assert not path.exists() + quarantined = list((path.parent / "quarantine").glob("*.quarantine")) + assert len(quarantined) == 1 + assert quarantined[0].stat().st_mode & 0o777 == 0o600 + + +def test_quarantine_avoids_symlink_and_mode_collision_destinations( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_collision"), 2) + path.parent.mkdir(parents=True) + path.write_bytes(b"invalid") + quarantine = path.parent / "quarantine" + quarantine.mkdir() + fingerprint = hashlib.sha256(b"invalid").hexdigest()[:16] + base = f"{path.name}.malformed-payload.{fingerprint}" + victim = tmp_path / "victim" + victim.write_text("unchanged", encoding="utf-8") + (quarantine / f"{base}.0.quarantine").symlink_to(victim) + collision = quarantine / f"{base}.1.quarantine" + collision.write_text("attacker collision", encoding="utf-8") + collision.chmod(0o644) + monkeypatch.setattr( + lanes.os, + "chmod", + lambda *_args, **_kwargs: pytest.fail("quarantine must not chmod foreign inodes"), + ) + + destination = lanes._quarantine_terminal( + path, + lanes._terminal_identity(path), + "malformed-payload", + ) + + assert not path.exists() + assert destination.name == f"{base}.2.quarantine" + assert destination.read_bytes() == b"invalid" + assert destination.stat().st_mode & 0o777 == 0o600 + assert victim.read_text(encoding="utf-8") == "unchanged" + assert collision.read_text(encoding="utf-8") == "attacker collision" + assert collision.stat().st_mode & 0o777 == 0o644 + + +def test_hardlinked_terminal_source_is_never_chmodded_or_copied_as_data( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_hardlink"), 5) + path.parent.mkdir(parents=True) + foreign = tmp_path / "foreign" + foreign.write_text("foreign inode contents", encoding="utf-8") + foreign.chmod(0o644) + os.link(foreign, path) + monkeypatch.setattr( + lanes.os, + "chmod", + lambda *_args, **_kwargs: pytest.fail("hardlinked source must not be chmodded"), + ) + + destination = lanes._quarantine_terminal( + path, + lanes._terminal_identity(path), + "malformed-payload", + ) + + assert not path.exists() + assert foreign.read_text(encoding="utf-8") == "foreign inode contents" + assert foreign.stat().st_mode & 0o777 == 0o644 + assert destination.stat().st_mode & 0o777 == 0o600 + diagnostic = json.loads(destination.read_text(encoding="utf-8")) + assert diagnostic["source_kind"] == "hardlink" + assert "foreign inode contents" not in destination.read_text(encoding="utf-8") + + def test_recovery_does_not_parse_committed_journals_every_tick( tmp_path: Path, monkeypatch, @@ -1127,6 +1416,8 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals( ] for artifact in old_artifacts: artifact.write_text("{}", encoding="utf-8") + quarantine_symlink = quarantine / "attacker.quarantine" + quarantine_symlink.symlink_to(tmp_path / "missing-target") pending = board / "t.run-1.terminal.pending.json" pending.write_text("{}", encoding="utf-8") old_time = 100.0 @@ -1136,8 +1427,9 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals( assert lanes.gc_lane_artifacts( now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000 - ) == len(old_artifacts) + ) == len(old_artifacts) + 1 assert not any(artifact.exists() for artifact in old_artifacts) + assert not quarantine_symlink.is_symlink() assert pending.exists() @@ -1189,9 +1481,10 @@ def test_artifact_gc_is_interval_gated( "status": "completed", "summary": "done", "changed_files": ["src/a.py"], - "tests_run": ["pytest -q"], - "artifacts": ["reports/result.json"], - "blockers": [], + "tests_run": ["pytest -q"], + "artifacts": ["reports/result.json"], + "findings": [], + "blockers": [], }, False, ), @@ -1205,9 +1498,10 @@ def test_artifact_gc_is_interval_gated( "status": "completed", "summary": "The full test suite is still running.", "changed_files": ["src/a.py"], - "tests_run": ["pytest -q — in progress"], - "artifacts": [], - "blockers": [], + "tests_run": ["pytest -q — in progress"], + "artifacts": [], + "findings": [], + "blockers": [], }, False, ), @@ -1369,6 +1663,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress( "changed_files": ["src/a.py"], "tests_run": ["pytest focused: passed"], "artifacts": [], + "findings": [], "blockers": [], }, False, @@ -1382,6 +1677,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress( "changed_files": ["src/a.py"], "tests_run": ["pytest full: passed"], "artifacts": [], + "findings": [], "blockers": [], }, False, @@ -1614,6 +1910,7 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P "changed_files": [], "tests_run": [], "artifacts": [], + "findings": [], "blockers": [], }, False,