From 18be5957009213f959c4ae7caf8dc2e6636057ec Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 20:27:12 -0300 Subject: [PATCH] hermes: harden replay and decomposition races --- .../hermes-execution-safety-regression.py | 45 +- dockerfiles/patch-hermes-execution-safety.py | 167 ++++++- services/hermes/scripts/cli_lane_runner.py | 472 +++++++++++++++--- testing/tests/test_hermes_cli_lanes.py | 438 +++++++++++++++- 4 files changed, 1019 insertions(+), 103 deletions(-) diff --git a/dockerfiles/hermes-execution-safety-regression.py b/dockerfiles/hermes-execution-safety-regression.py index e21bd703..a632e51b 100644 --- a/dockerfiles/hermes-execution-safety-regression.py +++ b/dockerfiles/hermes-execution-safety-regression.py @@ -167,22 +167,45 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): self.assertEqual(completions.calls, 1) self.assertEqual(self._status(task_id), "ready") - def test_execution_history_gained_during_llm_call_blocks_commit(self) -> None: + def _insert_run(self, task_id: str) -> None: + with kanban_db.connect_closing() as connection: + kanban_db._synthesize_ended_run( + connection, + task_id, + outcome="reclaimed", + summary="concurrent execution evidence", + ) + + def test_execution_history_gained_during_single_llm_call_blocks_commit(self) -> None: + task_id = kanban_db.create_task( + self.connection, + title="concurrent single objective", + triage=True, + ) + completions = self._client( + { + "fanout": False, + "title": "redundant rewrite", + "body": "must not be committed", + "assignee": "default", + }, + before_response=lambda: self._insert_run(task_id), + ) + + outcome = kanban_decompose.decompose_task(task_id, automatic=True) + + self.assertFalse(outcome.ok) + self.assertIn("gained execution history", outcome.reason) + self.assertEqual(completions.calls, 1) + self.assertEqual(self._status(task_id), "triage") + + def test_execution_history_gained_during_fanout_llm_call_blocks_commit(self) -> None: task_id = kanban_db.create_task( self.connection, title="concurrent triage objective", triage=True, ) - def add_run() -> None: - with kanban_db.connect_closing() as connection: - kanban_db._synthesize_ended_run( - connection, - task_id, - outcome="reclaimed", - summary="concurrent execution evidence", - ) - completions = self._client( { "fanout": True, @@ -195,7 +218,7 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): } ], }, - before_response=add_run, + before_response=lambda: self._insert_run(task_id), ) before = len(kanban_db.list_tasks(self.connection)) diff --git a/dockerfiles/patch-hermes-execution-safety.py b/dockerfiles/patch-hermes-execution-safety.py index bcf93c50..621d508d 100644 --- a/dockerfiles/patch-hermes-execution-safety.py +++ b/dockerfiles/patch-hermes-execution-safety.py @@ -15,6 +15,106 @@ def replace_once(source: str, before: str, after: str, label: str) -> str: 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") + +specify_signature_before = '''def specify_triage_task( + conn: sqlite3.Connection, + task_id: str, + *, + title: Optional[str] = None, + body: Optional[str] = None, + assignee: Optional[str] = None, + author: Optional[str] = None, +) -> bool: +''' +specify_signature_after = '''def specify_triage_task( + conn: sqlite3.Connection, + task_id: str, + *, + title: Optional[str] = None, + body: Optional[str] = None, + assignee: Optional[str] = None, + author: Optional[str] = None, + require_no_runs: bool = False, +) -> bool: +''' +db = replace_once( + db, + specify_signature_before, + specify_signature_after, + "transactional specify signature", +) + +specify_guard_before = ''' if existing is None: + return False + sets: list[str] = ["status = 'todo'"] +''' +specify_guard_after = ''' if existing is None: + return False + if require_no_runs and conn.execute( + "SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1", + (task_id,), + ).fetchone() is not None: + return False + sets: list[str] = ["status = 'todo'"] +''' +db = replace_once( + db, + specify_guard_before, + specify_guard_after, + "transactional specify execution-history guard", +) + +decompose_signature_before = '''def decompose_triage_task( + conn: sqlite3.Connection, + task_id: str, + *, + root_assignee: Optional[str], + children: list[dict], + author: Optional[str] = None, + auto_promote: bool = True, +) -> Optional[list[str]]: +''' +decompose_signature_after = '''def decompose_triage_task( + conn: sqlite3.Connection, + task_id: str, + *, + root_assignee: Optional[str], + children: list[dict], + author: Optional[str] = None, + auto_promote: bool = True, + require_no_runs: bool = False, +) -> Optional[list[str]]: +''' +db = replace_once( + db, + decompose_signature_before, + decompose_signature_after, + "transactional decomposition signature", +) + +decompose_guard_before = ''' if root_row["status"] != "triage": + return None + tenant = root_row["tenant"] +''' +decompose_guard_after = ''' if root_row["status"] != "triage": + return None + if require_no_runs and conn.execute( + "SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1", + (task_id,), + ).fetchone() is not None: + return None + tenant = root_row["tenant"] +''' +db = replace_once( + db, + decompose_guard_before, + decompose_guard_after, + "transactional decomposition execution-history guard", +) +db_path.write_text(db, encoding="utf-8") + decompose_path = source_root / "hermes_cli/kanban_decompose.py" decompose = decompose_path.read_text(encoding="utf-8") @@ -87,17 +187,20 @@ decompose = replace_once( "auto-decompose preflight guard", ) -single_before = ''' with kb.connect_closing() as conn: - ok = kb.specify_triage_task( -''' -single_after = ''' if automatic and _has_execution_history(task_id): - return DecomposeOutcome( - task_id, - False, - "task gained execution history and requires deliberate manual triage", +single_before = ''' author=audit_author, ) - with kb.connect_closing() as conn: - ok = kb.specify_triage_task( + if not ok: +''' +single_after = ''' author=audit_author, + require_no_runs=automatic, + ) + if not ok: + if automatic and _has_execution_history(task_id): + return DecomposeOutcome( + task_id, + False, + "task gained execution history and requires deliberate manual triage", + ) ''' decompose = replace_once( decompose, @@ -106,25 +209,43 @@ decompose = replace_once( "auto-decompose single-task commit guard", ) -fanout_before = ''' try: - with kb.connect_closing() as conn: - child_ids = kb.decompose_triage_task( +fanout_before = ''' author=audit_author, + auto_promote=auto_promote, + ) ''' -fanout_after = ''' if automatic and _has_execution_history(task_id): - return DecomposeOutcome( - task_id, - False, - "task gained execution history and requires deliberate manual triage", - ) - try: - with kb.connect_closing() as conn: - child_ids = kb.decompose_triage_task( +fanout_after = ''' author=audit_author, + auto_promote=auto_promote, + require_no_runs=automatic, + ) ''' decompose = replace_once( decompose, fanout_before, fanout_after, - "auto-decompose fanout commit guard", + "auto-decompose transactional fanout guard", +) + +fanout_outcome_before = ''' if child_ids is None: + return DecomposeOutcome( + task_id, False, "task moved out of triage before decomposition", + ) +''' +fanout_outcome_after = ''' if child_ids is None: + if automatic and _has_execution_history(task_id): + return DecomposeOutcome( + task_id, + False, + "task gained execution history and requires deliberate manual triage", + ) + return DecomposeOutcome( + task_id, False, "task moved out of triage before decomposition", + ) +''' +decompose = replace_once( + decompose, + fanout_outcome_before, + fanout_outcome_after, + "auto-decompose fanout rejection reason", ) decompose_path.write_text(decompose, encoding="utf-8") diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index c26f26ca..7fa30e34 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -4,12 +4,14 @@ from __future__ import annotations import concurrent.futures +import hashlib import json import os import re import selectors import signal import sqlite3 +import stat import subprocess import sys import threading @@ -43,12 +45,17 @@ HEARTBEAT_SECONDS = 20 PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60 PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60 KANBAN_STORAGE_ATTEMPTS = 5 +ARTIFACT_GC_INTERVAL_SECONDS = 5 * 60 +ARTIFACT_RETENTION_AGE_SECONDS = 30 * 24 * 60 * 60 +ARTIFACT_RETENTION_COUNT = 256 +ARTIFACT_RETENTION_BYTES = 128 * 1024 * 1024 PROVIDER_HEALTH_PATHS = { "codex": DATA_ROOT / "provider-health/codex.json", "claude": DATA_ROOT / "provider-health/claude.json", } WORKTREE_LOCK = threading.Lock() BOARD_CORRUPTION_ERRORS: dict[str, str] = {} +LAST_ARTIFACT_GC = 0.0 CAPACITY_PATTERN = re.compile( r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529|authentication|oauth|token.*expired)", re.I, @@ -127,22 +134,65 @@ class TerminalFinalizationPending(RuntimeError): """An accepted worker result is durable but not committed to Kanban yet.""" +@dataclass(frozen=True) +class TerminalIdentity: + """Replay authority derived only from a journal's directory and filename.""" + + board: str + task_id: str + run_id: int + state: str + + def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None: - """Durably replace a small non-secret state document.""" - path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") - temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") - temporary.chmod(mode) - os.replace(temporary, path) + """Durably replace a small state document without following temp symlinks.""" + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(path.parent, 0o700, follow_symlinks=False) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = None + try: + descriptor = os.open(temporary, flags, mode) + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + descriptor = None + json.dump(value, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, mode, follow_symlinks=False) + os.replace(temporary, path) + _fsync_directory(path.parent) + finally: + if descriptor is not None: + os.close(descriptor) + try: + temporary.unlink() + except FileNotFoundError: + pass + + +def _fsync_directory(directory: Path) -> None: + """Persist directory-entry changes after an atomic rename or quarantine.""" + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(directory, flags) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) def load_json(path: Path) -> dict[str, Any]: try: - value = json.loads(path.read_text(encoding="utf-8")) + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + with os.fdopen(descriptor, "r", encoding="utf-8") as stream: + value = json.load(stream) except (OSError, json.JSONDecodeError): return {} return value if isinstance(value, dict) else {} @@ -313,10 +363,80 @@ def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path: ) -def _terminal_path(state_file: Path, run_id: Any) -> Path: - """Return the replay journal path for an accepted terminal response.""" +def _terminal_path( + state_file: Path, + run_id: Any, + state: str = "pending", +) -> Path: + """Return the identity-bound replay path for an accepted terminal response.""" + if state not in {"pending", "committed"}: + raise ValueError(f"invalid terminal journal state: {state}") safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown")) - return state_file.with_name(f"{state_file.stem}.run-{safe_run}.terminal.json") + return state_file.with_name( + f"{state_file.stem}.run-{safe_run}.terminal.{state}.json" + ) + + +_TERMINAL_NAME = re.compile( + r"^(?P[a-zA-Z0-9_.-]+)\.run-(?P[0-9]+)\." + r"terminal\.(?Ppending|committed)\.json$" +) +_SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$") + + +def _terminal_identity(path: Path) -> TerminalIdentity | None: + """Parse replay authority lexically before opening a journal or board.""" + try: + relative = path.relative_to(STATE_ROOT) + except ValueError: + return None + if len(relative.parts) != 2: + return None + board, filename = relative.parts + match = _TERMINAL_NAME.fullmatch(filename) + try: + board_stat = path.parent.stat(follow_symlinks=False) + except OSError: + return None + if ( + not match + or not stat.S_ISDIR(board_stat.st_mode) + or not _SAFE_BOARD.fullmatch(board) + or board in {".", ".."} + or match.group("task") in {".", ".."} + ): + return None + return TerminalIdentity( + board=board, + task_id=match.group("task"), + run_id=int(match.group("run")), + state=match.group("state"), + ) + + +def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]: + """Read through the already-validated board directory without symlink hops.""" + if _terminal_identity(path) != identity: + return {} + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + directory_flags |= getattr(os, "O_NOFOLLOW", 0) + file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + directory = None + descriptor = None + 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): + return {} + finally: + if descriptor is not None: + os.close(descriptor) + if directory is not None: + os.close(directory) + return value if isinstance(value, dict) else {} def _persist_candidate( @@ -381,7 +501,10 @@ def _write_terminal_record( return path, record -def _terminal_record_valid(record: dict[str, Any]) -> bool: +def _terminal_record_valid( + record: dict[str, Any], + identity: TerminalIdentity | None = None, +) -> bool: """Reject malformed or non-terminal replay journals without side effects.""" if not isinstance(record, dict): return False @@ -399,10 +522,18 @@ def _terminal_record_valid(record: dict[str, Any]) -> bool: structured = json.loads(record["result"]) except (TypeError, json.JSONDecodeError): return False - return ( + valid = ( isinstance(structured, dict) and structured.get("status") == "completed" and not structured.get("blockers") + and record.get("kanban_state") in {"pending", "committed"} + ) + if not valid or identity is None: + return valid + return ( + record.get("board") == identity.board + and record.get("task_id") == identity.task_id + and record.get("expected_run_id") == identity.run_id ) @@ -945,21 +1076,142 @@ def _board_call( raise last_storage_error +def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path: + """Move an unusable journal aside with a deterministic, private name.""" + board_dir = path.parent + try: + board_stat = board_dir.stat(follow_symlinks=False) + except OSError: + board_stat = None + if board_stat is None or not stat.S_ISDIR(board_stat.st_mode): + print( + f"refused terminal journal path through unsafe board directory: {path}", + file=sys.stderr, + flush=True, + ) + return path + quarantine = board_dir / "quarantine" + quarantine.mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(quarantine, 0o700, follow_symlinks=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(), + }, + ) + print( + "quarantined terminal journal " + f"{path.name}: {reason}; identity={identity or 'unparseable'}", + file=sys.stderr, + flush=True, + ) + return destination + + +def _recover_quarantined_run( + kanban_db: Any, + identity: TerminalIdentity | None, + reason: str, +) -> bool: + """Make an exact external run retryable after its journal is quarantined.""" + if identity is None: + return False + + def operation(conn: Any) -> bool: + task = kanban_db.get_task(conn, identity.task_id) + if task is None: + return False + if ( + str(_task_value(task, "status", "")) != "running" + or _task_value(task, "current_run_id", None) != identity.run_id + or not _external(task) + ): + return False + return bool( + kanban_db.reclaim_task( + conn, + identity.task_id, + reason=f"terminal journal quarantined ({reason}); exact run may retry", + ) + ) + + try: + recovered = bool(_board_call(kanban_db, identity.board, operation)) + except Exception as error: + _record_board_access_error(identity.board, error) + return False + if recovered: + print( + f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} " + f"after terminal journal quarantine: {reason}", + file=sys.stderr, + flush=True, + ) + return recovered + + +def _commit_terminal_file(path: Path, identity: TerminalIdentity, document: dict[str, Any]) -> Path: + """Persist committed state, then atomically retire a pending journal.""" + committed = _terminal_path( + state_path(identity.board, identity.task_id), + identity.run_id, + "committed", + ) + document["kanban_state"] = "committed" + document["committed_at"] = utc_now() + atomic_json(path, document) + os.replace(path, committed) + os.chmod(committed, 0o600, follow_symlinks=False) + _fsync_directory(committed.parent) + return committed + + def _finalize_terminal_record( kanban_db: Any, path: Path, - record: dict[str, Any] | None = None, + _record: dict[str, Any] | None = None, ) -> str: """Commit one exact-run terminal journal, or leave it safely pending.""" - document = dict(record or load_json(path)) + identity = _terminal_identity(path) + if identity is None or identity.state != "pending": + return "invalid" + document = _load_terminal_json(path, identity) if not _terminal_record_valid(document): return "invalid" - board = str(document["board"]) - task_id = str(document["task_id"]) - expected_run_id = int(document["expected_run_id"]) + if not _terminal_record_valid(document, identity): + return "foreign" def operation(conn: Any) -> str: - task = kanban_db.get_task(conn, task_id) + task = kanban_db.get_task(conn, identity.task_id) if task is None: return "stale" status = str(_task_value(task, "status", "")) @@ -971,24 +1223,22 @@ def _finalize_terminal_record( ) if ( status != "running" - or _task_value(task, "current_run_id", None) != expected_run_id + or _task_value(task, "current_run_id", None) != identity.run_id ): return "stale" completed = kanban_db.complete_task( conn, - task_id, + identity.task_id, result=document["result"], summary=document["summary"], metadata=document["metadata"], - expected_run_id=expected_run_id, + expected_run_id=identity.run_id, ) return "committed" if completed else "pending" - outcome = str(_board_call(kanban_db, board, operation)) + outcome = str(_board_call(kanban_db, identity.board, operation)) if outcome == "committed": - document["kanban_state"] = "committed" - document["committed_at"] = utc_now() - atomic_json(path, document) + _commit_terminal_file(path, identity, document) return outcome @@ -997,18 +1247,30 @@ def recover_pending_finalizations() -> int: from hermes_cli import kanban_db recovered = 0 - for path in sorted(STATE_ROOT.glob("*/*.terminal.json")): - record = load_json(path) - if record.get("kanban_state") == "committed": + for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")): + identity = _terminal_identity(path) + record = _load_terminal_json(path, identity) if identity is not None else {} + if identity is None: + _quarantine_terminal(path, None, "malformed-name") + continue + if not _terminal_record_valid(record): + _quarantine_terminal(path, identity, "malformed-payload") + _recover_quarantined_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") continue try: outcome = _finalize_terminal_record(kanban_db, path, record) except Exception as error: - board = str(record.get("board") or "unknown") - _record_board_access_error(board, error) + _record_board_access_error(identity.board, error) continue if outcome == "committed": recovered += 1 + elif outcome in {"invalid", "foreign", "stale"}: + _quarantine_terminal(path, identity, outcome) + _recover_quarantined_run(kanban_db, identity, outcome) return recovered @@ -1016,22 +1278,101 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool: """Keep an exact run claimed while its accepted result awaits replay.""" if not isinstance(run_id, int): return False - path = _terminal_path(state_path(board, task_id), run_id) + path = _terminal_path(state_path(board, task_id), run_id, "pending") try: path.stat() except FileNotFoundError: return False except OSError: - return True - record = load_json(path) - if not _terminal_record_valid(record): - return True - identity = ( - record.get("board") == board - and record.get("task_id") == task_id - and record.get("expected_run_id") == run_id + return False + identity = _terminal_identity(path) + if identity is None: + return False + record = _load_terminal_json(path, identity) + return _terminal_record_valid(record, identity) + + +def _artifact_gc_candidates(board_dir: Path) -> list[Path]: + """List only bounded artifact classes; pending journals are excluded.""" + candidates: list[Path] = [] + patterns = ( + "*.provider-*.result.json", + "*.candidate-*.json", + "*.terminal.committed.json", ) - return not identity or record.get("kanban_state") != "committed" + for pattern in patterns: + candidates.extend(board_dir.glob(pattern)) + quarantine = board_dir / "quarantine" + if quarantine.is_dir(): + candidates.extend(quarantine.glob("*.quarantine")) + return candidates + + +def gc_lane_artifacts( + *, + now: float | None = None, + max_age_seconds: int = ARTIFACT_RETENTION_AGE_SECONDS, + max_count: int = ARTIFACT_RETENTION_COUNT, + max_bytes: int = ARTIFACT_RETENTION_BYTES, +) -> int: + """Bound non-pending lane artifacts by age, count, and bytes per board.""" + current = time.time() if now is None else now + removed = 0 + for board_dir in STATE_ROOT.iterdir() if STATE_ROOT.is_dir() else (): + if not board_dir.is_dir() or board_dir.is_symlink(): + continue + board_removed = 0 + entries: list[tuple[Path, os.stat_result]] = [] + for path in _artifact_gc_candidates(board_dir): + try: + file_stat = path.stat(follow_symlinks=False) + except OSError: + continue + if not stat.S_ISREG(file_stat.st_mode): + continue + if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds: + path.unlink() + removed += 1 + board_removed += 1 + else: + entries.append((path, file_stat)) + retained_count = 0 + retained_bytes = 0 + for path, file_stat in sorted( + entries, key=lambda item: item[1].st_mtime, reverse=True + ): + exceeds_count = max_count >= 0 and retained_count >= max_count + exceeds_bytes = ( + 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 + continue + retained_count += 1 + retained_bytes += file_stat.st_size + if board_removed: + _fsync_directory(board_dir) + return removed + + +def maybe_gc_lane_artifacts(*, now: float | None = None) -> int: + """Run artifact retention on a bounded cadence, not every dispatcher tick.""" + global LAST_ARTIFACT_GC + current = time.time() if now is None else now + if current - LAST_ARTIFACT_GC < ARTIFACT_GC_INTERVAL_SECONDS: + return 0 + LAST_ARTIFACT_GC = current + try: + return gc_lane_artifacts(now=current) + except OSError as error: + print( + f"lane artifact retention deferred: {type(error).__name__}: {error}", + file=sys.stderr, + flush=True, + ) + return 0 def execute_claim(board: str, task_id: str) -> None: @@ -1176,6 +1517,16 @@ def execute_claim(board: str, task_id: str) -> None: heartbeat, remaining, ) + candidate_file = None + if result.structured: + candidate_file = _persist_candidate( + state, + state_file, + dict(result.structured), + route=route, + returncode=result.returncode, + goal_turn=goal_turn, + ) if result.capacity_failure: unavailable_provider = route.provider retry_context = ( @@ -1206,6 +1557,15 @@ def execute_claim(board: str, task_id: str) -> None: max(1, int(deadline - time.monotonic())), ) route = fallback + if result.structured: + candidate_file = _persist_candidate( + state, + state_file, + dict(result.structured), + route=route, + returncode=result.returncode, + goal_turn=goal_turn, + ) structured = result.structured if structured: structured = dict(structured) @@ -1213,16 +1573,6 @@ def execute_claim(board: str, task_id: str) -> None: workspace, structured.get("artifacts"), ) - candidate_file = _persist_candidate( - state, - state_file, - structured, - route=route, - returncode=result.returncode, - goal_turn=goal_turn, - ) - else: - candidate_file = None metadata = { "executor": "direct-cli-lane", "provider": route.provider, @@ -1281,8 +1631,9 @@ 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) - metadata["terminal_record"] = str(terminal_file) + metadata["terminal_record"] = str( + _terminal_path(state_file, run_id, "committed") + ) terminal_file, terminal_record = _write_terminal_record( state_file, board=board, @@ -1292,11 +1643,17 @@ def execute_claim(board: str, task_id: str) -> None: summary=str(structured.get("summary") or "Completed"), metadata=metadata, ) - outcome = _finalize_terminal_record( - kanban_db, - terminal_file, - terminal_record, - ) + try: + outcome = _finalize_terminal_record( + kanban_db, + terminal_file, + terminal_record, + ) + except Exception as error: + raise TerminalFinalizationPending( + "accepted worker result is durably journaled; " + f"Kanban finalization raised {type(error).__name__}" + ) from error if outcome != "committed": raise TerminalFinalizationPending( "accepted worker result is durably journaled but " @@ -1537,6 +1894,7 @@ def main() -> int: print(f"worker future failed: {error}", file=sys.stderr, flush=True) del futures[future] recover_pending_finalizations() + maybe_gc_lane_artifacts() active = set(futures.values()) try: newly_claimed = claim_ready(active, workers - len(futures)) diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index de0c72f3..c4ae715a 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -727,7 +727,7 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run( lanes.execute_claim("cassandra", "t_terminal") candidates = list((state_root / "cassandra").glob("*.candidate-*.json")) - terminals = list((state_root / "cassandra").glob("*.terminal.json")) + terminals = list((state_root / "cassandra").glob("*.terminal.pending.json")) assert len(candidates) == 1 assert json.loads(candidates[0].read_text())["structured"] == structured assert len(terminals) == 1 @@ -739,7 +739,12 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run( assert lanes.recover_pending_finalizations() == 1 assert task.status == "done" assert len(completions) == 2 - terminal = json.loads(terminals[0].read_text()) + assert not terminals[0].exists() + committed = list( + (state_root / "cassandra").glob("*.terminal.committed.json") + ) + assert len(committed) == 1 + terminal = json.loads(committed[0].read_text()) assert terminal["kanban_state"] == "committed" assert completions[-1]["result"] == terminal["result"] @@ -842,7 +847,334 @@ def test_terminal_replay_never_crosses_into_a_replacement_run( assert lanes.recover_pending_finalizations() == 0 assert completions == [] - assert json.loads(terminal_path.read_text())["kanban_state"] == "pending" + assert not terminal_path.exists() + quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine")) + assert len(quarantined) == 1 + + +def _completed_result(summary: str = "done") -> dict: + return { + "status": "completed", + "summary": summary, + "changed_files": [], + "tests_run": [], + "artifacts": [], + "findings": [], + "blockers": [], + } + + +def test_complete_exception_after_journal_is_replayable_and_never_blocks( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + task = SimpleNamespace( + id="t_crash", + status="running", + result=None, + current_run_id=51, + assignee="cli-auto", + max_runtime_seconds=60, + ) + completion_raises = {"value": True} + blocks = [] + comments = [] + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + if completion_raises["value"]: + raise RuntimeError("crash between journal and commit") + 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_crash"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "Finish safely.", + 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("crash-safe result"), False + ), + ) + + lanes.execute_claim("cassandra", "t_crash") + + pending = list((state_root / "cassandra").glob("*.terminal.pending.json")) + assert len(pending) == 1 + assert blocks == [] + assert any("RuntimeError" in body for body in comments) + + completion_raises["value"] = False + assert lanes.recover_pending_finalizations() == 1 + assert task.status == "done" + assert not pending[0].exists() + + +def test_terminal_payload_cannot_select_a_different_board( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + path = lanes._terminal_path(lanes.state_path("alpha", "t_alpha"), 3) + lanes.atomic_json( + path, + { + "board": "beta", + "task_id": "t_beta", + "expected_run_id": 9, + "result": json.dumps(_completed_result()), + "summary": "forged", + "metadata": {}, + "kanban_state": "pending", + }, + ) + alpha_task = SimpleNamespace( + id="t_alpha", status="running", current_run_id=3, assignee="cli-auto" + ) + opened = [] + reclaimed = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: (opened.append(board) or Connection()), + get_task=lambda _conn, task_id: alpha_task if task_id == "t_alpha" else None, + reclaim_task=lambda _conn, task_id, **_kwargs: (reclaimed.append(task_id) or True), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 0 + + assert opened == ["alpha"] + assert reclaimed == ["t_alpha"] + assert not path.exists() + quarantined = list((state_root / "alpha/quarantine").glob("*.quarantine")) + assert len(quarantined) == 1 + assert quarantined[0].stat().st_mode & 0o777 == 0o600 + + +def test_terminal_finalize_rereads_journal_after_a_path_swap( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + state_file = lanes.state_path("alpha", "t_alpha") + path, safe_record = lanes._write_terminal_record( + state_file, + board="alpha", + task_id="t_alpha", + run_id=4, + structured=_completed_result("safe"), + summary="safe", + metadata={}, + ) + forged = dict(safe_record) + forged.update({"board": "beta", "task_id": "t_beta", "expected_run_id": 7}) + lanes.atomic_json(path, forged) + opened = [] + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: (opened.append(board) or pytest.fail("must not open DB")), + ) + + assert lanes._finalize_terminal_record(fake_db, path, safe_record) == "foreign" + assert opened == [] + + +@pytest.mark.parametrize("payload", [b"", b'{"board":"cassandra"']) +def test_malformed_exact_run_journal_is_quarantined_and_reclaimed( + tmp_path: Path, + monkeypatch, + payload: bytes, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + path = lanes._terminal_path(lanes.state_path("cassandra", "t_partial"), 12) + path.parent.mkdir(parents=True) + path.write_bytes(payload) + task = SimpleNamespace( + id="t_partial", status="running", current_run_id=12, assignee="cli-auto" + ) + 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, + 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 reclaimed == [True] + assert not path.exists() + assert lanes._has_pending_finalization("cassandra", "t_partial", 12) is False + quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine")) + assert len(quarantined) == 1 + assert quarantined[0].stat().st_mode & 0o777 == 0o600 + + +def test_recovery_does_not_parse_committed_journals_every_tick( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + committed = lanes._terminal_path( + lanes.state_path("cassandra", "t_done"), 1, "committed" + ) + lanes.atomic_json(committed, {"kanban_state": "committed"}) + reads = [] + monkeypatch.setattr(lanes, "load_json", lambda path: (reads.append(path) or {})) + monkeypatch.setitem( + sys.modules, + "hermes_cli", + SimpleNamespace(kanban_db=SimpleNamespace()), + ) + + assert lanes.recover_pending_finalizations() == 0 + assert reads == [] + + +def test_atomic_json_fsyncs_file_and_directory_and_uses_private_mode( + tmp_path: Path, + monkeypatch, +): + calls = [] + real_fsync = lanes.os.fsync + monkeypatch.setattr( + lanes.os, + "fsync", + lambda descriptor: (calls.append(descriptor), real_fsync(descriptor))[1], + ) + path = tmp_path / "state.json" + + lanes.atomic_json(path, {"safe": True}) + + assert json.loads(path.read_text()) == {"safe": True} + assert path.stat().st_mode & 0o777 == 0o600 + assert len(calls) >= 2 + + +def test_atomic_json_refuses_a_precreated_temp_symlink( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "state.json" + victim = tmp_path / "credential" + victim.write_text("do not overwrite", encoding="utf-8") + temporary = tmp_path / ".state.json.fixed.tmp" + temporary.symlink_to(victim) + monkeypatch.setattr(lanes.uuid, "uuid4", lambda: SimpleNamespace(hex="fixed")) + + with pytest.raises(FileExistsError): + lanes.atomic_json(path, {"unsafe": True}) + + assert victim.read_text(encoding="utf-8") == "do not overwrite" + assert not path.exists() + + +def test_artifact_gc_prunes_by_age_without_touching_pending_journals( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + board = lanes.STATE_ROOT / "cassandra" + board.mkdir(parents=True) + quarantine = board / "quarantine" + quarantine.mkdir() + old_artifacts = [ + board / "t.run-1.candidate-1.json", + board / "t.run-1.provider-1.result.json", + board / "t.run-1.terminal.committed.json", + quarantine / "t.invalid.1234.quarantine", + ] + for artifact in old_artifacts: + artifact.write_text("{}", encoding="utf-8") + pending = board / "t.run-1.terminal.pending.json" + pending.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)) + + assert lanes.gc_lane_artifacts( + now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000 + ) == len(old_artifacts) + assert not any(artifact.exists() for artifact in old_artifacts) + assert pending.exists() + + +def test_artifact_gc_prunes_oldest_by_count_and_total_bytes( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + board = lanes.STATE_ROOT / "cassandra" + board.mkdir(parents=True) + files = [] + for sequence in range(3): + path = board / f"t.run-1.provider-{sequence}.result.json" + path.write_text("x" * 40, encoding="utf-8") + os.utime(path, (100 + sequence, 100 + sequence)) + files.append(path) + + assert lanes.gc_lane_artifacts( + now=200.0, max_age_seconds=1000, max_count=2, max_bytes=45 + ) == 2 + assert [path.exists() for path in files] == [False, False, True] + + +def test_artifact_gc_is_interval_gated( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + monkeypatch.setattr(lanes, "LAST_ARTIFACT_GC", 0.0) + calls = [] + monkeypatch.setattr( + lanes, "gc_lane_artifacts", lambda **kwargs: (calls.append(kwargs) or 0) + ) + + assert lanes.maybe_gc_lane_artifacts(now=1000.0) == 0 + assert lanes.maybe_gc_lane_artifacts(now=1001.0) == 0 + assert len(calls) == 1 @pytest.mark.parametrize( @@ -936,7 +1268,7 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) - monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json") + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") monkeypatch.setattr( lanes, "select_route", @@ -1004,11 +1336,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress( block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) - monkeypatch.setattr( - lanes, - "state_path", - lambda _board, _task_id: tmp_path / "state.json", - ) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") claude_low = lanes.Route( "claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, () ) @@ -1087,7 +1415,9 @@ def test_goal_card_continues_after_local_judge_rejects_progress( assert route_calls[2][1]["exclude_provider"] == "claude" assert "prior rejected reports" in judge_contexts[1] assert "commit, push, and remote verification are missing" in judge_contexts[1] - candidates = sorted(tmp_path.glob("state.run-12.candidate-*.json")) + candidates = sorted( + (lanes.STATE_ROOT / "cassandra").glob("t_goal.run-12.candidate-*.json") + ) assert len(candidates) == 2 assert json.loads(candidates[0].read_text())["structured"]["summary"] == ( "Focused tests passed." @@ -1098,6 +1428,87 @@ def test_goal_card_continues_after_local_judge_rejects_progress( assert reports == [] +def test_capacity_fallback_preserves_first_claude_structured_response( + tmp_path: Path, + monkeypatch, +): + task = SimpleNamespace( + id="t_fallback", + status="running", + result=None, + current_run_id=14, + assignee="cli-auto", + max_runtime_seconds=60, + ) + + 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, + worker_log_path=lambda _task_id, board: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_fallback"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "Complete with a fallback if needed.", + heartbeat_worker=lambda *_args, **_kwargs: True, + add_comment=lambda *_args: None, + complete_task=lambda _conn, _task_id, **kwargs: ( + setattr(task, "status", "done") or setattr(task, "result", kwargs["result"]) or True + ), + block_task=lambda *_args, **_kwargs: pytest.fail("fallback should complete"), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + claude = lanes.Route( + "claude", "claude-fable-5", "high", "claude-high", "test", "test", 1, () + ) + codex = lanes.Route( + "codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, () + ) + monkeypatch.setattr( + lanes, + "select_route", + lambda _prompt, assignee, **_kwargs: codex + if assignee == "cli-codex-high" + else claude, + ) + reports = [ + lanes.ProcessResult( + 1, + "subscription capacity exhausted", + { + **_completed_result("Claude preserved evidence"), + "status": "incomplete", + "blockers": ["subscription capacity exhausted"], + }, + True, + ), + lanes.ProcessResult(0, "", _completed_result("Codex completed"), False), + ] + monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0)) + + lanes.execute_claim("cassandra", "t_fallback") + + candidates = sorted( + (lanes.STATE_ROOT / "cassandra").glob("t_fallback.run-14.candidate-*.json") + ) + assert len(candidates) == 2 + first, second = [json.loads(path.read_text()) for path in candidates] + assert (first["provider"], first["structured"]["summary"]) == ( + "claude", + "Claude preserved evidence", + ) + assert (second["provider"], second["structured"]["summary"]) == ( + "codex", + "Codex completed", + ) + assert task.status == "done" + + def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch): task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto") calls = [] @@ -1143,11 +1554,15 @@ def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path): def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: Path, monkeypatch): task = SimpleNamespace( id="t_resume", + status="running", + result=None, current_run_id=9, assignee="cli-auto", max_runtime_seconds=60, ) - state_file = tmp_path / "state.json" + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + state_file = lanes.state_path("cassandra", "t_resume") + state_file.parent.mkdir(parents=True) state_file.write_text( json.dumps({"current_route": {"provider": "claude"}}), encoding="utf-8", @@ -1173,7 +1588,6 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P block_task=lambda *_args, **_kwargs: None, ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) - monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file) monkeypatch.setattr( lanes, "select_route",