fix(scheduler): dispatch stranded peer handoffs

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 回执投影。

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,
+    }