diff --git a/docs/architecture/rfcs/same-goal-agent-handoff-inbox-v0.md b/docs/architecture/rfcs/same-goal-agent-handoff-inbox-v0.md
new file mode 100644
index 0000000000..6bca05aa1c
--- /dev/null
+++ b/docs/architecture/rfcs/same-goal-agent-handoff-inbox-v0.md
@@ -0,0 +1,102 @@
+# RFC: Same-Goal Agent Handoff Inbox v0 / 同 Goal Agent 交接收件箱 v0
+
+| Field / 字段 | Value / 值 |
+|---|---|
+| Status / 状态 | Implemented / 已实现 |
+| Date / 日期 | 2026-09-13 |
+| Core owner / Core 责任 | typed dispatch, durable receipt, Turn-start projection / 类型化派发、持久回执、Turn-start 投影 |
+| Agent owner / Agent 责任 | canonical Todo claim, bounded execution and result writeback / 规范 Todo 认领、有界执行与结果回写 |
+
+## Decision / 决策
+
+An `independent_handoff` Todo that excludes the current Agent is not a quiet
+wait. When a registered same-Goal peer remains eligible, the coordinator
+deterministically dispatches the first candidate to that peer. The handoff is
+stored under the existing manager-context inbox and projected by the same
+`manager.context_inbox` Turn-start hook used by quota heartbeats and managed
+`turn run-once`. It does not introduce a second message queue or a second UI
+source of truth.
+
+如果一个 `independent_handoff` Todo 排除了当前 Agent，它不能被当作静默等待。
+只要同一 Goal 内仍有符合条件的已注册 peer，协调者就把第一个候选确定性地派发给
+该 peer。交接记录复用现有 manager-context 收件箱，并由同一个
+`manager.context_inbox` Turn-start hook 投影给 quota heartbeat 与托管
+`turn run-once`；不新增第二套消息队列或 UI 真相源。
+
+Dispatch is admitted only after live readback proves that the Todo comes from
+canonical authority, remains open and unclaimed, uses `independent_handoff`,
+excludes the sender, and does not exclude the recipient. A message-delivery
+receipt is not execution authority. The recipient must run the exact projected
+`todo claim` command and then `manager-inbox acknowledge-handoff`. The latter
+re-reads canonical Todo state before changing the durable receipt from
+`dispatched` to `claimed`.
+
+派发前必须实时读回并证明：Todo 来自规范 authority、仍为 open 且未被认领、
+使用 `independent_handoff`、排除发送者且不排除接收者。消息送达回执不等于执行
+权限。接收者必须先运行投影出的精确 `todo claim` 命令，再运行
+`manager-inbox acknowledge-handoff`；后者会再次读取规范 Todo 状态，验证成功后
+才把持久回执从 `dispatched` 更新为 `claimed`。
+
+## Identity, isolation and replay / 身份、隔离与重放
+
+- The dispatch identity binds Goal, Todo, sender and recipient. Canonical state
+  is read immediately before the first dispatch; later retries replay the same
+  receipt instead of creating a new identity for unrelated provider revisions.
+  Replaying the same decision returns the existing receipt.
+- Inbox paths are keyed by both `goal_id` and `agent_id`; a same-named Agent in
+  another Goal cannot read or acknowledge the handoff.
+- The recipient is selected from the intersection of the candidate's eligible
+  peer set and the Goal's registered Agent set, excluding the sender.
+- Private inbox entries and receipts are local mode `0600`. Public quota/Turn
+  output contains only bounded identifiers and receipt state, never inbox
+  content or provider payloads.
+- A handoff grants no repository, provider, trading, payment, publishing or
+  other protected-operation authority. Existing Todo capability and write-scope
+  gates continue to apply after claim.
+
+- 派发身份绑定 Goal、Todo、发送者与接收者；首次派发前立即读取规范状态，后续
+  重试复用同一回执，不会因无关的 provider revision 变化生成新身份。
+- 收件箱路径同时按 `goal_id` 与 `agent_id` 分区；其他 Goal 下的同名 Agent
+  无法读取或确认该交接。
+- 接收者从“候选允许的 peer”与“该 Goal 已注册 Agent”的交集中确定，并排除
+  发送者。
+- 私有收件箱条目和回执在本地使用 `0600` 权限；公开 quota/Turn 输出只包含
+  有界标识与回执状态，不包含收件箱正文或 provider payload。
+- 交接不授予仓库、provider、交易、支付、发布或其他受保护操作权限；认领后仍
+  受 Todo 原有 capability 与写范围 gate 约束。
+
+When no eligible peer exists, Core emits the non-quiet typed state
+`handoff_dispatch_state=no_eligible_peer` and forbids spend until a peer is
+registered. It never lets the excluded coordinator execute the Todo.
+
+当不存在符合条件的 peer 时，Core 发出非静默类型化状态
+`handoff_dispatch_state=no_eligible_peer`，并在注册合适 peer 前禁止 spend；
+它不会让被排除的协调 Agent 执行该 Todo。
+
+If the Todo is claimed by another peer between dispatch and recipient
+acknowledgement, acknowledgement writes a terminal `claim_conflict` receipt
+with the observed claimant. It does not overwrite the canonical claim or leave
+the stale inbox item pending forever.
+
+如果 Todo 在派发和接收者确认之间被另一个 peer 抢先认领，确认操作会写入终态
+`claim_conflict` 回执并记录实际 claimant；它不会覆盖规范认领，也不会让过期
+收件箱条目永久悬挂。
+
+## Product surfaces and acceptance / 产品入口与验收
+
+- CLI/quota and managed Turn share the same Turn-start hook and dispatch code.
+- Lark remains an adapter over the same inbox/read/receipt model; it does not
+  own handoff state.
+- No frontend control is added in v0 because this is an internal Agent routing
+  transition, not user-configurable state. Existing frontend Todo and Turn
+  projections continue to consume the bounded shared contract capsule.
+- Acceptance covers deterministic replay, Goal/Agent isolation, canonical
+  claim-before-ack, private file modes, the no-peer frontier, and Turn-envelope
+  receipt projection.
+
+- CLI/quota 与托管 Turn 共享同一 Turn-start hook 和派发实现。
+- Lark 仍只是同一 inbox/read/receipt 模型的适配器，不持有独立交接状态。
+- v0 不增加前端控件，因为这是内部 Agent 路由转换，不是用户可配置状态；现有
+  前端 Todo 与 Turn 投影继续消费有界的共享 contract capsule。
+- 验收覆盖确定性重放、Goal/Agent 隔离、先规范认领再确认、私有文件权限、无
+  peer frontier，以及 Turn envelope 回执投影。
diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py
index c90a653fa8..5c9b73dbbf 100644
--- a/loopx/capabilities/manager_context/__init__.py
+++ b/loopx/capabilities/manager_context/__init__.py
@@ -192,14 +192,32 @@ def deliver(
     path = _root(runtime_root) / "entries" / _hash(request) / (request_id + ".json")
     with exclusive_file_lock(path.with_suffix(".lock")):
         exists = path.exists()
-        if exists and {k: v for k, v in _read(path).items() if k not in {"delivered_at", "source_channel"}} != value:
+        if (
+            exists
+            and {
+                k: v
+                for k, v in _read(path).items()
+                if k not in {"delivered_at", "source_channel"}
+            }
+            != value
+        ):
             raise ValueError("context request identity conflict")
         if not exists:
             from .tracking import _now
-            _write(path, value | {"delivered_at": _now(), "source_channel": session.get("channel_id")})
-        if {k: v for k, v in _read(path).items() if k not in {"delivered_at", "source_channel"}} != value:
+
+            _write(
+                path,
+                value
+                | {"delivered_at": _now(), "source_channel": session.get("channel_id")},
+            )
+        if {
+            k: v
+            for k, v in _read(path).items()
+            if k not in {"delivered_at", "source_channel"}
+        } != value:
             raise ValueError("context delivery readback failed")
     from .roundtrip import register
+
     register(runtime_root, value, session, turn)
     return {
         "request_id": request_id,
@@ -222,6 +240,7 @@ def pending(runtime_root: Path, goal_id: str, agent_id: str) -> dict:
     items = []
     for path in sorted(folder.glob("*.json")):
         from .roundtrip import needs_conclusion
+
         decided = (_root(runtime_root) / "decisions" / path.name).exists()
         if decided and not needs_conclusion(runtime_root, path.stem):
             continue
@@ -233,15 +252,24 @@ def pending(runtime_root: Path, goal_id: str, agent_id: str) -> dict:
         ):
             raise ValueError("context inbox scope mismatch")
         if decided:
-            item = {**item, "receiver_decision_recorded": True,
-                    "next_action": "Return the original audience a conclusion with manager-inbox report; do not repeat the recorded decision or reprioritize unrelated work."}
+            item = {
+                **item,
+                "receiver_decision_recorded": True,
+                "next_action": "Return the original audience a conclusion with manager-inbox report; do not repeat the recorded decision or reprioritize unrelated work.",
+            }
         items.append(item)
         if len(items) == 21:
             break
+    from .agent_handoff import pending_handoffs
+
+    handoffs = pending_handoffs(runtime_root, goal_id, agent_id)
+    handoff_has_more = len(handoffs) > 20
+    remaining = max(0, 20 - len(handoffs[:20]))
+    combined = handoffs[:20] + items[:remaining]
     return {
         "ok": True,
-        "items": items[:20],
-        "has_more": len(items) > 20,
+        "items": combined,
+        "has_more": handoff_has_more or len(items) > remaining,
         "instruction": INSTRUCTION,
     }
 
@@ -276,6 +304,7 @@ def acknowledge(
                 raise ValueError("context decision already recorded")
         else:
             from .tracking import _now
+
             _write(path, value | {"decided_at": _now()})
     return {"ok": True, **value}
 
@@ -334,12 +363,41 @@ def produce():
         required_read={
             "kind": "operator_inbox",
             "command": command,
-            "reason": "Review owner context and decide whether the current plan should change; no priority is imposed.",
+            "reason": (
+                "Review pending owner context or a typed same-Goal agent handoff before "
+                "work; owner context imposes no priority and a handoff requires canonical claim."
+            ),
             "ordering": "before_work",
         },
     )
 
 
+def extend_turn_start_dispatch(
+    dispatch: dict[str, Any],
+    *,
+    runtime_root: Path,
+    registry_path: Path,
+    goal_id: str,
+    agent_id: str | None,
+) -> dict[str, Any]:
+    """Merge the shared manager-context hook into any Turn entry point."""
+    if not agent_id:
+        return dispatch
+    from ...control_plane.capability_hooks import dispatch_turn_start_hooks
+
+    context_dispatch = dispatch_turn_start_hooks(
+        (turn_start_hook(runtime_root, registry_path, goal_id, agent_id),)
+    )
+    merged = dict(dispatch)
+    for key in ("results", "required_reads", "failures"):
+        merged[key] = list(merged.get(key) or []) + list(
+            context_dispatch.get(key) or []
+        )
+    for key in ("registered_count", "invoked_count"):
+        merged[key] = int(merged.get(key) or 0) + int(context_dispatch.get(key) or 0)
+    return merged
+
+
 def evidence_goal_scope(runtime_root: Path, channel: str) -> list[str] | None:
     """Audience-wide read grant; absent preserves the existing connection scope.
 
@@ -358,7 +416,8 @@ def evidence_goal_scope(runtime_root: Path, channel: str) -> list[str] | None:
             return None
         ids = source["evidence_goal_ids"]
         if not isinstance(ids, list) or any(
-            not isinstance(v, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,159}", v)
+            not isinstance(v, str)
+            or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,159}", v)
             for v in ids
         ):
             return []
@@ -367,8 +426,14 @@ def evidence_goal_scope(runtime_root: Path, channel: str) -> list[str] | None:
         return []
 
 
-def configure_evidence_scope(runtime_root: Path, registry_path: Path, *, channel: str,
-                             goal_ids: list[str], execute: bool = False) -> dict:
+def configure_evidence_scope(
+    runtime_root: Path,
+    registry_path: Path,
+    *,
+    channel: str,
+    goal_ids: list[str],
+    execute: bool = False,
+) -> dict:
     """Local operator grants only selected Goal summaries to an exact audience."""
     if not re.fullmatch(r"manager\.external\.[a-f0-9]{24}", channel):
         raise ValueError("an exact external manager channel is required")
@@ -381,13 +446,24 @@ def configure_evidence_scope(runtime_root: Path, registry_path: Path, *, channel
     if execute:
         path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
         with exclusive_file_lock(path.with_suffix(".lock")):
-            policy = _read(path) if path.exists() else {"schema_version": POLICY_SCHEMA, "sources": {}}
+            policy = (
+                _read(path)
+                if path.exists()
+                else {"schema_version": POLICY_SCHEMA, "sources": {}}
+            )
             if policy.get("schema_version") != POLICY_SCHEMA:
                 raise ValueError("invalid manager policy")
-            policy.setdefault("sources", {}).setdefault(channel, {})["evidence_goal_ids"] = ids
+            policy.setdefault("sources", {}).setdefault(channel, {})[
+                "evidence_goal_ids"
+            ] = ids
             _write(path, policy)
         if evidence_goal_scope(runtime_root, channel) != ids:
             raise ValueError("read scope verification failed")
-    return {"ok": True, "executed": execute, "channel_id": channel,
-            "evidence_goal_ids": ids, "scope": "audience_goal_summaries",
-            "delegation_authority_changed": False}
+    return {
+        "ok": True,
+        "executed": execute,
+        "channel_id": channel,
+        "evidence_goal_ids": ids,
+        "scope": "audience_goal_summaries",
+        "delegation_authority_changed": False,
+    }
diff --git a/loopx/capabilities/manager_context/agent_handoff.py b/loopx/capabilities/manager_context/agent_handoff.py
new file mode 100644
index 0000000000..e385e9b908
--- /dev/null
+++ b/loopx/capabilities/manager_context/agent_handoff.py
@@ -0,0 +1,381 @@
+"""Durable same-Goal agent handoffs over the manager-context inbox."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from pathlib import Path
+import re
+import shlex
+from typing import Any
+
+from ...agent_registry import registered_agent_ids_for_goal
+from ...file_lock import exclusive_file_lock
+from ...history import load_registry
+from ...todos import list_goal_todos
+from . import _hash, _read, _root, _write
+
+HANDOFF_ENTRY_SCHEMA = "loopx_agent_handoff_inbox_entry_v0"
+HANDOFF_RECEIPT_SCHEMA = "loopx_agent_handoff_dispatch_receipt_v0"
+_TOKEN_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,159}")
+_TODO_RE = re.compile(r"todo_[A-Za-z0-9_-]+")
+
+
+def _require_token(value: Any, label: str, *, todo: bool = False) -> str:
+    normalized = str(value or "").strip()
+    pattern = _TODO_RE if todo else _TOKEN_RE
+    if not pattern.fullmatch(normalized):
+        raise ValueError(f"invalid {label}")
+    return normalized
+
+
+def _goal_agents(registry_path: Path, goal_id: str) -> list[str]:
+    registry = load_registry(registry_path)
+    goal = next(
+        (
+            item
+            for item in registry.get("goals", [])
+            if isinstance(item, dict) and item.get("id") == goal_id
+        ),
+        None,
+    )
+    if goal is None:
+        raise ValueError("handoff Goal is not registered")
+    return registered_agent_ids_for_goal(goal)
+
+
+def _receipt_path(runtime_root: Path, dispatch_id: str) -> Path:
+    return _root(runtime_root) / "agent-handoffs" / "receipts" / f"{dispatch_id}.json"
+
+
+def _entry_path(runtime_root: Path, target: dict[str, str], dispatch_id: str) -> Path:
+    return (
+        _root(runtime_root)
+        / "agent-handoffs"
+        / "entries"
+        / _hash(target)
+        / f"{dispatch_id}.json"
+    )
+
+
+def dispatch_from_quota_decision(
+    runtime_root: Path,
+    registry_path: Path,
+    *,
+    goal_id: str,
+    from_agent_id: str,
+    decision: Mapping[str, Any],
+) -> dict[str, Any] | None:
+    """Dispatch the first typed handoff candidate to a deterministic peer."""
+    frontier = decision.get("agent_scope_frontier")
+    if (
+        not isinstance(frontier, Mapping)
+        or frontier.get("handoff_dispatch_required") is not True
+    ):
+        return None
+    goal_id = _require_token(goal_id, "goal id")
+    from_agent_id = _require_token(from_agent_id, "source agent id")
+    registered = _goal_agents(registry_path, goal_id)
+    if from_agent_id not in registered:
+        raise ValueError("handoff source agent is not registered for the Goal")
+
+    raw_candidates = frontier.get("executor_excluded_dispatchable_items")
+    candidates = raw_candidates if isinstance(raw_candidates, list) else []
+    candidate = next((item for item in candidates if isinstance(item, Mapping)), None)
+    if candidate is None:
+        raise ValueError("handoff dispatch frontier has no typed candidate")
+    todo_id = _require_token(candidate.get("todo_id"), "handoff todo id", todo=True)
+    if candidate.get("continuation_policy") != "independent_handoff":
+        raise ValueError("handoff candidate must use independent_handoff")
+
+    raw_peers = candidate.get("eligible_peer_ids")
+    if not isinstance(raw_peers, list):
+        raw_peers = frontier.get("eligible_peer_ids")
+    peers = raw_peers if isinstance(raw_peers, list) else []
+    eligible = sorted(
+        {
+            _require_token(peer, "eligible peer id")
+            for peer in peers
+            if str(peer or "").strip()
+        }
+        & set(registered) - {from_agent_id}
+    )
+    if not eligible:
+        raise ValueError("handoff dispatch frontier has no eligible registered peer")
+    to_agent_id = eligible[0]
+    target = {"goal_id": goal_id, "agent_id": to_agent_id}
+    identity = {
+        "goal_id": goal_id,
+        "todo_id": todo_id,
+        "from_agent_id": from_agent_id,
+        "to_agent_id": to_agent_id,
+    }
+    dispatch_id = _hash(identity)
+    entry_path = _entry_path(runtime_root, target, dispatch_id)
+    receipt_path = _receipt_path(runtime_root, dispatch_id)
+    with exclusive_file_lock(receipt_path.with_suffix(".lock")):
+        if receipt_path.exists():
+            current = _read(receipt_path)
+            if any(current.get(key) != value for key, value in identity.items()):
+                raise ValueError("agent handoff dispatch identity conflict")
+            if not entry_path.exists():
+                raise ValueError("agent handoff dispatch entry is missing")
+            entry = _read(entry_path)
+            if any(entry.get(key) != value for key, value in identity.items()):
+                raise ValueError("agent handoff entry identity conflict")
+            return {
+                key: current.get(key)
+                for key in (
+                    "schema_version",
+                    "dispatch_id",
+                    "goal_id",
+                    "todo_id",
+                    "from_agent_id",
+                    "to_agent_id",
+                    "status",
+                )
+            } | {"replayed": True}
+
+    todo_read = list_goal_todos(
+        registry_path=registry_path,
+        goal_id=goal_id,
+        role="agent",
+        todo_id=todo_id,
+        runtime_root_arg=str(runtime_root),
+    )
+    todos = todo_read.get("todos", [])
+    canonical_todo = next((item for item in todos if isinstance(item, dict)), None)
+    if (
+        not isinstance(todo_read.get("authority_read"), Mapping)
+        or canonical_todo is None
+    ):
+        raise ValueError("agent handoff requires a readable canonical Todo authority")
+    if (
+        canonical_todo.get("status") != "open"
+        or canonical_todo.get("claimed_by")
+        or canonical_todo.get("continuation_policy") != "independent_handoff"
+        or from_agent_id not in (canonical_todo.get("excluded_agents") or [])
+        or to_agent_id in (canonical_todo.get("excluded_agents") or [])
+    ):
+        raise ValueError(
+            "agent handoff candidate no longer matches canonical Todo eligibility"
+        )
+    claim_operation_id = f"agent-handoff-{dispatch_id[:32]}"
+    claim_command = shlex.join(
+        [
+            "loopx",
+            "--registry",
+            str(registry_path),
+            "--runtime-root",
+            str(runtime_root),
+            "todo",
+            "claim",
+            "--goal-id",
+            goal_id,
+            "--todo-id",
+            todo_id,
+            "--claimed-by",
+            to_agent_id,
+            "--agent-id",
+            to_agent_id,
+            "--role",
+            "agent",
+            "--claim-operation-id",
+            claim_operation_id,
+        ]
+    )
+    acknowledge_command = shlex.join(
+        [
+            "loopx",
+            "--registry",
+            str(registry_path),
+            "--runtime-root",
+            str(runtime_root),
+            "manager-inbox",
+            "acknowledge-handoff",
+            "--goal-id",
+            goal_id,
+            "--agent-id",
+            to_agent_id,
+            "--request-id",
+            dispatch_id,
+        ]
+    )
+    entry = {
+        "schema_version": HANDOFF_ENTRY_SCHEMA,
+        "inbox_kind": "agent_handoff",
+        "dispatch_id": dispatch_id,
+        **identity,
+        "claim_operation_id": claim_operation_id,
+        "claim_command": claim_command,
+        "acknowledge_command": acknowledge_command,
+        "instruction": (
+            "Claim the canonical same-Goal Todo with claim_command before work, then run "
+            "acknowledge_command so the sender receives durable claim readback. This handoff "
+            "grants no repository, provider, trading, payment, publishing, or other protected-operation authority."
+        ),
+    }
+    receipt = {
+        "schema_version": HANDOFF_RECEIPT_SCHEMA,
+        "dispatch_id": dispatch_id,
+        **identity,
+        "status": "dispatched",
+    }
+    with exclusive_file_lock(receipt_path.with_suffix(".lock")):
+        replayed = receipt_path.exists()
+        if replayed:
+            current = _read(receipt_path)
+            if any(
+                current.get(key) != value
+                for key, value in receipt.items()
+                if key != "status"
+            ):
+                raise ValueError("agent handoff dispatch identity conflict")
+        else:
+            _write(entry_path, entry)
+            _write(receipt_path, receipt)
+        if not entry_path.exists() or _read(entry_path) != entry:
+            raise ValueError("agent handoff dispatch readback failed")
+        current = _read(receipt_path)
+    return {
+        key: current.get(key)
+        for key in (
+            "schema_version",
+            "dispatch_id",
+            "goal_id",
+            "todo_id",
+            "from_agent_id",
+            "to_agent_id",
+            "status",
+        )
+    } | {"replayed": replayed}
+
+
+def pending_handoffs(
+    runtime_root: Path, goal_id: str, agent_id: str
+) -> list[dict[str, Any]]:
+    target = {"goal_id": goal_id, "agent_id": agent_id}
+    folder = _root(runtime_root) / "agent-handoffs" / "entries" / _hash(target)
+    items: list[dict[str, Any]] = []
+    for path in sorted(folder.glob("*.json")):
+        item = _read(path)
+        if (
+            item.get("schema_version") != HANDOFF_ENTRY_SCHEMA
+            or item.get("goal_id") != goal_id
+            or item.get("to_agent_id") != agent_id
+        ):
+            raise ValueError("agent handoff inbox scope mismatch")
+        receipt = _read(_receipt_path(runtime_root, str(item.get("dispatch_id") or "")))
+        if (
+            receipt.get("schema_version") != HANDOFF_RECEIPT_SCHEMA
+            or receipt.get("dispatch_id") != item.get("dispatch_id")
+            or receipt.get("goal_id") != goal_id
+            or receipt.get("to_agent_id") != agent_id
+        ):
+            raise ValueError("agent handoff receipt scope mismatch")
+        if receipt.get("status") in {"claimed", "completed", "claim_conflict"}:
+            continue
+        items.append(item)
+        if len(items) == 21:
+            break
+    return items
+
+
+def record_handoff_reads(runtime_root: Path, items: list[dict[str, Any]]) -> None:
+    """Record inbox visibility without treating delivery as a Todo claim."""
+    for item in items:
+        if item.get("inbox_kind") != "agent_handoff":
+            continue
+        dispatch_id = str(item.get("dispatch_id") or "")
+        if not re.fullmatch(r"[a-f0-9]{64}", dispatch_id):
+            raise ValueError("invalid handoff dispatch id")
+        receipt_path = _receipt_path(runtime_root, dispatch_id)
+        with exclusive_file_lock(receipt_path.with_suffix(".lock")):
+            receipt = _read(receipt_path)
+            if (
+                receipt.get("schema_version") != HANDOFF_RECEIPT_SCHEMA
+                or receipt.get("dispatch_id") != dispatch_id
+                or receipt.get("goal_id") != item.get("goal_id")
+                or receipt.get("to_agent_id") != item.get("to_agent_id")
+            ):
+                raise ValueError("agent handoff receipt scope mismatch")
+            if receipt.get("status") == "dispatched":
+                _write(receipt_path, {**receipt, "status": "read"})
+
+
+def acknowledge_claim(
+    runtime_root: Path,
+    registry_path: Path,
+    *,
+    goal_id: str,
+    agent_id: str,
+    dispatch_id: str,
+) -> dict[str, Any]:
+    goal_id = _require_token(goal_id, "goal id")
+    agent_id = _require_token(agent_id, "agent id")
+    if not re.fullmatch(r"[a-f0-9]{64}", dispatch_id):
+        raise ValueError("invalid handoff dispatch id")
+    target = {"goal_id": goal_id, "agent_id": agent_id}
+    entry = _read(_entry_path(runtime_root, target, dispatch_id))
+    if entry.get("goal_id") != goal_id or entry.get("to_agent_id") != agent_id:
+        raise ValueError("agent handoff inbox scope mismatch")
+    todos = list_goal_todos(
+        registry_path=registry_path,
+        goal_id=goal_id,
+        role="agent",
+        todo_id=str(entry.get("todo_id") or ""),
+        runtime_root_arg=str(runtime_root),
+    ).get("todos", [])
+    todo = next((item for item in todos if isinstance(item, dict)), None)
+    if todo is None or not todo.get("claimed_by"):
+        raise ValueError(
+            "canonical Todo must be claimed by the handoff recipient first"
+        )
+    receipt_path = _receipt_path(runtime_root, dispatch_id)
+    with exclusive_file_lock(receipt_path.with_suffix(".lock")):
+        receipt = _read(receipt_path)
+        if (
+            receipt.get("schema_version") != HANDOFF_RECEIPT_SCHEMA
+            or receipt.get("dispatch_id") != dispatch_id
+            or receipt.get("goal_id") != goal_id
+            or receipt.get("to_agent_id") != agent_id
+        ):
+            raise ValueError("agent handoff receipt scope mismatch")
+        if todo.get("claimed_by") != agent_id:
+            conflict = {
+                **receipt,
+                "status": "claim_conflict",
+                "observed_claimed_by": todo.get("claimed_by"),
+            }
+            replayed = receipt.get("status") == "claim_conflict"
+            if not replayed:
+                _write(receipt_path, conflict)
+            return {
+                "ok": True,
+                "schema_version": HANDOFF_RECEIPT_SCHEMA,
+                "dispatch_id": dispatch_id,
+                "goal_id": goal_id,
+                "todo_id": entry["todo_id"],
+                "from_agent_id": entry["from_agent_id"],
+                "to_agent_id": agent_id,
+                "status": "claim_conflict",
+                "canonical_claim_verified": False,
+                "observed_claimed_by": todo.get("claimed_by"),
+                "reroute_required": True,
+                "replayed": replayed,
+            }
+        replayed = receipt.get("status") == "claimed"
+        if not replayed:
+            receipt = {**receipt, "status": "claimed"}
+            _write(receipt_path, receipt)
+    return {
+        "ok": True,
+        "schema_version": HANDOFF_RECEIPT_SCHEMA,
+        "dispatch_id": dispatch_id,
+        "goal_id": goal_id,
+        "todo_id": entry["todo_id"],
+        "from_agent_id": entry["from_agent_id"],
+        "to_agent_id": agent_id,
+        "status": "claimed",
+        "canonical_claim_verified": True,
+        "replayed": replayed,
+    }
diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py
index 68e0fa0eb0..5174e8e69b 100644
--- a/loopx/cli_commands/manager_inbox.py
+++ b/loopx/cli_commands/manager_inbox.py
@@ -21,6 +21,7 @@ def register_manager_inbox(subparsers, add_format):
         choices=(
             "read",
             "acknowledge",
+            "acknowledge-handoff",
             "link",
             "report",
             "status",
@@ -51,8 +52,14 @@ def handle_manager_inbox(args, registry_path, runtime_root):
     try:
         if args.manager_inbox_action == "configure-ssh-read-scope":
             from ..capabilities.manager_context.ssh_evidence import configure
-            result = configure(runtime_root, channel=args.channel_id or "", host=args.ssh_host,
-                               goal_ids=args.read_goal_id, execute=args.execute)
+
+            result = configure(
+                runtime_root,
+                channel=args.channel_id or "",
+                host=args.ssh_host,
+                goal_ids=args.read_goal_id,
+                execute=args.execute,
+            )
             print(json.dumps(result, ensure_ascii=False, indent=2))
             return 0
         if args.manager_inbox_action == "configure-read-scope":
@@ -74,10 +81,39 @@ def handle_manager_inbox(args, registry_path, runtime_root):
         if args.manager_inbox_action == "read":
             result = pending(runtime_root, args.goal_id, args.agent_id)
             from ..capabilities.manager_context.tracking import record_read
+            from ..capabilities.manager_context.agent_handoff import (
+                record_handoff_reads,
+            )
 
-            record_read(runtime_root, result["items"])
+            owner_context_items = [
+                item
+                for item in result["items"]
+                if item.get("inbox_kind") != "agent_handoff"
+            ]
+            record_read(runtime_root, owner_context_items)
+            record_handoff_reads(runtime_root, result["items"])
+            if any(
+                item.get("inbox_kind") == "agent_handoff" for item in result["items"]
+            ):
+                result["handoff_followthrough"] = (
+                    "For every agent_handoff item, run its exact claim_command before work, "
+                    "then its acknowledge_command. A delivered message is not a claim receipt."
+                )
             result["followthrough"] = (
-                "After reading and deciding, associate Core work with manager-inbox link. Then use manager-inbox report --phase conclusion --reply-text to return this request's concrete result, replan decision, or explicit blocker/defer reason to its original audience automatically. Use optional --phase decision only for meaningful interim news during longer work. Adoption/linking alone is not a completed exchange. Do not wait for the owner to ask again. Write audience-ready text, not private deliberation."
+                "For owner-context items, after reading and deciding, associate Core work "
+                "with manager-inbox link. Then use manager-inbox report --phase conclusion "
+                "--reply-text to return the concrete result, replan decision, or explicit "
+                "blocker/defer reason automatically."
+            )
+        elif args.manager_inbox_action == "acknowledge-handoff":
+            from ..capabilities.manager_context.agent_handoff import acknowledge_claim
+
+            result = acknowledge_claim(
+                runtime_root,
+                registry_path,
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+                dispatch_id=args.request_id or "",
             )
         elif args.manager_inbox_action == "report":
             from ..capabilities.manager_context.roundtrip import report
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 2ab95b1e9c..070348cfd0 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -42,10 +42,13 @@
     render_existing_heartbeat_receipt_payload,
 )
 from ..control_plane.quota.turn_envelope import build_turn_envelope
+from ..control_plane.capability_hooks import InteractionProjectionHookRegistration
 from ..control_plane.coordination.legacy_writer_fence import (
     LegacyCoordinationWriterFenced,
 )
-from ..control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
+from ..control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityUnavailable,
+)
 from ..control_plane.effect_runtime import EffectRuntimeRejected
 from ..control_plane.scheduler.execution_context import (
     GUIDED_START_TURN_RUNTIME_PROFILES,
@@ -246,7 +249,9 @@ def _quota_failure_payload(
         )
         if error.agent_id is not None:
             payload["agent_id"] = error.agent_id
-    elif isinstance(error, (LegacyCoordinationWriterFenced, LocalCoordinationAuthorityUnavailable)):
+    elif isinstance(
+        error, (LegacyCoordinationWriterFenced, LocalCoordinationAuthorityUnavailable)
+    ):
         payload.update(
             {
                 "error_code": error.code,
@@ -471,11 +476,13 @@ def _dispatch_quota_turn_start_hooks(
         from ..control_plane.agents.capability_memory import (
             extend_turn_start_dispatch as extend_capability_memory_dispatch,
         )
-        from ..capabilities.manager_context import turn_start_hook
-        from ..control_plane.capability_hooks import dispatch_turn_start_hooks
+        from ..capabilities.manager_context import extend_turn_start_dispatch
         from ..history import load_registry
         from ..paths import resolve_runtime_root
-        root = resolve_runtime_root(load_registry(registry_path), runtime_root_arg, registry_path=registry_path)
+
+        root = resolve_runtime_root(
+            load_registry(registry_path), runtime_root_arg, registry_path=registry_path
+        )
         dispatch = extend_capability_memory_dispatch(
             dispatch,
             registry_path=registry_path,
@@ -484,15 +491,25 @@ def _dispatch_quota_turn_start_hooks(
             agent_id=args.agent_id,
             available=args.available_capabilities,
         )
-        context_dispatch = dispatch_turn_start_hooks((turn_start_hook(root, registry_path, args.goal_id, args.agent_id),))
-        dispatch = dict(dispatch)
-        for key in ("results", "required_reads", "failures"):
-            dispatch[key] = list(dispatch.get(key) or []) + list(context_dispatch.get(key) or [])
-        for key in ("registered_count", "invoked_count"):
-            dispatch[key] = int(dispatch.get(key) or 0) + int(context_dispatch.get(key) or 0)
-        from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch
-        dispatch = extend_cadence_turn_start_dispatch(dispatch, registry_path=registry_path,
-            runtime_root=root, goal_id=args.goal_id, agent_id=args.agent_id)
+        dispatch = extend_turn_start_dispatch(
+            dispatch,
+            registry_path=registry_path,
+            runtime_root=root,
+            goal_id=args.goal_id,
+            agent_id=args.agent_id,
+        )
+
+        from ..capabilities.periodic_report.cadence_runtime import (
+            extend_cadence_turn_start_dispatch,
+        )
+
+        dispatch = extend_cadence_turn_start_dispatch(
+            dispatch,
+            registry_path=registry_path,
+            runtime_root=root,
+            goal_id=args.goal_id,
+            agent_id=args.agent_id,
+        )
     local_private_state_mutated = any(
         isinstance(result, Mapping)
         and result.get("local_private_state_mutated") is True
@@ -509,6 +526,174 @@ def _attach_turn_start_hook_dispatch(
         payload["turn_start_capability_hook_dispatch"] = dict(dispatch)
 
 
+def _attach_agent_handoff_dispatch(
+    payload: dict[str, object],
+    *,
+    runtime_root: Path,
+    registry_path: Path,
+    goal_id: str,
+    from_agent_id: str | None,
+) -> None:
+    """Attach a same-Goal handoff receipt without growing quota orchestration."""
+
+    if not from_agent_id:
+        return
+
+    from ..capabilities.manager_context.agent_handoff import (
+        dispatch_from_quota_decision,
+    )
+
+    receipt = dispatch_from_quota_decision(
+        runtime_root,
+        registry_path,
+        goal_id=goal_id,
+        from_agent_id=from_agent_id,
+        decision=payload,
+    )
+    if receipt is not None:
+        payload["agent_handoff_dispatch_receipt"] = receipt
+
+
+def _reconcile_should_run_heartbeat(
+    args: argparse.Namespace,
+    *,
+    payload: dict[str, object],
+    context: QuotaCommandContext,
+    registry_path: Path,
+    runtime_root_arg: str | None,
+    heartbeat_receipt_existing: dict[str, object] | None,
+    receipt_bound_todo_id: str | None,
+    receipt_bound_replan_obligation_id: str | None,
+    interaction_projection_hooks: tuple[InteractionProjectionHookRegistration, ...],
+) -> tuple[
+    dict[str, object],
+    dict[str, object] | None,
+    str,
+    bool,
+    str,
+    bool,
+    dict[str, object],
+    dict[str, object] | None,
+]:
+    """Reconcile or persist one heartbeat receipt for a should-run decision."""
+
+    status_payload = context.status_payload
+    cache_metadata = context.cache_metadata
+    heartbeat_turn_id = context.heartbeat_turn_id
+    existing_status = "replayed"
+    existing_appended = False
+    stall_observation = "not_evaluated"
+    receipt_ready = False
+    if not heartbeat_turn_id:
+        return (
+            payload,
+            heartbeat_receipt_existing,
+            existing_status,
+            existing_appended,
+            stall_observation,
+            receipt_ready,
+            status_payload,
+            cache_metadata,
+        )
+    if heartbeat_receipt_existing:
+        (
+            heartbeat_receipt_existing,
+            existing_status,
+            existing_appended,
+            stall_observation,
+            receipt_ready,
+        ) = reconcile_existing_heartbeat_receipt_for_turn(
+            payload,
+            args,
+            runtime_root=context.runtime_root,
+            turn_instance_id=heartbeat_turn_id,
+            existing=heartbeat_receipt_existing,
+        )
+    else:
+        existing_stall = find_quota_monitor_poll_turn(
+            context.runtime_root,
+            goal_id=args.goal_id,
+            agent_id=args.agent_id,
+            turn_instance_id=heartbeat_turn_id,
+        )
+        if payload.get("effective_action") == "monitor_quiet_skip" or existing_stall:
+            poll = record_quota_monitor_poll(
+                status_payload,
+                goal_id=args.goal_id,
+                registry_path=registry_path,
+                execute=True,
+                source="heartbeat",
+                agent_id=args.agent_id,
+                available_capabilities=args.available_capabilities,
+                turn_instance_id=heartbeat_turn_id,
+                scheduler_execution_context=context.scheduler_context,
+                operator_inbox_urgency_projector=(
+                    context.operator_inbox_urgency_projector
+                ),
+                bounded_research_frontier_projector=(
+                    project_live_explore_composition_frontier
+                ),
+            )
+            if not poll.get("ok"):
+                raise RuntimeError(
+                    "heartbeat stall observation writeback failed: "
+                    f"{poll.get('reason') or 'missing follow-up quota decision'}"
+                )
+            status_payload = collect_status(
+                registry_path=registry_path,
+                runtime_root_override=runtime_root_arg,
+                scan_roots=context.scan_roots,
+                limit=context.status_limit,
+                goal_id=context.status_goal_id,
+                available_capabilities=args.available_capabilities,
+            )
+            payload = build_live_quota_should_run_decision(
+                status_payload,
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+                available_capabilities=args.available_capabilities,
+                include_scheduler_detail="scheduler" in context.detail_sections,
+                include_agent_todo_detail=(
+                    "agent-todos" in context.detail_sections
+                    and not bool(getattr(args, "turn_envelope", False))
+                ),
+                codex_app_current_rrule=args.codex_app_current_rrule,
+                registry_path=registry_path,
+                runtime_root=context.runtime_root,
+                host_observation_resolver=resolve_codex_app_automation_rrule,
+                scheduler_execution_context=context.scheduler_context,
+                operator_inbox_urgency_projector=(
+                    context.operator_inbox_urgency_projector
+                ),
+                bounded_research_frontier_projector=(
+                    project_live_explore_composition_frontier
+                ),
+                receipt_bound_todo_id=receipt_bound_todo_id,
+                receipt_bound_replan_obligation_id=(receipt_bound_replan_obligation_id),
+                turn_instance_id=heartbeat_turn_id,
+                interaction_projection_hooks=interaction_projection_hooks,
+            )
+            cache_metadata = None
+            stall_observation = "replayed" if poll.get("replayed") else "appended"
+            payload["heartbeat_stall_writeback"] = {
+                "turn_instance_id": heartbeat_turn_id,
+                "status": stall_observation,
+                "generated_at": poll.get("generated_at"),
+            }
+        else:
+            stall_observation = "not_applicable"
+        receipt_ready = True
+    return (
+        payload,
+        heartbeat_receipt_existing,
+        existing_status,
+        existing_appended,
+        stall_observation,
+        receipt_ready,
+        status_payload,
+        cache_metadata,
+    )
+
 
 def _render_turn_envelope_payload(
     payload: dict[str, object],
@@ -531,6 +716,7 @@ def _render_turn_envelope_payload(
         degraded["turn_envelope_skipped"] = str(envelope_error)[:200]
         return degraded
 
+
 def handle_quota_command(
     args: argparse.Namespace,
     *,
@@ -642,98 +828,33 @@ def handle_quota_command(
                 receipt_bound_todo_id=receipt_bound_todo_id,
                 receipt_bound_replan_obligation_id=(receipt_bound_replan_obligation_id),
             )
-            if heartbeat_turn_id:
-                if heartbeat_receipt_existing:
-                    (
-                        heartbeat_receipt_existing,
-                        heartbeat_receipt_existing_status,
-                        heartbeat_receipt_existing_appended,
-                        heartbeat_stall_observation,
-                        heartbeat_receipt_ready,
-                    ) = reconcile_existing_heartbeat_receipt_for_turn(
-                        payload,
-                        args,
-                        runtime_root=runtime_root,
-                        turn_instance_id=heartbeat_turn_id,
-                        existing=heartbeat_receipt_existing,
-                    )
-                else:
-                    existing_stall = find_quota_monitor_poll_turn(
-                        runtime_root,
-                        goal_id=args.goal_id,
-                        agent_id=args.agent_id,
-                        turn_instance_id=heartbeat_turn_id,
-                    )
-                    if (
-                        payload.get("effective_action") == "monitor_quiet_skip"
-                        or existing_stall is not None
-                    ):
-                        poll = record_quota_monitor_poll(
-                            status_payload,
-                            goal_id=args.goal_id,
-                            registry_path=registry_path,
-                            execute=True,
-                            source="heartbeat",
-                            agent_id=args.agent_id,
-                            available_capabilities=args.available_capabilities,
-                            turn_instance_id=heartbeat_turn_id,
-                            scheduler_execution_context=scheduler_context,
-                            operator_inbox_urgency_projector=operator_inbox_urgency_projector,
-                            bounded_research_frontier_projector=(
-                                project_live_explore_composition_frontier
-                            ),
-                        )
-                        if not poll.get("ok"):
-                            raise RuntimeError(
-                                "heartbeat stall observation writeback failed: "
-                                f"{poll.get('reason') or 'missing follow-up quota decision'}"
-                            )
-                        status_payload = collect_status(
-                            registry_path=registry_path,
-                            runtime_root_override=runtime_root_arg,
-                            scan_roots=scan_roots,
-                            limit=status_limit,
-                            goal_id=status_goal_id,
-                            available_capabilities=args.available_capabilities,
-                        )
-                        payload = build_live_quota_should_run_decision(
-                            status_payload,
-                            goal_id=args.goal_id,
-                            agent_id=args.agent_id,
-                            available_capabilities=args.available_capabilities,
-                            include_scheduler_detail="scheduler" in detail_sections,
-                            include_agent_todo_detail=(
-                                "agent-todos" in detail_sections
-                                and not bool(getattr(args, "turn_envelope", False))
-                            ),
-                            codex_app_current_rrule=args.codex_app_current_rrule,
-                            registry_path=registry_path,
-                            runtime_root=runtime_root,
-                            host_observation_resolver=resolve_codex_app_automation_rrule,
-                            scheduler_execution_context=scheduler_context,
-                            operator_inbox_urgency_projector=operator_inbox_urgency_projector,
-                            bounded_research_frontier_projector=(
-                                project_live_explore_composition_frontier
-                            ),
-                            receipt_bound_todo_id=receipt_bound_todo_id,
-                            receipt_bound_replan_obligation_id=(
-                                receipt_bound_replan_obligation_id
-                            ),
-                            turn_instance_id=heartbeat_turn_id,
-                            interaction_projection_hooks=(interaction_projection_hooks),
-                        )
-                        cache_metadata = None
-                        heartbeat_stall_observation = (
-                            "replayed" if poll.get("replayed") else "appended"
-                        )
-                        payload["heartbeat_stall_writeback"] = {
-                            "turn_instance_id": heartbeat_turn_id,
-                            "status": heartbeat_stall_observation,
-                            "generated_at": poll.get("generated_at"),
-                        }
-                    else:
-                        heartbeat_stall_observation = "not_applicable"
-                    heartbeat_receipt_ready = True
+            (
+                payload,
+                heartbeat_receipt_existing,
+                heartbeat_receipt_existing_status,
+                heartbeat_receipt_existing_appended,
+                heartbeat_stall_observation,
+                heartbeat_receipt_ready,
+                status_payload,
+                cache_metadata,
+            ) = _reconcile_should_run_heartbeat(
+                args,
+                payload=payload,
+                context=context,
+                registry_path=registry_path,
+                runtime_root_arg=runtime_root_arg,
+                heartbeat_receipt_existing=heartbeat_receipt_existing,
+                receipt_bound_todo_id=receipt_bound_todo_id,
+                receipt_bound_replan_obligation_id=(receipt_bound_replan_obligation_id),
+                interaction_projection_hooks=interaction_projection_hooks,
+            )
+            _attach_agent_handoff_dispatch(
+                payload,
+                runtime_root=runtime_root,
+                registry_path=registry_path,
+                goal_id=args.goal_id,
+                from_agent_id=args.agent_id,
+            )
         elif args.quota_command == "monitor-poll":
             payload = record_quota_monitor_poll_for_cli(
                 args,
diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index a8d2fbe003..365be713b7 100644
--- a/loopx/cli_commands/turn.py
+++ b/loopx/cli_commands/turn.py
@@ -14,11 +14,16 @@
 from ..capabilities.agent_turn_recall import (
     run_configured_agent_turn_recall_fail_open,
 )
+from ..capabilities.manager_context import extend_turn_start_dispatch
 from ..capabilities.reward_memory import (
     run_configured_turn_outcome_ingest_fail_open,
 )
-from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch
-from ..capabilities.periodic_report.pending_intent import periodic_report_pending_intent_interaction_hook
+from ..capabilities.periodic_report.cadence_runtime import (
+    extend_cadence_turn_start_dispatch,
+)
+from ..capabilities.periodic_report.pending_intent import (
+    periodic_report_pending_intent_interaction_hook,
+)
 from ..control_plane.quota.live_decision import build_live_quota_should_run_decision
 from ..control_plane.quota.heartbeat_receipt import (
     ensure_turn_heartbeat_settlement_receipt,
@@ -86,8 +91,6 @@
 FormatSelector = Callable[..., str]
 
 
-
-
 def handle_turn_command(
     args: argparse.Namespace,
     *,
@@ -125,9 +128,20 @@ def handle_turn_command(
                 goal_id=args.goal_id,
                 agent_id=args.agent_id,
             )
+            turn_start_hook_dispatch = extend_turn_start_dispatch(
+                turn_start_hook_dispatch,
+                registry_path=registry_path,
+                runtime_root=runtime_root,
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+            )
             turn_start_hook_dispatch = extend_cadence_turn_start_dispatch(
-                turn_start_hook_dispatch, registry_path=registry_path, runtime_root=runtime_root,
-                goal_id=args.goal_id, agent_id=args.agent_id)
+                turn_start_hook_dispatch,
+                registry_path=registry_path,
+                runtime_root=runtime_root,
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+            )
             from ..control_plane.agents.capability_memory import (
                 extend_turn_start_dispatch as extend_capability_memory_dispatch,
             )
@@ -156,6 +170,7 @@ def handle_turn_command(
             execution_mode=args.execution_mode,
             scheduler_owner=args.scheduler_owner,
         )
+
         def build_turn_decision(
             *, requested_action_todo_id: str | None = None
         ) -> dict[str, Any]:
@@ -176,9 +191,14 @@ def build_turn_decision(
                 ),
                 requested_action_todo_id=requested_action_todo_id,
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
-                interaction_projection_hooks=(periodic_report_pending_intent_interaction_hook(
-                    registry_path=registry_path, runtime_root=runtime_root,
-                    goal_id=args.goal_id, agent_id=args.agent_id),),
+                interaction_projection_hooks=(
+                    periodic_report_pending_intent_interaction_hook(
+                        registry_path=registry_path,
+                        runtime_root=runtime_root,
+                        goal_id=args.goal_id,
+                        agent_id=args.agent_id,
+                    ),
+                ),
             )
 
         decision = build_turn_decision()
@@ -197,6 +217,20 @@ def build_turn_decision(
                 )
             selected_todo["selected_by"] = "turn_controller_advisory_primary"
             decision["action_portfolio"] = advisory_portfolio
+        if args.turn_command == "run-once" and args.execute and args.agent_id:
+            from ..capabilities.manager_context.agent_handoff import (
+                dispatch_from_quota_decision,
+            )
+
+            handoff_receipt = dispatch_from_quota_decision(
+                runtime_root,
+                registry_path,
+                goal_id=args.goal_id,
+                from_agent_id=args.agent_id,
+                decision=decision,
+            )
+            if handoff_receipt is not None:
+                decision["agent_handoff_dispatch_receipt"] = handoff_receipt
         resume_identity = {
             "goal_id": args.resume_goal_id,
             "agent_id": args.resume_agent_id,
@@ -266,15 +300,24 @@ def build_turn_decision(
             # signed, provider-neutral intent untouched. This only projects
             # argv; the Turn driver never executes a projected shell string.
             command = shlex.split(capability_action["intent"]["command"])
-            capability_action["command_argv"] = [command[0], "--registry", str(registry_path),
-                "--runtime-root", str(runtime_root), *command[1:]]
+            capability_action["command_argv"] = [
+                command[0],
+                "--registry",
+                str(registry_path),
+                "--runtime-root",
+                str(runtime_root),
+                *command[1:],
+            ]
             capability_action["command"] = shlex.join(capability_action["command_argv"])
         if turn_start_hook_dispatch.get("registered_count") or (
             turn_start_hook_dispatch.get("failures")
         ):
             payload["turn_start_capability_hook_dispatch"] = turn_start_hook_dispatch
-            if any(isinstance(result, Mapping) and result.get("local_private_state_mutated") is True
-                   for result in turn_start_hook_dispatch.get("results", [])):
+            if any(
+                isinstance(result, Mapping)
+                and result.get("local_private_state_mutated") is True
+                for result in turn_start_hook_dispatch.get("results", [])
+            ):
                 # The pure plan builder has no effects, but live preflight
                 # hooks may journal an inbox or calendar admission. Disclose
                 # that write without claiming a host turn or report ran.
@@ -321,9 +364,15 @@ def build_turn_decision(
                 # The normal host transaction forbids Core mutations. A
                 # capability may prepare artifacts and require authored input;
                 # never run its command as an arbitrary host/shell adapter.
-                payload.update(mode="run_once", status="capability_action_required",
-                               execute=bool(args.execute), executed=False)
-                print_payload(payload, output_format(args), _render_loopx_turn_plan_markdown)
+                payload.update(
+                    mode="run_once",
+                    status="capability_action_required",
+                    execute=bool(args.execute),
+                    executed=False,
+                )
+                print_payload(
+                    payload, output_format(args), _render_loopx_turn_plan_markdown
+                )
                 return 0
             project = Path(args.project).expanduser().resolve()
             planned_host = (
@@ -346,9 +395,7 @@ def build_turn_decision(
                     )
             else:
                 if args.host_command_json:
-                    raise ValueError(
-                        f"{args.host} does not accept --host-command-json"
-                    )
+                    raise ValueError(f"{args.host} does not accept --host-command-json")
                 raw_argv = None
             if args.validation_command_json:
                 raw_validation_argv = json.loads(args.validation_command_json)
@@ -726,9 +773,7 @@ def spend(*, effect_ref: str) -> dict[str, object]:
                         replan_obligation_id=settlement_identity.replan_obligation_id,
                     )
                     if readback is None:
-                        raise RuntimeError(
-                            EXACT_SETTLEMENT_READBACK_NOT_FOUND
-                        )
+                        raise RuntimeError(EXACT_SETTLEMENT_READBACK_NOT_FOUND)
                     event = readback.spend_event
                     if event is None:
                         append_settlement_event(
@@ -790,9 +835,7 @@ def terminal_completion_readback() -> dict[str, object] | None:
                 readback = project_durable_terminal_completion_readback(
                     todo=durable_todo,
                     expected_todo_id=todo_id,
-                    expected_completion_turn_key=(
-                        settlement_identity.turn_instance_id
-                    ),
+                    expected_completion_turn_key=(settlement_identity.turn_instance_id),
                     projection_source=projection_source,
                     existing_todo_ids=existing_todo_ids,
                 )
@@ -820,9 +863,7 @@ def writeback_resolver(effect_ref: str) -> dict[str, object]:
                         replan_obligation_id=settlement_identity.replan_obligation_id,
                     )
                     if readback is None:
-                        raise RuntimeError(
-                            EXACT_SETTLEMENT_READBACK_NOT_FOUND
-                        )
+                        raise RuntimeError(EXACT_SETTLEMENT_READBACK_NOT_FOUND)
                     run = readback.writeback_run
                     event = readback.writeback_event
                     if run is None and event is None:
@@ -864,9 +905,7 @@ def spend_resolver(effect_ref: str) -> dict[str, object]:
                         replan_obligation_id=settlement_identity.replan_obligation_id,
                     )
                     if readback is None:
-                        raise RuntimeError(
-                            EXACT_SETTLEMENT_READBACK_NOT_FOUND
-                        )
+                        raise RuntimeError(EXACT_SETTLEMENT_READBACK_NOT_FOUND)
                     run = readback.spend_run
                     event = readback.spend_event
                     if run is not None and run.get("effect_ref") != effect_ref:
@@ -912,9 +951,7 @@ def terminal_closeout_resolver(effect_ref: str) -> dict[str, object]:
                         replan_obligation_id=settlement_identity.replan_obligation_id,
                     )
                     if readback is None:
-                        raise RuntimeError(
-                            EXACT_SETTLEMENT_READBACK_NOT_FOUND
-                        )
+                        raise RuntimeError(EXACT_SETTLEMENT_READBACK_NOT_FOUND)
                     event = readback.completion_event
                     completion = terminal_completion_readback()
                     if event is None and completion is None:
@@ -1070,7 +1107,11 @@ def post_settlement_reward_memory(
             raise ValueError("turn requires the `plan` or `run-once` subcommand")
     except Exception as exc:  # noqa: BLE001 - CLI boundary renders typed JSON failure
         payload = {
-            **({"error_code": exc.code, **getattr(exc, "payload", {})} if isinstance(getattr(exc, "code", None), str) else {}),
+            **(
+                {"error_code": exc.code, **getattr(exc, "payload", {})}
+                if isinstance(getattr(exc, "code", None), str)
+                else {}
+            ),
             "ok": False,
             "schema_version": (
                 LOOPX_TURN_EXECUTION_SCHEMA_VERSION
diff --git a/loopx/control_plane/agents/agent_scope.py b/loopx/control_plane/agents/agent_scope.py
index ce4dde62e3..b3e8402f67 100644
--- a/loopx/control_plane/agents/agent_scope.py
+++ b/loopx/control_plane/agents/agent_scope.py
@@ -83,12 +83,16 @@
 }
 
 AGENT_TASK_SCOPE = "goal_all_read_claimed_run_global_read_v0"
+
+
 def _agent_identity_has_scoped_lane(
     agent_identity: dict[str, Any] | None,
 ) -> TypeGuard[dict[str, object]]:
     return isinstance(agent_identity, dict) and bool(
         normalize_todo_claimed_by(agent_identity.get("agent_id"))
     )
+
+
 def _attach_agent_identity_contracts(
     *,
     payload: dict[str, Any],
@@ -103,6 +107,8 @@ def _attach_agent_identity_contracts(
 
 def _todo_task_class(item: dict[str, Any]) -> str:
     return todo_item_task_class(item)
+
+
 def _todo_projection_sort_key(item: dict[str, Any]) -> tuple[int, int, str, str]:
     return todo_presentation_sort_key(item)
 
@@ -177,7 +183,10 @@ def agent_scope_count_advancement_items(
         if normalized_claimed_by == "__unclaimed__":
             if item_claimed_by:
                 continue
-        elif normalized_claimed_by is not None and item_claimed_by != normalized_claimed_by:
+        elif (
+            normalized_claimed_by is not None
+            and item_claimed_by != normalized_claimed_by
+        ):
             continue
         count += 1
     return count
@@ -205,7 +214,9 @@ def _action_scope_tokens_from_text(text: str) -> set[str]:
     }
 
 
-def _todo_item_claimed_by_agent_or_unclaimed(item: dict[str, Any], *, agent_id: str) -> bool:
+def _todo_item_claimed_by_agent_or_unclaimed(
+    item: dict[str, Any], *, agent_id: str
+) -> bool:
     return agent_scope_item_claimed_by_agent_or_unclaimed(item, agent_id=agent_id)
 
 
@@ -221,11 +232,17 @@ def _scoped_user_gate_fallback(
     if not gates or not isinstance(agent_todo_summary, dict):
         return None
     raw_due_monitor_candidates = agent_todo_summary.get("monitor_due_items")
-    due_monitor_candidates = raw_due_monitor_candidates if isinstance(raw_due_monitor_candidates, list) else []
-    raw_ready_deferred_candidates = agent_todo_summary.get(
-        "deferred_resume_candidates"
+    due_monitor_candidates = (
+        raw_due_monitor_candidates
+        if isinstance(raw_due_monitor_candidates, list)
+        else []
+    )
+    raw_ready_deferred_candidates = agent_todo_summary.get("deferred_resume_candidates")
+    ready_deferred_candidates = (
+        raw_ready_deferred_candidates
+        if isinstance(raw_ready_deferred_candidates, list)
+        else []
     )
-    ready_deferred_candidates = raw_ready_deferred_candidates if isinstance(raw_ready_deferred_candidates, list) else []
 
     # An empty capability projection is authoritative for advancement work.
     # Due monitors are projected separately from advancement candidates and are
@@ -263,8 +280,11 @@ def _scoped_user_gate_fallback(
         else None
     )
     selection = select_scoped_gate_fallback(
-        gates, executable_items, agent_id=claim_scope.get("agent_id") if claim_scope else None,
-        allow_unrelated_gate=allow_unrelated_gate, monitor_debt_backoff_active=monitor_debt_backoff_active,
+        gates,
+        executable_items,
+        agent_id=claim_scope.get("agent_id") if claim_scope else None,
+        allow_unrelated_gate=allow_unrelated_gate,
+        monitor_debt_backoff_active=monitor_debt_backoff_active,
     )
     if selection is None:
         return None
@@ -274,8 +294,14 @@ def _scoped_user_gate_fallback(
     blocked_items = []
     for blocked in selection["blocked"][:3]:
         item = executable_items[blocked["candidate_index"]]
-        blocked_items.append({**compact_todo_summary_item(item, text=str(item.get("text") or "").strip()),
-                              "todo_gate_relation": blocked["relation"]})
+        blocked_items.append(
+            {
+                **compact_todo_summary_item(
+                    item, text=str(item.get("text") or "").strip()
+                ),
+                "todo_gate_relation": blocked["relation"],
+            }
+        )
     selected_text = str(selected.get("text") or "").strip()
     selected_item = compact_todo_summary_item(selected, text=selected_text)
     selected_is_deferred_replan = selection["deferred_replan"]
@@ -317,7 +343,9 @@ def _scoped_user_gate_fallback(
     }
 
 
-def _first_executable_todo_text(agent_todo_summary: dict[str, Any] | None) -> str | None:
+def _first_executable_todo_text(
+    agent_todo_summary: dict[str, Any] | None,
+) -> str | None:
     if not isinstance(agent_todo_summary, dict):
         return None
     raw_items = agent_todo_summary.get("first_executable_items")
@@ -423,7 +451,9 @@ def _agent_scoped_user_todo_override(
             "blocked_action_scope": None,
         },
         "item_patch": {
-            "status": "active_state_agent_todo" if selected_action else "agent_scope_wait",
+            "status": "active_state_agent_todo"
+            if selected_action
+            else "agent_scope_wait",
             "waiting_on": "codex" if selected_action else "agent_scope",
             "recommended_action": (
                 selected_action or "wait for work or a user action bound to this agent"
@@ -505,7 +535,9 @@ def build_hint(
 
     if isinstance(agent_lane_next_action, dict):
         selected_by = str(agent_lane_next_action.get("selected_by") or "")
-        claim_required = agent_lane_next_action.get("claim_required_before_work") is True
+        claim_required = (
+            agent_lane_next_action.get("claim_required_before_work") is True
+        )
         target_todo_id = normalize_todo_id(agent_lane_next_action.get("todo_id"))
         if selected_by == "unclaimed_todo" or claim_required:
             action = None
@@ -575,7 +607,9 @@ def build_hint(
                     else None
                 ),
             )
-        deferred_todo_id = _first_compact_todo_id(frontier.get("deferred_resume_candidates"))
+        deferred_todo_id = _first_compact_todo_id(
+            frontier.get("deferred_resume_candidates")
+        )
         route_todo_id = _first_compact_todo_id(
             frontier.get("route_continuation_replan_candidates")
         )
@@ -631,7 +665,9 @@ def build_hint(
             )
         blocker_todo_id = _first_compact_todo_id(frontier.get("blocking_handoff_gates"))
         if not blocker_todo_id:
-            blocker_todo_id = _first_compact_todo_id(frontier.get("other_agent_claimed_items"))
+            blocker_todo_id = _first_compact_todo_id(
+                frontier.get("other_agent_claimed_items")
+            )
         if blocker_todo_id:
             return build_hint(
                 AgentLaneFrontierHintDecision.QUIET_NOOP_BLOCKER,
@@ -660,10 +696,20 @@ def build_hint(
     current_advancement_count = int(
         agent_todo_summary.get("current_agent_claimed_advancement_count") or 0
     )
-    current_monitor_count = int(agent_todo_summary.get("current_agent_claimed_monitor_count") or 0)
+    current_monitor_count = int(
+        agent_todo_summary.get("current_agent_claimed_monitor_count") or 0
+    )
     unclaimed_count = _selectable_unclaimed_advancement_count(agent_todo_summary)
-    lane = str(work_lane_contract.get("lane") or "") if isinstance(work_lane_contract, dict) else ""
-    if current_advancement_count == 0 and unclaimed_count == 0 and current_monitor_count > 0:
+    lane = (
+        str(work_lane_contract.get("lane") or "")
+        if isinstance(work_lane_contract, dict)
+        else ""
+    )
+    if (
+        current_advancement_count == 0
+        and unclaimed_count == 0
+        and current_monitor_count > 0
+    ):
         return build_hint(
             AgentLaneFrontierHintDecision.QUIET_NOOP_BLOCKER,
             source="agent_todo_summary",
@@ -708,11 +754,15 @@ def _agent_scope_deferred_resume_candidates(
             continue
         if not todo_item_is_deferred(item):
             continue
-        identity = str(item.get("todo_id") or item.get("index") or item.get("text") or "")
+        identity = str(
+            item.get("todo_id") or item.get("index") or item.get("text") or ""
+        )
         if identity in seen:
             continue
         seen.add(identity)
-        unique.append(compact_todo_summary_item(item, text=str(item.get("text") or "").strip()))
+        unique.append(
+            compact_todo_summary_item(item, text=str(item.get("text") or "").strip())
+        )
     return sorted(unique, key=_todo_projection_sort_key)
 
 
@@ -749,11 +799,15 @@ def _agent_scope_monitor_blocked_resume_candidates(
         # The typed resume-planning owner has already diagnosed and selected
         # this repair lane. This consumer keeps executor scope and presentation,
         # not a second interpretation of condition target/class/status.
-        identity = str(item.get("todo_id") or item.get("index") or item.get("text") or "")
+        identity = str(
+            item.get("todo_id") or item.get("index") or item.get("text") or ""
+        )
         if identity in seen:
             continue
         seen.add(identity)
-        compact = compact_todo_summary_item(item, text=str(item.get("text") or "").strip())
+        compact = compact_todo_summary_item(
+            item, text=str(item.get("text") or "").strip()
+        )
         if item.get("blocking_monitor_todo_id"):
             compact["blocking_monitor_todo_id"] = item["blocking_monitor_todo_id"]
         unique.append(compact)
@@ -803,7 +857,9 @@ def _agent_scope_handoff_gates_by_state(
             continue
         if item.get("gate_state") != gate_state:
             continue
-        identity = str(item.get("todo_id") or item.get("index") or item.get("text") or "")
+        identity = str(
+            item.get("todo_id") or item.get("index") or item.get("text") or ""
+        )
         if identity in seen:
             continue
         seen.add(identity)
@@ -862,7 +918,9 @@ def _agent_scope_route_continuation_replan_candidates(
             for item in value:
                 if not isinstance(item, dict):
                     continue
-                if not _route_continuation_candidate_matches_agent(item, agent_id=agent_id):
+                if not _route_continuation_candidate_matches_agent(
+                    item, agent_id=agent_id
+                ):
                     continue
                 candidates.append(item)
 
@@ -887,7 +945,9 @@ def _agent_scope_route_continuation_replan_candidates(
         if not identity or identity in seen:
             continue
         seen.add(identity)
-        compact = compact_todo_summary_item(item, text=str(item.get("text") or "").strip())
+        compact = compact_todo_summary_item(
+            item, text=str(item.get("text") or "").strip()
+        )
         compact["route_continuation_replan_required"] = True
         if item.get("route_continuation_reason") is not None:
             compact["route_continuation_reason"] = item.get("route_continuation_reason")
@@ -997,8 +1057,7 @@ def _monitor_blocked_resume_frontier(
     first = candidates[0]
     candidate_id = str(first.get("todo_id") or "").strip() or "<todo_id>"
     monitor_id = (
-        str(first.get("blocking_monitor_todo_id") or "").strip()
-        or "<monitor_todo_id>"
+        str(first.get("blocking_monitor_todo_id") or "").strip() or "<monitor_todo_id>"
     )
     return build_agent_scope_frontier_payload(
         agent_id=context.agent_id,
@@ -1024,6 +1083,96 @@ def _monitor_blocked_resume_frontier(
     )
 
 
+def _executor_excluded_handoff_dispatch_frontier(
+    context: _AgentScopeNoCandidateContext,
+) -> dict[str, Any] | None:
+    claim_scope = context.summary.get("claim_scope")
+    if not isinstance(claim_scope, dict):
+        return None
+    items = claim_scope.get("executor_excluded_dispatchable_items")
+    if not isinstance(items, list):
+        return None
+    candidates = [item for item in items if isinstance(item, dict)]
+    if not candidates:
+        return None
+    first = candidates[0]
+    todo_id = str(first.get("todo_id") or "").strip() or "<todo_id>"
+    peers = [
+        str(value).strip()
+        for value in first.get("eligible_peer_ids", [])
+        if str(value).strip()
+    ]
+    peer_text = ", ".join(peers) or "an eligible registered peer"
+    return build_agent_scope_frontier_payload(
+        agent_id=context.agent_id,
+        action=AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED,
+        quiet_noop_allowed=False,
+        spend_policy="spend once after durable peer activation/claim receipt writeback",
+        reason=(
+            f"independent handoff {todo_id} excludes current agent {context.agent_id} "
+            f"but is dispatchable to registered peer {peer_text}; quiet wait would strand "
+            "the handoff"
+        ),
+        recommended_action=(
+            f"Activate or resume one eligible same-Goal peer ({peer_text}) for {todo_id}; "
+            "the peer must claim it before execution, and the coordinator must record a "
+            "durable dispatch/claim receipt. Do not claim or execute it as the excluded agent."
+        ),
+        requires_replan=True,
+        candidate_counts={
+            **context.candidate_counts,
+            "executor_excluded_dispatchable_count": len(candidates),
+        },
+        extra_fields={
+            "handoff_dispatch_required": True,
+            "handoff_dispatch_state": "dispatchable",
+            "executor_excluded_dispatchable_items": candidates[:3],
+            "eligible_peer_ids": peers,
+        },
+    )
+
+
+def _executor_excluded_handoff_no_peer_frontier(
+    context: _AgentScopeNoCandidateContext,
+) -> dict[str, Any] | None:
+    claim_scope = context.summary.get("claim_scope")
+    if not isinstance(claim_scope, dict):
+        return None
+    items = claim_scope.get("executor_excluded_no_eligible_peer_items")
+    if not isinstance(items, list):
+        return None
+    candidates = [item for item in items if isinstance(item, dict)]
+    if not candidates:
+        return None
+    todo_id = str(candidates[0].get("todo_id") or "").strip() or "<todo_id>"
+    return build_agent_scope_frontier_payload(
+        agent_id=context.agent_id,
+        action=AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED,
+        quiet_noop_allowed=False,
+        spend_policy="do not spend until an eligible same-Goal peer is registered",
+        reason=(
+            f"independent handoff {todo_id} excludes current agent {context.agent_id}, "
+            "and no registered same-Goal peer is eligible; dispatch cannot complete"
+        ),
+        recommended_action=(
+            f"Register an eligible same-Goal peer for {todo_id}, then let the coordinator "
+            "dispatch it and require a canonical claim receipt. The excluded agent must not "
+            "claim or execute it."
+        ),
+        requires_replan=True,
+        candidate_counts={
+            **context.candidate_counts,
+            "executor_excluded_no_eligible_peer_count": len(candidates),
+        },
+        extra_fields={
+            "handoff_dispatch_required": False,
+            "handoff_dispatch_state": "no_eligible_peer",
+            "executor_excluded_no_eligible_peer_items": candidates[:3],
+            "eligible_peer_ids": [],
+        },
+    )
+
+
 def _deferred_resume_frontier(
     context: _AgentScopeNoCandidateContext,
 ) -> dict[str, Any] | None:
@@ -1301,7 +1450,9 @@ def _other_agent_or_exhausted_frontier(
             "other_claimants": other_claimants,
             "blocking_handoff_claimants": blocking_claimants,
             "other_agent_claimed_items": [
-                compact_todo_summary_item(item, text=str(item.get("text") or "").strip())
+                compact_todo_summary_item(
+                    item, text=str(item.get("text") or "").strip()
+                )
                 for item in other_items[:3]
             ],
         },
@@ -1309,6 +1460,8 @@ def _other_agent_or_exhausted_frontier(
 
 
 _AGENT_SCOPE_NO_CANDIDATE_RULES = (
+    _executor_excluded_handoff_dispatch_frontier,
+    _executor_excluded_handoff_no_peer_frontier,
     _monitor_blocked_resume_frontier,
     _deferred_resume_frontier,
     _blocked_successor_wait_frontier,
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index bf95915173..bb9b72587b 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -60,6 +60,11 @@ const CONTRACT_CAPSULE_FIELDS: Readonly<Record<string, readonly string[]>> = {
   agent_scope_frontier: [
     "schema_version", "action", "effective_action", "blocks_delivery",
     "quiet_noop_allowed", "requires_replan", "recommended_action", "spend_policy",
+    "handoff_dispatch_required", "handoff_dispatch_state", "eligible_peer_ids",
+  ],
+  agent_handoff_dispatch_receipt: [
+    "schema_version", "dispatch_id", "goal_id", "todo_id", "from_agent_id",
+    "to_agent_id", "status", "replayed",
   ],
   automation_liveness: [
     "schema_version", "keep_active", "pause_allowed", "pause_policy",
diff --git a/loopx/control_plane/todos/quota_selection.py b/loopx/control_plane/todos/quota_selection.py
index 0e2df5c202..bd55048c8a 100644
--- a/loopx/control_plane/todos/quota_selection.py
+++ b/loopx/control_plane/todos/quota_selection.py
@@ -69,6 +69,7 @@ def active(key: str) -> list[dict[str, Any]]:
                 "active_items": active("active_next_action_items"),
                 "active_executable_items": active("active_next_action_executable_items"),
                 "agent_id": agent, "profile": profile,
+                "registered_agents": identity.get("registered_agents") or [],
                 "user_gate_scope": filter_user_gate_blocks_agent,
                 "monitor_supported": todo_summary_monitor_writeback_supported(value),
                 "source_open_count": source_open_count,
diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts
index c71e4a16bb..bc97c80c5c 100644
--- a/loopx/control_plane/todos/quota_selection.ts
+++ b/loopx/control_plane/todos/quota_selection.ts
@@ -98,8 +98,34 @@ function visibility(items: readonly Row[], agent: string | null, backlog: number
   return result;
 }
 
+function handoffDispatchScope(source: readonly Row[], agent: string,
+  registeredAgents: readonly string[], limit: number): JsonObject {
+  const candidates = source.filter(row => !row.claim && row.actionable && !row.removed &&
+    row.taskClass === "advancement_task" && row.excluded.includes(agent) &&
+    row.payload.continuation_policy === "independent_handoff");
+  // Keep the ordinary quota hot path byte-for-byte compact. Handoff
+  // diagnostics only belong in the claim scope when there is actually an
+  // executor-excluded handoff to route or fail closed.
+  if (!candidates.length) return {};
+  const projected = candidates.map(row => {
+    const eligiblePeerIds = registeredAgents.filter(peer => peer !== agent && !row.excluded.includes(peer));
+    return {...row.display, eligible_peer_ids: eligiblePeerIds};
+  });
+  const dispatchable = projected.filter(item => (item.eligible_peer_ids as string[]).length > 0);
+  const unavailable = projected.filter(item => (item.eligible_peer_ids as string[]).length === 0);
+  return {
+    executor_excluded_handoff_count: candidates.length,
+    executor_excluded_dispatchable_count: dispatchable.length,
+    executor_excluded_dispatchable_items: dispatchable.slice(0, limit),
+    executor_excluded_no_eligible_peer_count: unavailable.length,
+    executor_excluded_no_eligible_peer_items: unavailable.slice(0, limit),
+    executor_excluded_handoff_policy:
+      "independent handoffs excluded from this executor must be activated on an eligible registered peer or fail closed as no_eligible_peer",
+  };
+}
+
 function claimScope(source: readonly Row[], selected: readonly Row[], agent: string,
-  profile: JsonObject | null, limit: number): JsonObject {
+  registeredAgents: readonly string[], profile: JsonObject | null, limit: number): JsonObject {
   const current = selected.filter(row => row.claim === agent), unclaimed = selected.filter(row => !row.claim);
   const others = source.filter(row => bucket(row, agent) === 2);
   const excluded = source.filter(row => row.excluded.includes(agent)), removed = source.filter(row => row.removed);
@@ -112,6 +138,7 @@ function claimScope(source: readonly Row[], selected: readonly Row[], agent: str
     other_agent_claimed_items: otherItems, blocked_claimed_open_count: others.length, blocked_claimed_items: otherItems,
     executor_excluded_self_count: excluded.length, executor_excluded_self_items: compact(excluded, limit),
     executor_exclusion_policy: "excluded_agents_cannot_claim_or_execute",
+    ...handoffDispatchScope(source, agent, registeredAgents, limit),
     removed_continuation_blocked_count: removed.length, removed_continuation_blocked_items: compact(removed, limit),
     removed_continuation_policy: "legacy_review_handoffs_fail_closed_until_repaired",
     ...(profile ? {profile_routing: {schema_version: "agent_profile_routing_v0", applied: true,
@@ -125,6 +152,8 @@ export function projectQuotaSelection(value: unknown): JsonObject {
   const available = request.available === undefined ? undefined : requireStringArray(request.available, "available");
   const source = rows(request.items, available), active = rows(request.active_items, available), activeExecutable = rows(request.active_executable_items, available);
   const agent = optionalNonEmptyString(request.agent_id, "agent_id");
+  const registeredAgents = request.registered_agents === undefined
+    ? [] : requireStringArray(request.registered_agents, "registered_agents");
   const userMode = requireBoolean(request.user_gate_scope, "user_gate_scope");
   const supported = requireBoolean(request.monitor_supported, "monitor_supported");
   const limit = (key: string) => {
@@ -144,7 +173,8 @@ export function projectQuotaSelection(value: unknown): JsonObject {
   const open = userMode ? blocking : blocking.filter(row => executableBy(row, agent));
   if (agent && !userMode) open.sort((a, b) => bucket(a, agent) - bucket(b, agent) ||
     a.profileRank - b.profileRank || a.priority - b.priority || a.index - b.index);
-  const scope = agent && !userMode ? claimScope(blocking, open, agent, profile, diagnostic) : null;
+  const scope = agent && !userMode
+    ? claimScope(blocking, open, agent, registeredAgents, profile, diagnostic) : null;
   const monitors = open.filter(row => row.actionable && row.taskClass === "continuous_monitor");
   const due = supported ? monitors.filter(row => row.due && executableBy(row, agent)) : [];
   const activeVisible = (row: Row) => userMode ? (row.gate ? gateApplies(row, agent) : actionApplies(row, agent)) : executableBy(row, agent);
diff --git a/loopx/control_plane/todos/quota_summary.py b/loopx/control_plane/todos/quota_summary.py
index 58ac727a9c..35fc1eeb57 100644
--- a/loopx/control_plane/todos/quota_summary.py
+++ b/loopx/control_plane/todos/quota_summary.py
@@ -553,10 +553,25 @@ def _compact_quota_payload_claim_scope(value: Any) -> Any:
     compact: dict[str, Any] = {}
     for key, child in value.items():
         if isinstance(child, list):
-            compact[key] = _compact_quota_payload_item_list(
+            compact_items = _compact_quota_payload_item_list(
                 child,
                 limit=QUOTA_PAYLOAD_DIAGNOSTIC_LANE_LIMIT,
             )
+            if key in {
+                "executor_excluded_dispatchable_items",
+                "executor_excluded_no_eligible_peer_items",
+            }:
+                for raw_item, compact_item in zip(
+                    child[:QUOTA_PAYLOAD_DIAGNOSTIC_LANE_LIMIT], compact_items
+                ):
+                    if not isinstance(raw_item, dict) or not isinstance(compact_item, dict):
+                        continue
+                    peers = raw_item.get("eligible_peer_ids")
+                    if isinstance(peers, list):
+                        compact_item["eligible_peer_ids"] = [
+                            str(peer) for peer in peers[:QUOTA_PAYLOAD_DIAGNOSTIC_LANE_LIMIT]
+                        ]
+            compact[key] = compact_items
         else:
             compact[key] = child
     return compact
diff --git a/tests/control_plane/test_quota_selection.py b/tests/control_plane/test_quota_selection.py
index b10c6f481b..2167326249 100644
--- a/tests/control_plane/test_quota_selection.py
+++ b/tests/control_plane/test_quota_selection.py
@@ -1,33 +1,60 @@
 """Quota scope is distinct from execution ownership; all data are synthetic."""
+
 import pytest
 import json
 from pathlib import Path
 
 from canonical_authority_fixture import initialize_canonical_authority
-from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result
-from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.testing.canary_harness import (
+    write_fixture_registry,
+    run_json_cli_result,
+)
+from loopx.control_plane.coordination.runtime_shadow import (
+    build_todo_runtime_shadow_projection,
+)
 from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
 
-from loopx.control_plane.todos.quota_summary import summarize_user_todos_for_quota
+from loopx.control_plane.todos.quota_summary import (
+    compact_quota_todo_summary_for_payload,
+    summarize_user_todos_for_quota,
+)
+from loopx.control_plane.agents.agent_scope import _agent_scope_no_candidate_frontier
 
 
 def summary(items):
-    return {"schema_version": "todo_summary_v0", "source_section": "User Todo",
-            "items": items, "first_open_items": items, "total_count": len(items),
-            "open_count": len(items), "done_count": 0, "deferred_count": 0}
+    return {
+        "schema_version": "todo_summary_v0",
+        "source_section": "User Todo",
+        "items": items,
+        "first_open_items": items,
+        "total_count": len(items),
+        "open_count": len(items),
+        "done_count": 0,
+        "deferred_count": 0,
+    }
 
 
 def item(todo_id, **fields):
-    return {"todo_id": todo_id, "text": "Synthetic scoped work", "index": 1,
-            "priority": "P1", "status": "open", "task_class": "user_gate", **fields}
+    return {
+        "todo_id": todo_id,
+        "text": "Synthetic scoped work",
+        "index": 1,
+        "priority": "P1",
+        "status": "open",
+        "task_class": "user_gate",
+        **fields,
+    }
 
 
 @pytest.mark.parametrize("scope", [{"global_gate": True}, {"blocks_agent": "agent-b"}])
 @pytest.mark.parametrize("exclusions", [[], ["agent-b"]])
 def test_explicit_user_gate_scope_is_not_erased_by_executor_claim(scope, exclusions):
     gate = item("todo_gate", claimed_by="agent-a", excluded_agents=exclusions, **scope)
-    result = summarize_user_todos_for_quota(summary([gate]),
-        agent_identity={"agent_id": "agent-b"}, filter_user_gate_blocks_agent=True)
+    result = summarize_user_todos_for_quota(
+        summary([gate]),
+        agent_identity={"agent_id": "agent-b"},
+        filter_user_gate_blocks_agent=True,
+    )
     assert [row["todo_id"] for row in result["gate_open_items"]] == ["todo_gate"]
     assert result["open_count"] == 1
     assert result["first_executable_items"] == []
@@ -35,50 +62,317 @@ def test_explicit_user_gate_scope_is_not_erased_by_executor_claim(scope, exclusi
 
 def test_scope_precedence_keeps_other_lane_gate_diagnostic():
     gate = item("todo_gate", claimed_by="agent-b", blocks_agent="agent-a")
-    result = summarize_user_todos_for_quota(summary([gate]),
-        agent_identity={"agent_id": "agent-b"}, filter_user_gate_blocks_agent=True)
+    result = summarize_user_todos_for_quota(
+        summary([gate]),
+        agent_identity={"agent_id": "agent-b"},
+        filter_user_gate_blocks_agent=True,
+    )
     assert result["gate_open_items"] == []
     assert result["other_agent_scoped_open_count"] == 1
     assert result["open_count"] == 0
 
 
 def test_agent_execution_still_respects_claim_and_exclusion():
-    rows = [item("todo_peer", task_class="advancement_task", claimed_by="agent-a"),
-            item("todo_excluded", task_class="advancement_task", excluded_agents=["agent-b"]),
-            item("todo_free", task_class="advancement_task"),
-            item("todo_owned", task_class="advancement_task", claimed_by="agent-b")]
-    result = summarize_user_todos_for_quota(summary(rows), agent_identity={"agent_id": "agent-b"})
-    assert [row["todo_id"] for row in result["first_executable_items"]] == ["todo_owned", "todo_free"]
+    rows = [
+        item("todo_peer", task_class="advancement_task", claimed_by="agent-a"),
+        item(
+            "todo_excluded", task_class="advancement_task", excluded_agents=["agent-b"]
+        ),
+        item("todo_free", task_class="advancement_task"),
+        item("todo_owned", task_class="advancement_task", claimed_by="agent-b"),
+    ]
+    result = summarize_user_todos_for_quota(
+        summary(rows), agent_identity={"agent_id": "agent-b"}
+    )
+    assert [row["todo_id"] for row in result["first_executable_items"]] == [
+        "todo_owned",
+        "todo_free",
+    ]
     assert result["claim_scope"]["executor_excluded_self_count"] == 1
     assert result["claim_scope"]["other_agent_claimed_open_count"] == 1
 
 
+def test_executor_excluded_handoff_projects_dispatchable_and_no_peer_outcomes():
+    rows = [
+        item(
+            "todo_dispatch",
+            task_class="advancement_task",
+            continuation_policy="independent_handoff",
+            excluded_agents=["agent-b"],
+        ),
+        item(
+            "todo_no_peer",
+            task_class="advancement_task",
+            continuation_policy="independent_handoff",
+            excluded_agents=["agent-a", "agent-b"],
+        ),
+    ]
+    result = summarize_user_todos_for_quota(
+        summary(rows),
+        agent_identity={
+            "agent_id": "agent-b",
+            "registered_agents": ["agent-a", "agent-b"],
+        },
+    )
+    scope = result["claim_scope"]
+    assert scope["executor_excluded_dispatchable_count"] == 1
+    assert scope["executor_excluded_dispatchable_items"][0]["eligible_peer_ids"] == [
+        "agent-a"
+    ]
+    assert scope["executor_excluded_no_eligible_peer_count"] == 1
+    assert (
+        scope["executor_excluded_no_eligible_peer_items"][0]["eligible_peer_ids"] == []
+    )
+
+    frontier = _agent_scope_no_candidate_frontier(
+        agent_identity={"agent_id": "agent-b"},
+        agent_todo_summary=result,
+        agent_lane_next_action=None,
+        work_lane_contract={
+            "lane": "advancement_task",
+            "must_attempt_work": True,
+        },
+        candidate_should_run=True,
+    )
+    assert frontier is not None
+    assert frontier["action"] == "successor_replan_required"
+    assert frontier["handoff_dispatch_required"] is True
+    assert frontier["quiet_noop_allowed"] is False
+    assert frontier["eligible_peer_ids"] == ["agent-a"]
+    compact = compact_quota_todo_summary_for_payload(result)
+    assert compact["claim_scope"]["executor_excluded_dispatchable_items"][0][
+        "eligible_peer_ids"
+    ] == ["agent-a"]
+
+
+def test_executor_excluded_handoff_without_peer_is_nonquiet_and_not_dispatched():
+    result = summarize_user_todos_for_quota(
+        summary(
+            [
+                item(
+                    "todo_no_peer",
+                    task_class="advancement_task",
+                    continuation_policy="independent_handoff",
+                    excluded_agents=["agent-a", "agent-b"],
+                )
+            ]
+        ),
+        agent_identity={
+            "agent_id": "agent-b",
+            "registered_agents": ["agent-a", "agent-b"],
+        },
+    )
+    frontier = _agent_scope_no_candidate_frontier(
+        agent_identity={"agent_id": "agent-b"},
+        agent_todo_summary=result,
+        agent_lane_next_action=None,
+        work_lane_contract={"lane": "advancement_task", "must_attempt_work": True},
+        candidate_should_run=True,
+    )
+    assert frontier is not None
+    assert frontier["quiet_noop_allowed"] is False
+    assert frontier["handoff_dispatch_required"] is False
+    assert frontier["handoff_dispatch_state"] == "no_eligible_peer"
+    assert frontier["eligible_peer_ids"] == []
+
+
+def test_quota_dispatch_survives_process_restart_and_replays_one_peer_inbox(
+    tmp_path: Path,
+):
+    runtime = tmp_path / "runtime"
+    registry = tmp_path / "registry.json"
+    state = tmp_path / "state.md"
+    state.write_text(
+        "---\nstatus: active\n---\n# Goal\n## Objective\nDeliver a checked change.\n\n"
+        "## Agent Todo\n"
+        "- [ ] [P0] Implement the independent review.\n"
+        "  <!-- loopx:todo todo_id=todo_handoff status=open "
+        "task_class=advancement_task continuation_policy=independent_handoff "
+        "excluded_agents=agent-b -->\n"
+    )
+    write_fixture_registry(
+        project=tmp_path,
+        runtime_root=runtime,
+        registry_path=registry,
+        goal_id="goal-handoff",
+        domain="quota-handoff",
+        adapter_kind="generic_project_goal_v0",
+        state_file=str(state),
+        registered_agents=["agent-a", "agent-b"],
+        quota_allowed_slots=None,
+    )
+    goal = json.loads(registry.read_text())["goals"][0]
+    fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None)
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-handoff",
+        todos=(
+            fields["agent_todos"]["items"]
+            + fields.get("user_todos", {}).get("items", [])
+        ),
+        handoff_mode="soft_claim",
+    )
+    initialize_canonical_authority(
+        runtime,
+        "goal-handoff",
+        projection,
+        state_path=state,
+    )
+
+    args = (
+        "quota",
+        "should-run",
+        "--goal-id",
+        "goal-handoff",
+        "--agent-id",
+        "agent-b",
+        "--scan-path",
+        str(tmp_path),
+    )
+    code, first = run_json_cli_result(
+        *args,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, first
+    receipt = first["agent_handoff_dispatch_receipt"]
+    assert receipt["status"] == "dispatched" and receipt["replayed"] is False
+    assert receipt["to_agent_id"] == "agent-a"
+
+    code, replay = run_json_cli_result(
+        *args,
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, replay
+    assert (
+        replay["agent_handoff_dispatch_receipt"]["dispatch_id"]
+        == receipt["dispatch_id"]
+    )
+    assert replay["agent_handoff_dispatch_receipt"]["replayed"] is True
+
+    code, inbox = run_json_cli_result(
+        "manager-inbox",
+        "read",
+        "--goal-id",
+        "goal-handoff",
+        "--agent-id",
+        "agent-a",
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, inbox
+    assert [item["dispatch_id"] for item in inbox["items"]] == [receipt["dispatch_id"]]
+    assert inbox["items"][0]["inbox_kind"] == "agent_handoff"
+    assert inbox["handoff_followthrough"]
+
+    code, claimed = run_json_cli_result(
+        "todo",
+        "claim",
+        "--goal-id",
+        "goal-handoff",
+        "--todo-id",
+        "todo_handoff",
+        "--claimed-by",
+        "agent-a",
+        "--agent-id",
+        "agent-a",
+        "--role",
+        "agent",
+        "--claim-operation-id",
+        f"agent-handoff-{receipt['dispatch_id'][:32]}",
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, claimed
+    code, acknowledged = run_json_cli_result(
+        "manager-inbox",
+        "acknowledge-handoff",
+        "--goal-id",
+        "goal-handoff",
+        "--agent-id",
+        "agent-a",
+        "--request-id",
+        receipt["dispatch_id"],
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, acknowledged
+    assert acknowledged["canonical_claim_verified"] is True
+    assert acknowledged["status"] == "claimed"
+    code, empty = run_json_cli_result(
+        "manager-inbox",
+        "read",
+        "--goal-id",
+        "goal-handoff",
+        "--agent-id",
+        "agent-a",
+        registry_path=registry,
+        runtime_root=runtime,
+    )
+    assert code == 0, empty
+    assert empty["items"] == []
+
+
 @pytest.mark.parametrize("display", ["legacy", "missing", "stale"])
 @pytest.mark.parametrize("scope", ["global_gate=true", "blocks_agent=agent-b"])
-def test_public_quota_keeps_gate_from_real_canonical_provider(tmp_path: Path, display: str, scope: str):
-    runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
-    history = "\n".join(f"- [x] [P2] Completed synthetic work {i}.\n"
-        f"  <!-- loopx:todo todo_id=todo_history_{i} status=done task_class=advancement_task -->" for i in range(12))
-    state.write_text("---\nstatus: active\n---\n# Goal\n## Objective\nDeliver a checked change.\n\n"
+def test_public_quota_keeps_gate_from_real_canonical_provider(
+    tmp_path: Path, display: str, scope: str
+):
+    runtime, registry, state = (
+        tmp_path / "runtime",
+        tmp_path / "registry.json",
+        tmp_path / "state.md",
+    )
+    history = "\n".join(
+        f"- [x] [P2] Completed synthetic work {i}.\n"
+        f"  <!-- loopx:todo todo_id=todo_history_{i} status=done task_class=advancement_task -->"
+        for i in range(12)
+    )
+    state.write_text(
+        "---\nstatus: active\n---\n# Goal\n## Objective\nDeliver a checked change.\n\n"
         "## Agent Todo\n" + history + "\n\n## User Todo\n"
         "- [ ] [P0] Owner approval is required.\n"
-        f"  <!-- loopx:todo todo_id=todo_gate task_class=user_gate status=open claimed_by=agent-a {scope} -->\n")
-    write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry,
-        goal_id="goal-scope", domain="quota-scope", adapter_kind="generic_project_goal_v0",
-        state_file=str(state), registered_agents=["agent-a", "agent-b"], quota_allowed_slots=None)
+        f"  <!-- loopx:todo todo_id=todo_gate task_class=user_gate status=open claimed_by=agent-a {scope} -->\n"
+    )
+    write_fixture_registry(
+        project=tmp_path,
+        runtime_root=runtime,
+        registry_path=registry,
+        goal_id="goal-scope",
+        domain="quota-scope",
+        adapter_kind="generic_project_goal_v0",
+        state_file=str(state),
+        registered_agents=["agent-a", "agent-b"],
+        quota_allowed_slots=None,
+    )
     if display != "legacy":
         goal = json.loads(registry.read_text())["goals"][0]
         fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None)
-        projection = build_todo_runtime_shadow_projection(goal_id="goal-scope",
-            todos=fields["agent_todos"]["items"] + fields["user_todos"]["items"], handoff_mode="soft_claim")
-        initialize_canonical_authority(runtime, "goal-scope", projection, state_path=state)
+        projection = build_todo_runtime_shadow_projection(
+            goal_id="goal-scope",
+            todos=fields["agent_todos"]["items"] + fields["user_todos"]["items"],
+            handoff_mode="soft_claim",
+        )
+        initialize_canonical_authority(
+            runtime, "goal-scope", projection, state_path=state
+        )
         if display == "missing":
             state.unlink()
         else:
             state.write_text("# Stale projection\n## User Todo\n- [x] Old approval.\n")
     before = state.read_bytes() if state.exists() else None
-    code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-scope",
-        "--agent-id", "agent-b", "--scan-path", str(tmp_path), registry_path=registry, runtime_root=runtime)
+    code, packet = run_json_cli_result(
+        "quota",
+        "should-run",
+        "--goal-id",
+        "goal-scope",
+        "--agent-id",
+        "agent-b",
+        "--scan-path",
+        str(tmp_path),
+        registry_path=registry,
+        runtime_root=runtime,
+    )
     assert code == 0, packet
     assert packet["requires_user_action"] is True
     gates = packet["user_todo_summary"]["gate_open_items"]
diff --git a/tests/control_plane_ts/quota_selection.test.ts b/tests/control_plane_ts/quota_selection.test.ts
index eafec322e9..aee01b0da3 100644
--- a/tests/control_plane_ts/quota_selection.test.ts
+++ b/tests/control_plane_ts/quota_selection.test.ts
@@ -45,9 +45,39 @@ test("execution scope is shared with active-next-action, including removed-polic
   assert.deepEqual(ids(lanes.active_next_action_items), ["unclaimed", "mine"]);
   assert.equal((lanes.claim_scope as JsonObject).executor_excluded_self_count, 1);
   assert.equal((lanes.claim_scope as JsonObject).removed_continuation_blocked_count, 1);
+  assert.equal(
+    "executor_excluded_handoff_count" in (lanes.claim_scope as JsonObject), false,
+  );
   assert.deepEqual(ids((result.claim_visibility as JsonObject).claimed_by_others_items), ["peer"]);
 });
 
+test("executor-excluded independent handoffs identify eligible peers and fail closed without one", () => {
+  const items = [
+    row("dispatchable", {excluded: ["agent-a"], payload: {
+      todo_id: "dispatchable", continuation_policy: "independent_handoff",
+    }}),
+    row("unavailable", {excluded: ["agent-a", "agent-b", "agent-c"], payload: {
+      todo_id: "unavailable", continuation_policy: "independent_handoff",
+    }}),
+    row("ordinary", {excluded: ["agent-a"]}),
+  ];
+  const lanes = projectQuotaSelection(request(items, {
+    registered_agents: ["agent-a", "agent-b", "agent-c"],
+  })).lanes as JsonObject;
+  const scope = lanes.claim_scope as JsonObject;
+  assert.equal(scope.executor_excluded_handoff_count, 2);
+  assert.equal(scope.executor_excluded_dispatchable_count, 1);
+  assert.deepEqual(scope.executor_excluded_dispatchable_items, [{
+    todo_id: "dispatchable", continuation_policy: "independent_handoff",
+    eligible_peer_ids: ["agent-b", "agent-c"],
+  }]);
+  assert.equal(scope.executor_excluded_no_eligible_peer_count, 1);
+  assert.deepEqual(scope.executor_excluded_no_eligible_peer_items, [{
+    todo_id: "unavailable", continuation_policy: "independent_handoff",
+    eligible_peer_ids: [],
+  }]);
+});
+
 test("monitor eligibility preserves provider writeback and capability fences", () => {
   const items = [row("due", {task_class: "continuous_monitor", due: true}),
     row("missing", {task_class: "continuous_monitor", due: true, missing: ["network"]}),
diff --git a/tests/test_manager_context_handoff.py b/tests/test_manager_context_handoff.py
index 2714115120..8bdd0b8dcd 100644
--- a/tests/test_manager_context_handoff.py
+++ b/tests/test_manager_context_handoff.py
@@ -14,6 +14,10 @@
     register_ingress,
     turn_start_hook,
 )
+from loopx.capabilities.manager_context.agent_handoff import (
+    acknowledge_claim,
+    dispatch_from_quota_decision,
+)
 from loopx.control_plane.capability_hooks import dispatch_turn_start_hooks
 
 
@@ -189,6 +193,213 @@ def test_hook_keeps_decided_requests_open_until_receiver_returns_conclusion(
         )
 
 
+def test_agent_handoff_dispatch_is_idempotent_and_requires_canonical_claim(
+    fixture, monkeypatch
+):
+    root, registry, _session, _turn, _request = fixture
+    registry_value = json.loads(registry.read_text())
+    registry_value["goals"][0]["coordination"]["registered_agents"] = [
+        "coordinator",
+        "worker",
+    ]
+    registry.write_text(json.dumps(registry_value))
+    decision = {
+        "agent_scope_frontier": {
+            "handoff_dispatch_required": True,
+            "eligible_peer_ids": ["worker"],
+            "executor_excluded_dispatchable_items": [
+                {
+                    "todo_id": "todo_handoff",
+                    "continuation_policy": "independent_handoff",
+                    "eligible_peer_ids": ["worker"],
+                    "revision": 4,
+                }
+            ],
+        }
+    }
+    monkeypatch.setattr(
+        "loopx.capabilities.manager_context.agent_handoff.list_goal_todos",
+        lambda **_kwargs: {
+            "authority_read": {"source_authority": "canonical"},
+            "todos": [
+                {
+                    "todo_id": "todo_handoff",
+                    "status": "open",
+                    "claimed_by": None,
+                    "continuation_policy": "independent_handoff",
+                    "excluded_agents": ["coordinator"],
+                }
+            ],
+        },
+    )
+
+    first = dispatch_from_quota_decision(
+        root,
+        registry,
+        goal_id="research",
+        from_agent_id="coordinator",
+        decision=decision,
+    )
+    replay = dispatch_from_quota_decision(
+        root,
+        registry,
+        goal_id="research",
+        from_agent_id="coordinator",
+        decision=decision,
+    )
+    assert first is not None and replay is not None
+    assert first["status"] == "dispatched" and first["replayed"] is False
+    assert replay["dispatch_id"] == first["dispatch_id"]
+    assert replay["replayed"] is True
+    hook_dispatch = dispatch_turn_start_hooks(
+        [turn_start_hook(root, registry, "research", "worker")]
+    )
+    assert hook_dispatch["required_reads"][0]["ordering"] == "before_work"
+    assert hook_dispatch["results"][0]["agent_read_required"] is True
+    assert "todo_handoff" not in json.dumps(hook_dispatch)
+    inbox = pending(root, "research", "worker")
+    handoff = inbox["items"][0]
+    assert handoff["inbox_kind"] == "agent_handoff"
+    assert "todo claim" in handoff["claim_command"]
+    assert "acknowledge-handoff" in handoff["acknowledge_command"]
+    assert not pending(root, "other", "peer")["items"]
+    assert all(
+        path.stat().st_mode & 0o777 == 0o600
+        for path in (root / ".local" / "manager-context" / "agent-handoffs").rglob(
+            "*.json"
+        )
+    )
+
+    monkeypatch.setattr(
+        "loopx.capabilities.manager_context.agent_handoff.list_goal_todos",
+        lambda **_kwargs: {"todos": [{"todo_id": "todo_handoff", "claimed_by": None}]},
+    )
+    with pytest.raises(ValueError, match="must be claimed"):
+        acknowledge_claim(
+            root,
+            registry,
+            goal_id="research",
+            agent_id="worker",
+            dispatch_id=first["dispatch_id"],
+        )
+    monkeypatch.setattr(
+        "loopx.capabilities.manager_context.agent_handoff.list_goal_todos",
+        lambda **_kwargs: {
+            "todos": [{"todo_id": "todo_handoff", "claimed_by": "worker"}]
+        },
+    )
+    claimed = acknowledge_claim(
+        root,
+        registry,
+        goal_id="research",
+        agent_id="worker",
+        dispatch_id=first["dispatch_id"],
+    )
+    assert claimed["canonical_claim_verified"] is True
+    assert claimed["status"] == "claimed" and claimed["replayed"] is False
+    assert (
+        acknowledge_claim(
+            root,
+            registry,
+            goal_id="research",
+            agent_id="worker",
+            dispatch_id=first["dispatch_id"],
+        )["replayed"]
+        is True
+    )
+    assert not pending(root, "research", "worker")["items"]
+
+
+def test_agent_handoff_rejects_cross_goal_or_unregistered_source(fixture):
+    root, registry, _session, _turn, _request = fixture
+    decision = {
+        "agent_scope_frontier": {
+            "handoff_dispatch_required": True,
+            "executor_excluded_dispatchable_items": [
+                {
+                    "todo_id": "todo_handoff",
+                    "continuation_policy": "independent_handoff",
+                    "eligible_peer_ids": ["peer"],
+                }
+            ],
+        }
+    }
+    with pytest.raises(ValueError, match="source agent"):
+        dispatch_from_quota_decision(
+            root,
+            registry,
+            goal_id="research",
+            from_agent_id="coordinator",
+            decision=decision,
+        )
+    with pytest.raises(ValueError, match="no eligible registered peer"):
+        dispatch_from_quota_decision(
+            root,
+            registry,
+            goal_id="research",
+            from_agent_id="worker",
+            decision=decision,
+        )
+
+
+def test_agent_handoff_records_stale_claim_conflict(fixture, monkeypatch):
+    root, registry, _session, _turn, _request = fixture
+    registry_value = json.loads(registry.read_text())
+    registry_value["goals"][0]["coordination"]["registered_agents"] = [
+        "coordinator",
+        "worker",
+        "other-worker",
+    ]
+    registry.write_text(json.dumps(registry_value))
+    canonical = {
+        "authority_read": {"source_authority": "canonical"},
+        "todos": [
+            {
+                "todo_id": "todo_stale",
+                "status": "open",
+                "claimed_by": None,
+                "continuation_policy": "independent_handoff",
+                "excluded_agents": ["coordinator"],
+            }
+        ],
+    }
+    monkeypatch.setattr(
+        "loopx.capabilities.manager_context.agent_handoff.list_goal_todos",
+        lambda **_kwargs: canonical,
+    )
+    receipt = dispatch_from_quota_decision(
+        root,
+        registry,
+        goal_id="research",
+        from_agent_id="coordinator",
+        decision={
+            "agent_scope_frontier": {
+                "handoff_dispatch_required": True,
+                "executor_excluded_dispatchable_items": [
+                    {
+                        "todo_id": "todo_stale",
+                        "continuation_policy": "independent_handoff",
+                        "eligible_peer_ids": ["worker"],
+                    }
+                ],
+            }
+        },
+    )
+    assert receipt is not None
+    canonical["todos"][0]["claimed_by"] = "other-worker"
+    conflict = acknowledge_claim(
+        root,
+        registry,
+        goal_id="research",
+        agent_id="worker",
+        dispatch_id=receipt["dispatch_id"],
+    )
+    assert conflict["status"] == "claim_conflict"
+    assert conflict["observed_claimed_by"] == "other-worker"
+    assert conflict["reroute_required"] is True
+    assert not pending(root, "research", "worker")["items"]
+
+
 def test_actual_manager_turn_delivers_and_reports_host_receipt(fixture, monkeypatch):
     from loopx.chat_runtime import ChatRuntimeController
     from loopx.chat_store import ChatSessionStore
diff --git a/tests/test_turn_envelope.py b/tests/test_turn_envelope.py
index 3d703d74da..96325f6146 100644
--- a/tests/test_turn_envelope.py
+++ b/tests/test_turn_envelope.py
@@ -261,7 +261,9 @@ def test_turn_envelope_preserves_action_boundary_and_writeback() -> None:
     )
 
 
-def test_turn_envelope_compacts_replan_help_without_losing_successor_execution() -> None:
+def test_turn_envelope_compacts_replan_help_without_losing_successor_execution() -> (
+    None
+):
     source = _full_decision()
     successor_command = (
         "loopx todo add --goal-id fixture-goal --role agent "
@@ -438,9 +440,9 @@ def test_turn_envelope_preserves_signed_adaptive_orchestration_contract() -> Non
 
     envelope = build_turn_envelope(source)
 
-    assert envelope["task_orchestration_contract"] == source[
-        "task_orchestration_contract"
-    ]
+    assert (
+        envelope["task_orchestration_contract"] == source["task_orchestration_contract"]
+    )
     assert envelope["action_signature"]["matches"] is True
     assert quota_action_signature_document(
         source
@@ -457,7 +459,7 @@ def test_three_long_child_briefs_stay_within_turn_envelope_budget() -> None:
             "action_kind": "implement",
             "task_domain": "code",
             "text": "Implement the primary slice.",
-                "required_write_scopes": ["loopx/**"],
+            "required_write_scopes": ["loopx/**"],
         },
         *[
             {
@@ -481,7 +483,7 @@ def test_three_long_child_briefs_stay_within_turn_envelope_budget() -> None:
                 "spawn_allowed": True,
                 "max_children": 3,
                 "allowed_domains": ["code", "validation"],
-            }
+            },
         },
         agent_identity={
             "agent_id": "codex-fixture",
@@ -582,13 +584,9 @@ def test_turn_envelope_full_decision_compacts_first_class_profiles(
 
 
 def test_turn_envelope_unbound_full_decision_is_not_executable() -> None:
-    full_decision = build_turn_envelope(_full_decision())["detail_ref"][
-        "full_decision"
-    ]
+    full_decision = build_turn_envelope(_full_decision())["detail_ref"]["full_decision"]
 
-    assert full_decision == (
-        "rerun the typed quota_guard from the current host packet"
-    )
+    assert full_decision == ("rerun the typed quota_guard from the current host packet")
     assert "quota should-run" not in full_decision
 
 
@@ -830,9 +828,7 @@ def test_action_signature_detects_semantic_drift() -> None:
     ) != turn_envelope_action_signature_document(envelope)
 
     envelope = build_turn_envelope(source)
-    envelope["action"]["selected_todo"]["continuation_policy"] = (
-        "independent_handoff"
-    )
+    envelope["action"]["selected_todo"]["continuation_policy"] = "independent_handoff"
 
     assert quota_action_signature_document(
         source
@@ -920,9 +916,22 @@ def test_contract_capsule_stays_bounded_with_replan_and_vision_contracts() -> No
         "blocks_delivery": True,
         "quiet_noop_allowed": False,
         "requires_replan": True,
+        "handoff_dispatch_required": True,
+        "handoff_dispatch_state": "dispatchable",
+        "eligible_peer_ids": ["codex-peer"],
         "recommended_action": "write a concrete successor before quiet wait " * 20,
         "spend_policy": "spend only after successor writeback " * 20,
     }
+    source["agent_handoff_dispatch_receipt"] = {
+        "schema_version": "loopx_agent_handoff_dispatch_receipt_v0",
+        "dispatch_id": "a" * 64,
+        "goal_id": "fixture-goal",
+        "todo_id": "todo_fixture",
+        "from_agent_id": "codex-fixture",
+        "to_agent_id": "codex-peer",
+        "status": "dispatched",
+        "replayed": False,
+    }
     source["vision_continuation_audit"] = {
         "schema_version": "vision_continuation_audit_v0",
         "required": True,
@@ -945,6 +954,14 @@ def test_contract_capsule_stays_bounded_with_replan_and_vision_contracts() -> No
     assert (
         envelope["contract_capsule"]["agent_scope_frontier"]["requires_replan"] is True
     )
+    assert (
+        envelope["contract_capsule"]["agent_scope_frontier"]["handoff_dispatch_state"]
+        == "dispatchable"
+    )
+    assert (
+        envelope["contract_capsule"]["agent_handoff_dispatch_receipt"]["status"]
+        == "dispatched"
+    )
     assert envelope["contract_capsule"]["vision_continuation_audit"]["required"] is True
     assert envelope["compaction"]["within_budget"] is True
     assert envelope["compaction"]["envelope_json_bytes"] < TURN_ENVELOPE_BUDGET_BYTES