fix(collaboration): retain queued handoffs across native lane contention

docs/reference/local-delegation.md

@@ -165,6 +165,23 @@ original Inbox request, binding, conversation and governed Turn; it creates no
 replacement task, grant or scheduler. This remains the explicit task-bound
 route described here, not general automatic activation of an idle Inbox owner.
 
+A still-prepared replay also performs the same current native preflight as a
+first submission. A busy or unknown Turn lane, refused task admission,
+unavailable runtime or missing current acceptance leaves the original operation
+prepared and returns its identity and preflight diagnosis without submitting a
+worker. Once those observations recover, replay can rejoin that same operation
+under fresh source authority and the existing prepared-source startup fence.
+An operation that has already advanced is read back without probing its Todo
+for a new launch. The preflight remains a point-in-time observation; the native
+start fence still owns a later race. For task-bound CLI, MCP and Chat workers,
+if that fence refuses an operation with no prior Turn journal, the typed
+observation owner keeps it `prepared` only for the
+native `turn_lane_in_flight` result proving no Host invocation, state write,
+quota spend or scheduler ACK. Readback retains that reason; a later authorized
+retry uses the same request, operation and lease identity. A journaled,
+effectful or uncertain result follows existing recovery and cannot be rewound.
+This adds no automatic retry or wake pump.
+
 中文：Chat 发起的执行保留原 Inbox 请求、固定的对话来源和原 operation。worker 在
 入口、Host 启动前和完成结算前重新核对当前发送者、接收者及精确
 Goal/Agent/requester/binding grant；撤权、来源不可读或身份变化会拒绝在原
@@ -183,6 +200,18 @@ prepared operation 可在恢复的当前 grant 下续接原 owner，终态 opera
 不创建替代任务、授权或调度器。这仍是本文的精确任务绑定路径，不代表普通 Inbox
 空闲接收者的自动激活已完成。
 
+仍为 `prepared` 的重放也执行首次提交所用的当前原生预检。Turn lane 忙碌或未知、
+任务准入拒绝、运行时不可用或当前验收依据缺失时，原 operation 保持 `prepared`，
+返回其身份和预检原因，不提交 worker。观察恢复后，在新鲜来源授权和既有
+prepared-source 启动条件下续接同一 operation；已推进的 operation 直接读回，
+不按新启动重查其 Todo。预检仍只是当时观察，之后的竞争由原生启动 fence 拒绝。
+对精确任务绑定的 CLI、MCP 和 Chat worker，原 operation 尚无 Turn journal，
+且原生 `turn_lane_in_flight` 结果明确证明未调用
+Host、未写 Turn 状态、未扣额度、未 ACK scheduler 时，类型化观察 owner 将其保留为
+`prepared` 并读回原原因；之后获授权的重试沿用原请求、operation 和租约身份。
+已有 journal、已有副作用或结果未知均沿原恢复流程，不回退执行状态。
+本改动不新增自动重试或唤醒循环。
+
 Member completion uses the ordinary active-Goal continuation, including legacy
 non-hard-lease routes. It does not declare terminal `no_followup` for a
 requester-owned synthesis. This lets controller validation finish the Todo
@@ -1536,6 +1565,23 @@ never certify native start, consumption, Todo completion or original-audience
 result delivery. The internal `--codex-confirmed-operation-id` is only an exact
 resume fence, not a caller credential or domain permit.
 
+Before a new task-bound callback dispatch, the adapter uses the same native
+preflight as Chat and consumes its TypeScript-owned `dispatch_allowed` fact.
+Busy or unknown lanes, unavailable authority/acceptance and known unavailable
+runtimes return `blocked` with the preflight; no peer request or worker is
+submitted. A later authorized callback can retry the original confirmed
+operation. Existing operation replay remains read-only and does not require a
+new admission. An unprobed configured runtime may still attempt the bounded
+native launch; the final Turn fence decides any race after preview. This adds
+neither a retry pump nor ordinary Inbox activation without a Todo. The separate
+`original_operation` continuation retains its existing reservation owner.
+
+中文：新建 task-bound callback 委派与 Chat 共用原生预检，直接消费 TypeScript 的
+`dispatch_allowed`。忙碌、状态未知或已知不可执行时只返回 blocked 和预检事实，
+保留原确认，不发布 peer 请求或提交 worker；后续合法回调仍沿原操作重试。已有操作
+重放只读，不重新准入；未探测 runtime 仍可有界尝试，启动竞争继续由最终 Turn fence
+处理。不增加自动重试或无 Todo 的普通 Inbox 唤醒，独立 original_operation 路径保持。
+
 Remove `managed_turn_wake` or the exact delegation requester grant to stop new
 admission. The collector reloads the launch grant at each callback; already
 started work needs its ordinary stop/reconciliation path. Lost spawn ACKs do

loopx/capabilities/manager_context/execution.py

@@ -86,40 +86,45 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
             raise ValueError("execution binding not granted to this source")
         service, binding = _service(root, registry, grant)
         operation = "context-" + receipt["request_id"]
-        # Recover an existing operation instead of probing a now-completed Todo
-        # and misreporting an accepted result as a refused new launch.
-        if service.path(operation).exists():
+        replayed = service.path(operation).exists()
+        saved = {}
+        if replayed:
             row = read_delegation_source_execution(service, selected, operation,
                 source_request_id=receipt["request_id"], brief=request["brief"],
                 conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]})
-            if row["status"] == "prepared":
-                # The requester may have died after persisting the operation
-                # but before starting its worker. Rejoin that same owner under
-                # fresh source authority; never retry an uncertain running Host.
-                if not execution_allowed() or grant not in _grants(root, registry, session, turn):
-                    raise ValueError("source execution grant changed before recovery")
-                row = service.resume(operation, prepared_source_request_id=receipt["request_id"])
-            return {"submitted": True, "operation_id": operation, "todo_id": binding["todo_id"],
-                    "status": row["status"], "replayed": True}
+            saved = {"operation_id": operation, "todo_id": binding["todo_id"],
+                     "status": row["status"], "replayed": True}
+            # Only a still-prepared operation can submit a worker. Other
+            # states remain observations, including an accepted operation
+            # whose now-completed Todo would refuse a new admission.
+            if row["status"] != "prepared":
+                return {"submitted": True, **saved}
+        # A crash before spawn leaves the original prepared operation pending.
+        # Its recovery needs the same fresh native preflight as first dispatch;
+        # otherwise a busy/unknown lane can strand it in a premature rejection.
         preflight = service.inspect(selected)
         # Match the existing start owner: an unprobed configured runtime may
         # attempt bounded execution, but is never presented as ready/running.
         # Known refusal, missing canonical acceptance or an unavailable runtime
         # must stop before dispatch. Actual launch and acceptance remain owned
         # by the governed Turn, not this point-in-time preview.
-        if (preflight["state"] not in {"launchable", "runtime_unverified"}
-                or not all(preflight.get(key) is True for key in
-                           ("turn_eligible", "acceptance_ready", "authority_ready"))):
-            return {"submitted": False, "preflight": preflight, "reason": "execution_not_launchable"}
+        if preflight.get("dispatch_allowed") is not True:
+            return {"submitted": False, **saved, "preflight": preflight, "reason": "execution_not_launchable"}
         # Re-read the source and operator selection after the potentially slow
         # preview. Delegations.start itself rechecks the exact binding and Turn.
         if not execution_allowed() or grant not in _grants(root, registry, session, turn):
             raise ValueError("source execution grant changed before launch")
-        result = service.start(selected, operation, request["brief"],
-                               conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]},
-                               source_request_id=receipt["request_id"])
+        if replayed:
+            # Keep the prepared-source CAS through worker startup. A delayed
+            # child must not retry an operation that has already advanced.
+            result = service.resume(operation, prepared_source_request_id=receipt["request_id"])
+        else:
+            result = service.start(selected, operation, request["brief"],
+                                   conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]},
+                                   source_request_id=receipt["request_id"])
         return {"submitted": True, "operation_id": operation, "todo_id": binding["todo_id"],
-                "status": result["status"], "runtime_readiness": preflight["state"], "replayed": False}
+                "status": result["status"], "replayed": replayed,
+                **({} if replayed else {"runtime_readiness": preflight["state"]})}
     except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRejected):
         return {"submitted": False, "reason": "execution_binding_or_admission_unavailable"}
 

loopx/collaboration_mcp.py

@@ -1481,31 +1481,11 @@ def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
 
     def _record_turn_result(
         self, path: Path, row: dict, result: dict, *, publish: bool = True,
-        already_locked: bool = False,
+        already_locked: bool = False, unstarted_turn: bool = False,
     ) -> None:
-        turn_key = result.get("resume_turn_key")
-        if turn_key:
-            # Validate the public shape before persisting an address supplied
-            # by the CLI boundary.
-            turn_journal_path(self.root, goal_id=self.goal_id, turn_key=turn_key)
-            row["turn_key"] = turn_key
-        row["turn_result"] = {
-            key: result.get(key)
-            for key in (
-                "status",
-                "result_kind",
-                "resume_turn_key",
-                "reason",
-                "host_failure",
-                "error",
-            )
-        }
-        if publish:
-            self._observe(path, row, "turn_returned", already_locked=already_locked)
-        elif already_locked:
-            self._fenced_write_locked(path, row)
-        else:
-            self._fenced_write(path, row)
+        # Keep the bound service seam used by existing recovery/fault-injection callers.
+        delegation_validation.record_turn_result(self, path, row, result, publish=publish,
+            already_locked=already_locked, unstarted_turn=unstarted_turn)
 
     def _receiver_adopted(self, row: dict, binding: dict) -> bool:
         request_id = row["identity"]["request_id"]
@@ -1806,7 +1786,10 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                                    "--execute", timeout=binding["timeout_seconds"] + 60,
                                    delegated_lease=self._delegated_lease_context(row, binding),
                                    host_record=self._host_process_record(path))
-                self._record_turn_result(path, row, result)
+                self._record_turn_result(path, row, result,
+                    unstarted_turn=turn_key is None and "turn_key" not in row)
+                if row["status"] == "prepared":
+                    return  # Native lane refused before this operation launched a Host.
         finally:
             # The compatibility bootstrap is private host input.  Keeping it
             # after the host returns (including an exception or timeout) makes

loopx/control_plane/collaboration/delegation.ts

@@ -290,7 +290,7 @@ export function delegationPreflight(params: JsonObject): JsonObject {
       return {
         schema_version: "loopx_delegation_preflight_v0",
         binding: {id: binding.id, agent_id: binding.agent_id, todo_id: binding.todo_id},
-        state: "workspace_unavailable", workspace_state: workspace.state,
+        state: "workspace_unavailable", workspace_state: workspace.state, dispatch_allowed: false,
         workspace_next_action: "review_operator_workspace_binding",
         turn_eligible: false, turn_route: null, acceptance_ready: false,
         acceptance_reason_code: null, acceptance_next_action: "none",
@@ -324,7 +324,7 @@ export function delegationPreflight(params: JsonObject): JsonObject {
       scheduler_acknowledged: false};
     return {
       schema_version: "loopx_delegation_preflight_v0", binding,
-      state: "authority_unavailable", turn_eligible: false, turn_route: null,
+      state: "authority_unavailable", dispatch_allowed: false, turn_eligible: false, turn_route: null,
       acceptance_ready: false, authority_ready: false,
       acceptance_reason_code: null, acceptance_next_action: "none",
       authority_reason: boundedReason(authority.reason, "canonical authority unavailable"),
@@ -350,7 +350,7 @@ export function delegationPreflight(params: JsonObject): JsonObject {
     "delegation inspection requires a matching effect-free selection refusal");
     const acceptance = params.acceptance === null ? null : requireJsonObject(params.acceptance, "task acceptance");
     return {
-      schema_version: "loopx_delegation_preflight_v0", binding, state: "turn_blocked",
+      schema_version: "loopx_delegation_preflight_v0", binding, state: "turn_blocked", dispatch_allowed: false,
       turn_eligible: false, turn_route: null, turn_blocker: refusal,
       ...delegationAcceptanceFacts(binding.todo_id, acceptance, params.validation_files_current),
       authority_ready: true, authority_reason: null, authority_state: "promoted",
@@ -383,7 +383,8 @@ export function delegationPreflight(params: JsonObject): JsonObject {
     : executor.available === null ? "runtime_unverified" : "launchable";
   return {
     schema_version: "loopx_delegation_preflight_v0", binding,
-    state, turn_eligible: eligible, turn_route: route.kind,
+    state, dispatch_allowed: state === "launchable" || state === "runtime_unverified",
+    turn_eligible: eligible, turn_route: route.kind,
     ...acceptanceFacts, authority_ready: true, authority_reason: null,
     authority_state: "promoted", authority_next_action: "none",
     promotion_from_surface_allowed: false,
@@ -521,6 +522,23 @@ export function transitionDelegationObservation(params: JsonObject): JsonObject
   const from = params.from as Observation, to = params.to as Observation;
   requireThat(Object.hasOwn(transitions, from) && Object.hasOwn(transitions, to), "invalid delegation observation");
   requireThat(from === to || transitions[from].includes(to), "invalid delegation observation transition");
+  // Losing the native lane race before this operation has a Turn is queued
+  // work, not a failed execution. Only the native effect-free refusal permits
+  // returning to prepared; a partial/resumed/unknown Turn keeps normal recovery.
+  if (from === "running" && to === "turn_returned" && params.unstarted_turn === true
+      && params.turn_execution !== undefined) {
+    const result = requireJsonObject(params.turn_execution, "native Turn execution");
+    const effects = result.effects as JsonObject | undefined;
+    const host = result.host as JsonObject | undefined;
+    if (result.schema_version === "loopx_turn_execution_v0" && result.mode === "run_once"
+        && result.ok === false && result.dry_run === false && result.replayed === false
+        && result.status === "unavailable" && result.reason === "turn_lane_in_flight"
+        && result.result_kind === null && result.quota_slot_spend_count === 0
+        && host?.executable === "not_invoked" && effects !== undefined && effects !== null
+        && ["host_invoked", "state_written", "quota_spent", "scheduler_acknowledged"]
+          .every(key => effects[key] === false)
+        && Object.values(effects).every(value => value === false)) return {status: "prepared"};
+  }
   if (to === "accepted") requireThat(params.canonical_done === true
     && params.acceptance_ready === true && params.artifacts_current === true,
   "accepted return requires current canonical completion and artifacts");

loopx/control_plane/collaboration/delegation_validation.py

@@ -131,6 +131,52 @@ def record_revalidation_result(service, path, row, binding, result):
     service._fenced_write(path, row)
 
 
+def record_turn_result(
+    service, path: Path, row: dict, result: dict, *, publish: bool = True,
+    already_locked: bool = False, unstarted_turn: bool = False,
+) -> None:
+    """Persist native result facts through the existing typed observation/fence."""
+    turn_key = result.get("resume_turn_key")
+    if turn_key:
+        # Validate the public shape before persisting an address supplied
+        # by the CLI boundary.
+        turn_journal_path(service.root, goal_id=service.goal_id, turn_key=turn_key)
+    if publish:
+        # Only native receipt facts needed by the typed status owner cross
+        # this boundary. Evidence, settlement details and Host configuration
+        # remain with their original result instead of entering another RPC.
+        observation = {key: result.get(key) for key in (
+            "schema_version", "mode", "ok", "dry_run", "replayed", "status",
+            "reason", "result_kind", "quota_slot_spend_count", "effects",
+        )}
+        host = result.get("host")
+        observation["host"] = {"executable": host.get("executable")} if isinstance(host, dict) else None
+        decision = effect_runtime_result("collaboration.delegation.observe", {
+            "from": row["status"], "to": "turn_returned",
+            "turn_execution": observation, "unstarted_turn": unstarted_turn,
+        })
+        row["status"] = decision["status"]
+        if row["status"] == "prepared":
+            row["error"] = result["reason"]
+    if turn_key and (not publish or row["status"] != "prepared"):
+        row["turn_key"] = turn_key
+    row["turn_result"] = {
+        key: result.get(key)
+        for key in (
+            "status",
+            "result_kind",
+            "resume_turn_key",
+            "reason",
+            "host_failure",
+            "error",
+        )
+    }
+    if already_locked:
+        service._fenced_write_locked(path, row)
+    else:
+        service._fenced_write(path, row)
+
+
 def matching_turn_key(service, row: dict, binding: dict) -> str | None:
     """Find only the journal bound to this operation's settlement identity."""
 

loopx/control_plane/collaboration/operation_wake.py

@@ -214,6 +214,13 @@ def dispatch_confirmed_operation_wake(
                             "model": profile["model"],
                             "reasoning_effort": profile["reasoning_effort"]},
         )
+        # Confirmation does not make an occupied/unknown native lane idle.
+        # Reuse the same typed preflight as Chat; a refused callback retains
+        # its canonical confirmation without publishing peer work or a worker.
+        preflight = service.inspect(configuration["binding_id"])
+        if preflight.get("dispatch_allowed") is not True:
+            return {**base, "state": "blocked", "reason_code": "execution_not_launchable",
+                    "preflight": preflight}
         brief = {
             "schema_version": "collaboration_brief_v0",
             "purpose": "Resume the exact human-confirmed canonical operation",