From aa32154e4b3db32c75143a17504f1f8651420853 Mon Sep 17 00:00:00 2001 From: srojk34 <286497132+srojk34@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:43:55 +0300 Subject: [PATCH] fix(kanban): apply goal_mode judge gate to CLI complete command MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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. --- hermes_cli/kanban.py | 39 +++++++++ tests/hermes_cli/test_kanban_goal_mode.py | 99 +++++++++++++++++++++++ 2 files changed, 138 insertions(+) diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4de2f8fb8b668..81cc0e8c6d9f0 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -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 ` 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, diff --git a/tests/hermes_cli/test_kanban_goal_mode.py b/tests/hermes_cli/test_kanban_goal_mode.py index da0c2ae168f05..770e134161615 100644 --- a/tests/hermes_cli/test_kanban_goal_mode.py +++ b/tests/hermes_cli/test_kanban_goal_mode.py @@ -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