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

hermes_cli/delegation_status.py

@@ -0,0 +1,71 @@
+"""Read-only durable observer for background delegation state."""
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any, Optional
+
+
+class DelegationStatusReadError(RuntimeError):
+    """The durable delegation store could not be read reliably."""
+
+
+def _read_only_uri(path: Path) -> str:
+    resolved = path.expanduser().resolve(strict=False)
+    return f"{resolved.as_uri()}?mode=ro"
+
+
+def query_delegation_status(
+    delegation_id: str,
+    *,
+    db_path: str | Path,
+) -> Optional[dict[str, Any]]:
+    """Return one durable delegation row, ``None`` only for an absent id.
+
+    The SQLite connection uses ``mode=ro`` and never initializes or migrates
+    schema. Missing, unreadable, corrupt, or incompatible databases raise a
+    typed error so callers cannot mistake uncertainty for absence.
+    """
+    path = Path(db_path)
+    try:
+        conn = sqlite3.connect(_read_only_uri(path), uri=True, timeout=10)
+        conn.row_factory = sqlite3.Row
+        try:
+            row = conn.execute(
+                """SELECT delegation_id, origin_session, origin_ui_session_id,
+                          parent_session_id, state, dispatched_at,
+                          completed_at, updated_at, result_json,
+                          delivery_state, delivery_attempts, delivered_at,
+                          owner_pid, task_json
+                   FROM async_delegations
+                   WHERE delegation_id = ?""",
+                (delegation_id,),
+            ).fetchone()
+        finally:
+            conn.close()
+    except (OSError, sqlite3.Error) as exc:
+        raise DelegationStatusReadError(
+            f"cannot read delegation state from {path}: {exc}"
+        ) from exc
+
+    if row is None:
+        return None
+    result: dict[str, Any] = dict(row)
+    raw_result_json = result.pop("result_json", None)
+    if raw_result_json:
+        try:
+            result["result"] = json.loads(raw_result_json)
+        except (json.JSONDecodeError, TypeError):
+            result["result"] = raw_result_json
+    else:
+        result["result"] = None
+    raw_task_json = result.pop("task_json", None)
+    if raw_task_json:
+        try:
+            result["task"] = json.loads(raw_task_json)
+        except (json.JSONDecodeError, TypeError):
+            result["task"] = None
+    else:
+        result["task"] = None
+    return result

hermes_cli/kanban_db.py

@@ -2664,6 +2664,18 @@ class ArtifactPreservationError(RuntimeError):
     """Raised when a declared scratch deliverable cannot be preserved."""
 
 
+class CompletionEvidenceError(ValueError):
+    """Raised when a required independent completion receipt is invalid."""
+
+    def __init__(self, *, task_id: str, reason: str):
+        self.task_id = task_id
+        self.reason = reason
+        super().__init__(
+            f"completion blocked for task {task_id!r}: independent evidence "
+            f"is required ({reason})"
+        )
+
+
 class LiveClaimError(ValueError):
     """``complete_task`` refused: the task is ``running`` under a live claim and
     the caller neither owns its run (``expected_run_id``) nor passed ``force``.
@@ -2694,10 +2706,18 @@ def _claim_is_live(trow) -> bool:
 
 
 def complete_task(
-    conn: sqlite3.Connection, task_id: str, *, result: Optional[str] = None,
-    summary: Optional[str] = None, metadata: Optional[dict] = None,
-    created_cards: Optional[Iterable[str]] = None, expected_run_id: Optional[int] = None,
-    fire_lifecycle_hook: bool = True, force: bool = False,
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    result: Optional[str] = None,
+    summary: Optional[str] = None,
+    metadata: Optional[dict] = None,
+    created_cards: Optional[Iterable[str]] = None,
+    expected_run_id: Optional[int] = None,
+    require_completion_evidence: bool = False,
+    completion_evidence: Optional[Mapping[str, Any]] = None,
+    fire_lifecycle_hook: bool = True,
+    force: bool = False,
 ) -> bool:
     """``running|ready|blocked|review -> done``; records ``result``.
 
@@ -2711,13 +2731,69 @@ def complete_task(
     ``created_cards`` are verified first — a phantom id raises
     :class:`HallucinatedCardsError` after an auditable event; afterwards the
     prose is scanned for unresolvable ``t_<hex>`` refs (advisory event only).
+
+    When ``require_completion_evidence`` is true, ``completion_evidence``
+    must carry a passed terminal-verification receipt bound to this exact
+    ``task_id`` and ``expected_run_id``. Rejections are audited without
+    copying worker prose into the event. The worker tool enables this gate
+    for code workspaces and leaves non-code/manual completion compatible.
+    Missing or malformed evidence (no receipt, wrong task/run, untrusted
+    source, not ``passed``) always fails closed via
+    :class:`CompletionEvidenceError` — it never silently falls back to
+    fail-open completion.
     """
     now = int(time.time())
     # Cheap pre-check; re-checked inside the txn to close the parent-reopen race.
     if not _parents_satisfied(conn, task_id):
         return False
     from hermes_cli.kanban_pr_acceptance_store import prepare_acceptance, record_acceptance
+
+    # Gate: verify created_cards BEFORE the main write txn (main's helper —
+    # a rejected completion still needs an auditable event, emitted in its
+    # own tiny txn, then raised; this function never mutates task state on
+    # a phantom-card rejection).
     verified_cards = _gate_created_cards(conn, task_id, created_cards, summary or result)
+
+    # A worker's summary is a claim, never acceptance proof. Code-workspace
+    # callers can require a receipt captured by the terminal verification
+    # ledger. The trusted tool wrapper binds that receipt to this exact task
+    # and dispatcher run; stale evidence from an earlier retry fails closed.
+    verified_completion_evidence: Optional[dict[str, Any]] = None
+    if require_completion_evidence:
+        evidence = dict(completion_evidence or {})
+        if evidence.get("receipt_id") is None:
+            evidence_reason = "missing_receipt"
+        elif evidence.get("status") != "passed":
+            evidence_reason = "not_passed"
+        elif evidence.get("task_id") != task_id:
+            evidence_reason = "task_mismatch"
+        elif expected_run_id is None or evidence.get("run_id") != int(expected_run_id):
+            evidence_reason = "run_mismatch"
+        elif evidence.get("source", "").split(":", 1)[0] != "verification_evidence":
+            evidence_reason = "untrusted_source"
+        else:
+            evidence_reason = None
+        if evidence_reason is not None:
+            with write_txn(conn):
+                _append_event(
+                    conn,
+                    task_id,
+                    "completion_blocked_missing_evidence",
+                    {
+                        "reason": evidence_reason,
+                        "receipt_id": evidence.get("receipt_id"),
+                        "evidence_run_id": evidence.get("run_id"),
+                        "expected_run_id": expected_run_id,
+                    },
+                )
+            raise CompletionEvidenceError(task_id=task_id, reason=evidence_reason)
+        verified_completion_evidence = {
+            key: evidence.get(key)
+            for key in (
+                "receipt_id", "source", "status", "task_id", "run_id",
+                "session_id", "root", "created_at",
+            )
+        }
     metadata = _merge_completion_prose_artifacts(
         conn, task_id, metadata, summary=summary, result=result,
     )
@@ -2850,11 +2926,15 @@ def _cleaned_artifact_paths(metadata: Any) -> list[str]:
 
 def _completed_event_payload(
     result: Optional[str], event_summary: Optional[str], verified_cards: list[str], metadata: Any,
+    completion_evidence: Optional[dict] = None,
 ) -> dict:
     """``completed`` event payload: first summary line (400 chars) so gateway
     notifiers / dashboard WS render without a second round-trip; verified
-    cards; and ``metadata["artifacts"]`` promoted so the notifier can upload
-    them as native attachments without fetching the run row."""
+    cards; ``metadata["artifacts"]`` promoted so the notifier can upload
+    them as native attachments without fetching the run row; and, when the
+    caller required independent completion evidence, the verified receipt
+    (subset of fields only) so the audit trail shows the completion was
+    backed by proof, not just a worker's claim."""
     # Mirror CLI's _show_voice_status: include STT/TTS provider availability so the user can tell at a
     # glance *why* voice mode isn't working ("STT provider: MISSING ..." is the common case). ``record_key``
     # mirrors the configured ``voice.record_key`` so the TUI can both bind it (frontend
@@ -2866,6 +2946,8 @@ def _completed_event_payload(
     }
     if verified_cards:
         payload["verified_cards"] = verified_cards
+    if completion_evidence is not None:
+        payload["completion_evidence"] = completion_evidence
     if isinstance(metadata, dict):
         cleaned = _cleaned_artifact_paths(metadata)
         if cleaned:
@@ -4352,6 +4434,7 @@ def parent_results(conn: sqlite3.Connection, task_id: str) -> list[tuple[str, Op
     'KANBAN_TERMINAL_TIMEOUT_GRACE_SECONDS': ('hermes_cli.kanban_db_dispatch', 'KANBAN_TERMINAL_TIMEOUT_GRACE_SECONDS'),
     'KanbanDbCorruptError': ('hermes_cli.kanban_db_connect', 'KanbanDbCorruptError'),
     'MEMORY_GUARD_MB_PER_WORKER': ('hermes_cli.kanban_db_dispatch', 'MEMORY_GUARD_MB_PER_WORKER'),
+    '_NOTIFY_DELIVERY_MODES': ('hermes_cli.kanban_db_notify', '_NOTIFY_DELIVERY_MODES'),
     'RepairResult': ('hermes_cli.kanban_db_connect', 'RepairResult'),
     'add_notify_sub': ('hermes_cli.kanban_db_notify', 'add_notify_sub'),
     'advance_notify_cursor': ('hermes_cli.kanban_db_notify', 'advance_notify_cursor'),

tests/hermes_cli/test_kanban_completion_evidence.py

@@ -0,0 +1,128 @@
+"""Run-bound completion-evidence gates for kanban task completion."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+def _running_task(conn):
+    task_id = kb.create_task(conn, title="child", assignee="worker")
+    task = kb.claim_task(conn, task_id)
+    assert task is not None
+    assert task.current_run_id is not None
+    return task_id, task.current_run_id
+
+
+def _evidence(base_task_id: str, base_run_id: int, **overrides):
+    evidence = {
+        "receipt_id": 41,
+        "source": "verification_evidence",
+        "status": "passed",
+        "task_id": base_task_id,
+        "run_id": base_run_id,
+        "session_id": "session-1",
+        "root": "/repo",
+    }
+    evidence.update(overrides)
+    return evidence
+
+
+def test_matching_worker_prose_without_evidence_cannot_complete(kanban_home):
+    """A perfect success claim is not independent acceptance evidence."""
+    with kb.connect() as conn:
+        task_id, run_id = _running_task(conn)
+        with pytest.raises(kb.CompletionEvidenceError):
+            kb.complete_task(
+                conn,
+                task_id,
+                summary="All required checks passed and the feature is complete.",
+                expected_run_id=run_id,
+                require_completion_evidence=True,
+            )
+        assert kb.get_task(conn, task_id).status == "running"
+
+
+def test_passed_evidence_bound_to_exact_task_and_run_completes(kanban_home):
+    with kb.connect() as conn:
+        task_id, run_id = _running_task(conn)
+        assert kb.complete_task(
+            conn,
+            task_id,
+            summary="Implemented and verified.",
+            expected_run_id=run_id,
+            require_completion_evidence=True,
+            completion_evidence=_evidence(task_id, run_id),
+        )
+        assert kb.get_task(conn, task_id).status == "done"
+        event = conn.execute(
+            "SELECT payload FROM task_events WHERE task_id = ? AND kind = 'completed'",
+            (task_id,),
+        ).fetchone()
+        assert event is not None
+        receipt = json.loads(event["payload"])["completion_evidence"]
+        assert receipt["receipt_id"] == 41
+        assert receipt["run_id"] == run_id
+
+
+@pytest.mark.parametrize(
+    ("overrides", "reason"),
+    [
+        ({"task_id": "t_wrong"}, "task_mismatch"),
+        ({"run_id": 999999}, "run_mismatch"),
+        ({"status": "failed"}, "not_passed"),
+        ({"receipt_id": None}, "missing_receipt"),
+    ],
+)
+def test_stale_or_invalid_evidence_cannot_complete(kanban_home, overrides, reason):
+    with kb.connect() as conn:
+        task_id, run_id = _running_task(conn)
+        with pytest.raises(kb.CompletionEvidenceError) as exc_info:
+            kb.complete_task(
+                conn,
+                task_id,
+                summary="Implemented and verified.",
+                expected_run_id=run_id,
+                require_completion_evidence=True,
+                completion_evidence=_evidence(task_id, run_id, **overrides),
+            )
+        assert exc_info.value.reason == reason
+        assert kb.get_task(conn, task_id).status == "running"
+
+
+def test_evidence_rejection_is_audited_without_storing_worker_prose(kanban_home):
+    with kb.connect() as conn:
+        task_id, run_id = _running_task(conn)
+        with pytest.raises(kb.CompletionEvidenceError):
+            kb.complete_task(
+                conn,
+                task_id,
+                summary="Untrusted secret-looking worker prose",
+                expected_run_id=run_id,
+                require_completion_evidence=True,
+            )
+        event = conn.execute(
+            "SELECT payload FROM task_events WHERE task_id = ? AND kind = ?",
+            (task_id, "completion_blocked_missing_evidence"),
+        ).fetchone()
+        assert event is not None
+        assert "Untrusted secret-looking" not in event["payload"]
+
+
+def test_completion_without_evidence_requirement_is_backward_compatible(kanban_home):
+    with kb.connect() as conn:
+        task_id, run_id = _running_task(conn)
+        assert kb.complete_task(conn, task_id, result="done", expected_run_id=run_id)