diff --git a/evals/kanban_pr_acceptance_live.py b/evals/kanban_pr_acceptance_live.py
new file mode 100644
index 0000000000..0bcaac32a9
--- /dev/null
+++ b/evals/kanban_pr_acceptance_live.py
@@ -0,0 +1,66 @@
+"""Local HTTP/lifecycle probe; no external GitHub writes or inference.
+Run: python evals/kanban_pr_acceptance_live.py ROOT
+"""
+import importlib.util
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+
+root = Path(sys.argv[1]).resolve()
+sys.path.insert(0, str(root))
+os.environ.pop('HERMES_DELEGATED_CHILD_CONTEXT', None)
+import pytest
+from hermes_cli import kanban_db as kb
+from hermes_cli.kanban_db_connect import connect
+import inspect
+
+def create(conn, title):
+    if 'completion_contract' in inspect.signature(kb.create_task).parameters:
+        return kb.create_task(conn, title=title, completion_contract='acme/repo')
+    tid = kb.create_task(conn, title=title)
+    if 'completion_contract' not in {r[1] for r in conn.execute('PRAGMA table_info(tasks)')}:
+        conn.execute('ALTER TABLE tasks ADD COLUMN completion_contract TEXT')
+    conn.execute('UPDATE tasks SET completion_contract=? WHERE id=?', ('acme/repo', tid))
+    conn.commit()
+    return tid
+
+fixture_root = Path(sys.argv[2]) if len(sys.argv) > 2 else root
+spec = importlib.util.spec_from_file_location('http_fixture', fixture_root / 'tests/hermes_cli/test_kanban_pr_acceptance.py')
+fixture = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(fixture)
+patch = pytest.MonkeyPatch()
+with tempfile.TemporaryDirectory(prefix='hermes-pr-live-') as home:
+    setup = fixture.github.__wrapped__(Path(home), patch)
+    state = next(setup)
+    reports = []
+    try:
+        with connect() as conn:
+            for outcome in ('failure', 'cancelled', 'timed_out', 'success'):
+                state.update(conclusion=outcome)
+                tid = create(conn, 'PR acceptance live')
+                accepted = kb.complete_task(conn, tid, metadata={'published_pr': 'https://github.com/acme/repo/pull/7'})
+                row = conn.execute("SELECT payload FROM task_events WHERE task_id=? AND kind='pr_acceptance' ORDER BY id DESC", (tid,)).fetchone()
+                receipt = json.loads(row[0]) if row else None
+                reports.append({'case': outcome, 'accepted': accepted, 'status': kb.get_task(conn, tid).status, 'receipt': receipt})
+            for outcome in (('success', 'failure') if 'completion_contract' in inspect.signature(kb.create_task).parameters else ()):
+                tid = kb.create_task(conn, title='Concurrent ownership', completion_contract='acme/repo')
+                conn.execute('UPDATE tasks SET current_run_id=101 WHERE id=?', (tid,))
+                conn.commit()
+                def reclaim():
+                    with connect() as other:
+                        other.execute('UPDATE tasks SET current_run_id=102 WHERE id=?', (tid,))
+                        other.commit()
+                state.update(conclusion=outcome, race=reclaim)
+                accepted = kb.complete_task(conn, tid, expected_run_id=101, metadata={'published_pr': 'https://github.com/acme/repo/pull/7'})
+                reports.append({'case': 'CAS-'+outcome, 'accepted': accepted, 'run_id': kb.get_task(conn, tid).current_run_id,
+                    'receipts': conn.execute("SELECT count(*) FROM task_events WHERE task_id=? AND kind='pr_acceptance'", (tid,)).fetchone()[0]})
+                del state['race']
+        print(json.dumps({'reports': reports, 'requests': state['requests']}, indent=2))
+    finally:
+        try:
+            next(setup)
+        except StopIteration:
+            pass
+        patch.undo()
diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py
index a4e76bd2fa..6cc8111cb6 100644
--- a/hermes_cli/kanban.py
+++ b/hermes_cli/kanban.py
@@ -368,6 +368,7 @@ def _cmd_create(args: argparse.Namespace) -> int:
             provider_override=getattr(args, "provider_override", None),
             goal_mode=bool(getattr(args, "goal_mode", False)),
             goal_max_turns=getattr(args, "goal_max_turns", None),
+            completion_contract=getattr(args, "completion_contract", None),
             initial_status=getattr(args, "initial_status", "running"),
         )
         task = kb.get_task(conn, task_id)
diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py
index 47fdffc353..5337525d13 100644
--- a/hermes_cli/kanban_db.py
+++ b/hermes_cli/kanban_db.py
@@ -714,6 +714,7 @@ class Task:
     # VALID_BLOCK_KINDS or None (legacy); kept across unblock so a same-kind re-block reads as a loop.
     block_kind: Optional[str] = None
     block_recurrences: int = 0               # unblock-loop counter, see BLOCK_RECURRENCE_LIMIT
+    completion_contract: Optional[str] = None
 
     @classmethod
     def from_row(cls, row: sqlite3.Row) -> "Task":
@@ -743,7 +744,7 @@ _TASK_REQUIRED_COLUMNS = (
 _TASK_OPTIONAL_COLUMNS = (
     "branch_name", "project_id", "tenant", "result", "idempotency_key", "worker_pid",
     "max_runtime_seconds", "last_heartbeat_at", "current_run_id", "workflow_template_id",
-    "current_step_key", "max_retries", "session_id",
+    "current_step_key", "max_retries", "session_id", "completion_contract",
 )
 # Text columns where "" is stored/read as "not set".
 _TASK_EMPTY_IS_NULL_COLUMNS = (
@@ -1228,6 +1229,7 @@ def create_task(
     goal_mode: bool = False, goal_max_turns: Optional[int] = None, initial_status: str = "running",
     session_id: Optional[str] = None, board: Optional[str] = None, project_id: Optional[str] = None,
     project_source_task_id: Optional[str] = None,
+    completion_contract: Optional[str] = None,
 ) -> str:
     """Create a task (optionally under ``parents``); returns its id.
 
@@ -1240,6 +1242,8 @@ def create_task(
     ``project_source_task_id``: cross-profile fallback when ``project_id`` is not
     in the active profile's projects.db — see ``_resolve_project_link``.
     """
+    from hermes_cli.kanban_pr_acceptance import validate_contract
+    completion_contract = validate_contract(completion_contract)
     model_override, provider_override = _validate_model_override(model_override, provider_override)
     reasoning_effort = normalize_reasoning_effort(reasoning_effort)
     assignee = _canonical_assignee(assignee)
@@ -1316,8 +1320,8 @@ def create_task(
                         max_runtime_seconds,
                         skills, max_retries, model_override, provider_override,
                         reasoning_effort,
-                        goal_mode, goal_max_turns, session_id
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                        goal_mode, goal_max_turns, session_id, completion_contract
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                     """,
                     (
                         task_id, title.strip(), body, assignee, task_status, priority,
@@ -1326,7 +1330,7 @@ def create_task(
                         _opt_int(max_runtime_seconds),
                         json.dumps(skills_list) if skills_list is not None else None,
                         _opt_int(max_retries), model_override, provider_override, reasoning_effort,
-                        1 if goal_mode else 0, _opt_int(goal_max_turns), session_id,
+                        1 if goal_mode else 0, _opt_int(goal_max_turns), session_id, completion_contract,
                     ),
                 )
                 for pid in parents:
@@ -2553,16 +2557,22 @@ def complete_task(
     # 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
     verified_cards = _gate_created_cards(conn, task_id, created_cards, summary or result)
     metadata = _merge_completion_prose_artifacts(
         conn, task_id, metadata, summary=summary, result=result,
     )
     handoff_summary = summary if summary is not None else result
+    acceptance = prepare_acceptance(conn, task_id, expected_run_id, metadata)
+    if acceptance is False:
+        return False
     with write_txn(conn):
         # Hard invariant even for human review approval: a parent may have
         # reopened while this task waited.
         if not _parents_satisfied(conn, task_id):
             return False
+        if acceptance is not None and not record_acceptance(conn, task_id, acceptance):
+            return False
         prior_status = _task_status(conn, task_id)
         sql = """
                 UPDATE tasks
diff --git a/hermes_cli/kanban_db_connect.py b/hermes_cli/kanban_db_connect.py
index 9067dfd0e5..48e8945890 100644
--- a/hermes_cli/kanban_db_connect.py
+++ b/hermes_cli/kanban_db_connect.py
@@ -798,6 +798,7 @@ _LATER_TASK_COLUMNS = (
     # Ralph-style goal loop toggle; 0 = classic single-shot worker.
     ("goal_mode", "goal_mode INTEGER NOT NULL DEFAULT 0"),
     ("goal_max_turns", "goal_max_turns INTEGER"),
+    ("completion_contract", "completion_contract TEXT"),
     ("session_id", "session_id TEXT"),
     # Typed block reason (VALID_BLOCK_KINDS); NULL = generic human blocker.
     ("block_kind", "block_kind TEXT"),
diff --git a/hermes_cli/kanban_output.py b/hermes_cli/kanban_output.py
index a8ee535ede..c5abcce0c8 100644
--- a/hermes_cli/kanban_output.py
+++ b/hermes_cli/kanban_output.py
@@ -20,7 +20,7 @@ _TASK_DICT_FIELDS = (
     "workspace_kind", "workspace_path", "branch_name", "project_id",
     "created_by", "created_at", "started_at", "completed_at", "result",
     "skills", "max_retries", "model_override", "provider_override",
-    "session_id", "workflow_template_id", "current_step_key",
+    "session_id", "workflow_template_id", "current_step_key", "completion_contract", "last_failure_error",
 )
 _SHOW_RUN_FIELDS = (
     "id", "profile", "step_key", "status", "outcome", "summary", "error",
diff --git a/hermes_cli/kanban_parser.py b/hermes_cli/kanban_parser.py
index c7047a0956..6d0b413558 100644
--- a/hermes_cli/kanban_parser.py
+++ b/hermes_cli/kanban_parser.py
@@ -186,6 +186,8 @@ _SPECS = [
         _arg("--provider", dest="provider_override",
              help="Provider the --model belongs to (passed as --provider <name> to "
                   "the worker). Requires --model."),
+        _arg("--completion-contract", metavar="CONTRACT",
+             help="local-only (default), OWNER/REPO for publication, or exact GitHub PR URL; required CI gates done."),
         _arg("--goal", action="store_true", dest="goal_mode",
              help="Run the worker in a goal loop: after each turn a judge checks the "
                   "response against the card title/body and, if not done, the worker "
diff --git a/hermes_cli/kanban_pr_acceptance.py b/hermes_cli/kanban_pr_acceptance.py
new file mode 100644
index 0000000000..953dfd3c58
--- /dev/null
+++ b/hermes_cli/kanban_pr_acceptance.py
@@ -0,0 +1,117 @@
+"""Exact-head GitHub acceptance for explicitly declared PR tasks.
+
+Network work happens outside SQLite transactions. The lifecycle owner persists
+receipts only after rechecking the captured run/status/contract under its lock.
+"""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+from urllib.parse import quote
+
+_REPO = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+")
+_PR = re.compile(r"https://github\.com/([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)/pull/([1-9][0-9]*)")
+
+
+def validate_contract(value: str | None) -> str:
+    if value is None or value == "local-only":
+        return "local-only"
+    if not isinstance(value, str) or not (_REPO.fullmatch(value) or _PR.fullmatch(value)):
+        raise ValueError("completion_contract must be local-only, OWNER/REPO, or an exact GitHub PR URL")
+    return value
+
+
+def _api(endpoint: str, *, query: str | None = None, paginate: bool = False):
+    command = ["gh", "api", endpoint, "--hostname", "github.com"]
+    if query is not None:
+        command += ["-f", "query=" + query]
+    if paginate:
+        command += ["--paginate", "--slurp"]
+    result = subprocess.run(command, stdin=subprocess.DEVNULL, capture_output=True,
+                            text=True, timeout=30, check=True)
+    value = json.loads(result.stdout)
+    if isinstance(value, dict) and value.get("errors"):
+        raise ValueError("GitHub returned incomplete GraphQL evidence")
+    return value
+
+
+def collect_acceptance(contract: str, published_pr: str | None) -> dict:
+    receipt = {"ok": False, "classification": "missing", "head_sha": None,
+               "pr_url": published_pr, "checks": [],
+               "recovery": "Fix required failures, rerun infrastructure checks or wait, then retry completion. "
+                           "Use kanban_block if human input is needed; receipts remain on the task event log."}
+    try:
+        declared = _PR.fullmatch(contract)
+        url = contract if declared else published_pr
+        match = _PR.fullmatch(url or "")
+        if not match or (not declared and match[1] != contract) or (declared and published_pr and published_pr != contract):
+            receipt["detail"] = "Supply metadata.published_pr matching the persisted completion contract."
+            return receipt
+        repo, number = match[1], int(match[2])
+        receipt["pr_url"] = url
+        owner, name = repo.split("/")
+        query = '''{repository(owner:%s,name:%s){pullRequest(number:%d){headRefOid baseRefName state
+            baseRef{branchProtectionRule{requiredStatusChecks{context app{databaseId}}}}}}}''' % (
+                json.dumps(owner), json.dumps(name), number)
+        pr = _api("graphql", query=query)["data"]["repository"]["pullRequest"]
+        sha, branch = pr["headRefOid"], pr["baseRefName"]
+        receipt["head_sha"] = sha
+        if not re.fullmatch(r"[0-9a-f]{40}", sha) or pr["state"] not in {"OPEN", "MERGED"}:
+            raise ValueError("PR is closed or current head is unavailable")
+        protection = (pr.get("baseRef") or {}).get("branchProtectionRule") or {}
+        required = {(r["context"], (r.get("app") or {}).get("databaseId")) for r in protection.get("requiredStatusChecks", [])}
+        rules = _api(f"repos/{repo}/rules/branches/{quote(branch, safe='')}?per_page=100", paginate=True)
+        for page in rules:
+            for rule in page:
+                if rule["type"] == "required_status_checks":
+                    required.update((r["context"], r.get("integration_id"))
+                                    for r in rule["parameters"]["required_status_checks"])
+        receipt["required"] = [{"context": c, "app_id": a} for c, a in sorted(required, key=str)]
+        if not required:
+            receipt["detail"] = "No repository-required checks are configured; explicitly use a local-only contract for non-CI tasks."
+            return receipt
+        pages = _api(f"repos/{repo}/commits/{sha}/check-runs?per_page=100&filter=latest", paginate=True)
+        runs = [run for page in pages for run in page["check_runs"]]
+        if len({r["id"] for r in runs}) != pages[0]["total_count"]:
+            raise ValueError("Incomplete check-run pagination")
+        statuses = [{**s, "sha": sha} for page in _api(f"repos/{repo}/commits/{sha}/statuses?per_page=100", paginate=True) for s in page]
+        outcomes = []
+        for context, app_id in sorted(required, key=str):
+            matching = [r for r in runs if r["name"] == context and
+                        (app_id in (None, -1) or r["app"]["id"] == app_id)]
+            # A legacy status can satisfy an unpinned context, but never a check pinned to an app.
+            legacy = [s for s in statuses if s["context"] == context] if app_id in (None, -1) else []
+            selected = matching + ([max(legacy, key=lambda s: s["id"])] if legacy else [])
+            if not selected:
+                outcomes.append("missing")
+                receipt["checks"].append({"name": context, "classification": "missing", "head_sha": sha})
+            for check in selected:
+                is_run = "conclusion" in check
+                outcome = check.get("conclusion") if is_run else check["state"]
+                classification = _classify(check, sha, outcome, is_run)
+                outcomes.append(classification)
+                receipt["checks"].append({"name": context, "id": check["id"],
+                    "url": check.get("html_url") or check.get("target_url"),
+                    "head_sha": check.get("head_sha", check.get("sha")),
+                    "classification": classification, "conclusion": outcome})
+        # Re-read after all pages: old-head successes are never transferable.
+        current = _api(f"repos/{repo}/pulls/{number}")
+        if current["head"]["sha"] != sha or current["base"]["ref"] != branch or (current["state"] == "closed" and not current.get("merged")):
+            receipt.update(classification="stale", detail="PR head/base changed while collecting evidence; retry.")
+            return receipt
+        receipt["classification"] = next((x for x in outcomes if x != "success"), "missing" if not outcomes else "success")
+        receipt["ok"] = receipt["classification"] == "success"
+        return receipt
+    except (OSError, subprocess.SubprocessError, ValueError, KeyError, TypeError, IndexError):
+        # Never persist gh stderr (credentials/host details); the failed phase is actionable.
+        receipt.update(classification="infra", detail="GitHub acceptance evidence unavailable or incomplete; check gh authentication/API access and retry.")
+        return receipt
+
+
+def _classify(check: dict, sha: str, outcome: str | None, is_run: bool) -> str:
+    if check.get("head_sha", check.get("sha")) != sha:
+        return "stale"
+    if is_run and check.get("status") != "completed":
+        return "pending"
+    return {"success": "success", "failure": "failure", "error": "infra", "pending": "pending"}.get(outcome, "infra")
diff --git a/hermes_cli/kanban_pr_acceptance_store.py b/hermes_cli/kanban_pr_acceptance_store.py
new file mode 100644
index 0000000000..cf6a8327f3
--- /dev/null
+++ b/hermes_cli/kanban_pr_acceptance_store.py
@@ -0,0 +1,45 @@
+"""Persist acceptance with the same ownership snapshot as the terminal write."""
+from __future__ import annotations
+
+from hermes_cli.kanban_db_connect import write_txn
+from hermes_cli.kanban_pr_acceptance import _PR, collect_acceptance
+
+
+def _snapshot(conn, task_id):
+    row = conn.execute("SELECT current_run_id, status, completion_contract FROM tasks WHERE id=?", (task_id,)).fetchone()
+    return tuple(row) if row else None
+
+
+def prepare_acceptance(conn, task_id, expected_run_id, metadata):
+    snapshot = _snapshot(conn, task_id)
+    if snapshot is None:
+        return False
+    run_id, status, contract = snapshot
+    if not contract or contract == "local-only":
+        return None
+    if status not in {"running", "ready", "blocked", "review"} or (expected_run_id is not None and run_id != expected_run_id):
+        return False
+    published_pr = metadata.get("published_pr") if isinstance(metadata, dict) else None
+    match = _PR.fullmatch(published_pr) if isinstance(published_pr, str) else None
+    # Publication binds once. Retrying cannot replace the task's PR with a green sibling.
+    if match and contract == match[1]:
+        with write_txn(conn):
+            if _snapshot(conn, task_id) != snapshot:
+                return False
+            conn.execute("UPDATE tasks SET completion_contract=? WHERE id=?", (published_pr, task_id))
+        snapshot = (run_id, status, published_pr)
+        contract = published_pr
+    return snapshot, collect_acceptance(contract, published_pr)
+
+
+def record_acceptance(conn, task_id, acceptance):
+    """Called under complete_task's write_txn, before its terminal UPDATE."""
+    from hermes_cli.kanban_db import _append_event
+    snapshot, receipt = acceptance
+    if _snapshot(conn, task_id) != snapshot:
+        return False
+    _append_event(conn, task_id, "pr_acceptance", receipt, run_id=snapshot[0])
+    if not receipt["ok"]:
+        detail = f"PR acceptance {receipt['classification']}: {receipt.get('detail', '')} {receipt['recovery']}"
+        conn.execute("UPDATE tasks SET last_failure_error=? WHERE id=?", (detail, task_id))
+    return receipt["ok"]
diff --git a/tests/hermes_cli/test_kanban_pr_acceptance.py b/tests/hermes_cli/test_kanban_pr_acceptance.py
new file mode 100644
index 0000000000..279d2efa17
--- /dev/null
+++ b/tests/hermes_cli/test_kanban_pr_acceptance.py
@@ -0,0 +1,129 @@
+"""Two lifecycle invariants, using real SQLite and a local GitHub HTTP contract."""
+import json
+import os
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+from hermes_cli.kanban_db_connect import connect
+
+
+@pytest.fixture
+def github(tmp_path, monkeypatch):
+    state = {"conclusion": "success", "head": "a" * 40, "reads": 0, "requests": []}
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            state["requests"].append(self.path)
+            sha = state["head"]
+            if self.path == "/graphql":
+                value = {"data": {"repository": {"pullRequest": {
+                    "headRefOid": sha, "baseRefName": "main", "state": "OPEN",
+                    "baseRef": {"branchProtectionRule": {"requiredStatusChecks": [
+                        {"context": "required", "app": {"databaseId": 1}}]}}}}}}
+            elif "/rules/branches/" in self.path:
+                value = [[]]
+            elif "/check-runs" in self.path:
+                run = {"id": 42, "name": "required", "head_sha": sha,
+                       "app": {"id": 1}, "status": "completed", "conclusion": state["conclusion"],
+                       "html_url": "https://github.com/acme/repo/actions/runs/42"}
+                if state.get("stale"):
+                    run["head_sha"] = "b" * 40
+                runs = [] if state.get("missing") else [run]
+                value = [{"total_count": 101, "check_runs": [
+                    {**run, "id": i, "name": "optional", "conclusion": "skipped"}
+                    for i in range(100)]}, {"total_count": 101, "check_runs": runs}]
+                if state.get("race"):
+                    state["race"]()
+                if state.get("head_change"):
+                    state["head"] = "b" * 40
+            elif "/statuses" in self.path:
+                value = [[]]
+            elif "/pulls/" in self.path:
+                value = {"head": {"sha": sha}, "base": {"ref": "main"}, "state": "open"}
+            else:
+                self.send_error(404)
+                return
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(json.dumps(value).encode())
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    shim = tmp_path / "bin"
+    shim.mkdir()
+    gh = shim / "gh"
+    gh.write_text(f"#!{sys.executable}\nimport sys,urllib.request\n"
+                  f"u='http://127.0.0.1:{server.server_port}/'+sys.argv[2]\n"
+                  "print(urllib.request.urlopen(u).read().decode())\n")
+    gh.chmod(0o755)
+    monkeypatch.setenv("PATH", str(shim) + os.pathsep + os.environ["PATH"])
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
+    kb.init_db()
+    try:
+        yield state
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()
+
+
+@pytest.mark.linux_only
+def test_pr_completion_requires_current_required_evidence(github):
+    with connect() as conn:
+        for conclusion in ("failure", "cancelled", "timed_out", "action_required", "neutral", "skipped", None, "success"):
+            github.update(conclusion=conclusion, head="a" * 40)
+            tid = kb.create_task(conn, title="Publish", completion_contract="acme/repo")
+            ok = kb.complete_task(conn, tid, metadata={"published_pr": "https://github.com/acme/repo/pull/7"})
+            assert ok is (conclusion == "success")
+            task = kb.get_task(conn, tid)
+            assert (task.status == "done") is ok
+            receipts = [json.loads(r[0]) for r in conn.execute(
+                "SELECT payload FROM task_events WHERE task_id=? AND kind='pr_acceptance'", (tid,))]
+            assert receipts and receipts[-1]["head_sha"] == "a" * 40
+            if not ok:
+                assert task.status in {"running", "ready", "blocked", "review"}
+                assert "retry" in receipts[-1]["recovery"]
+                assert receipts[-1]["checks"][0]["id"] == 42
+        for fault in ("missing", "stale", "head_change"):
+            github.update(conclusion="success", head="a" * 40)
+            github[fault] = True
+            tid = kb.create_task(conn, title=fault, completion_contract="acme/repo")
+            assert not kb.complete_task(conn, tid, metadata={"published_pr": "https://github.com/acme/repo/pull/7"})
+            assert kb.get_task(conn, tid).status != "done"
+            github.pop(fault)
+        # Omission and a sibling repository cannot downgrade the stored declaration.
+        tid = kb.create_task(conn, title="publish", completion_contract="acme/repo")
+        assert not kb.complete_task(conn, tid, summary="local green")
+        assert not kb.complete_task(conn, tid, metadata={"published_pr": "https://github.com/other/repo/pull/7"})
+        before = len(github["requests"])
+        local = kb.create_task(conn, title="local", completion_contract="local-only")
+        assert kb.complete_task(conn, local, summary="https://github.com/acme/repo/pull/7 is background context")
+        assert len(github["requests"]) == before
+
+
+@pytest.mark.linux_only
+def test_acceptance_receipts_and_terminal_write_share_run_ownership(github):
+    with connect() as conn:
+        for conclusion in ("success", "failure"):
+            tid = kb.create_task(conn, title="race", completion_contract="acme/repo")
+            conn.execute("UPDATE tasks SET current_run_id=101 WHERE id=?", (tid,))
+            conn.commit()
+            def reclaim():
+                with connect() as rival:
+                    rival.execute("UPDATE tasks SET current_run_id=102 WHERE id=?", (tid,))
+                    rival.commit()
+            github.update(conclusion=conclusion, race=reclaim)
+            assert not kb.complete_task(conn, tid, expected_run_id=101,
+                metadata={"published_pr": "https://github.com/acme/repo/pull/7"})
+            assert kb.get_task(conn, tid).current_run_id == 102
+            assert kb.get_task(conn, tid).status != "done"
+            assert conn.execute("SELECT count(*) FROM task_events WHERE task_id=? AND kind='pr_acceptance'", (tid,)).fetchone()[0] == 0
+            github.pop("race")
diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py
index b776958a55..44c46a01ee 100644
--- a/tools/kanban_tools.py
+++ b/tools/kanban_tools.py
@@ -309,7 +309,7 @@ def _opt_int(value: Any, default: Optional[int] = None) -> Optional[int]:
 _TASK_FIELDS = tuple(
     "id title body assignee status tenant priority workspace_kind workspace_path created_by "
     "created_at started_at completed_at result current_run_id model_override "
-    "provider_override".split())
+    "provider_override completion_contract last_failure_error".split())
 _TASK_SUMMARY_FIELDS = tuple(
     "id title assignee status priority tenant workspace_kind workspace_path project_id created_by "
     "created_at started_at completed_at current_run_id model_override provider_override".split())
@@ -587,7 +587,9 @@ def _handle_complete(args: dict, **kw) -> str:
                 f"in-flight (no state change). Retry kanban_complete with the same "
                 f"summary/metadata and either drop these ids from created_cards, or pass "
                 f"created_cards=[] to skip the card-claim check entirely.")
-        _check(ok, f"could not complete {tid} (unknown id or already terminal)")
+        task = kb.get_task(conn, tid)
+        _check(ok, (task.last_failure_error if task else None) or
+               f"could not complete {tid} (unknown id, stale run, or already terminal)")
         run = kb.latest_run(conn, tid)
         return _ok(task_id=tid, run_id=run.id if run else None)
 
@@ -845,6 +847,7 @@ def _handle_create(args: dict, **kw) -> str:
             max_runtime_seconds=_opt_int(args.get("max_runtime_seconds")), skills=skills,
             model_override=model_override, provider_override=provider_override,
             goal_mode=goal_mode, goal_max_turns=_opt_int(args.get("goal_max_turns")),
+            completion_contract=args.get("completion_contract"),
             initial_status=str(args.get("initial_status") or "running"),
             created_by=os.environ.get("HERMES_PROFILE") or "worker", session_id=session_id)
         landed = _fields(kb.get_task(conn, new_tid), _CREATED_FIELDS)
diff --git a/tools/kanban_tools_schemas.py b/tools/kanban_tools_schemas.py
index c09a9afe6d..adcc30263c 100644
--- a/tools/kanban_tools_schemas.py
+++ b/tools/kanban_tools_schemas.py
@@ -458,6 +458,10 @@ KANBAN_CREATE_SCHEMA = _schema(
                 "open-ended cards where one shot rarely finishes the "
                 "work. Defaults to false (classic single-shot worker)."
         )),
+        "completion_contract": _prop("string", (
+            "Declare at creation: local-only (default), OWNER/REPO for PR publication, or an exact GitHub PR URL. "
+            "PR tasks cannot complete until repository-required exact-head CI passes. On publication pass metadata.published_pr."
+        )),
         "goal_max_turns": _prop("integer", (
                 "Turn budget for goal_mode workers. Caps how many "
                 "continuation turns the worker may take before the task "
diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md
index 084a36f3f8..f337128dab 100644
--- a/website/docs/user-guide/features/kanban.md
+++ b/website/docs/user-guide/features/kanban.md
@@ -29,6 +29,43 @@ This is the shape that covers the workloads `delegate_task` can't:
 
 For the full design rationale, comparative analysis against Cline Kanban / Paperclip / NanoClaw / Google Gemini Enterprise, and the eight canonical collaboration patterns, see `docs/hermes-kanban-v1-spec.pdf` in the repository.
 
+## PR completion contracts
+
+Declare PR work at creation with `--completion-contract OWNER/REPO` (or an exact
+`https://github.com/OWNER/REPO/pull/123` URL for existing work). `kanban_create`
+accepts the same `completion_contract`. Use `local-only` for intentionally local
+work; existing and undeclared cards retain that default. Prose URLs are not policy.
+
+After publishing, pass `metadata.published_pr` to completion. The first matching
+URL binds the card permanently; retries cannot substitute a green sibling PR.
+CLI `show --json` and `kanban_show` expose the persisted contract.
+
+The shared `complete_task` boundary covers worker tools, CLI, review approval and
+dashboard completion. It reads classic branch protection and active ruleset
+required contexts, paginates exact-head check runs and legacy statuses, then
+re-reads the PR head/base. Optional failed/skipped telemetry does not veto accepted
+required checks. Missing, pending, failed, cancelled, timed-out, stale, skipped or
+neutral **required** evidence cannot complete the card. Neither can zero-run
+acceptance, unreadable policy or GitHub API failures. A repository without required
+checks needs a local-only contract. `gh` must be authenticated with read access to
+the repository's checks and rules; no remote writes are performed by this gate.
+
+Rejection retains the active card and workspace. Durable `pr_acceptance` events
+store PR URL, SHA, required contexts, check IDs/URLs, classifications and recovery
+instructions; `last_failure_error` surfaces the next step. Fix failures, rerun
+infrastructure checks or wait, then retry completion. Use `kanban_block` when
+human action is needed. Generic GitHub `failure` cannot establish whether a test
+or artifact upload failed; inspect its retained URL. Explicit infrastructure
+conclusions and API failures are classified separately. No extra worker is spawned.
+
+Receipt persistence and the terminal write recheck run/status/contract ownership
+under one SQLite lock: a reclaimed worker cannot complete or attach acceptance to
+the new run. The final GitHub read is a completion-time snapshot, not a distributed
+transaction or a continuous post-completion monitor. This is a single-user lifecycle
+guard, not OS isolation against arbitrary direct database writes. GitHub Enterprise
+is not covered. Related publication/lifecycle work: #91230, #84254, #52311; local
+verification and publication alone are not remote acceptance.
+
 ## Kanban vs. `delegate_task`
 
 They look similar; they are not the same primitive.