diff --git a/dockerfiles/Dockerfile.hermes-agent b/dockerfiles/Dockerfile.hermes-agent index ce9ceb46..fc99307e 100644 --- a/dockerfiles/Dockerfile.hermes-agent +++ b/dockerfiles/Dockerfile.hermes-agent @@ -1544,6 +1544,13 @@ function RootRedirect() { } NODE +COPY dockerfiles/patch-hermes-execution-safety.py /tmp/patch-hermes-execution-safety.py +COPY dockerfiles/hermes-execution-safety-regression.py /tmp/hermes-execution-safety-regression.py +RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \ + && /opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py \ + && rm /tmp/patch-hermes-execution-safety.py \ + /tmp/hermes-execution-safety-regression.py + COPY dockerfiles/hermes-session-migrate.py /opt/hermes/bin/hermes-session-migrate RUN cd /opt/hermes/web \ @@ -1602,6 +1609,8 @@ RUN cd /opt/hermes/web \ /opt/hermes/agent/turn_context.py \ /opt/hermes/agent/conversation_loop.py \ /opt/hermes/hermes_cli/oneshot.py \ + /opt/hermes/hermes_cli/kanban_decompose.py \ + /opt/hermes/gateway/kanban_watchers.py \ /opt/hermes/tools/delegate_tool.py \ /opt/hermes/tools/web_tools.py \ /opt/hermes/tools/python_sandbox_tool.py \ diff --git a/dockerfiles/Dockerfile.hermes-agent.dockerignore b/dockerfiles/Dockerfile.hermes-agent.dockerignore index 1327bda4..1b3f891a 100644 --- a/dockerfiles/Dockerfile.hermes-agent.dockerignore +++ b/dockerfiles/Dockerfile.hermes-agent.dockerignore @@ -5,3 +5,5 @@ !dockerfiles/hermes-public-extract/** !dockerfiles/hermes-session-activity-panel.tsx !dockerfiles/hermes-session-migrate.py +!dockerfiles/patch-hermes-execution-safety.py +!dockerfiles/hermes-execution-safety-regression.py diff --git a/dockerfiles/hermes-execution-safety-regression.py b/dockerfiles/hermes-execution-safety-regression.py new file mode 100644 index 00000000..e21bd703 --- /dev/null +++ b/dockerfiles/hermes-execution-safety-regression.py @@ -0,0 +1,212 @@ +"""Build-time regressions for Hermes automatic decomposition safety.""" + +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from types import SimpleNamespace +from unittest import mock + + +_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-execution-test-home-") +os.environ["HERMES_HOME"] = _HERMES_HOME.name + +from agent import auxiliary_client # noqa: E402 +from hermes_cli import kanban_db # noqa: E402 +from hermes_cli import kanban_decompose # noqa: E402 + + +class _FakeCompletions: + def __init__(self, payload: dict, before_response=None) -> None: + self.payload = payload + self.before_response = before_response + self.calls = 0 + + def create(self, **_kwargs): + self.calls += 1 + if self.before_response is not None: + self.before_response() + return SimpleNamespace( + choices=[ + SimpleNamespace( + message=SimpleNamespace(content=json.dumps(self.payload)) + ) + ] + ) + + +class _FakeClient: + def __init__(self, completions: _FakeCompletions) -> None: + self.chat = SimpleNamespace(completions=completions) + + +class AutomaticDecompositionSafetyTests(unittest.TestCase): + """Exercise the patched upstream APIs against an isolated real database.""" + + def setUp(self) -> None: + self.connection = kanban_db.connect() + self.patches = [ + mock.patch.object( + kanban_decompose, + "_build_roster", + return_value=([], {"default"}), + ), + mock.patch.object( + kanban_decompose, + "_load_config", + return_value={"kanban": {"auto_promote_children": True}}, + ), + mock.patch.object( + auxiliary_client, + "get_auxiliary_extra_body", + return_value={}, + ), + ] + for patch in self.patches: + patch.start() + + def tearDown(self) -> None: + for patch in reversed(self.patches): + patch.stop() + self.connection.close() + + def _client(self, payload: dict, before_response=None) -> _FakeCompletions: + completions = _FakeCompletions(payload, before_response) + client = _FakeClient(completions) + patch = mock.patch.object( + auxiliary_client, + "get_text_auxiliary_client", + return_value=(client, "test-decomposer"), + ) + patch.start() + self.addCleanup(patch.stop) + return completions + + def _status(self, task_id: str) -> str: + task = kanban_db.get_task(self.connection, task_id) + self.assertIsNotNone(task) + return task.status + + def _route_executed_task_to_triage(self) -> str: + task_id = kanban_db.create_task( + self.connection, + title="already implemented parent", + ) + for attempt in range(kanban_db.BLOCK_RECURRENCE_LIMIT): + if attempt: + self.assertTrue(kanban_db.unblock_task(self.connection, task_id)) + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + self.assertTrue( + kanban_db.block_task( + self.connection, + task_id, + reason="same unavailable capability", + kind="capability", + ) + ) + self.assertEqual(self._status(task_id), "triage") + self.assertEqual( + len(kanban_db.list_runs(self.connection, task_id)), + kanban_db.BLOCK_RECURRENCE_LIMIT, + ) + return task_id + + def test_automatic_decomposer_skips_executed_triage_task(self) -> None: + task_id = self._route_executed_task_to_triage() + completions = self._client( + {"fanout": False, "title": "must not run", "body": "must not run"} + ) + before = len(kanban_db.list_tasks(self.connection)) + + outcome = kanban_decompose.decompose_task(task_id, automatic=True) + + self.assertFalse(outcome.ok) + self.assertIn("execution history", outcome.reason) + self.assertEqual(completions.calls, 0) + self.assertEqual(self._status(task_id), "triage") + self.assertEqual(len(kanban_db.list_tasks(self.connection)), before) + + def test_manual_decomposition_remains_available_after_execution(self) -> None: + task_id = self._route_executed_task_to_triage() + completions = self._client( + { + "fanout": False, + "title": "operator-approved retry", + "body": "Retain one bounded task.", + "assignee": "default", + } + ) + + outcome = kanban_decompose.decompose_task(task_id, author="operator") + + self.assertTrue(outcome.ok) + self.assertFalse(outcome.fanout) + self.assertEqual(completions.calls, 1) + self.assertEqual(self._status(task_id), "ready") + + def test_fresh_triage_task_still_auto_promotes(self) -> None: + task_id = kanban_db.create_task( + self.connection, + title="fresh objective", + triage=True, + ) + completions = self._client( + { + "fanout": False, + "title": "specified objective", + "body": "One bounded task.", + "assignee": "default", + } + ) + + outcome = kanban_decompose.decompose_task(task_id, automatic=True) + + self.assertTrue(outcome.ok) + self.assertEqual(completions.calls, 1) + self.assertEqual(self._status(task_id), "ready") + + def test_execution_history_gained_during_llm_call_blocks_commit(self) -> None: + task_id = kanban_db.create_task( + self.connection, + title="concurrent triage objective", + triage=True, + ) + + def add_run() -> None: + with kanban_db.connect_closing() as connection: + kanban_db._synthesize_ended_run( + connection, + task_id, + outcome="reclaimed", + summary="concurrent execution evidence", + ) + + completions = self._client( + { + "fanout": True, + "tasks": [ + { + "title": "redundant child", + "body": "must not be created", + "assignee": "default", + "parents": [], + } + ], + }, + before_response=add_run, + ) + before = len(kanban_db.list_tasks(self.connection)) + + outcome = kanban_decompose.decompose_task(task_id, automatic=True) + + self.assertFalse(outcome.ok) + self.assertIn("gained execution history", outcome.reason) + self.assertEqual(completions.calls, 1) + self.assertEqual(self._status(task_id), "triage") + self.assertEqual(len(kanban_db.list_tasks(self.connection)), before) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/dockerfiles/patch-hermes-execution-safety.py b/dockerfiles/patch-hermes-execution-safety.py new file mode 100644 index 00000000..bcf93c50 --- /dev/null +++ b/dockerfiles/patch-hermes-execution-safety.py @@ -0,0 +1,150 @@ +"""Patch upstream Hermes auto-decomposition to respect execution history.""" + +import os +from pathlib import Path + + +def replace_once(source: str, before: str, after: str, label: str) -> str: + """Replace one anchored upstream fragment and fail closed on source drift.""" + count = source.count(before) + if count != 1: + raise SystemExit( + f"Hermes {label} patch context changed: expected 1, found {count}" + ) + return source.replace(before, after, 1) + + +source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes")) +decompose_path = source_root / "hermes_cli/kanban_decompose.py" +decompose = decompose_path.read_text(encoding="utf-8") + +helper_anchor = ''' if chosen not in valid_names: + return default_assignee + return chosen + + +''' +helper_replacement = helper_anchor + '''def _has_execution_history(task_id: str) -> bool: + """Return whether a task has ever entered a worker run.""" + with kb.connect_closing() as conn: + return bool(kb.list_runs(conn, task_id)) + + +''' +decompose = replace_once( + decompose, + helper_anchor, + helper_replacement, + "auto-decompose execution-history helper", +) + +signature_before = '''def decompose_task( + task_id: str, + *, + author: Optional[str] = None, + timeout: Optional[int] = None, +) -> DecomposeOutcome: +''' +signature_after = '''def decompose_task( + task_id: str, + *, + author: Optional[str] = None, + timeout: Optional[int] = None, + automatic: bool = False, +) -> DecomposeOutcome: +''' +decompose = replace_once( + decompose, + signature_before, + signature_after, + "auto-decompose function signature", +) + +status_before = ''' if task.status != "triage": + return DecomposeOutcome( + task_id, False, f"task is not in triage (status={task.status!r})" + ) + + cfg = _load_config() +''' +status_after = ''' if task.status != "triage": + return DecomposeOutcome( + task_id, False, f"task is not in triage (status={task.status!r})" + ) + if automatic and _has_execution_history(task_id): + return DecomposeOutcome( + task_id, + False, + "task has execution history and requires deliberate manual triage", + ) + + cfg = _load_config() +''' +decompose = replace_once( + decompose, + status_before, + status_after, + "auto-decompose preflight guard", +) + +single_before = ''' with kb.connect_closing() as conn: + ok = kb.specify_triage_task( +''' +single_after = ''' if automatic and _has_execution_history(task_id): + return DecomposeOutcome( + task_id, + False, + "task gained execution history and requires deliberate manual triage", + ) + with kb.connect_closing() as conn: + ok = kb.specify_triage_task( +''' +decompose = replace_once( + decompose, + single_before, + single_after, + "auto-decompose single-task commit guard", +) + +fanout_before = ''' try: + with kb.connect_closing() as conn: + child_ids = kb.decompose_triage_task( +''' +fanout_after = ''' if automatic and _has_execution_history(task_id): + return DecomposeOutcome( + task_id, + False, + "task gained execution history and requires deliberate manual triage", + ) + try: + with kb.connect_closing() as conn: + child_ids = kb.decompose_triage_task( +''' +decompose = replace_once( + decompose, + fanout_before, + fanout_after, + "auto-decompose fanout commit guard", +) +decompose_path.write_text(decompose, encoding="utf-8") + + +watcher_path = source_root / "gateway/kanban_watchers.py" +watcher = watcher_path.read_text(encoding="utf-8") +call_before = ''' outcome = _decomp.decompose_task( + tid, author="auto-decomposer", + ) +''' +call_after = ''' outcome = _decomp.decompose_task( + tid, + author="auto-decomposer", + automatic=True, + ) +''' +watcher = replace_once( + watcher, + call_before, + call_after, + "gateway automatic-decomposition call", +) +watcher_path.write_text(watcher, encoding="utf-8") diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 3eba3453..c26f26ca 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -123,6 +123,10 @@ class ProcessResult: capacity_failure: bool +class TerminalFinalizationPending(RuntimeError): + """An accepted worker result is durable but not committed to Kanban yet.""" + + def utc_now() -> str: return datetime.now(timezone.utc).isoformat() @@ -293,6 +297,115 @@ def state_path(board: str, task_id: str) -> Path: return STATE_ROOT / safe_board / f"{safe_task}.json" +def _result_path(state_file: Path, run_id: Any, sequence: int) -> Path: + """Return a unique provider-result path; completed turns are never reused.""" + safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown")) + return state_file.with_name( + f"{state_file.stem}.run-{safe_run}.provider-{sequence}.result.json" + ) + + +def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path: + """Return the durable path for one exact structured worker response.""" + safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown")) + return state_file.with_name( + f"{state_file.stem}.run-{safe_run}.candidate-{sequence}.json" + ) + + +def _terminal_path(state_file: Path, run_id: Any) -> Path: + """Return the replay journal path for an accepted terminal response.""" + safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown")) + return state_file.with_name(f"{state_file.stem}.run-{safe_run}.terminal.json") + + +def _persist_candidate( + state: dict[str, Any], + state_file: Path, + structured: dict[str, Any], + *, + route: Route, + returncode: int, + goal_turn: int, +) -> Path: + """Persist every structured response before any judge can supersede it.""" + sequence = max(0, int(state.get("candidate_sequence", 0) or 0)) + 1 + path = _candidate_path(state_file, state.get("run_id"), sequence) + while path.exists(): + sequence += 1 + path = _candidate_path(state_file, state.get("run_id"), sequence) + atomic_json( + path, + { + "board": state.get("board"), + "task_id": state.get("task_id"), + "expected_run_id": state.get("run_id"), + "goal_turn": goal_turn, + "provider": route.provider, + "model": route.model, + "effort": route.effort, + "returncode": returncode, + "structured": structured, + "recorded_at": utc_now(), + }, + ) + state["candidate_sequence"] = sequence + state["last_candidate_file"] = str(path) + atomic_json(state_file, state) + return path + + +def _write_terminal_record( + state_file: Path, + *, + board: str, + task_id: str, + run_id: Any, + structured: dict[str, Any], + summary: str, + metadata: dict[str, Any], +) -> tuple[Path, dict[str, Any]]: + """Journal an accepted result before attempting the Kanban transaction.""" + path = _terminal_path(state_file, run_id) + record = { + "board": board, + "task_id": task_id, + "expected_run_id": run_id, + "result": json.dumps(structured, sort_keys=True), + "summary": summary, + "metadata": metadata, + "kanban_state": "pending", + "recorded_at": utc_now(), + } + atomic_json(path, record) + return path, record + + +def _terminal_record_valid(record: dict[str, Any]) -> bool: + """Reject malformed or non-terminal replay journals without side effects.""" + if not isinstance(record, dict): + return False + required_strings = ("board", "task_id", "result", "summary") + if not all( + isinstance(record.get(key), str) and record[key] + for key in required_strings + ): + return False + if not isinstance(record.get("expected_run_id"), int): + return False + if not isinstance(record.get("metadata"), dict): + return False + try: + structured = json.loads(record["result"]) + except (TypeError, json.JSONDecodeError): + return False + return ( + isinstance(structured, dict) + and structured.get("status") == "completed" + and not structured.get("blockers") + ) + + def build_prompt(context: str, workspace: Path, handoff: str = "") -> str: """Create a bounded worker contract with an explicit machine-readable result.""" return f"""You are a durable coding worker managed by Hermes Kanban. @@ -667,11 +780,22 @@ def run_provider( state["claude_session_id"] = str(uuid.uuid4()) state["current_route"] = asdict(route) state["updated_at"] = utc_now() - atomic_json(state_file, state) env = _base_env() - result_file = state_file.with_suffix(".result.json") if route.provider == "codex": - result_file.unlink(missing_ok=True) + result_sequence = max(0, int(state.get("result_sequence", 0) or 0)) + 1 + result_file = _result_path(state_file, state.get("run_id"), result_sequence) + while result_file.exists(): + result_sequence += 1 + result_file = _result_path( + state_file, + state.get("run_id"), + result_sequence, + ) + result_file.parent.mkdir(parents=True, exist_ok=True) + result_file.touch(mode=0o600, exist_ok=False) + state["result_sequence"] = result_sequence + state["current_result_file"] = str(result_file) + atomic_json(state_file, state) command = _codex_command(route, prompt, workspace, state, result_file) result = stream_process( command, @@ -690,8 +814,23 @@ def run_provider( and NO_CODEX_THREAD in result.output.lower() ): state.pop("codex_thread_id", None) + result_sequence += 1 + result_file = _result_path( + state_file, + state.get("run_id"), + result_sequence, + ) + while result_file.exists(): + result_sequence += 1 + result_file = _result_path( + state_file, + state.get("run_id"), + result_sequence, + ) + result_file.touch(mode=0o600, exist_ok=False) + state["result_sequence"] = result_sequence + state["current_result_file"] = str(result_file) atomic_json(state_file, state) - result_file.unlink(missing_ok=True) result = stream_process( _codex_command(route, prompt, workspace, state, result_file), provider="codex", @@ -706,8 +845,13 @@ 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: + result_file.chmod(0o600) + except OSError: + pass return result + atomic_json(state_file, state) resume = bool(state.get("claude_started")) result = stream_process( _claude_command(route, prompt, state, resume), @@ -801,6 +945,95 @@ def _board_call( raise last_storage_error +def _finalize_terminal_record( + kanban_db: Any, + path: Path, + record: dict[str, Any] | None = None, +) -> str: + """Commit one exact-run terminal journal, or leave it safely pending.""" + document = dict(record or load_json(path)) + if not _terminal_record_valid(document): + return "invalid" + board = str(document["board"]) + task_id = str(document["task_id"]) + expected_run_id = int(document["expected_run_id"]) + + def operation(conn: Any) -> str: + task = kanban_db.get_task(conn, task_id) + if task is None: + return "stale" + status = str(_task_value(task, "status", "")) + if status == "done": + return ( + "committed" + if str(_task_value(task, "result", "") or "") == document["result"] + else "stale" + ) + if ( + status != "running" + or _task_value(task, "current_run_id", None) != expected_run_id + ): + return "stale" + completed = kanban_db.complete_task( + conn, + task_id, + result=document["result"], + summary=document["summary"], + metadata=document["metadata"], + expected_run_id=expected_run_id, + ) + return "committed" if completed else "pending" + + outcome = str(_board_call(kanban_db, board, operation)) + if outcome == "committed": + document["kanban_state"] = "committed" + document["committed_at"] = utc_now() + atomic_json(path, document) + return outcome + + +def recover_pending_finalizations() -> int: + """Replay accepted exact-run results before scheduling more provider work.""" + from hermes_cli import kanban_db + + recovered = 0 + for path in sorted(STATE_ROOT.glob("*/*.terminal.json")): + record = load_json(path) + if record.get("kanban_state") == "committed": + continue + try: + outcome = _finalize_terminal_record(kanban_db, path, record) + except Exception as error: + board = str(record.get("board") or "unknown") + _record_board_access_error(board, error) + continue + if outcome == "committed": + recovered += 1 + return recovered + + +def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool: + """Keep an exact run claimed while its accepted result awaits replay.""" + if not isinstance(run_id, int): + return False + path = _terminal_path(state_path(board, task_id), run_id) + try: + path.stat() + except FileNotFoundError: + return False + except OSError: + return True + record = load_json(path) + if not _terminal_record_valid(record): + return True + identity = ( + record.get("board") == board + and record.get("task_id") == task_id + and record.get("expected_run_id") == run_id + ) + return not identity or record.get("kanban_state") != "committed" + + def execute_claim(board: str, task_id: str) -> None: """Execute one already-claimed task and commit its outcome to Kanban.""" from hermes_cli import kanban_db @@ -980,6 +1213,16 @@ def execute_claim(board: str, task_id: str) -> None: workspace, structured.get("artifacts"), ) + candidate_file = _persist_candidate( + state, + state_file, + structured, + route=route, + returncode=result.returncode, + goal_turn=goal_turn, + ) + else: + candidate_file = None metadata = { "executor": "direct-cli-lane", "provider": route.provider, @@ -992,6 +1235,8 @@ def execute_claim(board: str, task_id: str) -> None: "goal_mode": goal_mode, "goal_turn": goal_turn, } + if candidate_file is not None: + metadata["candidate_file"] = str(candidate_file) if structured: for key in ( "changed_files", @@ -1036,18 +1281,27 @@ def execute_claim(board: str, task_id: str) -> None: and result.returncode == 0 and completion_problem is None ): - _board_call( - kanban_db, - board, - lambda fresh: kanban_db.complete_task( - fresh, - task_id, - result=json.dumps(structured, sort_keys=True), - summary=str(structured.get("summary") or "Completed"), - metadata=metadata, - expected_run_id=run_id, - ), + terminal_file = _terminal_path(state_file, run_id) + metadata["terminal_record"] = str(terminal_file) + terminal_file, terminal_record = _write_terminal_record( + state_file, + board=board, + task_id=task_id, + run_id=run_id, + structured=structured, + summary=str(structured.get("summary") or "Completed"), + metadata=metadata, ) + outcome = _finalize_terminal_record( + kanban_db, + terminal_file, + terminal_record, + ) + if outcome != "committed": + raise TerminalFinalizationPending( + "accepted worker result is durably journaled but " + f"Kanban finalization is {outcome}" + ) break can_continue = ( @@ -1126,6 +1380,8 @@ def execute_claim(board: str, task_id: str) -> None: ), ) break + except TerminalFinalizationPending as error: + comment(str(error)) except Exception as error: failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}" _board_call( @@ -1178,6 +1434,7 @@ def recover_orphans() -> None: """Return external running tasks to ready after a runner/pod restart.""" from hermes_cli import kanban_db + recover_pending_finalizations() try: boards = kanban_db.list_boards(include_archived=False) except Exception as error: @@ -1194,9 +1451,13 @@ def recover_orphans() -> None: try: for task in kanban_db.list_tasks(conn): if _external(task) and str(_task_value(task, "status", "")) == "running": + task_id = str(_task_value(task, "id")) + run_id = _task_value(task, "current_run_id", None) + if _has_pending_finalization(board, task_id, run_id): + continue kanban_db.reclaim_task( conn, - str(_task_value(task, "id")), + task_id, reason="direct CLI lane restarted; provider session will resume", ) BOARD_CORRUPTION_ERRORS.pop(board, None) @@ -1275,6 +1536,7 @@ def main() -> int: except Exception as error: print(f"worker future failed: {error}", file=sys.stderr, flush=True) del futures[future] + recover_pending_finalizations() active = set(futures.values()) try: newly_claimed = claim_ready(active, workers - len(futures)) diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index fe5272da..b3f53f65 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -359,6 +359,66 @@ def test_codex_missing_server_thread_restarts_fresh(tmp_path: Path, monkeypatch) assert "codex_thread_id" not in json.loads(state_file.read_text()) +def test_codex_result_files_are_unique_and_prior_turns_are_retained( + tmp_path: Path, + monkeypatch, +): + state = {"run_id": 17} + state_file = tmp_path / "task.json" + result_paths = [] + + def fake_stream(command, **_kwargs): + result_path = Path(command[command.index("-o") + 1]) + result_path.write_text( + json.dumps( + { + "status": "completed", + "summary": f"turn {len(result_paths) + 1}", + "changed_files": [], + "tests_run": [], + "artifacts": [], + "findings": [], + "blockers": [], + } + ), + encoding="utf-8", + ) + result_paths.append(result_path) + return lanes.ProcessResult(0, "", None, False) + + monkeypatch.setattr(lanes, "stream_process", fake_stream) + route = lanes.Route( + "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, () + ) + + first = lanes.run_provider( + route, + "First turn.", + tmp_path, + state, + state_file, + tmp_path / "worker.log", + lambda _note: True, + 60, + ) + second = lanes.run_provider( + route, + "Second turn.", + tmp_path, + state, + state_file, + tmp_path / "worker.log", + lambda _note: True, + 60, + ) + + assert first.structured["summary"] == "turn 1" + assert second.structured["summary"] == "turn 2" + assert len(set(result_paths)) == 2 + assert all(path.exists() for path in result_paths) + assert all(path.stat().st_mode & 0o777 == 0o600 for path in result_paths) + + def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path): result = lanes.stream_process( [sys.executable, "-c", "print('authentication work completed')"], @@ -594,6 +654,197 @@ def test_board_call_retries_storage_faults_on_fresh_connections(): assert all(connection.closed for connection in connections) +def test_accepted_result_survives_failed_finalization_and_replays_exact_run( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + task = SimpleNamespace( + id="t_terminal", + status="running", + result=None, + current_run_id=41, + assignee="cli-auto", + max_runtime_seconds=60, + ) + completions = [] + comments = [] + permit_completion = {"value": False} + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + completions.append(kwargs) + if not permit_completion["value"]: + return False + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + worker_log_path=lambda _task_id, board: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_terminal"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "Finish and verify the objective.", + heartbeat_worker=lambda *_args, **_kwargs: True, + add_comment=lambda _conn, _task_id, _author, body: comments.append(body), + complete_task=complete_task, + block_task=lambda *_args, **_kwargs: pytest.fail( + "an accepted journaled result must not be converted into a block" + ), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr( + lanes, + "select_route", + lambda *_args, **_kwargs: lanes.Route( + "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, () + ), + ) + structured = { + "status": "completed", + "summary": "Verified exact terminal result.", + "changed_files": [], + "tests_run": ["pytest: passed"], + "artifacts": [], + "findings": [], + "blockers": [], + } + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult(0, "", structured, False), + ) + + lanes.execute_claim("cassandra", "t_terminal") + + candidates = list((state_root / "cassandra").glob("*.candidate-*.json")) + terminals = list((state_root / "cassandra").glob("*.terminal.json")) + assert len(candidates) == 1 + assert json.loads(candidates[0].read_text())["structured"] == structured + assert len(terminals) == 1 + assert json.loads(terminals[0].read_text())["kanban_state"] == "pending" + assert task.status == "running" + assert any("durably journaled" in body for body in comments) + + permit_completion["value"] = True + assert lanes.recover_pending_finalizations() == 1 + assert task.status == "done" + assert len(completions) == 2 + terminal = json.loads(terminals[0].read_text()) + assert terminal["kanban_state"] == "committed" + assert completions[-1]["result"] == terminal["result"] + + +def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + state_file = lanes.state_path("cassandra", "t_terminal") + terminal_path, _record = lanes._write_terminal_record( + state_file, + board="cassandra", + task_id="t_terminal", + run_id=8, + structured={ + "status": "completed", + "summary": "done", + "changed_files": [], + "tests_run": [], + "artifacts": [], + "findings": [], + "blockers": [], + }, + summary="done", + metadata={}, + ) + assert terminal_path.exists() + task = SimpleNamespace( + id="t_terminal", + status="running", + assignee="cli-auto", + current_run_id=8, + ) + reclaimed = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + list_boards=lambda include_archived=False: [{"slug": "cassandra"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + list_tasks=lambda _conn: [task], + reclaim_task=lambda *_args, **_kwargs: reclaimed.append(True), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + + lanes.recover_orphans() + + assert reclaimed == [] + + +def test_terminal_replay_never_crosses_into_a_replacement_run( + tmp_path: Path, + monkeypatch, +): + state_root = tmp_path / "cli-lanes" + monkeypatch.setattr(lanes, "STATE_ROOT", state_root) + state_file = lanes.state_path("cassandra", "t_terminal") + terminal_path, _record = lanes._write_terminal_record( + state_file, + board="cassandra", + task_id="t_terminal", + run_id=8, + structured={ + "status": "completed", + "summary": "old run", + "changed_files": [], + "tests_run": [], + "artifacts": [], + "findings": [], + "blockers": [], + }, + summary="old run", + metadata={}, + ) + replacement = SimpleNamespace( + id="t_terminal", + status="running", + result=None, + current_run_id=9, + ) + completions = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: replacement, + complete_task=lambda *_args, **_kwargs: completions.append(True), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 0 + assert completions == [] + assert json.loads(terminal_path.read_text())["kanban_state"] == "pending" + + @pytest.mark.parametrize( ("result", "expected_action"), [ @@ -640,6 +891,8 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( ): task = SimpleNamespace( id="t_worker", + status="running", + result=None, current_run_id=4, assignee="cli-auto", max_runtime_seconds=60, @@ -677,7 +930,9 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( heartbeats.append((note, expected_run_id)) or True ), add_comment=lambda *_args: None, - complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)), + complete_task=lambda *_args, **kwargs: ( + calls.append(("complete", kwargs)) or True + ), block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) @@ -717,6 +972,8 @@ def test_goal_card_continues_after_local_judge_rejects_progress( ): task = SimpleNamespace( id="t_goal", + status="running", + result=None, current_run_id=12, assignee="cli-auto", max_runtime_seconds=300, @@ -741,7 +998,9 @@ def test_goal_card_continues_after_local_judge_rejects_progress( build_worker_context=lambda *_args: "Run tests, commit, push, and verify remote HEAD.", heartbeat_worker=lambda *_args, **_kwargs: True, add_comment=lambda _conn, _task_id, _author, body: comments.append(body), - complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)), + complete_task=lambda *_args, **kwargs: ( + calls.append(("complete", kwargs)) or True + ), block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) @@ -828,6 +1087,14 @@ def test_goal_card_continues_after_local_judge_rejects_progress( assert route_calls[2][1]["exclude_provider"] == "claude" assert "prior rejected reports" in judge_contexts[1] assert "commit, push, and remote verification are missing" in judge_contexts[1] + candidates = sorted(tmp_path.glob("state.run-12.candidate-*.json")) + assert len(candidates) == 2 + assert json.loads(candidates[0].read_text())["structured"]["summary"] == ( + "Focused tests passed." + ) + assert json.loads(candidates[1].read_text())["structured"]["summary"].startswith( + "Full tests passed" + ) assert reports == [] @@ -902,7 +1169,7 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P set_workspace_path=lambda *_args: None, build_worker_context=lambda *_args: "resume objective", add_comment=lambda *_args: None, - complete_task=lambda *_args, **_kwargs: None, + complete_task=lambda *_args, **_kwargs: True, block_task=lambda *_args, **_kwargs: None, ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) @@ -1256,6 +1523,27 @@ def test_agent_dashboard_reconnects_all_transient_websockets(): assert "dashboard token rotated by a server restart" in dockerfile +def test_agent_image_runs_execution_safety_patch_and_regressions(): + dockerfiles = HERMES.parents[1] / "dockerfiles" + dockerfile = (dockerfiles / "Dockerfile.hermes-agent").read_text(encoding="utf-8") + dockerignore = (dockerfiles / "Dockerfile.hermes-agent.dockerignore").read_text( + encoding="utf-8" + ) + + for name in ( + "patch-hermes-execution-safety.py", + "hermes-execution-safety-regression.py", + ): + assert f"COPY dockerfiles/{name}" in dockerfile + assert f"!dockerfiles/{name}" in dockerignore + assert "/opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py" in ( + dockerfile + ) + assert "/opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py" in ( + dockerfile + ) + + def test_agent_refreshes_routes_after_restoring_cli_logins(): deployment = _agent_deployment() init_containers = {