fix(kanban): apply goal_mode judge gate to CLI complete command

The three-commit hardening series (Issue #38367, PR #55408) added a
pre-completion judge gate to `tools/kanban_tools.py:_handle_complete`
(the kanban_complete tool used by agent tool-calls).  The structurally
identical `hermes_cli/kanban.py:_cmd_complete` (the `hermes kanban
complete` CLI subcommand) was left unguarded.

A goal_mode worker with terminal tool access — the overwhelming default
for coding agents — can bypass the judge entirely by running:
    hermes kanban complete <task_id>
This transitions the task to `done` status with no judge verdict, making
the acceptance-criteria enforcement worthless on that path.

Fix: apply the same gate in _cmd_complete before calling kb.complete_task.
When a judge is reachable and returns anything other than "done", the
command prints an actionable rejection message and exits non-zero without
modifying the task.  The fail-open policy (no judge configured → allowed)
is preserved to match the tool-call path.
This commit is contained in:
srojk34 2026-06-30 22:43:55 +03:00 committed by Teknium
parent 73543744bc
commit aa32154e4b
2 changed files with 138 additions and 0 deletions

View File

@ -1996,6 +1996,45 @@ def _cmd_complete(args: argparse.Namespace) -> int:
failed: list[str] = []
with kb.connect_closing() as conn:
for tid in ids:
# Goal-mode pre-completion judge gate (mirrors the gate in
# tools/kanban_tools.py:_handle_complete — Issue #38367).
# Without this, a goal_mode worker can call
# `hermes kanban complete <id>` from the terminal tool and
# bypass the auxiliary judge that the tool-call path enforces.
task = kb.get_task(conn, tid)
if task and task.goal_mode:
judge_available = False
try:
from agent.auxiliary_client import get_text_auxiliary_client
_client, _model = get_text_auxiliary_client("goal_judge")
judge_available = _client is not None and bool(_model)
except Exception:
pass
if judge_available:
from hermes_cli.goals import judge_goal
verdict = "done"
reason = ""
try:
verdict, reason, _ = judge_goal(
goal=f"{task.title}\n\n{task.body or ''}".strip(),
last_response=(summary or args.result or "").strip(),
)
except Exception as judge_exc:
import logging as _logging
_logging.getLogger(__name__).warning(
"goal judge check failed, allowing completion: %s",
judge_exc,
exc_info=True,
)
if verdict != "done":
print(
f"kanban: goal completion of {tid} rejected by judge: {reason}. "
f"Provide evidence matching the task's acceptance criteria.",
file=sys.stderr,
)
failed.append(tid)
continue
if not kb.complete_task(
conn, tid,
result=args.result,

View File

@ -296,3 +296,102 @@ def test_loop_stops_if_task_reclaimed(monkeypatch):
first_response="x",
)
assert res["outcome"] == "stopped"
# ---------------------------------------------------------------------------
# CLI judge gate tests (hermes kanban complete bypass fix)
# ---------------------------------------------------------------------------
def _make_goal_task(tmp_path):
"""Create a SQLite kanban DB with one goal_mode task and return (db_path, task_id)."""
db_path = tmp_path / "kanban.db"
conn = sqlite3.connect(str(db_path))
conn.execute("""
CREATE TABLE tasks (
id TEXT PRIMARY KEY,
title TEXT,
body TEXT,
status TEXT DEFAULT 'todo',
goal_mode INTEGER DEFAULT 0,
result TEXT,
summary TEXT,
metadata TEXT,
run_id TEXT,
created_at TEXT,
updated_at TEXT
)
""")
conn.execute(
"INSERT INTO tasks (id, title, body, status, goal_mode) VALUES (?, ?, ?, ?, ?)",
("t-goal", "Finish the report", "Body details", "running", 1),
)
conn.commit()
conn.close()
return db_path, "t-goal"
class TestCLIJudgeGate:
"""hermes kanban complete must apply the same goal_mode judge gate as the
kanban_complete tool (Issue #38367 sibling gap).
Uses mocks for kb.get_task and kb.complete_task to avoid depending on the
full kanban_db schema; the gate logic is the unit under test.
"""
def _run(self, monkeypatch, *, goal_mode=True, judge_available=True,
verdict="done", reason="", complete_ok=True, summary="done"):
import argparse
import types
from unittest.mock import MagicMock, patch
from hermes_cli.kanban import _cmd_complete
fake_task = types.SimpleNamespace(
goal_mode=goal_mode,
title="Finish report",
body="acceptance: criteria",
)
fake_conn = MagicMock()
def fake_connect_closing():
from contextlib import contextmanager
@contextmanager
def _cm():
yield fake_conn
return _cm()
monkeypatch.setattr("hermes_cli.kanban.kb.get_task", lambda conn, tid: fake_task)
monkeypatch.setattr("hermes_cli.kanban.kb.complete_task",
lambda conn, tid, **kw: complete_ok)
monkeypatch.setattr("hermes_cli.kanban.kb.connect_closing", fake_connect_closing)
monkeypatch.setattr("hermes_cli.kanban._worker_run_id_for", lambda _: None)
_aux_client = (object(), "judge-model") if judge_available else (None, None)
monkeypatch.setattr(
"agent.auxiliary_client.get_text_auxiliary_client",
lambda name: _aux_client,
)
monkeypatch.setattr(
"hermes_cli.goals.judge_goal",
lambda **kw: (verdict, reason, {}),
)
args = argparse.Namespace(task_ids=["t1"], summary=summary, result=None, metadata=None)
return _cmd_complete(args)
def test_judge_rejects_premature_completion(self, monkeypatch):
rc = self._run(monkeypatch, verdict="continue", reason="criteria not met")
assert rc != 0, "judge rejection must produce non-zero exit code"
def test_judge_allows_accepted_completion(self, monkeypatch):
rc = self._run(monkeypatch, verdict="done")
assert rc == 0
def test_judge_unavailable_fails_open(self, monkeypatch):
"""No auxiliary client configured → gate skipped, task completes."""
rc = self._run(monkeypatch, judge_available=False)
assert rc == 0
def test_non_goal_mode_task_skips_gate(self, monkeypatch):
"""Plain (non-goal_mode) tasks are never sent to the judge."""
rc = self._run(monkeypatch, goal_mode=False)
assert rc == 0