hermes: enforce goal completion evidence

This commit is contained in:
jenkins 2026-08-16 10:51:34 -03:00
parent 468ab278ea
commit 1527fd1af0
6 changed files with 539 additions and 67 deletions

View File

@ -61,6 +61,7 @@ configMapGenerator:
- ai_usage_exporter.py=scripts/ai_usage_exporter.py - ai_usage_exporter.py=scripts/ai_usage_exporter.py
- claude=scripts/claude - claude=scripts/claude
- claude_command_policy.py=scripts/claude_command_policy.py - 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 - cli_lane_runner.py=scripts/cli_lane_runner.py
- codex=scripts/codex - codex=scripts/codex
- configure_agent_clients.py=scripts/configure_agent_clients.py - configure_agent_clients.py=scripts/configure_agent_clients.py

View File

@ -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}"

View File

@ -22,6 +22,8 @@ from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any, Callable from typing import Any, Callable
import cli_lane_goal
DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data")) DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data"))
SWITCHYARD_URL = os.environ.get( SWITCHYARD_URL = os.environ.get(
@ -60,7 +62,10 @@ RESULT_SCHEMA: dict[str, Any] = {
"additionalProperties": False, "additionalProperties": False,
"required": ["status", "summary", "changed_files", "tests_run", "artifacts", "blockers"], "required": ["status", "summary", "changed_files", "tests_run", "artifacts", "blockers"],
"properties": { "properties": {
"status": {"type": "string", "enum": ["completed", "blocked"]}, "status": {
"type": "string",
"enum": sorted(cli_lane_goal.RESULT_STATUSES),
},
"summary": {"type": "string"}, "summary": {"type": "string"},
"changed_files": {"type": "array", "items": {"type": "string"}}, "changed_files": {"type": "array", "items": {"type": "string"}},
"tests_run": {"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. 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: 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 return value
if not isinstance(value, str): if not isinstance(value, str):
return None return None
@ -333,7 +338,7 @@ def _extract_json(value: Any) -> dict[str, Any] | None:
parsed = json.loads(candidate) parsed = json.loads(candidate)
except json.JSONDecodeError: except json.JSONDecodeError:
continue 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 parsed
return None return None
@ -670,7 +675,7 @@ def run_provider(
max_runtime=max_runtime, max_runtime=max_runtime,
) )
file_result = load_json(result_file) 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 result.structured = file_result
return result return result
@ -890,83 +895,178 @@ def execute_claim(board: str, task_id: str) -> None:
max_runtime = int( max_runtime = int(
_task_value(task, "max_runtime_seconds", 0) or DEFAULT_MAX_RUNTIME _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) goal_mode = bool(_task_value(task, "goal_mode", False))
if result.capacity_failure: goal_max_turns = max(1, int(_task_value(task, "goal_max_turns", 1) or 1))
retry_context = ( goal_turn = max(1, int(state.get("goal_turn", 0) or 0) + 1)
context deadline = time.monotonic() + max_runtime
+ "\n\nRouting boundary: the first provider failed from capacity/authentication. " while True:
+ "Select the alternate hosted provider at an appropriate effort." state["goal_turn"] = goal_turn
) atomic_json(state_file, state)
alternate = "claude" if route.provider == "codex" else "codex" remaining = max(1, int(deadline - time.monotonic()))
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( result = run_provider(
fallback, route,
build_prompt(context, workspace, git_handoff(workspace, result.output)), prompt,
workspace, workspace,
state, state,
state_file, state_file,
log_path, log_path,
heartbeat, heartbeat,
max_runtime, remaining,
) )
route = fallback if result.capacity_failure:
structured = result.structured retry_context = (
if structured: context
structured = dict(structured) + "\n\nRouting boundary: the first provider failed from capacity/authentication. "
structured["artifacts"] = workspace_artifacts( + "Select the alternate hosted provider at an appropriate effort."
workspace, )
structured.get("artifacts"), alternate = "claude" if route.provider == "codex" else "codex"
) fallback = select_route(
metadata = { retry_context,
"executor": "direct-cli-lane", f"cli-{alternate}-{route.effort}",
"provider": route.provider, )
"model": route.model, comment(
"effort": route.effort, f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.",
"classifier": route.classifier, )
"state_file": str(state_file), result = run_provider(
"codex_thread_id": state.get("codex_thread_id"), fallback,
"claude_session_id": state.get("claude_session_id"), build_prompt(
} context,
if structured: workspace,
for key in ("changed_files", "tests_run", "artifacts", "blockers"): git_handoff(workspace, result.output),
value = structured.get(key) ),
metadata[key] = value if isinstance(value, list) else [] workspace,
if structured and structured.get("status") == "completed" and result.returncode == 0: state,
_board_call( state_file,
kanban_db, log_path,
board, heartbeat,
lambda fresh: kanban_db.complete_task( max(1, int(deadline - time.monotonic())),
fresh, )
task_id, route = fallback
result=json.dumps(structured, sort_keys=True), structured = result.structured
summary=str(structured.get("summary") or "Completed"), if structured:
metadata=metadata, structured = dict(structured)
expected_run_id=run_id, structured["artifacts"] = workspace_artifacts(
), workspace,
) structured.get("artifacts"),
else: )
reason = ( metadata = {
"; ".join(str(item) for item in (structured or {}).get("blockers", [])) "executor": "direct-cli-lane",
if structured "provider": route.provider,
else result.output[-4000:] "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( _board_call(
kanban_db, kanban_db,
board, board,
lambda fresh: kanban_db.block_task( lambda fresh: kanban_db.block_task(
fresh, fresh,
task_id, 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", kind="transient" if result.capacity_failure else "capability",
expected_run_id=run_id, expected_run_id=run_id,
), ),
) )
break
except Exception as error: except Exception as error:
failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}" failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}"
_board_call( _board_call(

View File

@ -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

View File

@ -8,9 +8,9 @@ import sys
from pathlib import Path from pathlib import Path
SCRIPT = ( SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
Path(__file__).parents[2] / "services/hermes/scripts/cli_lane_runner.py" sys.path.insert(0, str(SCRIPTS))
) SCRIPT = SCRIPTS / "cli_lane_runner.py"
SPEC = importlib.util.spec_from_file_location("cli_lane_routing", SCRIPT) SPEC = importlib.util.spec_from_file_location("cli_lane_routing", SCRIPT)
assert SPEC and SPEC.loader assert SPEC and SPEC.loader
lanes = importlib.util.module_from_spec(SPEC) lanes = importlib.util.module_from_spec(SPEC)

View File

@ -16,6 +16,7 @@ import yaml
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts" SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
sys.path.insert(0, str(SCRIPTS))
HERMES = Path(__file__).parents[2] / "services/hermes" HERMES = Path(__file__).parents[2] / "services/hermes"
KEYCLOAK = Path(__file__).parents[2] / "services/keycloak" KEYCLOAK = Path(__file__).parents[2] / "services/keycloak"
FLUX_HERMES = ( FLUX_HERMES = (
@ -590,6 +591,22 @@ def test_board_call_retries_storage_faults_on_fresh_connections():
), ),
"complete", "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( 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) 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): 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") task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto")
calls = [] calls = []