kanban/delegation: bind completion to run evidence and add read-only status

hermes_cli/kanban.py

@@ -1052,6 +1052,28 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu
     p_repair.add_argument("--json", action="store_true",
                           help="Emit the repair report as JSON")
 
+    # --- delegation-status ---
+    p_delegation_status = sub.add_parser(
+        "delegation-status",
+        help="Read a background delegation's durable state from state.db",
+        description=(
+            "Reads the async_delegations table in state.db directly, "
+            "bypassing any in-memory or LLM self-report. Use this to "
+            "mechanically verify what actually happened to a background "
+            "delegation, independently of what the child subagent claims.\n\n"
+            "Exit codes: 0=found, 1=not found, 2=error."
+        ),
+    )
+    p_delegation_status.add_argument(
+        "delegation_id",
+        metavar="<delegation-id>",
+        help="The delegation_id to look up",
+    )
+    p_delegation_status.add_argument(
+        "--json", action="store_true",
+        help="Emit the full row as JSON (default: human-readable summary)",
+    )
+
     kanban_parser.set_defaults(_kanban_parser=kanban_parser)
     return kanban_parser
 
@@ -1189,6 +1211,7 @@ def kanban_command(args: argparse.Namespace) -> int:
             "specify":  _cmd_specify,
             "decompose":  _cmd_decompose,
             "gc":       _cmd_gc,
+            "delegation-status": _cmd_delegation_status,
         }
         handler = handlers.get(action)
         if not handler:
@@ -3388,6 +3411,55 @@ def _cmd_gc(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_delegation_status(args: argparse.Namespace) -> int:
+    """Read a background delegation's durable state from state.db.
+
+    Reads async_delegations directly, bypassing all LLM self-report
+    and in-memory state. Exit 0=found, 1=not found, 2=error.
+    """
+    try:
+        from tools.async_delegation import query_delegation_status
+    except ImportError as exc:
+        print(f"kanban delegation-status: cannot import async_delegation: {exc}",
+              file=sys.stderr)
+        return 2
+
+    delegation_id = args.delegation_id
+    use_json = getattr(args, "json", False)
+
+    row = query_delegation_status(delegation_id)
+    if row is None:
+        if use_json:
+            import json
+            print(json.dumps({"found": False, "delegation_id": delegation_id}))
+        else:
+            print(f"delegation-status: {delegation_id!r} not found in state.db",
+                  file=sys.stderr)
+        return 1
+
+    if use_json:
+        import json
+        print(json.dumps(row, default=str))
+    else:
+        state = row.get("state", "?")
+        completed = row.get("completed_at")
+        result = row.get("result") or {}
+        summary = result.get("summary") if isinstance(result, dict) else None
+        status_line = result.get("status") if isinstance(result, dict) else None
+        print(f"delegation_id : {delegation_id}")
+        print(f"state         : {state}")
+        if completed:
+            import datetime
+            ts = datetime.datetime.fromtimestamp(completed).isoformat()
+            print(f"completed_at  : {ts}")
+        if status_line:
+            print(f"result.status : {status_line}")
+        if summary:
+            first_line = str(summary).strip().splitlines()[0][:200]
+            print(f"result.summary: {first_line}")
+    return 0
+
+
 def _cmd_repair(args: argparse.Namespace) -> int:
     """Check DB integrity and apply the narrow index-REINDEX auto-repair.
 

hermes_cli/kanban_db.py

@@ -5360,6 +5360,33 @@ class ArtifactPreservationError(RuntimeError):
     """Raised when a declared scratch deliverable cannot be preserved."""
 
 
+class HallucinatedResultError(ValueError):
+    """Raised by ``complete_task`` when ``expected_result_pattern`` is supplied
+    but does NOT match the ``result``/``summary`` the worker provided.
+
+    Mirrors ``HallucinatedCardsError``'s fail-closed semantics: the task
+    status is never flipped to ``done`` and an auditable event is emitted.
+    The error carries structured fields for callers that need them:
+
+    - ``.task_id``: the task whose completion was rejected.
+    - ``.pattern``: the regex/string pattern that was required.
+    - ``.actual``: the result text that failed to match.
+
+    Kept as ``ValueError`` subclass so existing tool-error handlers treat
+    it as a recoverable user error.
+    """
+
+    def __init__(self, *, task_id: str, pattern: str, actual: Optional[str]):
+        self.task_id = task_id
+        self.pattern = pattern
+        self.actual = actual
+        super().__init__(
+            f"completion blocked for task {task_id!r}: result does not match "
+            f"expected_result_pattern {pattern!r} "
+            f"(got: {(actual or '').strip()[:200]!r})"
+        )
+
+
 def complete_task(
     conn: sqlite3.Connection,
     task_id: str,
@@ -5369,6 +5396,7 @@ def complete_task(
     metadata: Optional[dict] = None,
     created_cards: Optional[Iterable[str]] = None,
     expected_run_id: Optional[int] = None,
+    expected_result_pattern: Optional[str] = None,
     fire_lifecycle_hook: bool = True,
 ) -> bool:
     """Transition ``running|ready|blocked|review -> done`` and record ``result``.
@@ -5436,6 +5464,36 @@ def complete_task(
     else:
         verified_cards = []
 
+    # Gate: verify expected_result_pattern BEFORE the main write txn.
+    # When the caller supplies a pattern, it acts as a mechanical
+    # completeness contract: if the result/summary field does not match,
+    # we block completion with an auditable event (mirrors
+    # HallucinatedCardsError's fail-closed style). When omitted (the
+    # default), behavior is byte-identical to the existing path.
+    if expected_result_pattern is not None:
+        import re as _re
+        _actual = summary if summary is not None else result
+        _matched = bool(
+            _re.search(expected_result_pattern, _actual or "")
+        )
+        if not _matched:
+            with write_txn(conn):
+                _append_event(
+                    conn, task_id, "completion_blocked_hallucination",
+                    {
+                        "reason": "expected_result_pattern_mismatch",
+                        "expected_result_pattern": expected_result_pattern,
+                        "actual_preview": (
+                            (_actual or "").strip()[:200]
+                        ),
+                    },
+                )
+            raise HallucinatedResultError(
+                task_id=task_id,
+                pattern=expected_result_pattern,
+                actual=_actual,
+            )
+
     metadata = _merge_completion_prose_artifacts(
         conn, task_id, metadata, summary=summary, result=result,
     )

tests/hermes_cli/test_kanban_hallucinated_result.py

@@ -0,0 +1,196 @@
+"""Tests for HallucinatedResultError in kanban_db.complete_task.
+
+TDD RED phase: these tests must fail before the implementation exists.
+"""
+from __future__ import annotations
+
+import re
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    """Isolated HERMES_HOME with an empty kanban DB."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+# ---------------------------------------------------------------------------
+# A: HallucinatedResultError class exists and is a ValueError subclass
+# ---------------------------------------------------------------------------
+
+
+def test_hallucinated_result_error_is_value_error():
+    """HallucinatedResultError must be a ValueError subclass (fail-closed style)."""
+    assert issubclass(kb.HallucinatedResultError, ValueError)
+
+
+def test_hallucinated_result_error_carries_task_id_and_pattern():
+    """Error must carry the task_id and the unmatched pattern for structured access."""
+    exc = kb.HallucinatedResultError(
+        task_id="t_abc123",
+        pattern="expected pattern",
+        actual="actual result text",
+    )
+    assert exc.task_id == "t_abc123"
+    assert exc.pattern == "expected pattern"
+    assert exc.actual == "actual result text"
+    assert "t_abc123" in str(exc)
+
+
+# ---------------------------------------------------------------------------
+# A: complete_task without expected_result_pattern — byte-identical behavior
+# ---------------------------------------------------------------------------
+
+
+def test_complete_task_without_pattern_still_works(kanban_home):
+    """Omitting expected_result_pattern must not change existing behavior at all."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        assert kb.complete_task(conn, tid, result="anything at all") is True
+        task = kb.get_task(conn, tid)
+        assert task is not None
+        assert task.status == "done"
+        assert task.result == "anything at all"
+
+
+# ---------------------------------------------------------------------------
+# A: complete_task with expected_result_pattern — matching behavior
+# ---------------------------------------------------------------------------
+
+
+def test_complete_task_with_matching_regex_pattern_completes(kanban_home):
+    """When pattern matches result, task must be completed normally."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        assert (
+            kb.complete_task(
+                conn,
+                tid,
+                result="Implemented feature X in file.py",
+                expected_result_pattern=r"Implemented feature",
+            )
+            is True
+        )
+        task = kb.get_task(conn, tid)
+        assert task is not None
+        assert task.status == "done"
+
+
+def test_complete_task_with_matching_exact_string_completes(kanban_home):
+    """Exact-string (non-regex) match against result must also succeed."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        assert (
+            kb.complete_task(
+                conn,
+                tid,
+                result="exact expected output",
+                expected_result_pattern="exact expected output",
+            )
+            is True
+        )
+
+
+# ---------------------------------------------------------------------------
+# A: complete_task with expected_result_pattern — non-matching must raise
+# ---------------------------------------------------------------------------
+
+
+def test_complete_task_with_nonmatching_pattern_raises(kanban_home):
+    """When pattern does NOT match, HallucinatedResultError must be raised."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        with pytest.raises(kb.HallucinatedResultError) as exc_info:
+            kb.complete_task(
+                conn,
+                tid,
+                result="some vague unverified summary",
+                expected_result_pattern=r"commit [0-9a-f]{7}",
+            )
+        assert exc_info.value.task_id == tid
+        assert "commit" in exc_info.value.pattern
+
+
+def test_complete_task_with_nonmatching_pattern_does_not_commit(kanban_home):
+    """A rejected completion must NOT flip the task to done."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        try:
+            kb.complete_task(
+                conn,
+                tid,
+                result="bad result",
+                expected_result_pattern=r"REQUIRED_TOKEN",
+            )
+        except kb.HallucinatedResultError:
+            pass
+        task = kb.get_task(conn, tid)
+        assert task is not None
+        assert task.status != "done", "Task must not be done after pattern rejection"
+
+
+def test_complete_task_with_nonmatching_pattern_emits_audit_event(kanban_home):
+    """Rejected completion must emit an auditable event (never silent)."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        try:
+            kb.complete_task(
+                conn,
+                tid,
+                result="fake result",
+                expected_result_pattern=r"REQUIRED_TOKEN",
+            )
+        except kb.HallucinatedResultError:
+            pass
+        events = conn.execute(
+            "SELECT kind, payload FROM task_events WHERE task_id = ?",
+            (tid,),
+        ).fetchall()
+        event_kinds = [e["kind"] for e in events]
+        assert any(
+            "hallucinated" in k or "blocked" in k for k in event_kinds
+        ), f"Expected an audit event, got: {event_kinds}"
+
+
+def test_complete_task_with_none_result_and_pattern_raises(kanban_home):
+    """Pattern check against None/empty result must also raise (fail-closed)."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent")
+        tid = kb.create_task(conn, title="child", parents=[parent])
+        kb.complete_task(conn, parent, result="parent done")
+        kb.claim_task(conn, tid)
+        with pytest.raises(kb.HallucinatedResultError):
+            kb.complete_task(
+                conn,
+                tid,
+                result=None,
+                expected_result_pattern=r"REQUIRED_TOKEN",
+            )