fix(kanban): judge goal deliverables before completion; fail open for operator errors

hermes_cli/goals.py

@@ -40,7 +40,7 @@
 import time
 from dataclasses import dataclass, field, asdict
 from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 logger = logging.getLogger(__name__)
 
@@ -1174,6 +1174,7 @@ def judge_goal(
     subgoals: Optional[List[str]] = None,
     background_processes: Optional[List[Dict[str, Any]]] = None,
     contract: Optional[GoalContract] = None,
+    completion_handoff: bool = False,
 ) -> Tuple[str, str, bool, Optional[Dict[str, Any]], bool]:
     """Ask the auxiliary model whether the goal is satisfied.
 
@@ -1274,10 +1275,21 @@ def judge_goal(
         # Route through call_llm so auxiliary.goal_judge.* config
         # (provider/model/base_url, extra_body, reasoning_effort, retries)
         # all apply — the direct-create path dropped extra_body (#35566).
+        system_prompt = JUDGE_SYSTEM_PROMPT
+        if completion_handoff:
+            system_prompt += (
+                "\nKANBAN COMPLETION HANDOFF: This is the first attempt to complete "
+                "the task. Judge only the deliverables and verification evidence "
+                "in the proposed summary against the task criteria. Do not require "
+                "a prior kanban_complete call, a completed board state, or a "
+                "completion receipt; those cannot exist until after your verdict. "
+                "Ignore lifecycle-call requirements in the goal text when deciding "
+                "whether the substantive work is done.\n"
+            )
         resp = call_llm(
             task="goal_judge",
             messages=[
-                {"role": "system", "content": JUDGE_SYSTEM_PROMPT},
+                {"role": "system", "content": system_prompt},
                 {"role": "user", "content": prompt},
             ],
             temperature=0,
@@ -1302,6 +1314,81 @@ def judge_goal(
     return verdict, reason, parse_failed, wait_directive, False
 
 
+def goal_judge_available() -> bool:
+    """True when an auxiliary client is configured for the goal judge.
+
+    ``judge_goal`` is fail-open at the source: with no reachable auxiliary
+    model it returns a ``"continue"`` verdict indistinguishable from a real
+    "not done yet". Kanban handoff gates probe this first so an unconfigured
+    judge never wedges a ``goal_mode`` worker out of closing its own task.
+    """
+    try:
+        from agent.auxiliary_client import get_text_auxiliary_client
+        client, model = get_text_auxiliary_client("goal_judge")
+    except Exception:
+        return False
+    return client is not None and bool(model)
+
+
+def kanban_handoff_rejection(
+    task: Any,
+    evidence: str,
+    *,
+    conn: Any = None,
+    task_id: Optional[str] = None,
+    worker_run_id_for: Callable[[str], Optional[int]],
+    judge_available: Callable[[], bool],
+    judge: Optional[Callable[..., Tuple[Any, ...]]] = None,
+) -> Optional[str]:
+    """The ONE goal-mode gate for kanban complete / request-review handoffs.
+
+    Shared by ``tools.kanban_tools`` and ``hermes_cli.kanban`` (the CLI) so the two
+    surfaces cannot drift (Issue #38367 was two copies of one gate). Each
+    surface injects only its own seams: its run-ownership resolver, its judge
+    availability probe, and optionally the judge callable it exposes for tests.
+
+    Contract:
+      * The judge grades the proposed deliverables with
+        ``completion_handoff=True`` — it must never demand a prior
+        kanban_complete receipt (that receipt cannot exist before this call).
+      * A real verdict gates: ``done`` -> None, anything else -> the reason.
+      * A judge ERROR (transport failure / exception / unparseable reply):
+          - the owning worker (resolver returns a run id) retries once, then
+            the card is blocked ``transient`` with the error named;
+          - an operator (no owned run) fails open, with a ``judge_error``
+            event recorded on the caller's ``conn`` for audit.
+    Uses the caller's ``conn``; never opens its own.
+    """
+    if not task or not getattr(task, "goal_mode", False) or not judge_available():
+        return None
+    from hermes_cli import kanban_db as kb
+
+    if judge is None:
+        judge = judge_goal
+    worker_run_id = worker_run_id_for(task_id) if task_id else None
+    reason = "judge unavailable"
+    for _ in range(2 if worker_run_id is not None else 1):
+        try:
+            verdict, reason, parse_failed, _, transport_failed = judge(
+                goal=f"{task.title}\n\n{task.body or ''}".strip(),
+                last_response=(evidence or "").strip(),
+                completion_handoff=True,
+            )
+        except Exception as exc:
+            verdict, reason, parse_failed, transport_failed = (
+                "continue", f"judge error: {type(exc).__name__}", False, True,
+            )
+        if not (parse_failed or transport_failed):
+            return reason if verdict != "done" else None
+    if conn is not None and task_id:
+        kb._append_event(conn, task_id, "judge_error", {"reason": reason}, run_id=worker_run_id)
+        conn.commit()
+        if worker_run_id is not None:
+            blocked = kb.block_task(conn, task_id, reason=reason, kind="transient", expected_run_id=worker_run_id)
+            return f"{reason}; task {'blocked transient after judge retry' if blocked else 'not blocked (run ownership changed)'}"
+    return None
+
+
 def gather_background_processes(task_id: Optional[str] = None) -> List[Dict[str, Any]]:
     """Return the live background-process snapshot for the goal judge.
 

hermes_cli/kanban.py

@@ -3685,37 +3685,16 @@ def _worker_run_id_for(task_id: str) -> Optional[int]:
         return None
 
 
-def _goal_mode_handoff_rejection(task: Optional[kb.Task], evidence: str) -> Optional[str]:
-    """Apply the goal judge to every terminal worker handoff, including review."""
-    if task is None or not task.goal_mode:
-        return None
-    try:
-        from agent.auxiliary_client import get_text_auxiliary_client
-
-        client, model = get_text_auxiliary_client("goal_judge")
-    except Exception:
-        return None
-    if client is None or not model:
-        return None
+def _goal_mode_handoff_rejection(task: Optional[kb.Task], evidence: str, *, conn=None, task_id=None) -> Optional[str]:
+    """CLI-surface wiring of the shared goal-mode handoff gate (complete + review)."""
+    from hermes_cli import goals
 
-    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=evidence.strip(),
-        )
-    except Exception as judge_exc:
-        import logging as _logging
-
-        _logging.getLogger(__name__).warning(
-            "goal judge check failed, allowing lifecycle handoff: %s",
-            judge_exc,
-            exc_info=True,
-        )
-    return reason if verdict != "done" else None
+    return goals.kanban_handoff_rejection(
+        task, evidence, conn=conn, task_id=task_id,
+        worker_run_id_for=_worker_run_id_for,
+        judge_available=goals.goal_judge_available,
+        judge=goals.judge_goal,
+    )
 
 
 def _cmd_complete(args: argparse.Namespace) -> int:
@@ -3772,6 +3751,7 @@ def _cmd_complete(args: argparse.Namespace) -> int:
             rejection = None if superseded_by is not None else _goal_mode_handoff_rejection(
                 task,
                 (summary or args.result or "").strip(),
+                conn=conn, task_id=tid,
             )
             if rejection is not None:
                 print(
@@ -4022,6 +4002,7 @@ def _cmd_request_review(args: argparse.Namespace) -> int:
         rejection = _goal_mode_handoff_rejection(
             kb.get_task(conn, tid),
             summary or "",
+            conn=conn, task_id=tid,
         )
         if rejection is not None:
             print(

tests/hermes_cli/test_kanban_goal_mode.py

@@ -288,6 +288,205 @@ def _continue_judge(goal, response, **_kw):
 # CLI judge gate tests (hermes kanban complete bypass fix)
 # ---------------------------------------------------------------------------
 
+def test_completion_handoff_judge_does_not_require_prior_completion(monkeypatch):
+    from types import SimpleNamespace
+    from agent import auxiliary_client
+
+    prompts = []
+
+    def fake_call_llm(**kwargs):
+        system = kwargs["messages"][0]["content"]
+        prompts.append(system)
+        # A rubric that omits the lifecycle exception reproduces the circular
+        # rejection: evidence exists, but no completion receipt can exist yet.
+        done = "Do not require a prior kanban_complete call" in system
+        content = '{"verdict":"done","reason":"deliverables verified"}' if done else '{"verdict":"continue","reason":"kanban_complete not called"}'
+        return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=content))])
+
+    monkeypatch.setattr(auxiliary_client, "call_llm", fake_call_llm)
+    verdict, reason, *_ = goals.judge_goal(
+        goal="Print two canary phases and then call kanban_complete",
+        last_response="CANARY_PHASE1_NEW exit 0; CANARY_PHASE2_NEW exit 0",
+        completion_handoff=True,
+    )
+    assert verdict == "done", reason
+    assert len(prompts) == 1
+
+
+def test_cli_operator_completion_survives_judge_500(kanban_home, monkeypatch):
+    import argparse
+    from hermes_cli import kanban as cli
+    from agent import auxiliary_client
+
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="Verified artifact", assignee="builder", goal_mode=True)
+        assert kb.claim_task(conn, tid)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(auxiliary_client, "get_text_auxiliary_client", lambda name: (object(), "judge"))
+    monkeypatch.setattr(goals, "judge_goal", lambda **kw: ("continue", "judge error: InternalServerError", False, None, True))
+    args = argparse.Namespace(task_ids=[tid], summary="artifact and test passed", result=None, metadata=None)
+    assert cli._cmd_complete(args) == 0
+    with kb.connect() as conn:
+        assert kb.get_task(conn, tid).status == "done"
+        events = conn.execute("SELECT kind FROM task_events WHERE task_id = ?", (tid,)).fetchall()
+        assert any(e["kind"] == "judge_error" for e in events)
+
+
+def _cli_goal_card(conn, title="Print two canary phases and then call kanban_complete"):
+    tid = kb.create_task(conn, title=title, assignee="builder", goal_mode=True)
+    claimed = kb.claim_task(conn, tid)
+    assert claimed
+    return tid, claimed.current_run_id
+
+
+def _kanban_argv(argv):
+    """Drive the real ``hermes kanban ...`` argv path: build_parser -> parse_args
+    -> kanban_command, exactly as the top-level CLI dispatches it."""
+    import argparse
+    from hermes_cli import kanban as cli
+
+    wrap = argparse.ArgumentParser(prog="wrap", add_help=False)
+    parser = cli.build_parser(wrap.add_subparsers(dest="_top"))
+    return cli.kanban_command(parser.parse_args(argv))
+
+
+def test_cli_complete_first_call_passes_completion_handoff_and_closes(kanban_home, monkeypatch):
+    """(a') ``kanban complete`` argv path: the judge must be asked to grade the
+    deliverables (completion_handoff=True) and the card closes on the first call."""
+    from agent import auxiliary_client
+
+    with kb.connect() as conn:
+        tid, _ = _cli_goal_card(conn)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(auxiliary_client, "get_text_auxiliary_client", lambda name: (object(), "judge"))
+    calls = []
+
+    def judge(**kwargs):
+        calls.append(kwargs)
+        ok = kwargs.get("completion_handoff") is True
+        return ("done" if ok else "continue", "deliverables verified" if ok else "kanban_complete not called", False, None, False)
+
+    monkeypatch.setattr(goals, "judge_goal", judge)
+    rc = _kanban_argv(["complete", tid, "--summary", "CANARY_PHASE1_NEW exit 0; CANARY_PHASE2_NEW exit 0"])
+    assert rc == 0
+    assert len(calls) == 1 and calls[0]["completion_handoff"] is True
+    with kb.connect() as conn:
+        assert kb.get_task(conn, tid).status == "done"
+
+
+def test_cli_complete_real_judge_rubric_accepts_first_completion(kanban_home, monkeypatch):
+    """(a')+(c) end to end on the argv path with the REAL judge_goal prompt.
+    A rubric that demands a prior kanban_complete receipt makes this RED."""
+    from types import SimpleNamespace
+    from agent import auxiliary_client
+
+    with kb.connect() as conn:
+        tid, _ = _cli_goal_card(conn)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(auxiliary_client, "get_text_auxiliary_client", lambda name: (object(), "judge"))
+
+    def fake_call_llm(**kwargs):
+        system = kwargs["messages"][0]["content"]
+        done = "Do not require a prior kanban_complete call" in system
+        content = '{"verdict":"done","reason":"deliverables verified"}' if done else '{"verdict":"continue","reason":"kanban_complete not called"}'
+        return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=content))])
+
+    monkeypatch.setattr(auxiliary_client, "call_llm", fake_call_llm)
+    assert _kanban_argv(["complete", tid, "--summary", "CANARY_PHASE1_NEW exit 0; CANARY_PHASE2_NEW exit 0"]) == 0
+    with kb.connect() as conn:
+        assert kb.get_task(conn, tid).status == "done"
+
+
+def test_cli_owned_worker_judge_500_retries_then_blocks_transient(kanban_home, monkeypatch, capsys):
+    """(b') ``kanban complete`` argv path, owning worker: a judge error is retried
+    once, then the card blocks transient with the error named; never completes."""
+    from hermes_cli import kanban as cli
+    from agent import auxiliary_client
+
+    with kb.connect() as conn:
+        tid, run_id = _cli_goal_card(conn, title="Verified artifact")
+    monkeypatch.delenv("HERMES_KANBAN_OWNER_PID", raising=False)
+    monkeypatch.setenv("HERMES_KANBAN_TASK", tid)
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", str(run_id))
+    assert cli._worker_run_id_for(tid) == run_id
+    monkeypatch.setattr(auxiliary_client, "get_text_auxiliary_client", lambda name: (object(), "judge"))
+    calls = []
+
+    def failing_judge(**kwargs):
+        calls.append(kwargs)
+        return ("continue", "judge error: InternalServerError", False, None, True)
+
+    monkeypatch.setattr(goals, "judge_goal", failing_judge)
+    rc = _kanban_argv(["complete", tid, "--summary", "artifact and test passed"])
+    err = capsys.readouterr().err
+    assert rc != 0
+    assert len(calls) == 2
+    assert "InternalServerError" in err
+    with kb.connect() as conn:
+        task = kb.get_task(conn, tid)
+        assert task.status == "blocked"
+        assert task.block_kind == "transient"
+        events = conn.execute("SELECT kind FROM task_events WHERE task_id = ?", (tid,)).fetchall()
+        assert any(e["kind"] == "judge_error" for e in events)
+
+
+def test_cli_request_review_uses_shared_gate(kanban_home, monkeypatch):
+    """``kanban request-review`` argv path reaches the same gate: a real
+    ``continue`` verdict blocks the handoff and the judge saw completion_handoff."""
+    from agent import auxiliary_client
+
+    with kb.connect() as conn:
+        tid, _ = _cli_goal_card(conn)
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(auxiliary_client, "get_text_auxiliary_client", lambda name: (object(), "judge"))
+    calls = []
+
+    def judge(**kwargs):
+        calls.append(kwargs)
+        return ("continue", "phase 2 output missing", False, None, False)
+
+    monkeypatch.setattr(goals, "judge_goal", judge)
+    assert _kanban_argv(["request-review", tid, "--summary", "phase 1 only"]) != 0
+    assert len(calls) == 1 and calls[0]["completion_handoff"] is True
+    with kb.connect() as conn:
+        assert kb.get_task(conn, tid).status == "running"
+
+
+def test_goal_handoff_predicate_is_defined_exactly_once():
+    """AST contract (Issue #38367 class): the goal-mode handoff rejection
+    predicate — the one function that asks the judge with
+    ``completion_handoff`` — exists exactly once in the tree, and every
+    surface wrapper delegates to it instead of re-implementing it."""
+    import ast
+
+    root = Path(goals.__file__).resolve().parents[1]
+    definers = []
+    delegating = {}
+    for pkg in ("hermes_cli", "tools", "gateway", "agent", "plugins"):
+        base = root / pkg
+        if not base.is_dir():
+            continue
+        for path in base.rglob("*.py"):
+            try:
+                tree = ast.parse(path.read_text(encoding="utf-8"))
+            except (SyntaxError, UnicodeDecodeError):
+                continue
+            for fn in ast.walk(tree):
+                if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                    continue
+                calls = [n for n in ast.walk(fn) if isinstance(n, ast.Call)]
+                if any(k.arg == "completion_handoff" for c in calls for k in c.keywords):
+                    definers.append((path.relative_to(root).as_posix(), fn.name))
+                if fn.name == "_goal_mode_handoff_rejection":
+                    names = {
+                        getattr(c.func, "attr", None) or getattr(c.func, "id", None)
+                        for c in calls
+                    }
+                    delegating[path.relative_to(root).as_posix()] = "kanban_handoff_rejection" in names
+    assert definers == [("hermes_cli/goals.py", "kanban_handoff_rejection")], definers
+    assert delegating == {"tools/kanban_tools.py": True, "hermes_cli/kanban.py": True}, delegating
+
+
 class TestCLIJudgeGate:
     """hermes kanban complete must apply the same goal_mode judge gate as the
     kanban_complete tool (Issue #38367 sibling gap).