diff --git a/dockerfiles/Dockerfile.hermes-agent b/dockerfiles/Dockerfile.hermes-agent index e0024e44..1416aae7 100644 --- a/dockerfiles/Dockerfile.hermes-agent +++ b/dockerfiles/Dockerfile.hermes-agent @@ -1552,8 +1552,15 @@ COPY dockerfiles/patch_hermes_decomposition_safety.py /tmp/patch_hermes_decompos COPY dockerfiles/hermes_execution_regression_support.py /tmp/hermes_execution_regression_support.py COPY dockerfiles/hermes_run_safety_regression.py /tmp/hermes_run_safety_regression.py COPY dockerfiles/hermes_decomposition_safety_regression.py /tmp/hermes_decomposition_safety_regression.py +COPY dockerfiles/hermes_lane_compatibility_regression.py /tmp/hermes_lane_compatibility_regression.py COPY services/hermes/scripts/cli_lane_*.py /tmp/hermes-lane-regression/ -RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \ +RUN HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression \ + HERMES_COMPATIBILITY_MODE=legacy \ + /opt/hermes/.venv/bin/python /tmp/hermes_lane_compatibility_regression.py \ + && /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \ + && HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression \ + HERMES_COMPATIBILITY_MODE=patched \ + /opt/hermes/.venv/bin/python /tmp/hermes_lane_compatibility_regression.py \ && HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression \ /opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py \ && rm /tmp/patch-hermes-execution-safety.py \ @@ -1564,6 +1571,7 @@ RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \ /tmp/hermes_execution_regression_support.py \ /tmp/hermes_run_safety_regression.py \ /tmp/hermes_decomposition_safety_regression.py \ + /tmp/hermes_lane_compatibility_regression.py \ && find /tmp/hermes-lane-regression -depth -delete COPY dockerfiles/hermes-session-migrate.py /opt/hermes/bin/hermes-session-migrate diff --git a/dockerfiles/Dockerfile.hermes-agent.dockerignore b/dockerfiles/Dockerfile.hermes-agent.dockerignore index e6cb3493..0abc8045 100644 --- a/dockerfiles/Dockerfile.hermes-agent.dockerignore +++ b/dockerfiles/Dockerfile.hermes-agent.dockerignore @@ -13,6 +13,7 @@ !dockerfiles/hermes_execution_regression_support.py !dockerfiles/hermes_run_safety_regression.py !dockerfiles/hermes_decomposition_safety_regression.py +!dockerfiles/hermes_lane_compatibility_regression.py !services/ !services/hermes/ !services/hermes/scripts/ diff --git a/dockerfiles/hermes_lane_compatibility_regression.py b/dockerfiles/hermes_lane_compatibility_regression.py new file mode 100644 index 00000000..7ed69c63 --- /dev/null +++ b/dockerfiles/hermes_lane_compatibility_regression.py @@ -0,0 +1,120 @@ +"""Exercise mounted CLI scripts against the exact image Kanban API.""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +_HOME = tempfile.TemporaryDirectory(prefix="hermes-lane-compatibility-") +os.environ["HERMES_HOME"] = _HOME.name +os.environ["HERMES_CLI_LANE_CONCURRENCY"] = "1" +lane_source = Path(os.environ["HERMES_CLI_LANE_SOURCE"]) +sys.path.insert(0, str(lane_source)) + +from cli_lane_capabilities import ( # noqa: E402 + detect_kanban_capabilities, + initialize_kanban_capabilities, + runtime_health, +) +from cli_lane_config import TerminalIdentity # noqa: E402 +from cli_lane_dispatch import recover_orphans # noqa: E402 +from cli_lane_finalization import _finalize_document_db # noqa: E402 +from hermes_cli import kanban_db # noqa: E402 + + +class MountedLaneCompatibilityTests(unittest.TestCase): + """Verify safe behavior before and after the image API patch.""" + + def setUp(self) -> None: + self.connection = kanban_db.connect() + + def tearDown(self) -> None: + self.connection.close() + + @staticmethod + def _document(summary: str) -> dict: + return { + "result": json.dumps( + { + "status": "completed", + "summary": summary, + "changed_files": [], + "tests_run": ["mixed source compatibility"], + "artifacts": [], + "findings": [], + "blockers": [], + }, + sort_keys=True, + ), + "summary": summary, + "metadata": {"matrix": os.environ["HERMES_COMPATIBILITY_MODE"]}, + } + + def test_exact_image_api_matrix(self) -> None: + mode = os.environ["HERMES_COMPATIBILITY_MODE"] + capabilities = initialize_kanban_capabilities(kanban_db) + self.assertTrue(capabilities.exact_run_completion) + self.assertEqual(capabilities.ready, mode == "patched") + self.assertEqual(runtime_health()["state"], "ready" if mode == "patched" else "deferred") + self.assertLess(Path(_HOME.name, "cli-lanes/runtime-health.json").stat().st_size, 2048) + + def test_current_active_run_completion_stays_safe(self) -> None: + task_id = kanban_db.create_task(self.connection, title="active accepted result") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + run_id = kanban_db.get_task(self.connection, task_id).current_run_id + outcome = _finalize_document_db( + kanban_db, + TerminalIdentity("default", task_id, run_id, "pending"), + self._document("active result"), + ) + self.assertEqual(outcome, "committed") + self.assertEqual(kanban_db.get_task(self.connection, task_id).status, "done") + + def test_ended_run_replay_is_enabled_only_after_patch(self) -> None: + mode = os.environ["HERMES_COMPATIBILITY_MODE"] + task_id = kanban_db.create_task(self.connection, title="ended accepted 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="simulate post-journal error", + expected_run_id=run_id, + ) + ) + outcome = _finalize_document_db( + kanban_db, + TerminalIdentity("default", task_id, run_id, "pending"), + self._document("ended result"), + ) + self.assertEqual(outcome, "committed" if mode == "patched" else "deferred") + + def test_old_reclaim_api_is_never_called_unguarded(self) -> None: + capabilities = detect_kanban_capabilities(kanban_db) + if capabilities.exact_run_reclaim: + self.skipTest("patched API exercises guarded reclaim in the exact-run suite") + with mock.patch.object( + kanban_db, + "reclaim_task", + side_effect=AssertionError("legacy unguarded reclaim called"), + ), mock.patch("cli_lane_dispatch.recover_pending_finalizations", return_value=0): + recover_orphans() + + +def main() -> int: + """Run the compatibility matrix for the selected exact image source.""" + result = unittest.TextTestRunner(verbosity=2).run( + unittest.defaultTestLoader.loadTestsFromTestCase(MountedLaneCompatibilityTests) + ) + return 0 if result.wasSuccessful() else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 1de6b74b..6becbaf6 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -66,6 +66,7 @@ configMapGenerator: namespace: hermes files: - cli_lane_board.py=scripts/cli_lane_board.py + - cli_lane_capabilities.py=scripts/cli_lane_capabilities.py - cli_lane_config.py=scripts/cli_lane_config.py - cli_lane_dispatch.py=scripts/cli_lane_dispatch.py - cli_lane_evidence.py=scripts/cli_lane_evidence.py @@ -98,6 +99,7 @@ configMapGenerator: - claude=scripts/claude - claude_command_policy.py=scripts/claude_command_policy.py - cli_lane_board.py=scripts/cli_lane_board.py + - cli_lane_capabilities.py=scripts/cli_lane_capabilities.py - cli_lane_config.py=scripts/cli_lane_config.py - cli_lane_dispatch.py=scripts/cli_lane_dispatch.py - cli_lane_evidence.py=scripts/cli_lane_evidence.py diff --git a/services/hermes/scripts/cli_lane_capabilities.py b/services/hermes/scripts/cli_lane_capabilities.py new file mode 100644 index 00000000..c54c8bb0 --- /dev/null +++ b/services/hermes/scripts/cli_lane_capabilities.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Detect and publish the Kanban API boundary required by CLI lanes.""" + +from __future__ import annotations + +import inspect +import sys +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Callable + +from cli_lane_config import STATE_ROOT, utc_now +from cli_lane_files import atomic_json, load_json + + +_CAPABILITIES: KanbanCapabilities | None = None +_CAPABILITY_SOURCE: tuple[int, int] | None = None + + +@dataclass(frozen=True) +class KanbanCapabilities: + """Safety-relevant keyword support exposed by the image's Kanban DB.""" + + exact_run_completion: bool + ended_run_replay: bool + exact_run_reclaim: bool + + @property + def ready(self) -> bool: + """Return whether dispatch and every recovery transition are safe.""" + return ( + self.exact_run_completion + and self.ended_run_replay + and self.exact_run_reclaim + ) + + @property + def deferred_features(self) -> tuple[str, ...]: + """List fixed, bounded capability names absent from the image API.""" + missing = [] + if not self.exact_run_completion: + missing.append("exact-run-completion") + if not self.ended_run_replay: + missing.append("ended-run-replay") + if not self.exact_run_reclaim: + missing.append("exact-run-reclaim") + return tuple(missing) + + +def _explicit_keyword(function: Callable[..., Any], keyword: str) -> bool: + """Require an explicit keyword parameter rather than trusting ``**kwargs``.""" + try: + parameter = inspect.signature(function).parameters.get(keyword) + except (TypeError, ValueError): + return False + return parameter is not None and parameter.kind in { + inspect.Parameter.POSITIONAL_OR_KEYWORD, + inspect.Parameter.KEYWORD_ONLY, + } + + +def detect_kanban_capabilities(kanban_db: Any) -> KanbanCapabilities: + """Inspect the loaded image API without invoking a mutating DB operation.""" + complete_task = getattr(kanban_db, "complete_task", None) + reclaim_task = getattr(kanban_db, "reclaim_task", None) + return KanbanCapabilities( + exact_run_completion=callable(complete_task) + and _explicit_keyword(complete_task, "expected_run_id"), + ended_run_replay=callable(complete_task) + and _explicit_keyword(complete_task, "replay_ended_run_id"), + exact_run_reclaim=callable(reclaim_task) + and _explicit_keyword(reclaim_task, "expected_run_id"), + ) + + +def _source_identity(kanban_db: Any) -> tuple[int, int]: + """Bind cached capability evidence to the two inspected callables.""" + return ( + id(getattr(kanban_db, "complete_task", None)), + id(getattr(kanban_db, "reclaim_task", None)), + ) + + +def _health_document(capabilities: KanbanCapabilities) -> dict[str, Any]: + """Return a bounded operator-readable compatibility health document.""" + return { + "component": "cli-lane-runner", + "state": "ready" if capabilities.ready else "deferred", + "ready": capabilities.ready, + "active_run_completion_safe": capabilities.exact_run_completion, + "deferred_features": list(capabilities.deferred_features), + "capabilities": asdict(capabilities), + "observed_at": utc_now(), + } + + +def initialize_kanban_capabilities( + kanban_db: Any, + *, + health_path: Path | None = None, +) -> KanbanCapabilities: + """Detect startup compatibility and durably expose ready/deferred health.""" + global _CAPABILITIES, _CAPABILITY_SOURCE + capabilities = detect_kanban_capabilities(kanban_db) + _CAPABILITIES = capabilities + _CAPABILITY_SOURCE = _source_identity(kanban_db) + destination = health_path or STATE_ROOT / "runtime-health.json" + destination.parent.mkdir(parents=True, exist_ok=True) + atomic_json(destination, _health_document(capabilities), 0o600) + if not capabilities.ready: + print( + "CLI lane safety compatibility deferred; missing image APIs: " + + ", ".join(capabilities.deferred_features), + file=sys.stderr, + flush=True, + ) + return capabilities + + +def kanban_capabilities(kanban_db: Any) -> KanbanCapabilities: + """Return startup evidence, redetecting only when the loaded API changes.""" + if _CAPABILITIES is None or _source_identity(kanban_db) != _CAPABILITY_SOURCE: + return initialize_kanban_capabilities(kanban_db) + return _CAPABILITIES + + +def runtime_health(path: Path | None = None) -> dict[str, Any]: + """Read the bounded health document for probes and operator diagnostics.""" + document = load_json(path or STATE_ROOT / "runtime-health.json") + return document if isinstance(document, dict) else {} diff --git a/services/hermes/scripts/cli_lane_dispatch.py b/services/hermes/scripts/cli_lane_dispatch.py index 44f2b573..0a059911 100644 --- a/services/hermes/scripts/cli_lane_dispatch.py +++ b/services/hermes/scripts/cli_lane_dispatch.py @@ -11,6 +11,7 @@ from collections.abc import Callable from typing import Any from cli_lane_board import _external, _record_board_access_error, _task_value +from cli_lane_capabilities import initialize_kanban_capabilities, kanban_capabilities from cli_lane_config import ( BOARD_CORRUPTION_ERRORS, DEFAULT_CLAIM_TTL, @@ -52,6 +53,8 @@ def recover_orphans() -> None: from hermes_cli import kanban_db recover_pending_finalizations() + if not kanban_capabilities(kanban_db).exact_run_reclaim: + return try: boards = kanban_db.list_boards(include_archived=False) except Exception as error: @@ -154,8 +157,11 @@ def claim_ready( def main() -> int: """Continuously bridge external Kanban lanes to provider CLIs.""" + from hermes_cli import kanban_db + RESULT_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True) atomic_json(RESULT_SCHEMA_PATH, RESULT_SCHEMA, 0o644) + capabilities = initialize_kanban_capabilities(kanban_db) recover_orphans() workers = max(1, min(int(os.environ.get("HERMES_CLI_LANE_CONCURRENCY", "4")), 8)) futures: dict[concurrent.futures.Future[None], tuple[str, str]] = {} @@ -172,7 +178,11 @@ def main() -> int: maybe_gc_lane_artifacts() active = set(futures.values()) try: - newly_claimed = claim_ready(active, workers - len(futures)) + newly_claimed = ( + claim_ready(active, workers - len(futures)) + if capabilities.ready + else [] + ) BOARD_CORRUPTION_ERRORS.pop("board-registry", None) except Exception as error: _record_board_access_error("board-registry", error) diff --git a/services/hermes/scripts/cli_lane_evidence.py b/services/hermes/scripts/cli_lane_evidence.py index f261e870..d85abb0e 100644 --- a/services/hermes/scripts/cli_lane_evidence.py +++ b/services/hermes/scripts/cli_lane_evidence.py @@ -4,6 +4,7 @@ from __future__ import annotations import ctypes +import contextlib import errno import hashlib import json @@ -75,7 +76,7 @@ def _retire_terminal_entry( (authority_name or path.name).encode("utf-8") ).hexdigest()[:16] source_digest = hashlib.sha256( - f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8") + f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode() ).hexdigest()[:16] for sequence in range(32): staging = f".retire.{path_digest}.{source_digest}.{sequence}" @@ -157,15 +158,13 @@ def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool: if descriptor is not None: os.close(descriptor) if temporary_stat is not None: - try: + with contextlib.suppress(OSError): _retire_terminal_entry( Path(temporary), temporary_stat, board_descriptor=directory, quarantine_descriptor=directory, ) - except OSError: - pass os.close(directory) def _terminal_document_digest(document: dict[str, Any]) -> str: diff --git a/services/hermes/scripts/cli_lane_execution.py b/services/hermes/scripts/cli_lane_execution.py index 3b24b468..46a3d7b0 100644 --- a/services/hermes/scripts/cli_lane_execution.py +++ b/services/hermes/scripts/cli_lane_execution.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import json import sqlite3 import time @@ -98,7 +99,7 @@ def execute_claim(board: str, task_id: str) -> None: return True def comment(body: str) -> None: - try: + with contextlib.suppress(OSError, sqlite3.Error): _board_call( kanban_db, board, @@ -109,10 +110,8 @@ def execute_claim(board: str, task_id: str) -> None: body, ), ) - except (OSError, sqlite3.Error): - # Route state is also written to the durable lane-state file; - # a later heartbeat or terminal result remains authoritative. - pass + # Route state is also written to the durable lane-state file; + # a later heartbeat or terminal result remains authoritative. try: previous_route = state.get("current_route") @@ -395,15 +394,20 @@ def execute_claim(board: str, task_id: str) -> None: reason = reason or str(structured.get("summary") or "") if reason is None: reason = result.output[-4000:] + failure_reason = reason or ( + f"{route.provider} worker failed with exit {result.returncode}" + ) + failure_kind = ( + "transient" if result.capacity_failure else "capability" + ) _board_call( kanban_db, board, - lambda fresh: kanban_db.block_task( + lambda fresh, failure_reason=failure_reason, failure_kind=failure_kind: kanban_db.block_task( fresh, task_id, - reason=reason - or f"{route.provider} worker failed with exit {result.returncode}", - kind="transient" if result.capacity_failure else "capability", + reason=failure_reason, + kind=failure_kind, expected_run_id=run_id, ), ) diff --git a/services/hermes/scripts/cli_lane_files.py b/services/hermes/scripts/cli_lane_files.py index 9d160da0..64422a1e 100644 --- a/services/hermes/scripts/cli_lane_files.py +++ b/services/hermes/scripts/cli_lane_files.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import json import os import re @@ -36,10 +37,8 @@ def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None: finally: if descriptor is not None: os.close(descriptor) - try: + with contextlib.suppress(FileNotFoundError): temporary.unlink() - except FileNotFoundError: - pass def _fsync_directory(directory: Path) -> None: """Persist directory-entry changes after an atomic rename or quarantine.""" diff --git a/services/hermes/scripts/cli_lane_finalization.py b/services/hermes/scripts/cli_lane_finalization.py index 5599f830..ef39ffc5 100644 --- a/services/hermes/scripts/cli_lane_finalization.py +++ b/services/hermes/scripts/cli_lane_finalization.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any from cli_lane_board import _board_call, _external, _record_board_access_error, _task_value +from cli_lane_capabilities import kanban_capabilities from cli_lane_config import ( TerminalIdentity, TerminalRecoverySnapshot, @@ -39,6 +40,8 @@ def _recover_exact_run( """Make an exact external run retryable after journal recovery fails.""" if identity is None or canonical_run_id(identity.run_id) is None: return False + if not kanban_capabilities(kanban_db).exact_run_reclaim: + return False def operation(conn: Any) -> bool: task = kanban_db.get_task(conn, identity.task_id) @@ -178,11 +181,15 @@ def _finalize_document_db( 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: + if not kanban_capabilities(kanban_db).exact_run_completion: + return "deferred" completion_guard = {"expected_run_id": identity.run_id} elif status in {"ready", "blocked", "triage"} 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. + if not kanban_capabilities(kanban_db).ended_run_replay: + return "deferred" completion_guard = {"replay_ended_run_id": identity.run_id} else: return "stale" diff --git a/services/hermes/scripts/cli_lane_provider.py b/services/hermes/scripts/cli_lane_provider.py index 53a05bba..07a3d02b 100644 --- a/services/hermes/scripts/cli_lane_provider.py +++ b/services/hermes/scripts/cli_lane_provider.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import json import os import selectors @@ -166,17 +167,13 @@ def _signal_worker_tree( if _process_identity_matches(pid, start_time): groups.add(process_group) for process_group in groups: - try: + with contextlib.suppress(ProcessLookupError): os.killpg(process_group, sig) - except ProcessLookupError: - pass for pid, (_, start_time) in descendants.items(): if not _process_identity_matches(pid, start_time): continue - try: + with contextlib.suppress(ProcessLookupError): os.kill(pid, sig) - except ProcessLookupError: - pass def _terminate_worker_process( process: subprocess.Popen[str], @@ -187,10 +184,8 @@ def _terminate_worker_process( descendants.update(_descendant_processes(process.pid)) if process.poll() is None: _signal_worker_tree(process.pid, descendants, signal.SIGTERM) - try: + with contextlib.suppress(subprocess.TimeoutExpired): process.wait(timeout=10) - except subprocess.TimeoutExpired: - pass # Provider terminal tools create their own sessions, so killing only the # native CLI's process group leaves test/build subprocesses orphaned. The @@ -357,10 +352,8 @@ def run_provider( file_result = load_json(result_file) if file_result.get("status") in cli_lane_goal.RESULT_STATUSES: result.structured = file_result - try: + with contextlib.suppress(OSError): result_file.chmod(0o600) - except OSError: - pass return result atomic_json(state_file, state) diff --git a/services/hermes/scripts/cli_lane_quarantine.py b/services/hermes/scripts/cli_lane_quarantine.py index 21dfddb1..8c599184 100644 --- a/services/hermes/scripts/cli_lane_quarantine.py +++ b/services/hermes/scripts/cli_lane_quarantine.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import hashlib import json import os @@ -136,10 +137,8 @@ def _quarantine_terminal( quarantine_descriptor = os.dup(board_descriptor) quarantine_dir = board_dir else: - try: + with contextlib.suppress(FileExistsError): os.mkdir("quarantine", 0o700, dir_fd=board_descriptor) - except FileExistsError: - pass quarantine_descriptor = os.open( "quarantine", board_flags, diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 5a51e928..ed32e3b7 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -12,6 +12,14 @@ import subprocess import uuid import cli_lane_goal +from cli_lane_capabilities import ( + KanbanCapabilities, + _explicit_keyword, + detect_kanban_capabilities, + initialize_kanban_capabilities, + kanban_capabilities, + runtime_health, +) from cli_lane_board import ( _board_call, _external, diff --git a/testing/quality_contract.json b/testing/quality_contract.json index 06289d49..c5598914 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -39,6 +39,23 @@ "services/hermes/scripts/node_account_hardening.py", "services/hermes/scripts/node_account_io.py", "services/hermes/scripts/stage_runtime_access.py", + "services/hermes/scripts/cli_lane_board.py", + "services/hermes/scripts/cli_lane_capabilities.py", + "services/hermes/scripts/cli_lane_config.py", + "services/hermes/scripts/cli_lane_dispatch.py", + "services/hermes/scripts/cli_lane_evidence.py", + "services/hermes/scripts/cli_lane_execution.py", + "services/hermes/scripts/cli_lane_files.py", + "services/hermes/scripts/cli_lane_finalization.py", + "services/hermes/scripts/cli_lane_goal.py", + "services/hermes/scripts/cli_lane_prompt.py", + "services/hermes/scripts/cli_lane_provider.py", + "services/hermes/scripts/cli_lane_quarantine.py", + "services/hermes/scripts/cli_lane_records.py", + "services/hermes/scripts/cli_lane_recovery.py", + "services/hermes/scripts/cli_lane_retention.py", + "services/hermes/scripts/cli_lane_routing.py", + "services/hermes/scripts/cli_lane_runner.py", "testing/__init__.py", "testing/quality_contract.py", "testing/quality_docs.py", @@ -76,6 +93,23 @@ "services/hermes/scripts/node_account_hardening.py", "services/hermes/scripts/node_account_io.py", "services/hermes/scripts/stage_runtime_access.py", + "services/hermes/scripts/cli_lane_board.py", + "services/hermes/scripts/cli_lane_capabilities.py", + "services/hermes/scripts/cli_lane_config.py", + "services/hermes/scripts/cli_lane_dispatch.py", + "services/hermes/scripts/cli_lane_evidence.py", + "services/hermes/scripts/cli_lane_execution.py", + "services/hermes/scripts/cli_lane_files.py", + "services/hermes/scripts/cli_lane_finalization.py", + "services/hermes/scripts/cli_lane_goal.py", + "services/hermes/scripts/cli_lane_prompt.py", + "services/hermes/scripts/cli_lane_provider.py", + "services/hermes/scripts/cli_lane_quarantine.py", + "services/hermes/scripts/cli_lane_records.py", + "services/hermes/scripts/cli_lane_recovery.py", + "services/hermes/scripts/cli_lane_retention.py", + "services/hermes/scripts/cli_lane_routing.py", + "services/hermes/scripts/cli_lane_runner.py", "testing/tests", "testing" ], @@ -168,6 +202,7 @@ "services/hermes/scripts/node_account_hardening.py", "services/hermes/scripts/node_account_io.py", "services/hermes/scripts/stage_runtime_access.py", + "services/hermes/scripts/cli_lane_*.py", "services/mailu/scripts/mailu_sync.py", "services/mailu/scripts/mailu_sync_listener.py" ], @@ -232,6 +267,23 @@ "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", "services/hermes/scripts/execution_pool_worker.py", + "services/hermes/scripts/cli_lane_board.py", + "services/hermes/scripts/cli_lane_capabilities.py", + "services/hermes/scripts/cli_lane_config.py", + "services/hermes/scripts/cli_lane_dispatch.py", + "services/hermes/scripts/cli_lane_evidence.py", + "services/hermes/scripts/cli_lane_execution.py", + "services/hermes/scripts/cli_lane_files.py", + "services/hermes/scripts/cli_lane_finalization.py", + "services/hermes/scripts/cli_lane_goal.py", + "services/hermes/scripts/cli_lane_prompt.py", + "services/hermes/scripts/cli_lane_provider.py", + "services/hermes/scripts/cli_lane_quarantine.py", + "services/hermes/scripts/cli_lane_records.py", + "services/hermes/scripts/cli_lane_recovery.py", + "services/hermes/scripts/cli_lane_retention.py", + "services/hermes/scripts/cli_lane_routing.py", + "services/hermes/scripts/cli_lane_runner.py", "testing/quality_coverage.py" ], "tracked_files": [ @@ -260,6 +312,23 @@ "services/hermes/scripts/node_account_hardening.py", "services/hermes/scripts/node_account_io.py", "services/hermes/scripts/stage_runtime_access.py", + "services/hermes/scripts/cli_lane_board.py", + "services/hermes/scripts/cli_lane_capabilities.py", + "services/hermes/scripts/cli_lane_config.py", + "services/hermes/scripts/cli_lane_dispatch.py", + "services/hermes/scripts/cli_lane_evidence.py", + "services/hermes/scripts/cli_lane_execution.py", + "services/hermes/scripts/cli_lane_files.py", + "services/hermes/scripts/cli_lane_finalization.py", + "services/hermes/scripts/cli_lane_goal.py", + "services/hermes/scripts/cli_lane_prompt.py", + "services/hermes/scripts/cli_lane_provider.py", + "services/hermes/scripts/cli_lane_quarantine.py", + "services/hermes/scripts/cli_lane_records.py", + "services/hermes/scripts/cli_lane_recovery.py", + "services/hermes/scripts/cli_lane_retention.py", + "services/hermes/scripts/cli_lane_routing.py", + "services/hermes/scripts/cli_lane_runner.py", "testing/quality_contract.py", "testing/quality_docs.py", "testing/quality_hygiene.py", diff --git a/testing/tests/conftest.py b/testing/tests/conftest.py new file mode 100644 index 00000000..09dcb0a1 --- /dev/null +++ b/testing/tests/conftest.py @@ -0,0 +1,26 @@ +"""Shared pytest boundaries for repository automation tests.""" + +from __future__ import annotations + +import sys + +import pytest + + +@pytest.fixture(autouse=True) +def _patched_hermes_cli_db_contract(request, monkeypatch): + """Keep legacy lane test doubles focused on their original safety boundary.""" + module_name = request.module.__name__.rsplit(".", 1)[-1] + if not module_name.startswith("test_hermes_cli_") or module_name.endswith( + "capabilities" + ): + return + capability_module = sys.modules.get("cli_lane_capabilities") + if capability_module is None: + return + safe = capability_module.KanbanCapabilities(True, True, True) + for name, module in tuple(sys.modules.items()): + if ( + name == "cli_lane_runner" or name.startswith("cli_lane_") + ) and hasattr(module, "kanban_capabilities"): + monkeypatch.setattr(module, "kanban_capabilities", lambda _db: safe) diff --git a/testing/tests/test_hermes_agent_layout.py b/testing/tests/test_hermes_agent_layout.py index f5bd5074..2e2ef65e 100644 --- a/testing/tests/test_hermes_agent_layout.py +++ b/testing/tests/test_hermes_agent_layout.py @@ -308,6 +308,35 @@ def test_cli_lane_domain_modules_are_all_mounted_with_the_runner(): assert expected <= mounted +def test_cli_lane_config_refresh_does_not_restart_active_work(): + """Keep projected script refreshes separate from pod lifecycle changes.""" + manifest = yaml.safe_load((HERMES / "kustomization.yaml").read_text()) + coordinator = next( + item + for item in manifest["configMapGenerator"] + if item["name"] == "hermes-coordinator" + ) + deployment = _agent_deployment() + lane = next( + item + for item in deployment["spec"]["template"]["spec"]["containers"] + if item["name"] == "cli-lane-runner" + ) + mount = next( + item for item in lane["volumeMounts"] if item["name"] == "coordinator" + ) + + assert coordinator["options"]["disableNameSuffixHash"] is True + assert mount == { + "name": "coordinator", + "mountPath": "/opt/coordinator", + "readOnly": True, + } + assert "checksum/hermes-coordinator" not in deployment["spec"]["template"].get( + "metadata", {} + ).get("annotations", {}) + + def test_agent_dashboard_reconnects_all_transient_websockets(): dockerfile = ( HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent" diff --git a/testing/tests/test_hermes_cli_capabilities.py b/testing/tests/test_hermes_cli_capabilities.py new file mode 100644 index 00000000..adc6e3ca --- /dev/null +++ b/testing/tests/test_hermes_cli_capabilities.py @@ -0,0 +1,272 @@ +"""Mixed-image capability gates for mounted Hermes CLI lane scripts.""" + +from __future__ import annotations + +import json +import sys +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_cli_support import _completed_result, lanes + + +def _old_complete( + _conn, + _task_id, + *, + result=None, + summary=None, + metadata=None, + expected_run_id=None, +): + return bool(result or summary or metadata or expected_run_id) + + +def _new_complete( + _conn, + _task_id, + *, + result=None, + summary=None, + metadata=None, + expected_run_id=None, + replay_ended_run_id=None, +): + return bool(result or summary or metadata or expected_run_id or replay_ended_run_id) + + +def _old_reclaim(_conn, _task_id, *, reason, signal_fn=None): + return bool(reason or signal_fn) + + +def _new_reclaim( + _conn, + _task_id, + *, + reason, + signal_fn=None, + expected_run_id=None, +): + return bool(reason or signal_fn or expected_run_id) + + +def _db(*, patched: bool): + return SimpleNamespace( + complete_task=_new_complete if patched else _old_complete, + reclaim_task=_new_reclaim if patched else _old_reclaim, + ) + + +def _terminal_document(run_id: int = 7) -> dict: + structured = _completed_result("accepted result") + return { + "board": "cassandra", + "task_id": "t_compat", + "expected_run_id": run_id, + "result": json.dumps(structured, sort_keys=True), + "summary": "accepted result", + "metadata": {}, + "kanban_state": "pending", + "recorded_at": lanes.utc_now(), + } + + +def test_capability_detection_requires_explicit_safety_keywords(): + old = lanes.detect_kanban_capabilities(_db(patched=False)) + new = lanes.detect_kanban_capabilities(_db(patched=True)) + + assert old.exact_run_completion is True + assert old.ended_run_replay is False + assert old.exact_run_reclaim is False + assert old.ready is False + assert old.deferred_features == ("ended-run-replay", "exact-run-reclaim") + assert new.ready is True + assert new.deferred_features == () + + variadic = SimpleNamespace( + complete_task=lambda *_args, **_kwargs: True, + reclaim_task=lambda *_args, **_kwargs: True, + ) + assert lanes.detect_kanban_capabilities(variadic).ready is False + assert lanes._explicit_keyword(object(), "expected_run_id") is False + + +def test_startup_health_moves_from_bounded_deferred_to_ready( + tmp_path: Path, + capsys, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + health = tmp_path / "runtime-health.json" + old = lanes.initialize_kanban_capabilities(_db(patched=False), health_path=health) + old_document = lanes.runtime_health(health) + + assert old.ready is False + assert old_document["state"] == "deferred" + assert old_document["ready"] is False + assert old_document["active_run_completion_safe"] is True + assert old_document["deferred_features"] == [ + "ended-run-replay", + "exact-run-reclaim", + ] + assert health.stat().st_size < 2048 + assert health.stat().st_mode & 0o777 == 0o600 + assert "compatibility deferred" in capsys.readouterr().err + + new_db = _db(patched=True) + new = lanes.kanban_capabilities(new_db) + lanes.initialize_kanban_capabilities(new_db, health_path=health) + assert new.ready is True + assert lanes.runtime_health(health)["state"] == "ready" + + +@pytest.mark.parametrize( + ("status", "run_id", "expected"), + [("running", 7, "committed"), ("blocked", None, "deferred")], +) +def test_old_image_completes_only_the_exact_active_run( + tmp_path: Path, + monkeypatch, + status: str, + run_id: int | None, + expected: str, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + task = SimpleNamespace( + id="t_compat", + status=status, + current_run_id=run_id, + completed_run_id=None, + result=None, + assignee="cli-auto", + ) + calls = [] + + def complete( + _conn, + _task_id, + *, + result=None, + summary=None, + metadata=None, + expected_run_id=None, + ): + calls.append(expected_run_id) + task.status = "done" + task.current_run_id = None + task.completed_run_id = expected_run_id + task.result = result + return True + + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: SimpleNamespace(close=lambda: None), + get_task=lambda _conn, _task_id: task, + complete_task=complete, + reclaim_task=_old_reclaim, + ) + identity = lanes.TerminalIdentity("cassandra", "t_compat", 7, "pending") + + outcome = lanes._finalize_document_db(db, identity, _terminal_document()) + + assert outcome == expected + assert calls == ([7] if status == "running" else []) + if expected == "deferred": + assert task.status == "blocked" + + +def test_old_image_preserves_pending_and_prepared_journals( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + task = SimpleNamespace( + id="t_compat", + status="blocked", + current_run_id=None, + completed_run_id=None, + result=None, + assignee="cli-auto", + ) + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: SimpleNamespace(close=lambda: None), + get_task=lambda _conn, _task_id: task, + complete_task=_old_complete, + reclaim_task=_old_reclaim, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + pending, _record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_compat"), + board="cassandra", + task_id="t_compat", + run_id=7, + structured=_completed_result("accepted result"), + summary="accepted result", + metadata={}, + ) + + assert lanes.recover_pending_finalizations() == 0 + assert pending.exists() + assert len(list(pending.parent.glob("*.terminal.prepared-*.json"))) == 1 + assert task.status == "blocked" + + +def test_old_image_never_calls_unguarded_reclaim(tmp_path: Path, monkeypatch): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + task = SimpleNamespace( + id="t_running", + status="running", + current_run_id=11, + assignee="cli-auto", + ) + calls = [] + + def reclaim(_conn, _task_id, *, reason, signal_fn=None): + calls.append((reason, signal_fn)) + return True + + db = SimpleNamespace( + complete_task=_old_complete, + reclaim_task=reclaim, + list_boards=lambda include_archived=False: [{"slug": "cassandra"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: SimpleNamespace(close=lambda: None), + list_tasks=lambda _conn: [task], + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + + lanes.recover_orphans() + + assert calls == [] + + +def test_main_detects_capabilities_before_startup_recovery( + tmp_path: Path, + monkeypatch, +): + events = [] + db = _db(patched=False) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "result.schema.json") + monkeypatch.setattr( + lanes, + "initialize_kanban_capabilities", + lambda candidate: events.append(("detect", candidate)) + or lanes.KanbanCapabilities(True, False, False), + ) + + def stop_after_detection(): + events.append(("recover", db)) + raise RuntimeError("stop after startup boundary") + + monkeypatch.setattr(lanes, "recover_orphans", stop_after_detection) + + with pytest.raises(RuntimeError, match="startup boundary"): + lanes.main() + + assert events == [("detect", db), ("recover", db)] diff --git a/testing/tests/test_hermes_cli_dispatch_runtime.py b/testing/tests/test_hermes_cli_dispatch_runtime.py new file mode 100644 index 00000000..74684105 --- /dev/null +++ b/testing/tests/test_hermes_cli_dispatch_runtime.py @@ -0,0 +1,197 @@ +"""Dispatcher startup, degraded-board, and worker-loop regressions.""" + +from __future__ import annotations + +import sys +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_cli_support import lanes + + +def test_board_slug_and_connection_failure_paths(monkeypatch): + assert lanes._board_slug(SimpleNamespace(slug="cassandra")) == "cassandra" + assert lanes._board_slug(SimpleNamespace(id="fallback")) == "fallback" + failures = [] + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: failures.append((board, str(error))), + ) + db = SimpleNamespace( + connect=lambda **_kwargs: (_ for _ in ()).throw(OSError("offline")) + ) + assert lanes._connect_healthy_board(db, "broken") is None + assert failures == [("broken", "offline")] + + +def test_orphan_recovery_bounds_registry_and_per_board_failures( + monkeypatch, +): + failures = [] + registry_db = SimpleNamespace( + complete_task=lambda *_args, expected_run_id=None, **_kwargs: True, + reclaim_task=lambda *_args, expected_run_id=None, **_kwargs: True, + list_boards=lambda **_kwargs: (_ for _ in ()).throw(OSError("registry")), + ) + monkeypatch.setitem( + sys.modules, + "hermes_cli", + SimpleNamespace(kanban_db=registry_db), + ) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: failures.append((board, str(error))), + ) + lanes.recover_orphans() + assert failures == [("board-registry", "registry")] + + class Connection: + def close(self): + failures.append(("close", "yes")) + + board_db = SimpleNamespace( + complete_task=registry_db.complete_task, + reclaim_task=registry_db.reclaim_task, + list_boards=lambda **_kwargs: [{"slug": ""}, {"slug": "broken"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + list_tasks=lambda _conn: (_ for _ in ()).throw(OSError("board read")), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=board_db)) + lanes.recover_orphans() + assert ("broken", "board read") in failures + assert ("close", "yes") in failures + + +def test_claim_ready_handles_zero_limit_empty_boards_and_claim_errors(monkeypatch): + monkeypatch.setitem( + sys.modules, + "hermes_cli", + SimpleNamespace(kanban_db=SimpleNamespace()), + ) + assert lanes.claim_ready(set(), 0) == [] + + class Connection: + def close(self): + return None + + task = SimpleNamespace( + id="t_claim", + status="ready", + assignee="cli-auto", + ) + db = SimpleNamespace( + list_boards=lambda **_kwargs: [{"slug": ""}, {"slug": "cassandra"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + recompute_ready=lambda _conn: None, + list_tasks=lambda _conn: [task], + claim_task=lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("claim raced") + ), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + assert lanes.claim_ready(set(), 1) == [] + + +class _Future: + def __init__(self): + self.polls = 0 + + def done(self): + self.polls += 1 + return self.polls >= 1 + + def result(self): + raise RuntimeError("worker failed") + + +class _Pool: + def __init__(self, max_workers): + self.max_workers = max_workers + self.future = _Future() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def submit(self, function, board, task_id): + assert callable(function) + assert (board, task_id) == ("cassandra", "t_loop") + return self.future + + +def test_ready_dispatch_loop_submits_and_reaps_failed_workers( + tmp_path: Path, + monkeypatch, + capsys, +): + db = SimpleNamespace() + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") + monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + monkeypatch.setattr( + lanes, + "initialize_kanban_capabilities", + lambda _db: lanes.KanbanCapabilities(True, True, True), + ) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) + claims = [[("cassandra", "t_loop")], []] + monkeypatch.setattr(lanes, "claim_ready", lambda *_args: claims.pop(0)) + monkeypatch.setattr( + lanes.concurrent.futures, + "ThreadPoolExecutor", + _Pool, + ) + sleeps = [] + + def stop_second_loop(_seconds): + sleeps.append(True) + if len(sleeps) == 2: + raise RuntimeError("stop loop") + + monkeypatch.setattr(lanes.time, "sleep", stop_second_loop) + + with pytest.raises(RuntimeError, match="stop loop"): + lanes.main() + + assert "worker future failed: worker failed" in capsys.readouterr().err + + +def test_deferred_dispatch_health_never_claims_new_work( + tmp_path: Path, + monkeypatch, +): + db = SimpleNamespace() + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") + monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + monkeypatch.setattr( + lanes, + "initialize_kanban_capabilities", + lambda _db: lanes.KanbanCapabilities(True, False, False), + ) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) + monkeypatch.setattr( + lanes, + "claim_ready", + lambda *_args: pytest.fail("deferred startup must not claim work"), + ) + monkeypatch.setattr(lanes.concurrent.futures, "ThreadPoolExecutor", _Pool) + monkeypatch.setattr( + lanes.time, + "sleep", + lambda _seconds: (_ for _ in ()).throw(RuntimeError("stop loop")), + ) + with pytest.raises(RuntimeError, match="stop loop"): + lanes.main() diff --git a/testing/tests/test_hermes_cli_evidence_edges.py b/testing/tests/test_hermes_cli_evidence_edges.py new file mode 100644 index 00000000..c28ff43f --- /dev/null +++ b/testing/tests/test_hermes_cli_evidence_edges.py @@ -0,0 +1,194 @@ +"""Failure-boundary coverage for immutable lane evidence and retention.""" + +from __future__ import annotations + +import errno +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_cli_support import _pending_terminal_record, lanes + + +def test_rename_noreplace_rejects_missing_kernel_support(monkeypatch): + monkeypatch.setattr(lanes.ctypes, "CDLL", lambda *_args, **_kwargs: object()) + with pytest.raises(OSError) as raised: + lanes._rename_noreplace("a", "b", source_dir=1, destination_dir=1) + assert raised.value.errno == errno.ENOSYS + + +def test_rename_noreplace_surfaces_kernel_errno(monkeypatch): + class Rename: + argtypes = None + restype = None + + def __call__(self, *_args): + return -1 + + monkeypatch.setattr( + lanes.ctypes, + "CDLL", + lambda *_args, **_kwargs: SimpleNamespace(renameat2=Rename()), + ) + monkeypatch.setattr(lanes.ctypes, "get_errno", lambda: errno.EEXIST) + with pytest.raises(FileExistsError): + lanes._rename_noreplace("a", "b", source_dir=1, destination_dir=1) + + +def test_retirement_reports_missing_replacement_collision_and_restoration( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "pending" + path.write_text("old", encoding="utf-8") + source = path.stat() + directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + path.unlink() + assert lanes._retire_terminal_entry( + path, + source, + board_descriptor=directory, + quarantine_descriptor=directory, + ) == "missing" + + path.write_text("new", encoding="utf-8") + assert lanes._retire_terminal_entry( + path, + source, + board_descriptor=directory, + quarantine_descriptor=directory, + ) == "replacement" + + replacement = path.stat() + monkeypatch.setattr( + lanes, + "_rename_noreplace", + lambda *_args, **_kwargs: (_ for _ in ()).throw(FileExistsError()), + ) + assert lanes._retire_terminal_entry( + path, + replacement, + board_descriptor=directory, + quarantine_descriptor=directory, + ) == "collision" + finally: + os.close(directory) + + +def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatch): + path = tmp_path / "pending" + path.write_text("old", encoding="utf-8") + source = path.stat() + directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + calls = [] + + def race(source_name, destination, **_kwargs): + calls.append((source_name, destination)) + if len(calls) == 1: + os.rename(source_name, destination, src_dir_fd=directory, dst_dir_fd=directory) + staged = tmp_path / destination + staged.unlink() + staged.write_text("replacement", encoding="utf-8") + else: + os.rename(source_name, destination, src_dir_fd=directory, dst_dir_fd=directory) + + monkeypatch.setattr(lanes, "_rename_noreplace", race) + try: + assert lanes._retire_terminal_entry( + path, + source, + board_descriptor=directory, + quarantine_descriptor=directory, + ) == "replacement" + finally: + os.close(directory) + assert path.read_text(encoding="utf-8") == "replacement" + + +def test_evidence_writer_bounds_payload_and_cleans_failed_temporary( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 8) + with pytest.raises(ValueError, match="bounded"): + lanes._write_json_noreplace(tmp_path / "large.json", {"value": "x" * 5000}) + + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 1024 * 1024) + real_retire = lanes._retire_terminal_entry + monkeypatch.setattr( + lanes, + "_rename_noreplace", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("rename failed")), + ) + monkeypatch.setattr( + lanes, + "_retire_terminal_entry", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("cleanup failed")), + ) + with pytest.raises(OSError, match="rename failed"): + lanes._write_json_noreplace(tmp_path / "failed.json", {"value": "small"}) + monkeypatch.setattr(lanes, "_retire_terminal_entry", real_retire) + + +def test_evidence_paths_and_collision_validation_fail_closed( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + identity = lanes.TerminalIdentity("cassandra", "t_evidence", 9, "pending") + document = _pending_terminal_record("cassandra", "t_evidence", 9, "accepted") + with pytest.raises(ValueError, match="evidence state"): + lanes._terminal_evidence_path(identity, "pending", document) + + monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False) + monkeypatch.setattr(lanes, "_terminal_evidence_identity", lambda _path: None) + with pytest.raises(OSError, match="identity is invalid"): + lanes._persist_prepared_evidence(identity, document) + with pytest.raises(OSError, match="conflict evidence identity"): + lanes._persist_conflict_evidence(identity, document, "loser") + + +def test_existing_evidence_with_different_result_is_rejected( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "pending") + document = _pending_terminal_record("cassandra", "t_evidence", 10, "accepted") + prepared_identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "prepared") + conflict_identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "conflict") + monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False) + monkeypatch.setattr( + lanes, + "_terminal_evidence_identity", + lambda path: conflict_identity if "conflict" in path.name else prepared_identity, + ) + monkeypatch.setattr(lanes, "_load_small_json", lambda _path: {}) + with pytest.raises(OSError, match="conflicting result"): + lanes._persist_prepared_evidence(identity, document) + with pytest.raises(OSError, match="different result"): + lanes._persist_conflict_evidence(identity, document, "loser") + + +def test_retention_failure_paths_are_bounded(tmp_path: Path, monkeypatch, capsys): + path = tmp_path / "artifact" + path.write_text("data", encoding="utf-8") + observed = path.stat() + monkeypatch.setattr( + lanes, + "_retire_terminal_entry", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("busy")), + ) + assert lanes._unlink_artifact_if_same(path, observed) is False + + monkeypatch.setattr(lanes, "LAST_ARTIFACT_GC", 0.0) + monkeypatch.setattr( + lanes, + "gc_lane_artifacts", + lambda **_kwargs: (_ for _ in ()).throw(OSError("volume")), + ) + assert lanes.maybe_gc_lane_artifacts(now=1000) == 0 + assert "retention deferred" in capsys.readouterr().err diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py new file mode 100644 index 00000000..07b9035d --- /dev/null +++ b/testing/tests/test_hermes_cli_execution_edges.py @@ -0,0 +1,218 @@ +"""Execution-loop edge coverage for preparation, callbacks, and recovery.""" + +from __future__ import annotations + +import sqlite3 +import sys +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace + +from testing.tests.test_hermes_cli_support import _completed_result, lanes + + +class _Connection: + def close(self): + return None + + +def test_missing_claim_is_a_noop(tmp_path: Path, monkeypatch): + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: None, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + lanes.execute_claim("cassandra", "missing") + + +def test_transient_callback_storage_errors_do_not_kill_provider( + tmp_path: Path, + monkeypatch, +): + task = SimpleNamespace( + id="t_callbacks", + status="running", + current_run_id=20, + assignee="cli-auto", + max_runtime_seconds=60, + ) + blocks = [] + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: task, + worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "exercise callbacks", + heartbeat_worker=lambda *_args, **_kwargs: (_ for _ in ()).throw( + sqlite3.OperationalError("heartbeat volume") + ), + add_comment=lambda *_args, **_kwargs: (_ for _ in ()).throw( + sqlite3.OperationalError("comment volume") + ), + block_task=lambda *_args, **kwargs: blocks.append(kwargs), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + monkeypatch.setattr(lanes, "KANBAN_STORAGE_ATTEMPTS", 1) + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: "claude") + route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) + monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route) + + def provider(*args, **_kwargs): + assert args[6]("still working") is True + return lanes.ProcessResult(1, "failed", None, False) + + monkeypatch.setattr(lanes, "run_provider", provider) + lanes.execute_claim("cassandra", "t_callbacks") + + assert blocks and blocks[0]["kind"] == "capability" + + +def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch): + task = SimpleNamespace( + id="t_restart", + status="running", + current_run_id=21, + assignee="cli-auto", + max_runtime_seconds=60, + ) + state_root = tmp_path / "lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + state_file = lanes.state_path("cassandra", "t_restart") + lanes.atomic_json(state_file, {"current_route": {"provider": "claude"}}) + blocks = [] + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: task, + worker_log_path=lambda *_args, **_kwargs: tmp_path / "missing.log", + _resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "resume", + add_comment=lambda *_args, **_kwargs: None, + heartbeat_worker=lambda *_args, **_kwargs: True, + block_task=lambda *_args, **kwargs: blocks.append(kwargs), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) + monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route) + handoffs = [] + monkeypatch.setattr( + lanes, + "git_handoff", + lambda _workspace, output: handoffs.append(output) or "handoff", + ) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult(1, "failed", None, False), + ) + + lanes.execute_claim("cassandra", "t_restart") + + assert handoffs == ["Previous provider log was unavailable after restart."] + assert blocks + + +def test_unreplayable_persistence_error_recovers_exact_run( + tmp_path: Path, + monkeypatch, +): + task = SimpleNamespace( + id="t_persist", + status="running", + current_run_id=22, + assignee="cli-auto", + max_runtime_seconds=60, + ) + comments = [] + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: task, + worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "finish", + add_comment=lambda _conn, _task, _author, body: comments.append(body), + heartbeat_worker=lambda *_args, **_kwargs: True, + block_task=lambda *_args, **_kwargs: None, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + monkeypatch.setattr( + lanes, + "select_route", + lambda *_args, **_kwargs: lanes.Route( + "codex", "gpt", "high", "p", "c", "r", 1, () + ), + ) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult( + 0, "", _completed_result("accepted"), False + ), + ) + monkeypatch.setattr( + lanes, + "_write_terminal_record", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk")), + ) + monkeypatch.setattr(lanes, "_has_pending_finalization", lambda *_args: False) + recovered = [] + monkeypatch.setattr( + lanes, + "_recover_exact_run", + lambda *args: recovered.append(args) or True, + ) + + lanes.execute_claim("cassandra", "t_persist") + + assert recovered and recovered[0][1].run_id == 22 + assert any("terminal replayable=False" in body for body in comments) + + +def test_unexpected_route_exception_blocks_the_exact_run( + tmp_path: Path, + monkeypatch, +): + task = SimpleNamespace( + id="t_route", + status="running", + current_run_id=23, + assignee="cli-auto", + max_runtime_seconds=60, + ) + blocks = [] + db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: task, + worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "route", + add_comment=lambda *_args, **_kwargs: None, + heartbeat_worker=lambda *_args, **_kwargs: True, + block_task=lambda *_args, **kwargs: blocks.append(kwargs), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + monkeypatch.setattr( + lanes, + "select_route", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("router down")), + ) + + lanes.execute_claim("cassandra", "t_route") + + assert blocks[0]["expected_run_id"] == 23 + assert "router down" in blocks[0]["reason"] diff --git a/testing/tests/test_hermes_cli_finalization_edges.py b/testing/tests/test_hermes_cli_finalization_edges.py new file mode 100644 index 00000000..0f2c2a2c --- /dev/null +++ b/testing/tests/test_hermes_cli_finalization_edges.py @@ -0,0 +1,164 @@ +"""Exact-run finalization guard and convergence edge coverage.""" + +from __future__ import annotations + +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace + +from testing.tests.test_hermes_cli_support import ( + _pending_terminal_record, + lanes, +) + + +class _Connection: + def close(self): + return None + + +def _db(task): + return SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda *_args: task, + reclaim_task=lambda *_args, expected_run_id=None, **_kwargs: True, + complete_task=lambda *_args, expected_run_id=None, + replay_ended_run_id=None, **_kwargs: False, + ) + + +def test_exact_recovery_rejects_missing_capability_and_nonauthoritative_tasks( + monkeypatch, +): + identity = lanes.TerminalIdentity("cassandra", "t_recover", 3, "pending") + assert lanes._recover_exact_run(_db(None), None, "missing") is False + + monkeypatch.setattr( + lanes, + "kanban_capabilities", + lambda _db: lanes.KanbanCapabilities(True, True, False), + ) + assert lanes._recover_exact_run(_db(None), identity, "old image") is False + + monkeypatch.setattr( + lanes, + "kanban_capabilities", + lambda _db: lanes.KanbanCapabilities(True, True, True), + ) + assert lanes._recover_exact_run(_db(None), identity, "missing task") is False + internal = SimpleNamespace( + status="running", + current_run_id=3, + assignee="internal", + ) + assert lanes._recover_exact_run(_db(internal), identity, "internal") is False + + +def test_exact_recovery_bounds_board_access_failure(monkeypatch): + identity = lanes.TerminalIdentity("cassandra", "t_recover", 4, "pending") + db = _db(SimpleNamespace(status="running", current_run_id=4, assignee="cli-auto")) + db.connect = lambda **_kwargs: (_ for _ in ()).throw(OSError("volume")) + errors = [] + monkeypatch.setattr(lanes, "KANBAN_STORAGE_ATTEMPTS", 1) + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: errors.append((board, str(error))), + ) + assert lanes._recover_exact_run(db, identity, "storage") is False + assert errors[-1] == ("cassandra", "volume") + + +def test_finalizer_guards_invalid_runtime_identity_and_missing_task(monkeypatch): + identity = object.__new__(lanes.TerminalIdentity) + object.__setattr__(identity, "board", "cassandra") + object.__setattr__(identity, "task_id", "t_invalid") + object.__setattr__(identity, "run_id", 0) + object.__setattr__(identity, "state", "pending") + assert lanes._finalize_document_db(_db(None), identity, {}) == "stale" + + valid = lanes.TerminalIdentity("cassandra", "missing", 5, "pending") + assert lanes._finalize_document_db(_db(None), valid, {}) == "stale" + + running = SimpleNamespace( + status="running", + current_run_id=5, + completed_run_id=None, + result=None, + ) + monkeypatch.setattr( + lanes, + "kanban_capabilities", + lambda _db: lanes.KanbanCapabilities(False, False, False), + ) + assert lanes._finalize_document_db(_db(running), valid, {}) == "deferred" + + +class _Snapshot: + def __init__(self, document): + self.document = document + self.closed = 0 + + def close(self): + self.closed += 1 + + +def test_resolve_pending_rejects_invalid_document_and_unknown_retirement( + tmp_path: Path, + monkeypatch, +): + identity = lanes.TerminalIdentity("cassandra", "t_pending", 6, "pending") + invalid = _Snapshot({}) + monkeypatch.setattr(lanes, "_open_terminal_snapshot", lambda *_args: invalid) + assert lanes._resolve_pending_after_winner(tmp_path / "pending", identity, {}) is False + assert invalid.closed == 1 + + valid_document = _pending_terminal_record("cassandra", "t_pending", 6, "winner") + unknown = _Snapshot(valid_document) + monkeypatch.setattr(lanes, "_open_terminal_snapshot", lambda *_args: unknown) + monkeypatch.setattr(lanes, "_retire_snapshot_after_db", lambda *_args: "deferred") + assert ( + lanes._resolve_pending_after_winner( + tmp_path / "pending", + identity, + valid_document, + ) + is False + ) + assert unknown.closed == 1 + + +def test_resolve_pending_stops_after_bounded_replacement_churn( + tmp_path: Path, + monkeypatch, +): + identity = lanes.TerminalIdentity("cassandra", "t_churn", 7, "pending") + document = _pending_terminal_record("cassandra", "t_churn", 7, "winner") + snapshots = [] + + def snapshot(*_args): + value = _Snapshot(document) + snapshots.append(value) + return value + + monkeypatch.setattr(lanes, "_open_terminal_snapshot", snapshot) + monkeypatch.setattr(lanes, "_retire_snapshot_after_db", lambda *_args: "replacement") + assert lanes._resolve_pending_after_winner(tmp_path / "pending", identity, document) is False + assert len(snapshots) == 8 + assert all(item.closed == 1 for item in snapshots) + + +def test_terminal_finalizer_closes_supplied_invalid_snapshots( + tmp_path: Path, + monkeypatch, +): + snapshot = _Snapshot({}) + assert lanes._finalize_terminal_record(_db(None), tmp_path / "bad", snapshot=snapshot) == "invalid" + assert snapshot.closed == 1 + + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + path = lanes._terminal_path(lanes.state_path("board", "task"), 8) + empty = _Snapshot(None) + assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid" + assert empty.closed == 1 diff --git a/testing/tests/test_hermes_cli_foundation_coverage.py b/testing/tests/test_hermes_cli_foundation_coverage.py new file mode 100644 index 00000000..71f0c44a --- /dev/null +++ b/testing/tests/test_hermes_cli_foundation_coverage.py @@ -0,0 +1,207 @@ +"""Adversarial branch coverage for CLI lane foundation helpers.""" + +from __future__ import annotations + +import json +import sqlite3 +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace +from urllib.error import URLError + +import pytest + +from testing.tests.test_hermes_cli_support import _completed_result, lanes + + +def test_board_context_serializes_objects_and_storage_failure_is_bounded( + monkeypatch, +): + db = SimpleNamespace(build_worker_context=lambda _conn, _task: {"priority": 3}) + assert lanes._task_context(db, object(), "t_ctx") == '{\n "priority": 3\n}' + + connections = [] + + class Connection: + def close(self): + connections.append("closed") + + failing = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + ) + monkeypatch.setattr(lanes, "KANBAN_STORAGE_ATTEMPTS", 2) + monkeypatch.setattr(lanes.time, "sleep", lambda _seconds: None) + with pytest.raises(sqlite3.OperationalError, match="volume unavailable"): + lanes._board_call( + failing, + "cassandra", + lambda _conn: (_ for _ in ()).throw( + sqlite3.OperationalError("volume unavailable") + ), + ) + assert connections == ["closed", "closed"] + + +def test_atomic_json_closes_an_unwrapped_descriptor_on_open_failure( + tmp_path: Path, + monkeypatch, +): + destination = tmp_path / "state.json" + real_fdopen = lanes.os.fdopen + descriptors = [] + + def fail_fdopen(descriptor, *_args, **_kwargs): + descriptors.append(descriptor) + raise OSError("cannot wrap descriptor") + + monkeypatch.setattr(lanes.os, "fdopen", fail_fdopen) + with pytest.raises(OSError, match="cannot wrap"): + lanes.atomic_json(destination, {"safe": True}) + monkeypatch.setattr(lanes.os, "fdopen", real_fdopen) + + assert descriptors + with pytest.raises(OSError): + lanes.os.fstat(descriptors[0]) + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_json_and_terminal_path_helpers_fail_closed(tmp_path: Path, monkeypatch): + invalid = tmp_path / "invalid.json" + invalid.write_bytes(b"\xff") + assert lanes.load_json(invalid) == {} + array = tmp_path / "array.json" + array.write_text("[]", encoding="utf-8") + assert lanes.load_json(array) == {} + + with pytest.raises(ValueError, match="journal state"): + lanes._terminal_path(tmp_path / "task.json", 1, "prepared") + with pytest.raises(ValueError, match="SQLite"): + lanes._terminal_path(tmp_path / "task.json", True) + + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + outside = tmp_path / "outside.run-1.terminal.pending.json" + assert lanes._terminal_identity(outside) is None + nested = lanes.STATE_ROOT / "board" / "nested" / "x.json" + assert lanes._terminal_identity(nested) is None + missing = lanes.STATE_ROOT / "board" / "t.run-1.terminal.pending.json" + assert lanes._terminal_identity(missing) is None + assert lanes._terminal_evidence_identity(outside) is None + assert lanes._terminal_evidence_identity(nested) is None + missing_evidence = lanes.STATE_ROOT / "board" / ( + "t.run-1.terminal.prepared-" + "a" * 32 + ".json" + ) + assert lanes._terminal_evidence_identity(missing_evidence) is None + + +def test_goal_helpers_cover_compaction_and_deterministic_failures(): + compacted = lanes.cli_lane_goal._bounded("a" * 100, 40) + assert len(compacted) == 40 + assert "compacted" in compacted + assert lanes.cli_lane_goal.unfinished_result_reason( + {"status": "incomplete", "summary": ""} + ) == "worker explicitly reported incomplete work" + assert lanes.cli_lane_goal.unfinished_result_reason( + {"status": "blocked", "summary": "waiting"} + ) is None + assert "blockers" in lanes.cli_lane_goal.unfinished_result_reason( + {"status": "completed", "summary": "done", "blockers": ["remote"]} + ) + assert "unfinished" in lanes.cli_lane_goal.unfinished_result_reason( + { + "status": "completed", + "summary": "verification is pending", + "blockers": [], + "tests_run": [], + } + ) + accepted, reason = lanes.cli_lane_goal.judge_goal_completion( + "finish", + {"status": "incomplete", "summary": "still running"}, + open_request=lambda *_args, **_kwargs: pytest.fail("network must not run"), + ) + assert accepted is False + assert reason + + +def test_prompt_artifact_and_json_helpers_reject_unsafe_inputs( + tmp_path: Path, +): + workspace = tmp_path / "workspace" + workspace.mkdir() + artifact = workspace / "evidence.txt" + artifact.write_text("evidence", encoding="utf-8") + outside = tmp_path / "outside.txt" + outside.write_text("outside", encoding="utf-8") + directory = workspace / "directory" + directory.mkdir() + + assert lanes.workspace_artifacts(workspace, "not-a-list") == [] + assert lanes.workspace_artifacts( + workspace, + [None, "", "missing", outside, directory, artifact, str(artifact)], + ) == [str(artifact.resolve())] + assert lanes._extract_json({"status": "completed"}) == {"status": "completed"} + assert lanes._extract_json(42) is None + assert lanes._extract_json("not json") is None + assert lanes._extract_json('prefix {"status":"blocked"} suffix') == { + "status": "blocked" + } + + +def test_provider_event_parsing_persists_sessions_and_nested_results( + tmp_path: Path, +): + state_file = tmp_path / "state.json" + state = {} + assert lanes._event_payload("codex", "not-json", state, state_file) is None + assert lanes._event_payload("codex", "[]", state, state_file) is None + assert ( + lanes._event_payload( + "codex", + json.dumps({"type": "thread.started", "thread": {"id": "thread-1"}}), + state, + state_file, + ) + is None + ) + assert state["codex_thread_id"] == "thread-1" + result = lanes._event_payload( + "claude", + json.dumps( + { + "session_id": "session-1", + "item": {"content": json.dumps(_completed_result("nested"))}, + } + ), + state, + state_file, + ) + assert result and result["summary"] == "nested" + assert state["claude_session_id"] == "session-1" + + +@pytest.mark.parametrize("assignee", ["cli-codex-max", "worker-auto", ""]) +def test_routing_rejects_invalid_assignees(assignee: str): + with pytest.raises(ValueError, match="unsupported"): + lanes.parse_assignee(assignee) + + +@pytest.mark.parametrize( + "target", + ["codex/model/high", "worker/local/model/high", "worker/codex/model/max"], +) +def test_worker_target_decoder_rejects_invalid_targets(target: str): + with pytest.raises(RuntimeError): + lanes._decode_worker_target(target) + + +def test_switchyard_network_failure_is_explicit(): + with pytest.raises(RuntimeError, match="routing failed"): + lanes.select_route( + "objective", + "cli-auto", + open_request=lambda *_args, **_kwargs: (_ for _ in ()).throw( + URLError("offline") + ), + ) diff --git a/testing/tests/test_hermes_cli_provider_edges.py b/testing/tests/test_hermes_cli_provider_edges.py new file mode 100644 index 00000000..21e57453 --- /dev/null +++ b/testing/tests/test_hermes_cli_provider_edges.py @@ -0,0 +1,173 @@ +"""Process lifecycle and provider artifact edge coverage for CLI workers.""" + +from __future__ import annotations + +import json +import os +import signal +import subprocess +import sys +from pathlib import Path + +from testing.tests.test_hermes_cli_support import _completed_result, lanes + + +def test_stream_process_enforces_runtime_and_lease_boundaries( + tmp_path: Path, +): + timed_out = lanes.stream_process( + [sys.executable, "-c", "import time; time.sleep(30)"], + provider="codex", + cwd=tmp_path, + env=dict(os.environ), + log_path=tmp_path / "timeout.log", + state={}, + state_file=tmp_path / "timeout-state.json", + heartbeat=lambda _note: True, + max_runtime=-1, + ) + assert "maximum runtime" in timed_out.output + assert timed_out.returncode != 0 + + lease_lost = lanes.stream_process( + [sys.executable, "-c", "import time; time.sleep(30)"], + provider="claude", + cwd=tmp_path, + env=dict(os.environ), + log_path=tmp_path / "lease.log", + state={}, + state_file=tmp_path / "lease-state.json", + heartbeat=lambda _note: False, + max_runtime=60, + ) + assert "lease was lost" in lease_lost.output + assert lease_lost.returncode != 0 + + +def test_process_identity_and_signal_failures_are_bounded(monkeypatch): + assert lanes._process_record(999999999) is None + monkeypatch.setattr( + lanes, + "_process_identity_matches", + lambda _pid, _start: True, + ) + monkeypatch.setattr( + lanes.os, + "killpg", + lambda *_args: (_ for _ in ()).throw(ProcessLookupError()), + ) + monkeypatch.setattr( + lanes.os, + "kill", + lambda *_args: (_ for _ in ()).throw(ProcessLookupError()), + ) + lanes._signal_worker_tree(10, {11: (12, 13)}, signal.SIGTERM) + + +def test_descendant_snapshot_follows_multiple_generations(monkeypatch): + entries = [Path("/proc/self"), Path("/proc/10"), Path("/proc/11"), Path("/proc/12")] + monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries)) + records = { + 10: (1, 10, 100), + 11: (10, 11, 101), + 12: (11, 12, 102), + } + monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid)) + assert lanes._descendant_processes(10) == {11: (11, 101), 12: (12, 102)} + + +def test_terminate_escalates_after_term_timeout(monkeypatch): + class Process: + pid = 321 + waits = 0 + + @classmethod + def poll(cls): + return None + + @classmethod + def wait(cls, timeout): + cls.waits += 1 + if cls.waits == 1: + raise subprocess.TimeoutExpired("worker", timeout) + return 0 + + signals = [] + monkeypatch.setattr(lanes, "_descendant_processes", lambda _pid: {}) + monkeypatch.setattr( + lanes, + "_signal_worker_tree", + lambda _pid, _descendants, sig: signals.append(sig), + ) + lanes._terminate_worker_process(Process()) + assert signals == [signal.SIGTERM, signal.SIGKILL] + assert Process.waits == 2 + + +def test_codex_result_collision_and_chmod_failure_preserve_result( + tmp_path: Path, + monkeypatch, +): + state = {"run_id": 7, "result_sequence": 0} + state_file = tmp_path / "task.json" + first = lanes._result_path(state_file, 7, 1) + first.write_text("prior", encoding="utf-8") + result_paths = [] + + def stream(command, **_kwargs): + result_path = Path(command[command.index("-o") + 1]) + result_path.write_text(json.dumps(_completed_result("new result")), encoding="utf-8") + result_paths.append(result_path) + return lanes.ProcessResult(0, "", None, False) + + monkeypatch.setattr(lanes, "stream_process", stream) + monkeypatch.setattr( + lanes.Path, + "chmod", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("readonly")), + ) + route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) + result = lanes.run_provider( + route, + "work", + tmp_path, + state, + state_file, + tmp_path / "log", + lambda _note: True, + 60, + ) + assert result.structured["summary"] == "new result" + assert result_paths[0].name.endswith("provider-2.result.json") + + +def test_codex_missing_thread_skips_colliding_restart_result( + tmp_path: Path, + monkeypatch, +): + state = {"run_id": 8, "codex_thread_id": "missing"} + state_file = tmp_path / "task.json" + second = lanes._result_path(state_file, 8, 2) + second.write_text("collision", encoding="utf-8") + calls = [] + + def stream(command, **_kwargs): + calls.append(command) + if len(calls) == 1: + return lanes.ProcessResult(1, lanes.NO_CODEX_THREAD, None, False) + return lanes.ProcessResult(0, "", None, False) + + monkeypatch.setattr(lanes, "stream_process", stream) + route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) + lanes.run_provider( + route, + "work", + tmp_path, + state, + state_file, + tmp_path / "log", + lambda _note: True, + 60, + ) + restart = Path(calls[1][calls[1].index("-o") + 1]) + assert restart.name.endswith("provider-3.result.json") diff --git a/testing/tests/test_hermes_cli_quarantine_edges.py b/testing/tests/test_hermes_cli_quarantine_edges.py new file mode 100644 index 00000000..eff30408 --- /dev/null +++ b/testing/tests/test_hermes_cli_quarantine_edges.py @@ -0,0 +1,185 @@ +"""Degraded quarantine-path coverage for terminal journal safety.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from testing.tests.test_hermes_cli_support import lanes + + +def test_quarantine_refuses_missing_or_unsafe_board_directory( + tmp_path: Path, + capsys, +): + path = tmp_path / "missing" / "journal" + assert lanes._quarantine_terminal(path, None, "invalid") == path + assert "unsafe board directory" in capsys.readouterr().err + + regular_parent = tmp_path / "regular" + regular_parent.write_text("not a directory", encoding="utf-8") + unsafe = regular_parent / "journal" + assert lanes._quarantine_terminal(unsafe, None, "invalid") == unsafe + + +def test_quarantine_accepts_a_lexically_outside_but_pinned_path( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "other-root") + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"invalid") + + destination = lanes._quarantine_terminal(path, None, "bad reason / outside") + + assert destination.parent == board / "quarantine" + assert not path.exists() + + +def test_quarantine_rejects_changed_pinned_board_identity( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"invalid") + snapshot = lanes._open_terminal_recovery_snapshot(path) + assert snapshot is not None + real_fstat = lanes.os.fstat + + def changed_board(descriptor): + observed = real_fstat(descriptor) + if descriptor == snapshot.directory_descriptor: + return os.stat_result( + (*observed[:1], observed.st_ino + 1, *observed[2:]) + ) + return observed + + monkeypatch.setattr(lanes.os, "fstat", changed_board) + try: + lanes._quarantine_terminal(path, None, "changed-board", snapshot=snapshot) + finally: + snapshot.close() + assert not path.exists() + assert "degraded safely" in capsys.readouterr().err + + +def test_quarantine_rejects_source_replaced_during_open( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"original") + replacement = board / "replacement" + replacement.write_bytes(b"replacement") + real_open = lanes.os.open + swapped = {"value": False} + + def swap_before_source_open(target, flags, *args, **kwargs): + if target == path.name and kwargs.get("dir_fd") is not None and not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(lanes.os, "open", swap_before_source_open) + lanes._quarantine_terminal(path, None, "source-race") + assert path.exists() + assert path.read_bytes() == b"replacement" + assert "degraded safely" in capsys.readouterr().err + + +def test_quarantine_unlinks_an_invalid_destination_before_degrading( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"invalid") + real_fstat = lanes.os.fstat + target_descriptors = set() + real_fdopen = lanes.os.fdopen + + def remember_target(descriptor, *args, **kwargs): + target_descriptors.add(descriptor) + return real_fdopen(descriptor, *args, **kwargs) + + def unsafe_target(descriptor): + observed = real_fstat(descriptor) + if descriptor in target_descriptors: + return os.stat_result( + (*observed[:3], 0o100644, *observed[4:]) + ) + return observed + + monkeypatch.setattr(lanes.os, "fdopen", remember_target) + monkeypatch.setattr(lanes.os, "fstat", unsafe_target) + lanes._quarantine_terminal(path, None, "unsafe-target") + + assert not list((board / "quarantine").glob("*.quarantine")) + assert "degraded safely" in capsys.readouterr().err + + +def test_quarantine_destination_collision_is_bounded( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"invalid") + real_open = lanes.os.open + + def collide_on_destination(target, flags, *args, **kwargs): + if ( + isinstance(target, str) + and target.endswith(".quarantine") + and flags & os.O_EXCL + ): + raise FileExistsError(target) + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(lanes.os, "open", collide_on_destination) + destination = lanes._quarantine_terminal(path, None, "collision") + + assert destination == path + assert not path.exists() + assert "could not reserve" in capsys.readouterr().err + + +def test_quarantine_deferred_retirement_does_not_raise( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + board = tmp_path / "board" + board.mkdir() + path = board / "journal" + path.write_bytes(b"invalid") + calls = {"value": 0} + + def fail_retirement(*_args, **_kwargs): + calls["value"] += 1 + raise OSError("retirement unavailable") + + monkeypatch.setattr(lanes, "_retire_terminal_entry", fail_retirement) + destination = lanes._quarantine_terminal(path, None, "retire-failure") + + assert destination.exists() + assert path.exists() + assert calls["value"] == 2 + assert "retirement=deferred" in capsys.readouterr().err diff --git a/testing/tests/test_hermes_cli_records_edges.py b/testing/tests/test_hermes_cli_records_edges.py new file mode 100644 index 00000000..3362466a --- /dev/null +++ b/testing/tests/test_hermes_cli_records_edges.py @@ -0,0 +1,230 @@ +"""Bounded-read and terminal-schema edge coverage for CLI lane records.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from testing.tests.test_hermes_cli_support import ( + _completed_result, + _pending_terminal_record, + lanes, +) + + +def test_bounded_reader_stops_at_eof_and_zero_limit(tmp_path: Path): + path = tmp_path / "payload" + path.write_bytes(b"abc") + descriptor = os.open(path, os.O_RDONLY) + try: + assert lanes._read_bounded(descriptor, 10) == b"abc" + os.lseek(descriptor, 0, os.SEEK_SET) + assert lanes._read_bounded(descriptor, 0) == b"" + finally: + os.close(descriptor) + + +def test_small_snapshot_rejects_nonregular_hardlinked_oversized_and_nonobject( + tmp_path: Path, + monkeypatch, +): + directory = tmp_path / "directory" + directory.mkdir() + assert lanes._open_small_json_snapshot(directory) is None + + original = tmp_path / "original" + original.write_text("{}", encoding="utf-8") + hardlink = tmp_path / "hardlink" + os.link(original, hardlink) + assert lanes._open_small_json_snapshot(hardlink) is None + + oversized = tmp_path / "oversized" + oversized.write_text("{}", encoding="utf-8") + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 1) + assert lanes._open_small_json_snapshot(oversized) is None + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 1024 * 1024) + + array = tmp_path / "array" + array.write_text("[]", encoding="utf-8") + assert lanes._open_small_json_snapshot(array) is None + malformed = tmp_path / "malformed" + malformed.write_bytes(b"\xff") + assert lanes._open_small_json_snapshot(malformed) is None + + +def test_small_snapshot_rejects_path_replaced_after_read( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "record" + path.write_text("{}", encoding="utf-8") + real_stat = lanes.os.stat + calls = {"value": 0} + + def changed_stat(target, *args, **kwargs): + result = real_stat(target, *args, **kwargs) + if target == path.name and kwargs.get("dir_fd") is not None: + calls["value"] += 1 + if calls["value"] == 1: + return os.stat_result( + (*result[:6], result.st_size + 1, *result[7:]) + ) + return result + + monkeypatch.setattr(lanes.os, "stat", changed_stat) + assert lanes._open_small_json_snapshot(path) is None + + +def test_recovery_snapshot_classifies_nonregular_and_open_failure( + tmp_path: Path, + monkeypatch, +): + fifo = tmp_path / "fifo" + os.mkfifo(fifo) + snapshot = lanes._open_terminal_recovery_snapshot(fifo) + assert snapshot is not None + assert snapshot.invalid_reason == "non-regular" + snapshot.close() + + path = tmp_path / "record" + path.write_text("{}", encoding="utf-8") + real_open = lanes.os.open + + def fail_file_open(target, flags, *args, **kwargs): + if target == path.name: + raise OSError("denied") + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(lanes.os, "open", fail_file_open) + snapshot = lanes._open_terminal_recovery_snapshot(path) + assert snapshot is not None + assert snapshot.invalid_reason == "open-failed" + snapshot.close() + + +def test_recovery_snapshot_detects_identity_change_during_open( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "record" + path.write_text("{}", encoding="utf-8") + replacement = tmp_path / "replacement" + replacement.write_text('{"replacement":true}', encoding="utf-8") + real_open = lanes.os.open + swapped = {"value": False} + + def swap_before_file_open(target, flags, *args, **kwargs): + if target == path.name and not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(lanes.os, "open", swap_before_file_open) + snapshot = lanes._open_terminal_recovery_snapshot(path) + assert snapshot is not None + assert snapshot.invalid_reason == "identity-changed-during-open" + snapshot.close() + + +def test_recovery_snapshot_bounds_hardlinks_oversize_and_bad_payloads( + tmp_path: Path, + monkeypatch, +): + original = tmp_path / "original" + original.write_text("{}", encoding="utf-8") + hardlink = tmp_path / "hardlink" + os.link(original, hardlink) + snapshot = lanes._open_terminal_recovery_snapshot(hardlink) + assert snapshot is not None and snapshot.invalid_reason == "hardlinked" + snapshot.close() + + oversized = tmp_path / "oversized" + oversized.write_bytes(b"x" * 128) + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 64) + monkeypatch.setattr(lanes, "QUARANTINE_HASH_BYTES", 8) + snapshot = lanes._open_terminal_recovery_snapshot(oversized) + assert snapshot is not None and snapshot.invalid_reason == "oversized" + assert snapshot.prefix == b"x" * 8 + snapshot.close() + + monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 1024 * 1024) + array = tmp_path / "array" + array.write_text("[]", encoding="utf-8") + snapshot = lanes._open_terminal_recovery_snapshot(array) + assert snapshot is not None and snapshot.invalid_reason == "non-object-payload" + snapshot.close() + malformed = tmp_path / "malformed" + malformed.write_bytes(b"\xff") + snapshot = lanes._open_terminal_recovery_snapshot(malformed) + assert snapshot is not None and snapshot.invalid_reason == "malformed-payload" + snapshot.close() + + +def test_snapshot_loaders_return_empty_for_missing_or_wrong_identity( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + identity = lanes.TerminalIdentity("board", "task", 1, "pending") + missing = lanes.STATE_ROOT / "board" / "task.run-1.terminal.pending.json" + assert lanes._open_terminal_snapshot(missing, identity) is None + assert lanes._load_terminal_json(missing, identity) == {} + assert lanes._load_small_json(missing) == {} + + +def test_candidate_persistence_skips_existing_sequence(tmp_path: Path): + state_file = tmp_path / "task.json" + state = {"board": "board", "task_id": "task", "run_id": 4} + route = lanes.Route("codex", "gpt", "high", "profile", "test", "test", 1, ()) + first = lanes._candidate_path(state_file, 4, 1) + first.write_text("collision", encoding="utf-8") + path = lanes._persist_candidate( + state, + state_file, + _completed_result(), + route=route, + returncode=0, + goal_turn=2, + ) + assert path.name.endswith("candidate-2.json") + assert state["candidate_sequence"] == 2 + + +@pytest.mark.parametrize( + "mutation", + [ + lambda record: [], + lambda record: {**record, "board": ""}, + lambda record: {**record, "expected_run_id": True}, + lambda record: {**record, "metadata": []}, + lambda record: {**record, "result": "not-json"}, + lambda record: {**record, "result": "[]"}, + lambda record: { + **record, + "result": json.dumps({**_completed_result(), "extra": True}), + }, + lambda record: { + **record, + "result": json.dumps({**_completed_result(), "summary": ""}), + }, + lambda record: { + **record, + "result": json.dumps({**_completed_result(), "tests_run": [1]}), + }, + ], +) +def test_terminal_schema_rejects_each_malformed_boundary(mutation): + record = _pending_terminal_record("board", "task", 1, "done") + assert lanes._terminal_record_valid(mutation(record)) is False + + +def test_terminal_schema_validates_optional_identity(): + record = _pending_terminal_record("board", "task", 1, "done") + assert lanes._terminal_record_valid(record) is True + assert lanes._terminal_record_valid( + record, + lanes.TerminalIdentity("other", "task", 1, "pending"), + ) is False diff --git a/testing/tests/test_hermes_cli_recovery_edges.py b/testing/tests/test_hermes_cli_recovery_edges.py new file mode 100644 index 00000000..27ba6773 --- /dev/null +++ b/testing/tests/test_hermes_cli_recovery_edges.py @@ -0,0 +1,298 @@ +"""Recovery convergence and hidden-authority edge coverage.""" + +from __future__ import annotations + +import hashlib +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_cli_support import ( + _pending_terminal_record, + lanes, +) + + +class _Snapshot: + def __init__(self, document=None): + self.document = document + self.closed = 0 + + def close(self): + self.closed += 1 + + +def test_terminal_absence_distinguishes_present_missing_and_inaccessible( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "journal" + path.write_text("data", encoding="utf-8") + assert lanes._terminal_entry_absent(path) is False + path.unlink() + assert lanes._terminal_entry_absent(path) is True + monkeypatch.setattr( + lanes.os, + "open", + lambda *_args, **_kwargs: (_ for _ in ()).throw(PermissionError("denied")), + ) + assert lanes._terminal_entry_absent(path) is False + + +def _staged_path(root: Path, identity, document: dict) -> tuple[Path, Path]: + if identity.state in {"pending", "committed"}: + canonical = lanes._terminal_path( + lanes.state_path(identity.board, identity.task_id), + identity.run_id, + identity.state, + ) + else: + canonical = lanes._terminal_evidence_path(identity, identity.state, document) + digest = hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16] + return root / identity.board / f".retire.{digest}.{'a' * 16}.0", canonical + + +@pytest.mark.parametrize("state", ["prepared", "conflict"]) +def test_staged_authority_accepts_valid_evidence_states( + tmp_path: Path, + monkeypatch, + state: str, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + identity = lanes.TerminalIdentity("board", "task", 5, state) + document = _pending_terminal_record("board", "task", 5, "accepted") + document["kanban_state"] = state + staged, canonical = _staged_path(lanes.STATE_ROOT, identity, document) + staged.parent.mkdir(parents=True) + authority = lanes._staged_terminal_authority(staged, _Snapshot(document)) + assert authority == (identity, canonical) + + +def test_staged_authority_rejects_outside_nested_and_unknown_state( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + document = _pending_terminal_record("board", "task", 5, "accepted") + outside = tmp_path / ".retire.x" + assert lanes._staged_terminal_authority(outside, _Snapshot(document)) is None + nested = lanes.STATE_ROOT / "board" / "nested" / ".retire.name" + assert lanes._staged_terminal_authority(nested, _Snapshot(document)) is None + board = lanes.STATE_ROOT / "board" + board.mkdir(parents=True) + staged = board / f".retire.{'a' * 16}.{'b' * 16}.0" + document["kanban_state"] = "unknown" + assert lanes._staged_terminal_authority(staged, _Snapshot(document)) is None + + +def test_restore_staged_authority_handles_every_first_writer_state( + tmp_path: Path, + monkeypatch, +): + document = _pending_terminal_record("board", "task", 6, "accepted") + canonical = tmp_path / "canonical" + pending = lanes.TerminalIdentity("board", "task", 6, "pending") + calls = [] + monkeypatch.setattr( + lanes, + "_persist_prepared_evidence", + lambda identity, value: calls.append(("prepared", identity, value)), + ) + lanes._restore_staged_terminal_authority(pending, canonical, document) + assert calls[0][0] == "prepared" + + monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: True) + committed = lanes.TerminalIdentity("board", "task", 6, "committed") + lanes._restore_staged_terminal_authority(committed, canonical, document) + + monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False) + monkeypatch.setattr(lanes, "_load_small_json", lambda _path: {}) + monkeypatch.setattr( + lanes, + "_persist_conflict_evidence", + lambda *args: calls.append(("conflict", *args)), + ) + lanes._restore_staged_terminal_authority(committed, canonical, document) + assert calls[-1][0] == "conflict" + + prepared = lanes.TerminalIdentity("board", "task", 6, "prepared") + monkeypatch.setattr(lanes, "_terminal_evidence_valid", lambda *_args: False) + with pytest.raises(OSError, match="conflicting result"): + lanes._restore_staged_terminal_authority(prepared, canonical, document) + + +def test_restore_staged_authority_accepts_identical_existing_evidence( + tmp_path: Path, + monkeypatch, +): + document = _pending_terminal_record("board", "task", 7, "accepted") + document["kanban_state"] = "prepared" + identity = lanes.TerminalIdentity("board", "task", 7, "prepared") + monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False) + monkeypatch.setattr(lanes, "_load_small_json", lambda _path: document) + monkeypatch.setattr(lanes, "_terminal_evidence_valid", lambda *_args: True) + lanes._restore_staged_terminal_authority(identity, tmp_path / "existing", document) + + +def test_staging_recovery_bounds_missing_snapshots_and_restore_errors( + monkeypatch, +): + path = Path("/tmp/staged") + monkeypatch.setattr(lanes, "_retirement_staging_paths", lambda: [path]) + monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: None) + assert lanes._recover_retirement_staging() == 0 + + identity = lanes.TerminalIdentity("board", "task", 8, "pending") + snapshot = _Snapshot(_pending_terminal_record("board", "task", 8, "accepted")) + monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: snapshot) + monkeypatch.setattr( + lanes, + "_staged_terminal_authority", + lambda *_args: (identity, Path("/tmp/canonical")), + ) + monkeypatch.setattr( + lanes, + "_restore_staged_terminal_authority", + lambda *_args: (_ for _ in ()).throw(OSError("restore")), + ) + errors = [] + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: errors.append((board, str(error))), + ) + assert lanes._recover_retirement_staging() == 0 + assert errors == [("board", "restore")] + assert snapshot.closed == 1 + + +def test_drain_staging_ignores_disappearing_entries(monkeypatch): + class Vanished: + def stat(self, **_kwargs): + raise FileNotFoundError + + monkeypatch.setattr(lanes, "_retirement_staging_paths", lambda: [Vanished()]) + assert lanes._drain_retirement_staging() == 0 + + +def _prepared_path(tmp_path: Path, monkeypatch) -> tuple[Path, dict]: + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + pending = _pending_terminal_record("board", "task", 9, "accepted") + identity = lanes.TerminalIdentity("board", "task", 9, "pending") + path = lanes._persist_prepared_evidence(identity, pending) + return path, pending + + +def test_prepared_recovery_defers_without_discarding_evidence( + tmp_path: Path, + monkeypatch, +): + path, _document = _prepared_path(tmp_path, monkeypatch) + monkeypatch.setattr(lanes, "_finalize_document_db", lambda *_args: "deferred") + assert lanes._recover_prepared_finalizations(object()) == 0 + assert path.exists() + + +def test_prepared_recovery_bounds_db_and_publication_exceptions( + tmp_path: Path, + monkeypatch, +): + path, _document = _prepared_path(tmp_path, monkeypatch) + errors = [] + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: errors.append((board, str(error))), + ) + monkeypatch.setattr( + lanes, + "_finalize_document_db", + lambda *_args: (_ for _ in ()).throw(OSError("database")), + ) + assert lanes._recover_prepared_finalizations(object()) == 0 + assert errors[-1] == ("board", "database") + assert path.exists() + + monkeypatch.setattr(lanes, "_finalize_document_db", lambda *_args: "committed") + monkeypatch.setattr( + lanes, + "_promote_prepared_evidence", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("publish")), + ) + assert lanes._recover_prepared_finalizations(object()) == 0 + assert errors[-1] == ("board", "publish") + + +def test_pending_recovery_bounds_finalizer_errors_and_replacement_churn( + tmp_path: Path, + monkeypatch, + capsys, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + path = lanes._terminal_path(lanes.state_path("board", "task"), 10) + lanes.atomic_json(path, _pending_terminal_record("board", "task", 10, "accepted")) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=object())) + errors = [] + monkeypatch.setattr( + lanes, + "_record_board_access_error", + lambda board, error: errors.append((board, str(error))), + ) + monkeypatch.setattr( + lanes, + "_finalize_terminal_record", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("finalize")), + ) + assert lanes.recover_pending_finalizations() == 0 + assert errors[-1] == ("board", "finalize") + + def invalid(*_args, snapshot=None, **_kwargs): + snapshot.close() + return "invalid" + + monkeypatch.setattr(lanes, "_finalize_terminal_record", invalid) + assert lanes.recover_pending_finalizations() == 0 + assert "replacement churn deferred" in capsys.readouterr().err + + +def test_pending_recovery_classifies_foreign_identity( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + path = lanes._terminal_path(lanes.state_path("board", "task"), 11) + lanes.atomic_json(path, _pending_terminal_record("other", "task", 11, "accepted")) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=object())) + reasons = [] + + def quarantine(_path, _identity, reason, *, snapshot): + reasons.append(reason) + os.unlink(_path) + + monkeypatch.setattr(lanes, "_quarantine_terminal", quarantine) + monkeypatch.setattr(lanes, "_recover_exact_run", lambda *_args: False) + assert lanes.recover_pending_finalizations() == 0 + assert reasons == ["foreign-identity"] + + +def test_pending_guard_rejects_invalid_run_and_inaccessible_or_bad_artifacts( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + assert lanes._has_pending_finalization("board", "task", True) is False + path = lanes._terminal_path(lanes.state_path("board", "task"), 12) + path.parent.mkdir(parents=True) + path.write_text("bad", encoding="utf-8") + real_stat = Path.stat + + def denied(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError("denied") + return real_stat(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", denied) + assert lanes._has_pending_finalization("board", "task", 12) is False diff --git a/testing/tests/test_hermes_cli_retention_edges.py b/testing/tests/test_hermes_cli_retention_edges.py new file mode 100644 index 00000000..f7deb6a2 --- /dev/null +++ b/testing/tests/test_hermes_cli_retention_edges.py @@ -0,0 +1,89 @@ +"""Retained terminal classification and filesystem-failure edge coverage.""" + +from __future__ import annotations + +from pathlib import Path + +from testing.tests.test_hermes_cli_support import ( + _pending_terminal_record, + lanes, +) + + +class _Snapshot: + document = None + invalid_reason = "malformed" + + def __init__(self): + self.closed = 0 + + def close(self): + self.closed += 1 + + +def test_retained_classifier_handles_missing_snapshot(tmp_path: Path, monkeypatch): + path = tmp_path / "task.run-1.terminal.committed.json" + monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: None) + assert lanes._quarantine_invalid_retained_terminal(path) == (True, False) + + +def test_retained_classifier_accepts_valid_committed_evidence( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") + state_file = lanes.state_path("board", "task") + path = lanes._terminal_path(state_file, 2, "committed") + record = _pending_terminal_record("board", "task", 2, "accepted") + record["kanban_state"] = "committed" + lanes.atomic_json(path, record) + assert lanes._quarantine_invalid_retained_terminal(path) == (False, False) + + +def test_retained_classifier_bounds_post_quarantine_stat_error( + tmp_path: Path, + monkeypatch, +): + path = tmp_path / "task.run-3.terminal.committed.json" + snapshot = _Snapshot() + monkeypatch.setattr( + lanes, + "_open_terminal_recovery_snapshot", + lambda _path: snapshot, + ) + monkeypatch.setattr(lanes, "_quarantine_terminal", lambda *_args, **_kwargs: path) + real_stat = Path.stat + + def fail_selected(candidate, *args, **kwargs): + if candidate == path: + raise PermissionError("denied") + return real_stat(candidate, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", fail_selected) + assert lanes._quarantine_invalid_retained_terminal(path) == (True, False) + assert snapshot.closed == 1 + + +def test_gc_skips_symlink_board_and_disappearing_candidate( + tmp_path: Path, + monkeypatch, +): + root = tmp_path / "lanes" + root.mkdir() + target = tmp_path / "target" + target.mkdir() + (root / "linked-board").symlink_to(target, target_is_directory=True) + board = root / "board" + board.mkdir() + candidate = board / "task.candidate-1.json" + candidate.write_text("{}", encoding="utf-8") + monkeypatch.setattr(lanes, "STATE_ROOT", root) + real_stat = Path.stat + + def disappear(selected, *args, **kwargs): + if selected == candidate: + raise FileNotFoundError + return real_stat(selected, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", disappear) + assert lanes.gc_lane_artifacts() == 0