fix(kanban): refuse a receipt-less worker completion (no_receipt, rc 4)

hermes_cli/config_defaults.py

@@ -3273,6 +3273,11 @@
         # DEPLOYED", "STOP finding", outcome=partial, ...) goes to review
         # (human:apollo) instead of done. See kanban_negative_handoff.py.
         "negative_handoff_review": False,
+        # True: a worker's completion of a running card with no receipt (no
+        # PR, survivor, attachment or structured metadata beyond bookkeeping)
+        # is refused with reason code no_receipt (t_e21aa11c). Lint the board:
+        # python -m hermes_cli.kanban_receipt --days 7
+        "receipt_gate": True,
         # "all" (default): every request_review routes to review_assignee.
         # "milestone_only": only cards whose title/body carry "[milestone]"
         # or "qa:required" get a reviewer session (being a task_links parent

hermes_cli/kanban.py

@@ -33,6 +33,7 @@
 from hermes_cli.kanban_pr_freshness import DraftPrError
 from hermes_cli.kanban_branch_base import StaleBaseError
 from hermes_cli.kanban_open_pr import ClosedUnmergedPrError
+from hermes_cli.kanban_receipt import EXIT_NO_RECEIPT, ReceiptRequiredError
 from hermes_cli.kanban_identity import safe_comment_provenance
 from hermes_constants import get_default_hermes_root
 
@@ -4935,6 +4936,7 @@ def _cmd_complete(args: argparse.Namespace) -> int:
             )
             return 2
     failed: list[str] = []
+    no_receipt = False
     with kb.connect_closing() as conn:
         for tid in ids:
             # Goal-mode judge gate (mirrors tools/kanban_tools.py). Apply it
@@ -4973,6 +4975,11 @@ def _cmd_complete(args: argparse.Namespace) -> int:
                     superseded_by=superseded_by,
                     draft_ok=draft_ok,
                 )
+            except ReceiptRequiredError as receipt_err:
+                failed.append(tid)
+                no_receipt = True
+                print(f"cannot complete {tid}: {receipt_err}.", file=sys.stderr)
+                continue
             except (kb.EmptySupersedeError, kb.EmptyDraftOverrideError) as supersede_err:
                 failed.append(tid)
                 print(f"cannot complete {tid}: {supersede_err}.", file=sys.stderr)
@@ -5000,6 +5007,8 @@ def _cmd_complete(args: argparse.Namespace) -> int:
                 if override:
                     print(f"  draft override recorded for {', '.join(override['prs'])}: "
                           f"{override['reason']}")
+    if no_receipt:
+        return EXIT_NO_RECEIPT
     return 0 if not failed else 1
 
 
@@ -5437,6 +5446,9 @@ def _cmd_request_review(args: argparse.Namespace) -> int:
                 allow_same_actor=bool(getattr(args, "allow_same_actor", False)),
                 with_reason=True,
             )
+        except ReceiptRequiredError as receipt_err:
+            print(f"cannot request review for {tid}: {receipt_err}", file=sys.stderr)
+            return EXIT_NO_RECEIPT
         except (DraftPrError, StaleBaseError, ClosedUnmergedPrError) as draft_err:
             print(f"cannot request review for {tid}: {draft_err}", file=sys.stderr)
             return 1

hermes_cli/kanban_db.py

@@ -11512,6 +11512,31 @@ def complete_task(
         and candidate.current_run_id is not None
         and _retry_status_for_run(conn, task_id, candidate.current_run_id) == "review"
     )
+    # Receipt gate (t_e21aa11c): a dispatcher-owned worker handoff
+    # (``expected_run_id`` set) that closes ``done`` with prose only -- no PR,
+    # survivor, attachment or structured metadata -- is refused before any
+    # mutation. Operator closes are not gated. Knob ``kanban.receipt_gate``.
+    if (
+        expected_run_id is not None
+        and candidate.status == 'running' and not review_claimed
+        and not approve_head_sha and not superseded_by
+        and configured_receipt_gate()
+    ):
+        from hermes_cli import kanban_receipt as _receipt
+        if _receipt.missing(
+            summary=summary, result=result, metadata=metadata,
+            survivor_pr=survivor_pr, survivor_ref=survivor_ref,
+            survivor_none=survivor_none,
+            attachments=_receipt.attachment_count(conn, task_id),
+        ):
+            with write_txn(conn):
+                _append_event(
+                    conn, task_id, _receipt.EVENT,
+                    {"reason": _receipt.REASON_CODE,
+                     "metadata_keys": sorted((metadata or {}).keys())
+                     if isinstance(metadata, dict) else []},
+                )
+            raise _receipt.ReceiptRequiredError(task_id)
     negative_trigger: Optional[str] = None
     if (
         candidate.status == 'running' and not review_claimed
@@ -14025,6 +14050,17 @@ def configured_negative_handoff_review() -> bool:
     return value is True or str(value).strip().casefold() in ("1", "true", "yes", "on")
 
 
+def configured_receipt_gate() -> bool:
+    """``kanban.receipt_gate`` — refuse a receipt-less implementer completion (default on)."""
+    try:
+        value, _source = _kanban_review_setting("receipt_gate", True)
+    except Exception:
+        return True
+    if isinstance(value, bool):
+        return value
+    return str(value).strip().casefold() not in ("0", "false", "no", "off")
+
+
 def configured_review_policy() -> str:
     """``kanban.review_policy`` — ``all`` (default), ``milestone_only`` or ``none``.
 

hermes_cli/kanban_receipt.py

@@ -0,0 +1,190 @@
+"""Completion receipt gate: a worker's ``done`` handoff must carry evidence.
+
+Incident 2026-10-01 (t_e21aa11c): deploy cards closed ``done`` with nothing a
+reader could check but prose, and the session overview printed "no receipt on
+card" for them. A receipt is one of:
+
+* a PR the handoff owns or names (``owner/repo#N`` / PR URL in summary, result,
+  ``metadata.pr_url|pr_urls|pr`` or ``survivor_pr``),
+* a survivor claim (``survivor_ref`` / ``survivor_none``),
+* an attachment on the card, or declared ``metadata.artifacts``,
+* structured handoff metadata: any key outside :data:`BOOKKEEPING_KEYS`
+  (what was deployed, a sha table, a read-back).
+
+:func:`missing` is the single predicate. ``complete_task`` refuses a worker's
+dispatcher-owned (``expected_run_id``) completion with
+:class:`ReceiptRequiredError` (reason code ``no_receipt``; CLI exit
+:data:`EXIT_NO_RECEIPT`) when the board knob ``kanban.receipt_gate`` is on, and
+``python -m hermes_cli.kanban_receipt --days 7`` lists the receipt-less
+``done`` cards already on a board.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+import sys
+import time
+from typing import Any, Iterable, Optional
+
+SETTING = "receipt_gate"
+REASON_CODE = "no_receipt"
+# ``hermes kanban complete`` / ``request-review`` exit status for this refusal
+# (0 ok, 1 other refusal, 2 usage error).
+EXIT_NO_RECEIPT = 4
+EVENT = "completion_blocked_no_receipt"
+
+# Keys the runtime stamps, or that only say "no PR" without saying what was
+# done. A handoff whose metadata holds nothing else has no structured receipt.
+BOOKKEEPING_KEYS = frozenset({
+    "worker_session_id",
+    "pool",
+    "review_skipped",
+    "no_pr",
+    "no_pr_reason",
+})
+
+
+class ReceiptRequiredError(ValueError):
+    """``complete_task`` refused a worker handoff that carries no receipt.
+
+    ``ValueError`` so tool-error handlers treat it as recoverable: nothing was
+    mutated except the ``completion_blocked_no_receipt`` audit event.
+    """
+
+    code = REASON_CODE
+
+    def __init__(self, task_id: str):
+        self.task_id = task_id
+        super().__init__(
+            f"completion blocked ({REASON_CODE}): {task_id} has no receipt. Prose "
+            f"alone is not one. Attach the evidence (kanban_attach / artifacts=[...]), "
+            f"or put a structured handback in metadata (what was deployed, a "
+            f"before/after sha table, the read-back), or name the PR. "
+            f"{task_id} is still in-flight (no state change)"
+        )
+
+
+def _has_pr(summary, result, metadata, survivor_pr) -> bool:
+    from hermes_cli import kanban_open_pr as _open_pr
+
+    return bool(_open_pr.extract_pr_refs(
+        result, summary, metadata=metadata if isinstance(metadata, dict) else None,
+        survivor_pr=survivor_pr,
+    ))
+
+
+def structured_keys(metadata: Any) -> list:
+    """Metadata keys that count as a structured receipt (sorted)."""
+    if not isinstance(metadata, dict):
+        return []
+    return sorted(
+        k for k, v in metadata.items()
+        if k not in BOOKKEEPING_KEYS and v not in (None, "", [], {})
+    )
+
+
+def missing(
+    *,
+    summary: Optional[str] = None,
+    result: Optional[str] = None,
+    metadata: Any = None,
+    survivor_pr=None,
+    survivor_ref=None,
+    survivor_none: bool = False,
+    attachments: int = 0,
+) -> bool:
+    """True when the handoff carries no receipt (see module docstring)."""
+    if attachments or survivor_ref or survivor_none:
+        return False
+    if structured_keys(metadata):
+        return False
+    return not _has_pr(summary, result, metadata, survivor_pr)
+
+
+def attachment_count(conn: sqlite3.Connection, task_id: str) -> int:
+    row = conn.execute(
+        "SELECT count(*) FROM task_attachments WHERE task_id = ?", (task_id,)
+    ).fetchone()
+    return int(row[0]) if row else 0
+
+
+# ---------------------------------------------------------------- lint
+def lint(conn: sqlite3.Connection, *, days: float = 7, now: Optional[int] = None) -> list:
+    """Receipt-less ``done`` cards completed in the last ``days`` days.
+
+    Reads each card's latest ENDED run (the closing handoff). Superseded closes
+    carry their pointer as evidence and are skipped. Returns dicts sorted by
+    completion time.
+    """
+    now = int(now if now is not None else time.time())
+    since = now - int(days * 86400)
+    rows = conn.execute(
+        """
+        SELECT t.id, t.title, t.result, t.completed_at,
+               r.profile, r.outcome, r.summary, r.metadata,
+               (SELECT count(*) FROM task_attachments a WHERE a.task_id = t.id)
+          FROM tasks t
+          LEFT JOIN task_runs r ON r.id = (
+                SELECT max(id) FROM task_runs
+                 WHERE task_id = t.id AND ended_at IS NOT NULL)
+         WHERE t.status = 'done' AND t.completed_at >= ?
+         ORDER BY t.completed_at, t.id
+        """,
+        (since,),
+    ).fetchall()
+    out = []
+    for tid, title, result, completed_at, profile, outcome, summary, meta, att in rows:
+        if outcome == "superseded":
+            continue
+        try:
+            metadata = json.loads(meta) if meta else None
+        except (TypeError, ValueError):
+            metadata = None
+        if not missing(summary=summary, result=result, metadata=metadata, attachments=att):
+            continue
+        out.append({
+            "id": tid,
+            "completed_at": completed_at,
+            "closed_by": profile or "-",
+            "worker_handoff": isinstance(metadata, dict) and "worker_session_id" in metadata,
+            "title": (title or "").strip(),
+            "summary": ((summary or result or "").strip().splitlines() or [""])[0],
+        })
+    return out
+
+
+def _main(argv: Optional[Iterable[str]] = None) -> int:
+    ap = argparse.ArgumentParser(
+        prog="python -m hermes_cli.kanban_receipt",
+        description="List receipt-less done cards (rc 0 none, 1 some found, 2 error).",
+    )
+    ap.add_argument("--days", type=float, default=7)
+    ap.add_argument("--db", default=None, help="board DB (default: the active board)")
+    ap.add_argument("--json", action="store_true")
+    args = ap.parse_args(list(argv) if argv is not None else None)
+    try:
+        if args.db:
+            conn = sqlite3.connect(f"file:{args.db}?mode=ro", uri=True)
+        else:
+            from hermes_cli import kanban_db as kb
+            conn = kb.connect()
+        with conn:
+            found = lint(conn, days=args.days)
+    except (sqlite3.Error, OSError) as exc:
+        print(f"kanban_receipt: cannot read board: {exc}", file=sys.stderr)
+        return 2
+    if args.json:
+        print(json.dumps(found, indent=1))
+    else:
+        for c in found:
+            ts = time.strftime("%Y-%m-%d %H:%M", time.localtime(c["completed_at"]))
+            who = c["closed_by"] + (" (worker)" if c["worker_handoff"] else "")
+            print(f"{ts}  {c['id']}  {who:<22}  {c['title'][:70]}  | {c['summary'][:80]}")
+        print(f"{len(found)} receipt-less done card(s) in the last {args.days:g} day(s)")
+    return 1 if found else 0
+
+
+if __name__ == "__main__":
+    sys.exit(_main())

tests/agent/test_kanban_stop.py

@@ -366,7 +366,8 @@ def test_non_worker_does_not_nudge(clear_kanban_env):
 def test_closed_run_needs_no_tool_history(worker_run, handoff):
     conn, task = worker_run
     if handoff == "complete":
-        assert kb.complete_task(conn, task.id, expected_run_id=task.current_run_id)
+        assert kb.complete_task(conn, task.id, metadata={"tests_run": 1},
+                                expected_run_id=task.current_run_id)
     elif handoff == "block":
         assert kb.block_task(
             conn, task.id, reason="Missing input", expected_run_id=task.current_run_id,

tests/hermes_cli/test_kanban_cli_non_owner_complete.py

@@ -115,6 +115,7 @@ def test_owner_still_completes_its_own_card(home):
     task_id, run_id = _running_card(home)
 
     out = _run(home, "complete", task_id, "--summary", "owner done",
+               "--metadata", '{"tests_run": 1}',
                extra=_worker_env(task_id, run_id, "pending"))
 
     assert out.returncode == 0, out.stdout + out.stderr

tests/hermes_cli/test_kanban_complete_refusal.py

@@ -33,7 +33,8 @@ def test_already_done_names_who_and_when(kanban_home):
         t = kb.create_task(conn, title="x", assignee="daedalus")
         claimed = kb.claim_task(conn, t)
         assert claimed is not None
-        assert kb.complete_task(conn, t, summary="done", expected_run_id=claimed.current_run_id)
+        assert kb.complete_task(conn, t, summary="done", metadata={"tests_run": 1},
+                                expected_run_id=claimed.current_run_id)
         assert not kb.complete_task(conn, t, summary="again")
         msg = kb.explain_complete_refusal(conn, t)
     assert msg.startswith("already done by daedalus at 20"), msg

tests/hermes_cli/test_kanban_link_kinds.py

@@ -181,7 +181,8 @@ def test_running_child_completes_while_derived_from_parent_is_open(kanban_home):
         run_id = kb.get_task(conn, child).current_run_id
 
         assert kb.complete_task(
-            conn, child, summary="shipped", expected_run_id=run_id,
+            conn, child, summary="shipped", metadata={"tests_run": 1},
+            expected_run_id=run_id,
         ) is True
         assert kb.get_task(conn, child).status == "done"
         assert kb.get_task(conn, parent).status == "ready"

tests/hermes_cli/test_kanban_open_pr_route.py

@@ -329,6 +329,7 @@ def test_milestone_only_in_place_completion_uses_worker_run_id(kanban_home, monk
         assert ok is False and "expected_run_id" in reason
         assert _status(conn, tid) == "running"
         ok, reason = kb.request_review(conn, tid, summary="done",
+                                       metadata={"tests_run": 1},
                                        expected_run_id=run_id, with_reason=True)
         assert ok is True, reason
         assert _status(conn, tid) == "done"