From 1527fd1af04aa41cc1e2704c2f3fe130613fcef2 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 10:51:34 -0300 Subject: [PATCH] hermes: enforce goal completion evidence --- services/hermes/kustomization.yaml | 1 + services/hermes/scripts/cli_lane_goal.py | 150 ++++++++++++ services/hermes/scripts/cli_lane_runner.py | 228 +++++++++++++----- testing/tests/test_hermes_cli_lane_goal.py | 111 +++++++++ testing/tests/test_hermes_cli_lane_routing.py | 6 +- testing/tests/test_hermes_cli_lanes.py | 110 +++++++++ 6 files changed, 539 insertions(+), 67 deletions(-) create mode 100644 services/hermes/scripts/cli_lane_goal.py create mode 100644 testing/tests/test_hermes_cli_lane_goal.py diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index f3c45d92..ae7d93ad 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -61,6 +61,7 @@ configMapGenerator: - ai_usage_exporter.py=scripts/ai_usage_exporter.py - claude=scripts/claude - claude_command_policy.py=scripts/claude_command_policy.py + - cli_lane_goal.py=scripts/cli_lane_goal.py - cli_lane_runner.py=scripts/cli_lane_runner.py - codex=scripts/codex - configure_agent_clients.py=scripts/configure_agent_clients.py diff --git a/services/hermes/scripts/cli_lane_goal.py b/services/hermes/scripts/cli_lane_goal.py new file mode 100644 index 00000000..230d2a43 --- /dev/null +++ b/services/hermes/scripts/cli_lane_goal.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +"""Fail-closed completion checks for durable external Kanban workers.""" + +from __future__ import annotations + +import json +import os +import re +import urllib.error +import urllib.request +from typing import Any, Callable + + +GOAL_JUDGE_URL = os.environ.get( + "HERMES_GOAL_JUDGE_URL", + "http://hermes-model-gate.hermes.svc.cluster.local:11434/v1/chat/completions", +) +GOAL_JUDGE_MODEL = os.environ.get( + "HERMES_GOAL_JUDGE_MODEL", + "qwen2.5:14b-instruct-q4_0", +) +RESULT_STATUSES = frozenset({"completed", "blocked", "incomplete"}) +GOAL_JUDGE_SCHEMA: dict[str, Any] = { + "type": "object", + "additionalProperties": False, + "required": ["verdict", "reason"], + "properties": { + "verdict": {"type": "string", "enum": ["complete", "continue"]}, + "reason": {"type": "string"}, + }, +} +UNFINISHED_EVIDENCE = re.compile( + r"(?:\bin[ -]?progress\b|\bstill\s+(?:active|pending|running)\b|" + r"\bremains?\s+(?:active|pending|running|unfinished)\b|" + r"\b(?:tests?|checks?|build|validation|verification|commit|push)\b" + r"\s+(?:is|are|remains?|still|currently|has|have)\s+" + r"(?:pending|running|unfinished|not\s+(?:yet\s+)?(?:done|finished|run|complete))\b|" + r"\b(?:pending|running|unfinished)\s+" + r"\b(?:tests?|checks?|build|validation|verification|commit|push)\b)", + re.IGNORECASE | re.DOTALL, +) + + +def _bounded(value: str, limit: int) -> str: + """Retain both ends of evidence while bounding local-judge context.""" + if len(value) <= limit: + return value + marker = "\n...[goal evidence compacted]...\n" + remaining = max(0, limit - len(marker)) + head = remaining // 2 + return f"{value[:head]}{marker}{value[-(remaining - head):]}" + + +def unfinished_result_reason(result: dict[str, Any]) -> str | None: + """Reject self-contradictory completion reports without model inference.""" + status = str(result.get("status") or "") + summary = str(result.get("summary") or "").strip() + blockers = result.get("blockers") + if status == "incomplete": + return summary or "worker explicitly reported incomplete work" + if status != "completed": + return None + if isinstance(blockers, list) and any(str(item).strip() for item in blockers): + return "worker reported blockers while claiming completion" + tests = result.get("tests_run") + evidence = [summary] + if isinstance(tests, list): + evidence.extend(str(item) for item in tests) + match = UNFINISHED_EVIDENCE.search("\n".join(evidence)) + if match: + return f"completion evidence says work is unfinished: {match.group(0).strip()}" + return None + + +def judge_goal_completion( + objective: str, + result: dict[str, Any], + *, + url: str = GOAL_JUDGE_URL, + open_request: Callable[..., Any] = urllib.request.urlopen, +) -> tuple[bool, str]: + """Use the local model as a fail-closed judge for explicit goal cards.""" + deterministic_reason = unfinished_result_reason(result) + if deterministic_reason: + return False, deterministic_reason + payload = { + "model": GOAL_JUDGE_MODEL, + "messages": [ + { + "role": "system", + "content": ( + "You are a fail-closed completion judge for an engineering task. " + "Compare the objective and acceptance criteria with the worker report. " + "Return complete only when the report gives concrete evidence that every " + "explicit requirement finished. Return continue when any work, test, command, " + "commit, push, review, or verification is pending, in progress, omitted, or " + "inconclusive. Do not trust the report status field." + ), + }, + { + "role": "user", + "content": json.dumps( + { + "objective": _bounded(objective, 6000), + "worker_report": _bounded( + json.dumps(result, default=str), 4000 + ), + }, + separators=(",", ":"), + ), + }, + ], + "stream": False, + "temperature": 0, + "max_tokens": 160, + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "goal_completion_verdict", + "strict": True, + "schema": GOAL_JUDGE_SCHEMA, + }, + }, + } + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with open_request(request, timeout=60) as response: + document = json.load(response) + content = document["choices"][0]["message"]["content"] + verdict = json.loads(content) + if verdict.get("verdict") not in {"complete", "continue"}: + raise ValueError("local judge returned an invalid verdict") + reason = str(verdict.get("reason") or "local judge supplied no reason") + return verdict.get("verdict") == "complete", reason + except ( + AttributeError, + IndexError, + KeyError, + OSError, + TypeError, + ValueError, + json.JSONDecodeError, + urllib.error.URLError, + ) as error: + return False, f"local completion judge unavailable: {type(error).__name__}: {error}" diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 768702a0..2b63b42c 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -22,6 +22,8 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable +import cli_lane_goal + DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data")) SWITCHYARD_URL = os.environ.get( @@ -60,7 +62,10 @@ RESULT_SCHEMA: dict[str, Any] = { "additionalProperties": False, "required": ["status", "summary", "changed_files", "tests_run", "artifacts", "blockers"], "properties": { - "status": {"type": "string", "enum": ["completed", "blocked"]}, + "status": { + "type": "string", + "enum": sorted(cli_lane_goal.RESULT_STATUSES), + }, "summary": {"type": "string"}, "changed_files": {"type": "array", "items": {"type": "string"}}, "tests_run": {"type": "array", "items": {"type": "string"}}, @@ -272,7 +277,7 @@ Workspace: {workspace} Operate autonomously inside the workspace. Inspect before editing, preserve unrelated user changes, run proportionate tests, and do not claim completion without evidence. You have owner-level Kubernetes access in every namespace. Prefer Flux-tracked manifests for durable changes, but use kubectl, Flux, exec, port-forwarding, rollout operations, and existing Vault workflows when the objective or incident requires them. Persist any desired-state mutation back to Git. Do not force-push, hard-reset, clean untracked files, or expose credentials. -Return a final JSON object matching the supplied schema. Use status=blocked only for a concrete unresolved blocker. List changed files, tests run, durable artifact paths, and blockers explicitly. +Return a final JSON object matching the supplied schema. Use status=incomplete when required work, tests, commands, commits, pushes, or verification are still running or remain to be done. Use status=blocked only for a concrete unresolved blocker. Never use status=completed for a progress report. List changed files, tests run, durable artifact paths, and blockers explicitly. """ @@ -320,7 +325,7 @@ def workspace_artifacts(workspace: Path, values: Any) -> list[str]: def _extract_json(value: Any) -> dict[str, Any] | None: - if isinstance(value, dict) and value.get("status") in {"completed", "blocked"}: + if isinstance(value, dict) and value.get("status") in cli_lane_goal.RESULT_STATUSES: return value if not isinstance(value, str): return None @@ -333,7 +338,7 @@ def _extract_json(value: Any) -> dict[str, Any] | None: parsed = json.loads(candidate) except json.JSONDecodeError: continue - if isinstance(parsed, dict) and parsed.get("status") in {"completed", "blocked"}: + if isinstance(parsed, dict) and parsed.get("status") in cli_lane_goal.RESULT_STATUSES: return parsed return None @@ -670,7 +675,7 @@ def run_provider( max_runtime=max_runtime, ) file_result = load_json(result_file) - if file_result.get("status") in {"completed", "blocked"}: + if file_result.get("status") in cli_lane_goal.RESULT_STATUSES: result.structured = file_result return result @@ -890,83 +895,178 @@ def execute_claim(board: str, task_id: str) -> None: max_runtime = int( _task_value(task, "max_runtime_seconds", 0) or DEFAULT_MAX_RUNTIME ) - result = run_provider(route, prompt, workspace, state, state_file, log_path, heartbeat, max_runtime) - if result.capacity_failure: - retry_context = ( - context - + "\n\nRouting boundary: the first provider failed from capacity/authentication. " - + "Select the alternate hosted provider at an appropriate effort." - ) - alternate = "claude" if route.provider == "codex" else "codex" - fallback = select_route( - retry_context, - f"cli-{alternate}-{route.effort}", - ) - comment( - f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.", - ) + goal_mode = bool(_task_value(task, "goal_mode", False)) + goal_max_turns = max(1, int(_task_value(task, "goal_max_turns", 1) or 1)) + goal_turn = max(1, int(state.get("goal_turn", 0) or 0) + 1) + deadline = time.monotonic() + max_runtime + while True: + state["goal_turn"] = goal_turn + atomic_json(state_file, state) + remaining = max(1, int(deadline - time.monotonic())) result = run_provider( - fallback, - build_prompt(context, workspace, git_handoff(workspace, result.output)), + route, + prompt, workspace, state, state_file, log_path, heartbeat, - max_runtime, + remaining, ) - route = fallback - structured = result.structured - if structured: - structured = dict(structured) - structured["artifacts"] = workspace_artifacts( - workspace, - structured.get("artifacts"), - ) - metadata = { - "executor": "direct-cli-lane", - "provider": route.provider, - "model": route.model, - "effort": route.effort, - "classifier": route.classifier, - "state_file": str(state_file), - "codex_thread_id": state.get("codex_thread_id"), - "claude_session_id": state.get("claude_session_id"), - } - if structured: - for key in ("changed_files", "tests_run", "artifacts", "blockers"): - value = structured.get(key) - metadata[key] = value if isinstance(value, list) else [] - if structured and structured.get("status") == "completed" and result.returncode == 0: - _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, - ), - ) - else: - reason = ( - "; ".join(str(item) for item in (structured or {}).get("blockers", [])) - if structured - else result.output[-4000:] + if result.capacity_failure: + retry_context = ( + context + + "\n\nRouting boundary: the first provider failed from capacity/authentication. " + + "Select the alternate hosted provider at an appropriate effort." + ) + alternate = "claude" if route.provider == "codex" else "codex" + fallback = select_route( + retry_context, + f"cli-{alternate}-{route.effort}", + ) + comment( + f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.", + ) + result = run_provider( + fallback, + build_prompt( + context, + workspace, + git_handoff(workspace, result.output), + ), + workspace, + state, + state_file, + log_path, + heartbeat, + max(1, int(deadline - time.monotonic())), + ) + route = fallback + structured = result.structured + if structured: + structured = dict(structured) + structured["artifacts"] = workspace_artifacts( + workspace, + structured.get("artifacts"), + ) + metadata = { + "executor": "direct-cli-lane", + "provider": route.provider, + "model": route.model, + "effort": route.effort, + "classifier": route.classifier, + "state_file": str(state_file), + "codex_thread_id": state.get("codex_thread_id"), + "claude_session_id": state.get("claude_session_id"), + "goal_mode": goal_mode, + "goal_turn": goal_turn, + } + if structured: + for key in ("changed_files", "tests_run", "artifacts", "blockers"): + value = structured.get(key) + metadata[key] = value if isinstance(value, list) else [] + + completion_problem = None + if structured and result.returncode == 0: + completion_problem = cli_lane_goal.unfinished_result_reason(structured) + if ( + structured.get("status") == "completed" + and completion_problem is None + and goal_mode + ): + heartbeat("local goal-completion judge active") + accepted, judge_reason = cli_lane_goal.judge_goal_completion( + context, + structured, + ) + metadata["goal_judge_reason"] = judge_reason + if not accepted: + completion_problem = f"local goal judge requested continuation: {judge_reason}" + + if ( + structured + and structured.get("status") == "completed" + 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, + ), + ) + break + + can_continue = ( + goal_mode + and completion_problem is not None + and goal_turn < goal_max_turns + and deadline - time.monotonic() > 30 ) + if can_continue: + goal_turn += 1 + comment( + f"Goal completion rejected; continuing turn {goal_turn}/{goal_max_turns}: {completion_problem}", + ) + escalation_context = ( + context + + "\n\nThe previous worker turn missed its completion quality mark: " + + completion_problem + + "\nSelect a route that can finish and verify the remaining work." + ) + excluded = ( + fresh_unavailable_provider() if assignee == "cli-auto" else None + ) + next_route = select_route( + escalation_context, + assignee, + exclude_provider=excluded, + exclude_reason="is unavailable according to fresh native health" + if excluded + else None, + ) + handoff = ( + git_handoff(workspace, result.output) + if next_route.provider != route.provider + else "" + ) + route = next_route + prompt = build_prompt( + context, + workspace, + handoff + + "\nGoal-loop continuation: the previous final report was rejected because " + + completion_problem + + ". Reinspect live state, finish the outstanding work, and return new final evidence.", + ) + continue + + reason = completion_problem + if reason is None and structured: + blockers = structured.get("blockers", []) + reason = "; ".join(str(item) for item in blockers) + reason = reason or str(structured.get("summary") or "") + if reason is None: + reason = result.output[-4000:] _board_call( kanban_db, board, lambda fresh: kanban_db.block_task( fresh, task_id, - reason=reason or f"{route.provider} worker failed with exit {result.returncode}", + reason=reason + or f"{route.provider} worker failed with exit {result.returncode}", kind="transient" if result.capacity_failure else "capability", expected_run_id=run_id, ), ) + break except Exception as error: failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}" _board_call( diff --git a/testing/tests/test_hermes_cli_lane_goal.py b/testing/tests/test_hermes_cli_lane_goal.py new file mode 100644 index 00000000..48a864ea --- /dev/null +++ b/testing/tests/test_hermes_cli_lane_goal.py @@ -0,0 +1,111 @@ +"""Completion-gate tests for Hermes external Kanban workers.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +SCRIPT = ( + Path(__file__).parents[2] + / "services/hermes/scripts/cli_lane_goal.py" +) +SPEC = importlib.util.spec_from_file_location("cli_lane_goal_test", SCRIPT) +assert SPEC and SPEC.loader +goal = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = goal +SPEC.loader.exec_module(goal) + + +def _result(**overrides): + value = { + "status": "completed", + "summary": "All acceptance criteria passed and the branch was pushed.", + "changed_files": ["src/example.py"], + "tests_run": ["pytest -q: 12 passed"], + "artifacts": [], + "blockers": [], + } + value.update(overrides) + return value + + +@pytest.mark.parametrize( + "result", + [ + _result(status="incomplete", summary="The full suite is still running."), + _result(summary="The broad rerun remains active before the final push."), + _result(tests_run=["Full suite — in progress (11m)"]), + _result(blockers=["remote head was not verified"]), + ], +) +def test_unfinished_completion_evidence_is_rejected(result): + assert goal.unfinished_result_reason(result) + + +def test_completed_pending_state_test_name_is_not_a_false_positive(): + result = _result(tests_run=["pytest tests/test_pending_build_state.py: 8 passed"]) + + assert goal.unfinished_result_reason(result) is None + + +class JudgeResponse: + def __init__(self, verdict: str, reason: str): + self.body = json.dumps( + { + "choices": [ + { + "message": { + "content": json.dumps( + {"verdict": verdict, "reason": reason} + ) + } + } + ] + } + ).encode() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self): + return self.body + + +def test_goal_judge_uses_local_structured_verdict(): + observed = {} + + def request(req, timeout): + observed["payload"] = json.loads(req.data) + observed["timeout"] = timeout + return JudgeResponse("continue", "remote head evidence is missing") + + accepted, reason = goal.judge_goal_completion( + "Run tests, push, and verify the remote head.", + _result(summary="Tests passed."), + open_request=request, + ) + + assert accepted is False + assert reason == "remote head evidence is missing" + assert observed["payload"]["model"] == "qwen2.5:14b-instruct-q4_0" + assert observed["payload"]["response_format"]["json_schema"]["strict"] is True + assert observed["timeout"] == 60 + + +def test_goal_judge_fails_closed_on_invalid_response(): + accepted, reason = goal.judge_goal_completion( + "Finish the task.", + _result(), + open_request=lambda *_args, **_kwargs: JudgeResponse("unknown", "bad"), + ) + + assert accepted is False + assert "local completion judge unavailable" in reason diff --git a/testing/tests/test_hermes_cli_lane_routing.py b/testing/tests/test_hermes_cli_lane_routing.py index 5aac2f4e..14f880fc 100644 --- a/testing/tests/test_hermes_cli_lane_routing.py +++ b/testing/tests/test_hermes_cli_lane_routing.py @@ -8,9 +8,9 @@ import sys from pathlib import Path -SCRIPT = ( - Path(__file__).parents[2] / "services/hermes/scripts/cli_lane_runner.py" -) +SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts" +sys.path.insert(0, str(SCRIPTS)) +SCRIPT = SCRIPTS / "cli_lane_runner.py" SPEC = importlib.util.spec_from_file_location("cli_lane_routing", SCRIPT) assert SPEC and SPEC.loader lanes = importlib.util.module_from_spec(SPEC) diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 7e015b02..31327578 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -16,6 +16,7 @@ import yaml SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts" +sys.path.insert(0, str(SCRIPTS)) HERMES = Path(__file__).parents[2] / "services/hermes" KEYCLOAK = Path(__file__).parents[2] / "services/keycloak" FLUX_HERMES = ( @@ -590,6 +591,22 @@ def test_board_call_retries_storage_faults_on_fresh_connections(): ), "complete", ), + ( + lanes.ProcessResult( + 0, + "", + { + "status": "completed", + "summary": "The full test suite is still running.", + "changed_files": ["src/a.py"], + "tests_run": ["pytest -q — in progress"], + "artifacts": [], + "blockers": [], + }, + False, + ), + "block", + ), ], ) def test_claim_requires_structured_evidence_and_surfaces_artifacts( @@ -671,6 +688,99 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts( assert all(connection.closed for connection in connections) +def test_goal_card_continues_after_local_judge_rejects_progress( + tmp_path: Path, + monkeypatch, +): + task = SimpleNamespace( + id="t_goal", + current_run_id=12, + assignee="cli-auto", + max_runtime_seconds=300, + goal_mode=True, + goal_max_turns=3, + ) + calls = [] + comments = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + worker_log_path=lambda _task_id, board: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_goal"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + 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)), + block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr( + lanes, + "state_path", + lambda _board, _task_id: tmp_path / "state.json", + ) + route = lanes.Route( + "codex", "gpt-5.6-sol", "xhigh", "codex-xhigh", "jetson", "vote", 1, () + ) + monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route) + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None) + reports = [ + lanes.ProcessResult( + 0, + "first turn", + { + "status": "completed", + "summary": "Focused tests passed.", + "changed_files": ["src/a.py"], + "tests_run": ["pytest focused: passed"], + "artifacts": [], + "blockers": [], + }, + False, + ), + lanes.ProcessResult( + 0, + "second turn", + { + "status": "completed", + "summary": "Full tests passed; commit pushed and remote HEAD verified.", + "changed_files": ["src/a.py"], + "tests_run": ["pytest full: passed"], + "artifacts": [], + "blockers": [], + }, + False, + ), + ] + monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0)) + verdicts = iter( + [ + (False, "commit, push, and remote verification are missing"), + (True, "all explicit acceptance criteria have evidence"), + ] + ) + monkeypatch.setattr( + lanes.cli_lane_goal, + "judge_goal_completion", + lambda *_args, **_kwargs: next(verdicts), + ) + + lanes.execute_claim("cassandra", "t_goal") + + assert calls[0][0] == "complete" + assert calls[0][1]["metadata"]["goal_turn"] == 2 + assert any("Goal completion rejected; continuing turn 2/3" in item for item in comments) + assert reports == [] + + def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch): task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto") calls = []