From ca4d3745974fe0d984f51f075f46baa9d88cc4fc Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 05:34:02 +0800
Subject: [PATCH 1/9] fix(collaboration): preflight prepared source handoff
 recovery

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            | 17 ++++
 .../capabilities/manager_context/execution.py | 41 ++++++----
 .../test_delegation_source_replay.py          | 77 ++++++++++++++++++-
 tests/test_manager_context_execution.py       | 63 +++++++++++++++
 4 files changed, 180 insertions(+), 18 deletions(-)

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index fe5f228010..9ec8cd29c1 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -165,6 +165,16 @@ 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. This adds no automatic retry or wake pump.
+
 中文：Chat 发起的执行保留原 Inbox 请求、固定的对话来源和原 operation。worker 在
 入口、Host 启动前和完成结算前重新核对当前发送者、接收者及精确
 Goal/Agent/requester/binding grant；撤权、来源不可读或身份变化会拒绝在原
@@ -183,6 +193,13 @@ prepared operation 可在恢复的当前 grant 下续接原 owner，终态 opera
 不创建替代任务、授权或调度器。这仍是本文的精确任务绑定路径，不代表普通 Inbox
 空闲接收者的自动激活已完成。
 
+仍为 `prepared` 的重放也执行首次提交所用的当前原生预检。Turn lane 忙碌或未知、
+任务准入拒绝、运行时不可用或当前验收依据缺失时，原 operation 保持 `prepared`，
+返回其身份和预检原因，不提交 worker。观察恢复后，在新鲜来源授权和既有
+prepared-source 启动条件下续接同一 operation；已推进的 operation 直接读回，
+不按新启动重查其 Todo。预检仍只是当时观察，之后的竞争由原生启动 fence 拒绝；
+本改动不新增自动重试或唤醒循环。
+
 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
diff --git a/loopx/capabilities/manager_context/execution.py b/loopx/capabilities/manager_context/execution.py
index 3b68433dd6..6154b24e93 100644
--- a/loopx/capabilities/manager_context/execution.py
+++ b/loopx/capabilities/manager_context/execution.py
@@ -86,21 +86,22 @@ 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.
@@ -110,16 +111,22 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
         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"}
+            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"}
 
diff --git a/tests/control_plane/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
index 1ec5caf6c5..413f05b0bf 100644
--- a/tests/control_plane/test_delegation_source_replay.py
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -14,8 +14,9 @@
 from loopx.control_plane.collaboration.inbox import _read
 from loopx.control_plane.collaboration.peers import request as peer_request
 from loopx.control_plane.collaboration.source_grant_observation import read_delegation_source_execution
+from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight
 from test_delegation_source_authority import saved_handoff  # noqa: F401
-from test_local_delegation import service as delegation_service  # noqa: F401
+from test_local_delegation import demo, service as delegation_service  # noqa: F401
 
 
 def durable_bytes(root):
@@ -102,3 +103,77 @@ def test_chat_handoff_replay_reports_an_original_return_mismatch(saved_handoff):
     assert response["message"].startswith(answer + "\n\n")
     assert worker.path(operation).read_bytes() == before
     assert not (Path(binding["workspace"]) / "host-invocations").exists()
+
+
+def test_prepared_replay_retains_original_work_while_native_lane_is_busy(saved_handoff, monkeypatch):  # noqa: F811
+    root, worker, binding, (store, session, turn, request, receipt, _), operation = saved_handoff
+    submitted = []
+    inspected = []
+    inspect = execution.Delegations.inspect
+
+    def native_preflight(service, binding_id):
+        inspected.append(binding_id)
+        return inspect(service, binding_id)
+
+    monkeypatch.setattr(execution.Delegations, "inspect", native_preflight)
+    monkeypatch.setattr(execution.Delegations, "_spawn",
+        lambda _self, operation_id, **kwargs: submitted.append((operation_id, kwargs)))
+    original = worker.path(operation).read_bytes()
+    tasks = demo.canonical_tasks(root)
+    with turn_lane_singleflight(runtime_root=worker.root, goal_id=worker.goal_id,
+            plan={"turn_envelope": {"agent_id": binding["agent_id"]}}) as holder:
+        assert holder is not None
+        held = holder.read_bytes()
+        for _ in range(2):
+            refused = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+                request=request, receipt=receipt, execution_allowed=lambda: True)
+            assert refused["submitted"] is False, refused
+            assert refused["preflight"]["reason"] == "turn_lane_in_flight"
+            assert refused["operation_id"] == operation and refused["status"] == "prepared"
+        assert not submitted
+        assert holder.read_bytes() == held
+        assert worker.path(operation).read_bytes() == original
+        assert demo.canonical_tasks(root) == tasks
+        assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    recovered = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert recovered["submitted"] and recovered["operation_id"] == operation
+    assert inspected == [binding["id"]] * 3
+    assert submitted == [(operation, {"prepared_source_request_id": receipt["request_id"]})]
+    # The same real fixture Host now reads and answers the original source,
+    # not merely the separate peer execution request.
+    host = root / "fixture-host.py"
+    host.write_text(host.read_text().replace("print(json.dumps(build_result", """
+from loopx.control_plane.collaboration.peers import read_inbox
+from loopx.capabilities.manager_context.roundtrip import report
+source_id = delegation['source_request_id']
+items = read_inbox(root / 'runtime', root / 'registry.json', envelope['goal_id'], actor)['items']
+assert any(item['request_id'] == source_id for item in items)
+acknowledge(root / 'runtime', envelope['goal_id'], actor, source_id, 'adopt', 'Read original request')
+report(root / 'runtime', envelope['goal_id'], actor, source_id, 'conclusion', 'Original request handled')
+print(json.dumps(build_result"""))
+    worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    assert worker.read(operation)["status"] == "accepted"
+    terminal = worker.path(operation).read_bytes()
+    # Terminal recovery observes the accepted original operation; a completed
+    # task's new-admission refusal cannot hide it or cause another submission.
+    replay = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert replay["submitted"] and replay["status"] == "accepted"
+    assert worker.path(operation).read_bytes() == terminal
+    assert len(submitted) == 1
+    assert len(inspected) == 3
+    store.update_turn(session["session_id"], turn["turn_id"], status="completing",
+        response={"context_handoff_receipt": receipt})
+    store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    from loopx.capabilities.manager_context.roundtrip import drain
+    sends = []
+
+    def transport(route, *_args, **_kwargs):
+        sends.append(route["session_id"])
+        return {"message_id": "fixture-return", "reply_verified": True}
+
+    drain(worker.root, worker.registry, store, transport)
+    drain(worker.root, worker.registry, store, transport)
+    assert sends == [session["session_id"]]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
diff --git a/tests/test_manager_context_execution.py b/tests/test_manager_context_execution.py
index d4736ccade..78655c6c1b 100644
--- a/tests/test_manager_context_execution.py
+++ b/tests/test_manager_context_execution.py
@@ -262,6 +262,69 @@ def test_preflight_never_becomes_a_launch_permit(flow, monkeypatch, state):
     assert not flow[3]
 
 
+@pytest.mark.parametrize("state", ["turn_blocked", "acceptance_unavailable", "runtime_unavailable"])
+def test_prepared_recovery_uses_the_same_refusal_without_submitting_a_worker(flow, monkeypatch, state):
+    root, registry, (_, session, turn, request, receipt, _), started = flow
+    first = dispatch(flow)
+    monkeypatch.setattr(execution.Delegations, "read", lambda *_: {"status": "prepared"})
+    monkeypatch.setattr(execution.Delegations, "inspect", lambda *_: {"state": state})
+    monkeypatch.setattr(execution.Delegations, "resume",
+        lambda *_args, **_kwargs: pytest.fail("a refused preflight must retain prepared work"), raising=False)
+    result = execution.dispatch(root, registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert result["submitted"] is False and result["preflight"]["state"] == state
+    assert result["operation_id"] == first["operation_id"]
+    assert result["status"] == "prepared" and result["replayed"]
+    assert len(started) == 1
+
+
+@pytest.mark.parametrize("change", ["source_stop", "source_revocation"])
+def test_source_change_during_prepared_recovery_preview_does_not_resume(flow, monkeypatch, change):
+    root, registry, (_, session, turn, request, receipt, policy), started = flow
+    dispatch(flow)
+    monkeypatch.setattr(execution.Delegations, "read", lambda *_: {"status": "prepared"})
+    active = [True]
+
+    def preview(*_):
+        if change == "source_stop":
+            active[0] = False
+        else:
+            policy["sources"][session["channel_id"]]["execution_bindings"] = []
+            _write(_root(root) / "policy.json", policy)
+        return {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
+
+    monkeypatch.setattr(execution.Delegations, "inspect", preview)
+    monkeypatch.setattr(execution.Delegations, "resume",
+        lambda *_args, **_kwargs: pytest.fail("source stop must prevent prepared recovery"), raising=False)
+    result = execution.dispatch(root, registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: active[0])
+    assert result == {"submitted": False, "reason": "execution_binding_or_admission_unavailable"}
+    assert len(started) == 1
+
+
+@pytest.mark.parametrize("missing", ["turn_eligible", "acceptance_ready", "authority_ready"])
+def test_prepared_recovery_requires_every_original_preflight_fact(flow, monkeypatch, missing):
+    dispatch(flow)
+    monkeypatch.setattr(execution.Delegations, "read", lambda *_: {"status": "prepared"})
+    facts = {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
+    facts.pop(missing)
+    monkeypatch.setattr(execution.Delegations, "inspect", lambda *_: facts)
+    monkeypatch.setattr(execution.Delegations, "resume",
+        lambda *_args, **_kwargs: pytest.fail("an unavailable fact cannot permit recovery"), raising=False)
+    assert dispatch(flow)["submitted"] is False
+
+
+@pytest.mark.parametrize("status", ["running", "turn_returned", "accepted", "rejected", "stopped"])
+def test_advanced_source_replay_observes_without_new_admission(flow, monkeypatch, status):
+    original = dispatch(flow)
+    monkeypatch.setattr(execution.Delegations, "read", lambda *_: {"status": status})
+    monkeypatch.setattr(execution.Delegations, "inspect",
+        lambda *_: pytest.fail("advanced work is observed, not submitted again"))
+    result = dispatch(flow)
+    assert result == {"submitted": True, "operation_id": original["operation_id"],
+        "todo_id": "todo_current", "status": status, "replayed": True}
+
+
 def test_stop_during_preview_and_context_only_selection_do_not_launch(flow, monkeypatch):
     root, registry, (_, session, turn, request, receipt, _), started = flow
     active = [True]

From f7896ac27482f7b437631b003cc4057467669f19 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 06:24:24 +0800
Subject: [PATCH 2/9] fix(collaboration): retain unstarted work after native
 lane refusal

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            |  16 +-
 loopx/collaboration_mcp.py                    |  29 +++-
 .../control_plane/collaboration/delegation.ts |  17 ++
 .../test_delegation_source_replay.py          | 148 +++++++++++++++++-
 tests/control_plane_ts/delegation.test.ts     |  26 +++
 5 files changed, 228 insertions(+), 8 deletions(-)

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 9ec8cd29c1..0d0603ab99 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -173,7 +173,14 @@ 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. This adds no automatic retry or wake pump.
+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 启动前和完成结算前重新核对当前发送者、接收者及精确
@@ -197,7 +204,12 @@ prepared operation 可在恢复的当前 grant 下续接原 owner，终态 opera
 任务准入拒绝、运行时不可用或当前验收依据缺失时，原 operation 保持 `prepared`，
 返回其身份和预检原因，不提交 worker。观察恢复后，在新鲜来源授权和既有
 prepared-source 启动条件下续接同一 operation；已推进的 operation 直接读回，
-不按新启动重查其 Todo。预检仍只是当时观察，之后的竞争由原生启动 fence 拒绝；
+不按新启动重查其 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
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 9139716f2f..d1f12783b2 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -1481,13 +1481,31 @@ 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)
+        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)
@@ -1500,9 +1518,7 @@ def _record_turn_result(
                 "error",
             )
         }
-        if publish:
-            self._observe(path, row, "turn_returned", already_locked=already_locked)
-        elif already_locked:
+        if already_locked:
             self._fenced_write_locked(path, row)
         else:
             self._fenced_write(path, row)
@@ -1806,7 +1822,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
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 3a5c016a82..4d201f4385 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -521,6 +521,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");
diff --git a/tests/control_plane/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
index 413f05b0bf..b8a7219a7f 100644
--- a/tests/control_plane/test_delegation_source_replay.py
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -7,6 +7,7 @@
 import copy
 import json
 from pathlib import Path
+import subprocess
 
 import pytest
 
@@ -15,8 +16,9 @@
 from loopx.control_plane.collaboration.peers import request as peer_request
 from loopx.control_plane.collaboration.source_grant_observation import read_delegation_source_execution
 from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight
+from test_independent_delegation_validation import independent_binding
 from test_delegation_source_authority import saved_handoff  # noqa: F401
-from test_local_delegation import demo, service as delegation_service  # noqa: F401
+from test_local_delegation import brief, demo, service as delegation_service  # noqa: F401
 
 
 def durable_bytes(root):
@@ -105,6 +107,150 @@ def test_chat_handoff_replay_reports_an_original_return_mismatch(saved_handoff):
     assert not (Path(binding["workspace"]) / "host-invocations").exists()
 
 
+@pytest.mark.parametrize("hard_lease", [False, True])
+def test_busy_lane_after_dispatch_keeps_the_original_operation_retryable(saved_handoff, monkeypatch, hard_lease):  # noqa: F811
+    root, worker, binding, (store, session, turn, request, receipt, _), operation = saved_handoff
+    if hard_lease:
+        # Configure only this newly created disposable authority. The active
+        # Goal's policy migration/fence and execution grants are never changed.
+        program = '''
+import {openLocalAuthorityStore} from "./loopx/control_plane/coordination/local_authority_provider.ts";
+const [root,goal] = process.argv.slice(1);
+const store = await openLocalAuthorityStore(root,goal);
+const state = await store.loadAuthority();
+const result = await store.commitAuthority({expected_provider_revision:state.provider_revision,
+ operation_id:"fixture-lane-race",events:[],receipts:[],
+ next_projection:{...state.head,handoff_mode:"hard_lease"}});
+if(result.status!=="applied") throw new Error(JSON.stringify(result));
+'''
+        subprocess.run(["node", "--no-warnings", "--experimental-sqlite", "--experimental-strip-types",
+            "--input-type=module", "-e", program, str(worker.root), worker.goal_id],
+            capture_output=True, text=True, timeout=30, check=True)
+    original = _read(worker.path(operation))
+    tasks = demo.canonical_tasks(root)
+    # Dispatch already observed an idle lane and persisted the operation. The
+    # actual native Turn now loses its single-flight race, before any Host work.
+    with turn_lane_singleflight(runtime_root=worker.root, goal_id=worker.goal_id,
+            plan={"turn_envelope": {"agent_id": binding["agent_id"]}}) as holder:
+        assert holder is not None
+        held = holder.read_bytes()
+        queued_leases = []
+        for _ in range(2):
+            worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+            observed = _read(worker.path(operation))
+            assert observed["status"] == "prepared"
+            if hard_lease:
+                queued_leases.append(observed["task_lease"]["lease"])
+        queued = _read(worker.path(operation))
+        assert queued["status"] == "prepared", queued
+        assert queued["turn_result"]["reason"] == "turn_lane_in_flight"
+        assert queued["identity"] == original["identity"]
+        assert queued["conversation"] == original["conversation"]
+        assert "turn_key" not in queued
+        assert holder.read_bytes() == held
+        if hard_lease:
+            lease = queued["task_lease"]["lease"]
+            assert queued_leases[0] == queued_leases[1]
+            assert lease["status"] == "active"
+            assert lease["owner"] == binding["agent_id"]
+            assert lease["idempotency_key"] == worker._turn_instance_id(queued)
+        else:
+            assert demo.canonical_tasks(root) == tasks
+        assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+        assert not (Path(binding["workspace"]) / "host-invocations").exists()
+        assert not (Path(binding["workspace"]) / "DELEGATION.json").exists()
+    submitted = []
+    monkeypatch.setattr(execution.Delegations, "_spawn",
+        lambda _self, operation_id, **kwargs: submitted.append((operation_id, kwargs)))
+    resumed = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert resumed["submitted"] and resumed["operation_id"] == operation
+    assert submitted == [(operation, {"prepared_source_request_id": receipt["request_id"]})]
+    host = root / "fixture-host.py"
+    host.write_text(host.read_text().replace("print(json.dumps(build_result", """
+from loopx.capabilities.manager_context.roundtrip import report
+source_id = delegation['source_request_id']
+acknowledge(root / 'runtime', envelope['goal_id'], actor, source_id, 'adopt', 'Read original request')
+report(root / 'runtime', envelope['goal_id'], actor, source_id, 'conclusion', 'Original request handled')
+print(json.dumps(build_result"""))
+    worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    assert worker.read(operation)["status"] == "accepted"
+    completed = _read(worker.path(operation))
+    journals = list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert len(journals) == 1
+    journal = json.loads(journals[0].read_text())
+    assert journal["completed_phases"].count("quota_spend") == 1
+    if hard_lease:
+        proof = worker._cli(binding, "task-lease", "inspect", "--goal-id", worker.goal_id,
+            "--todo-id", binding["todo_id"])["lease"]
+        assert proof["status"] == "released"
+        assert proof["idempotency_key"] == lease["idempotency_key"]
+        assert proof["lease_epoch"] == lease["lease_epoch"]
+    worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    assert _read(worker.path(operation)) == completed
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    store.update_turn(session["session_id"], turn["turn_id"], status="completing",
+        response={"context_handoff_receipt": receipt})
+    store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    from loopx.capabilities.manager_context.roundtrip import drain
+    sends = []
+
+    def transport(route, *_args, **_kwargs):
+        sends.append(route["session_id"])
+        return {"message_id": "fixture-return", "reply_verified": True}
+
+    drain(worker.root, worker.registry, store, transport)
+    drain(worker.root, worker.registry, store, transport)
+    assert sends == [session["session_id"]]
+
+
+def test_native_cli_worker_keeps_a_late_busy_request_without_chat_source(delegation_service, monkeypatch):  # noqa: F811
+    root, worker = delegation_service
+    independent_binding(delegation_service)
+    monkeypatch.setattr(worker, "_spawn", lambda _: None)
+    operation = "cli-lane-race"
+    worker.start("analysis", operation, brief())
+    original = _read(worker.path(operation))
+    binding = worker.binding("analysis")
+    assert "source_request_id" not in original["identity"]
+    with turn_lane_singleflight(runtime_root=worker.root, goal_id=worker.goal_id,
+            plan={"turn_envelope": {"agent_id": binding["agent_id"]}}):
+        worker.execute(operation)
+        queued = worker.read(operation)
+        assert queued["status"] == "prepared"
+        assert queued["error"] == "turn_lane_in_flight"
+        assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert worker.resume(operation)["operation_id"] == operation
+    worker.execute(operation)
+    assert worker.read(operation)["status"] == "accepted"
+    worker.execute(operation)
+    assert _read(worker.path(operation))["identity"] == original["identity"]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+def test_unrelated_large_native_result_does_not_expand_the_observation(saved_handoff, monkeypatch):  # noqa: F811
+    _, worker, binding, (_, _, _, _, receipt, _), operation = saved_handoff
+    native = worker._cli
+
+    def result_with_auxiliary_details(*args, **kwargs):
+        result = native(*args, **kwargs)
+        if args[1:3] == ("turn", "run-once"):
+            # A later native result may carry large evidence/settlement detail;
+            # it is not input to delegation's status rule or another RPC grant.
+            result["auxiliary_detail"] = "x" * (3 * 1024 * 1024)
+            result["host"]["auxiliary_detail"] = "y" * (3 * 1024 * 1024)
+        return result
+
+    monkeypatch.setattr(worker, "_cli", result_with_auxiliary_details)
+    with turn_lane_singleflight(runtime_root=worker.root, goal_id=worker.goal_id,
+            plan={"turn_envelope": {"agent_id": binding["agent_id"]}}):
+        worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    queued = worker.read(operation)
+    assert queued["status"] == "prepared", queued
+    assert queued["error"] == "turn_lane_in_flight"
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+
+
 def test_prepared_replay_retains_original_work_while_native_lane_is_busy(saved_handoff, monkeypatch):  # noqa: F811
     root, worker, binding, (store, session, turn, request, receipt, _), operation = saved_handoff
     submitted = []
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index 9ff0c2dabd..78c314da27 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -221,6 +221,32 @@ test("message receipt and model return do not imply accepted work", () => {
   assert.equal(accepted.status, "accepted");
 });
 
+test("only an unstarted native lane refusal keeps delegation queued", () => {
+  const effects = {host_invoked: false, state_written: false, quota_spent: false,
+    scheduler_acknowledged: false};
+  const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false,
+    dry_run: false, replayed: false, status: "unavailable", reason: "turn_lane_in_flight",
+    result_kind: null, quota_slot_spend_count: 0, host: {executable: "not_invoked"}, effects};
+  const input = {from: "running", to: "turn_returned", unstarted_turn: true, turn_execution: result};
+  assert.deepEqual(transitionDelegationObservation(input), {status: "prepared"});
+  // Existing callers and already journaled operations keep their old lifecycle.
+  for (const unstarted_turn of [undefined, false, null, "true"])
+    assert.deepEqual(transitionDelegationObservation({...input, unstarted_turn}), {status: "turn_returned"});
+  assert.deepEqual(transitionDelegationObservation({from: "running", to: "turn_returned"}),
+    {status: "turn_returned"});
+  for (const patch of [
+    {schema_version: "other"}, {mode: "preview"}, {ok: true}, {dry_run: true}, {replayed: true},
+    {status: "committed"}, {reason: "runtime_unavailable"}, {result_kind: "host_failure"},
+    {quota_slot_spend_count: 1}, {host: {executable: "started"}}, {effects: null},
+    ...Object.keys(effects).flatMap(key => [true, null, undefined, "false"]
+      .map(value => ({effects: {...effects, [key]: value}}))),
+    {effects: {...effects, unknown_effect: true}},
+  ]) assert.deepEqual(transitionDelegationObservation({...input, turn_execution: {...result, ...patch}}),
+    {status: "turn_returned"});
+  assert.throws(() => transitionDelegationObservation({...input, from: "rejected"}), /transition/);
+  assert.throws(() => transitionDelegationObservation({...input, to: "prepared"}), /transition/);
+});
+
 test("only a conversation-origin accepted result leaves a wake intent for the exact requester", () => {
   // An ordinary CLI/MCP delegation has no conversation. It must keep the
   // transition it always had: no intent, no wake state, no widened readback.

From d0a532ec862878dcf77c90db7c71c94a8fe4e8c5 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 06:51:16 +0800
Subject: [PATCH 3/9] refactor(collaboration): keep turn receipt IO in
 validation owner

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/collaboration_mcp.py                    | 42 ++---------------
 .../collaboration/delegation_validation.py    | 46 +++++++++++++++++++
 2 files changed, 49 insertions(+), 39 deletions(-)

diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index d1f12783b2..56a607049c 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -1483,45 +1483,9 @@ def _record_turn_result(
         self, path: Path, row: dict, result: dict, *, publish: bool = True,
         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)
-        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:
-            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"]
diff --git a/loopx/control_plane/collaboration/delegation_validation.py b/loopx/control_plane/collaboration/delegation_validation.py
index ee4c0c843a..f30145b010 100644
--- a/loopx/control_plane/collaboration/delegation_validation.py
+++ b/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."""
 

From fd6033cd66d4a99eb5a66873136adcf39ff885ef Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 07:21:42 +0800
Subject: [PATCH 4/9] fix(collaboration): gate confirmed callbacks on native
 preflight

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            |  17 ++
 .../capabilities/manager_context/execution.py |   4 +-
 .../control_plane/collaboration/delegation.ts |   9 +-
 .../collaboration/operation_wake.py           |   7 +
 .../test_operation_wake_preflight.py          | 146 ++++++++++++++++++
 tests/control_plane_ts/delegation.test.ts     |   8 +
 .../test_lark_goal_channel_operation.py       |   3 +
 tests/test_manager_context_execution.py       |  13 +-
 8 files changed, 194 insertions(+), 13 deletions(-)
 create mode 100644 tests/control_plane/test_operation_wake_preflight.py

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 0d0603ab99..04fce5d671 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -1565,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
diff --git a/loopx/capabilities/manager_context/execution.py b/loopx/capabilities/manager_context/execution.py
index 6154b24e93..0208f5ba3f 100644
--- a/loopx/capabilities/manager_context/execution.py
+++ b/loopx/capabilities/manager_context/execution.py
@@ -108,9 +108,7 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
         # 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"))):
+        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.
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 4d201f4385..f832aa6745 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/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,
diff --git a/loopx/control_plane/collaboration/operation_wake.py b/loopx/control_plane/collaboration/operation_wake.py
index f56d304645..a5896f865d 100644
--- a/loopx/control_plane/collaboration/operation_wake.py
+++ b/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",
diff --git a/tests/control_plane/test_operation_wake_preflight.py b/tests/control_plane/test_operation_wake_preflight.py
new file mode 100644
index 0000000000..8883738e85
--- /dev/null
+++ b/tests/control_plane/test_operation_wake_preflight.py
@@ -0,0 +1,146 @@
+"""A confirmed callback retains its original intent when the receiver cannot run.
+
+Exercise the real File/SQLite preflight and native operation session. The local
+app-server fixture supplies protocol responses; it is not a model or App proof.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from examples import operation_action_fixtures as fixtures
+from loopx.chat_action_store import ChatActionStore
+from loopx.cli import build_parser
+from loopx.control_plane.collaboration.operation_wake import dispatch_confirmed_operation_wake
+from loopx.control_plane.turn_driver.codex_cli import load_codex_cli_session
+from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler, run_codex_operation_host
+from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight
+from loopx.file_lock import lock_holder_path
+from test_codex_operation_host import FAKE_SERVER
+from test_local_delegation import service, wait  # noqa: F401
+from test_loopx_turn_codex_cli import _request
+
+
+def confirmed_callback(delegation):
+    root, runner = delegation
+    executable = root / "fixture-codex"
+    starts = root / "native-starts"
+    executable.write_text(FAKE_SERVER.replace('thread = "owned-app-server-thread"',
+        "from pathlib import Path\n"
+        f"counter = Path({str(starts)!r})\n"
+        "counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n"
+        'thread = "owned-app-server-thread"'))
+    executable.chmod(0o700)
+    config = json.loads(runner.config.read_text())
+    binding = config["bindings"][0]
+    binding["host_args"] = ["--host", "codex-cli", "--codex-bin", str(executable),
+        "--codex-operation-tools", "--codex-model", "test-model", "--codex-reasoning-effort", "xhigh"]
+    runner.config.write_text(json.dumps(config))
+    selected = build_parser().parse_args(["turn", "run-once", "--goal-id", runner.goal_id,
+        "--agent-id", binding["agent_id"], "--todo-id", binding["todo_id"],
+        *runner._execution_arguments(binding, "preparation")])
+    request = _request()
+    request["turn_envelope"].update(goal_id=runner.goal_id, agent_id=binding["agent_id"])
+    request["turn_envelope"]["action"]["selected_todo"]["todo_id"] = binding["todo_id"]
+    run_codex_operation_host(request, runtime_root=runner.root, registry_path=runner.registry,
+        project=root / "analyst/initial", codex_bin=str(executable), model="test-model",
+        reasoning_effort="xhigh", mcp_server=selected.codex_mcp_server_json, timeout_seconds=5)
+    lineage = {"goal_id": runner.goal_id, "agent_id": binding["agent_id"], "todo_id": binding["todo_id"]}
+    session = load_codex_cli_session(runner.root, lineage=lineage)
+    handler = operation_tool_handler(runtime_root=runner.root, registry_path=runner.registry,
+        lineage=lineage, session_id=session["session_id"], profile_digest=session["operation_profile_digest"],
+        model="test-model", reasoning_effort="xhigh")
+    intent = fixtures.request(goal_id=runner.goal_id,
+        payload={"schema_version": "qualification_v0", "marker": "synthetic"})
+    terms = intent["normalized_parameters"]
+    terms.pop("executor")
+    terms.update(agent_id=binding["agent_id"], domain="qualification",
+        operation_kind="qualification.observe", operation_schema="qualification_v0")
+    terms["projection"].update(simulated=False, title="Synthetic callback qualification")
+    prepared = handler("loopx_operation", {"action": "prepare", "request": intent},
+        {"thread_id": session["session_id"], "host_turn_id": "preparing-turn"})
+    assert prepared["ok"], prepared
+    store = ChatActionStore(runner.root / "chat/actions")
+    proposal = prepared["proposal"]
+    delivered = store.record_operation_delivery(proposal["proposal_id"], delivery=fixtures.delivery(proposal))
+    confirmed = store.decide_operation(proposal["proposal_id"], decision="confirm",
+        confirmation=fixtures.confirmation(delivered))
+    wake = {"registry_path": str(runner.registry), "goal_id": runner.goal_id,
+        "requester_agent_id": runner.agent_id, "project": str(root),
+        "execution_config": "delegations.json", "binding_id": binding["id"]}
+    return runner, store, confirmed, wake, starts
+
+
+@pytest.mark.parametrize("lane", ["live", "foreign_host", "unreadable"])
+def test_unavailable_callback_retains_confirmation_then_starts_original_once(service, monkeypatch, lane):  # noqa: F811
+    runner, store, proposal, wake, starts = confirmed_callback(service)
+    original = store.load(proposal["proposal_id"])
+    journals = list((runner.root / "goals" / runner.goal_id / "turns").glob("*.json"))
+
+    def dispatch():
+        return dispatch_confirmed_operation_wake(proposal, runtime_root=runner.root, configuration=wake)
+
+    submitted = []
+    with monkeypatch.context() as no_worker, turn_lane_singleflight(runtime_root=runner.root, goal_id=runner.goal_id,
+            plan={"turn_envelope": {"agent_id": "analyst"}}) as held:
+        no_worker.setattr(type(runner), "_spawn", lambda self, operation, **kwargs: submitted.append(operation))
+        holder = lock_holder_path(runner._lane_target(runner.binding("analysis")))
+        if lane != "live":
+            record = json.loads(holder.read_text())
+            record["host"] = "different-fixture-host"
+            holder.write_text(json.dumps(record) if lane == "foreign_host" else "{unreadable")
+        retained = holder.read_bytes()
+        for _ in range(2):
+            result = dispatch()
+            assert result["state"] == "blocked", result
+            assert result["preflight"]["turn_lane"]["state"] == lane
+            assert not result["native_start_verified"] and not result["execution_allowed"]
+        assert held is not None and holder.read_bytes() == retained
+        assert submitted == []
+        assert store.load(proposal["proposal_id"]) == original
+        assert not list((runner.root / "collaboration").rglob("operation-wake-*.json"))
+        assert list((runner.root / "goals" / runner.goal_id / "turns").glob("*.json")) == journals
+        assert starts.read_text() == "1"  # Only the original preparation session.
+
+    result = dispatch()
+    assert result["state"] == "delegation_requested", result
+    completed = wait(runner, result["operation_id"])
+    stored = store.load(proposal["proposal_id"])
+    assert starts.read_text() == "2"
+    assert stored["operation"]["host_start"] is not None, completed
+    assert stored["operation"]["confirmation"] == original["operation"]["confirmation"]
+    assert dispatch()["state"] == "existing_delegation"
+    assert starts.read_text() == "2"
+    # The fixture waits; native start is not a verified domain outcome.
+    assert stored["operation"]["outcome"] is None
+
+
+@pytest.mark.parametrize("fault", ["workspace_missing", "grant_revoked_during_preview"])
+def test_callback_preflight_refusal_or_late_revocation_has_no_worker(service, monkeypatch, fault):  # noqa: F811
+    runner, store, proposal, wake, starts = confirmed_callback(service)
+    original = store.load(proposal["proposal_id"])
+    if fault == "workspace_missing":
+        binding = runner.binding("analysis")
+        workspace = Path(binding["workspace"])
+        workspace.rename(workspace.with_name("unavailable-workspace"))
+    else:
+        inspect = type(runner).inspect
+
+        def revoke_after_preview(self, binding_id):
+            result = inspect(self, binding_id)
+            config = json.loads(self.config.read_text())
+            config["bindings"][0]["requesters"] = []
+            self.config.write_text(json.dumps(config))
+            return result
+
+        monkeypatch.setattr(type(runner), "inspect", revoke_after_preview)
+    submitted = []
+    monkeypatch.setattr(type(runner), "_spawn", lambda self, operation, **kwargs: submitted.append(operation))
+    result = dispatch_confirmed_operation_wake(proposal, runtime_root=runner.root, configuration=wake)
+    assert result["state"] == "blocked", result
+    if fault == "workspace_missing":
+        assert result["preflight"]["state"] == "workspace_unavailable"
+    assert submitted == [] and starts.read_text() == "1"
+    assert store.load(proposal["proposal_id"]) == original
+    assert not list((runner.root / "collaboration").rglob("operation-wake-*.json"))
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index 78c314da27..c6dbd49471 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -69,6 +69,7 @@ test("preflight preserves a quota refusal, but rejects effectful or retargeted e
   const input = {binding,preview,acceptance:{todo_id:binding.todo_id,state:"ready"},validation_files_current:true};
   const observed = delegationPreflight(input);
   assert.equal(observed.state,"turn_blocked"); assert.equal(observed.turn_eligible,false);
+  assert.equal(observed.dispatch_allowed, false);
   assert.equal(observed.acceptance_ready,true); assert.equal(observed.executor,null);
   assert.deepEqual(observed.turn_blocker,projected.selection_rejection);
   assert.deepEqual(delegationPreflight({...input,preview:{...preview,selection_rejection:{
@@ -458,12 +459,16 @@ test("preflight separates task admission, acceptance binding and runtime availab
     managed_executor: {executor: "dsh", available: true, unavailable_reason: null, execution_profile: "explicit-profile"}};
   const params = {binding, preview, validation_files_current: true, acceptance: {todo_id: "todo_review", state: "ready"}};
   assert.equal(delegationPreflight(params).state, "launchable");
+  assert.equal(delegationPreflight(params).dispatch_allowed, true);
   assert.equal(delegationPreflight(params).authority_state, "promoted");
   for (const [available, expected] of [[null, "runtime_unverified"], [false, "runtime_unavailable"]] as const) {
     assert.equal(delegationPreflight({...params, preview: {...preview,
       managed_executor: {...preview.managed_executor, available}}}).state, expected);
+    assert.equal(delegationPreflight({...params, preview: {...preview,
+      managed_executor: {...preview.managed_executor, available}}}).dispatch_allowed, available !== false);
   }
   assert.equal(delegationPreflight({...params, acceptance: null}).state, "acceptance_unavailable");
+  assert.equal(delegationPreflight({...params, acceptance: null}).dispatch_allowed, false);
   assert.equal(delegationPreflight({...params, acceptance: {...params.acceptance, state: "stale"}}).state, "acceptance_unavailable");
   assert.equal(delegationPreflight({...params, preview: {...preview, route: {...preview.route,
     selected_todo_id: "other"}}}).state, "turn_blocked");
@@ -480,6 +485,7 @@ test("preflight uses scoped native lane facts without inventing App presence", (
       managed_executor: {executor: "generic-cli", available: null, unavailable_reason: null, execution_profile: null}}};
   const original = delegationPreflight(params);
   assert.equal(original.state, "runtime_unverified");
+  assert.equal(original.dispatch_allowed, true);
   assert.equal(original.turn_lane, undefined);
   for (const state of ["absent", "released", "dead"]) {
     const {turn_lane, ...observed} = delegationPreflight({...params, lane: {state, holder: "withheld"}});
@@ -490,6 +496,7 @@ test("preflight uses scoped native lane facts without inventing App presence", (
     const observed = delegationPreflight({...params, lane: {state}});
     assert.equal(observed.state, "turn_blocked");
     assert.equal(observed.turn_eligible, false);
+    assert.equal(observed.dispatch_allowed, false);
     assert.equal(observed.reason, state === "live" ? "turn_lane_in_flight" : "turn_lane_state_unknown");
     assert.equal(observed.acceptance_ready, true);
     assert.deepEqual(observed.effects, original.effects);
@@ -586,6 +593,7 @@ test("preflight reports unavailable current authority without granting execution
   acceptance: null, validation_files_current: false});
   assert.equal(result.state, "authority_unavailable");
   assert.equal(result.authority_ready, false);
+  assert.equal(result.dispatch_allowed, false);
   assert.equal(result.authority_state, "unavailable");
   assert.equal(result.authority_next_action, "repair_canonical_authority");
   assert.equal(result.promotion_from_surface_allowed, false);
diff --git a/tests/extensions/test_lark_goal_channel_operation.py b/tests/extensions/test_lark_goal_channel_operation.py
index def1a9cfa0..085cfb5975 100644
--- a/tests/extensions/test_lark_goal_channel_operation.py
+++ b/tests/extensions/test_lark_goal_channel_operation.py
@@ -150,6 +150,9 @@ def spawn(self, operation_id):
         if fault == "dispatch":
             raise RuntimeError("private argv must not be returned")
     monkeypatch.setattr(Delegations, "_spawn", spawn)
+    # This wire fixture owns callback identity/replay, not native Turn admission.
+    # Real File/SQLite preflight and worker execution are covered separately.
+    monkeypatch.setattr(Delegations, "inspect", lambda self, binding_id: {"dispatch_allowed": True})
     cards = {}
     runner = _runner([], cards)
     deliver_goal_channel_operation_card(proposal_id=proposal["proposal_id"], action_store_root=store.root,
diff --git a/tests/test_manager_context_execution.py b/tests/test_manager_context_execution.py
index 78655c6c1b..866b0b5b41 100644
--- a/tests/test_manager_context_execution.py
+++ b/tests/test_manager_context_execution.py
@@ -64,7 +64,7 @@ def path(self, operation):
             return tmp_path / operation
 
         def inspect(self, binding_id):
-            return {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
+            return {"state": "launchable", "dispatch_allowed": True, "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
 
         def start(self, binding_id, operation, semantic_brief, **kwargs):
             started.append((binding_id, operation, semantic_brief, kwargs))
@@ -291,7 +291,7 @@ def preview(*_):
         else:
             policy["sources"][session["channel_id"]]["execution_bindings"] = []
             _write(_root(root) / "policy.json", policy)
-        return {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
+        return {"state": "launchable", "dispatch_allowed": True, "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
 
     monkeypatch.setattr(execution.Delegations, "inspect", preview)
     monkeypatch.setattr(execution.Delegations, "resume",
@@ -302,12 +302,13 @@ def preview(*_):
     assert len(started) == 1
 
 
-@pytest.mark.parametrize("missing", ["turn_eligible", "acceptance_ready", "authority_ready"])
-def test_prepared_recovery_requires_every_original_preflight_fact(flow, monkeypatch, missing):
+@pytest.mark.parametrize("decision", ["missing", None, False, "true", 1])
+def test_prepared_recovery_requires_current_typed_dispatch_decision(flow, monkeypatch, decision):
     dispatch(flow)
     monkeypatch.setattr(execution.Delegations, "read", lambda *_: {"status": "prepared"})
     facts = {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
-    facts.pop(missing)
+    if decision != "missing":
+        facts["dispatch_allowed"] = decision
     monkeypatch.setattr(execution.Delegations, "inspect", lambda *_: facts)
     monkeypatch.setattr(execution.Delegations, "resume",
         lambda *_args, **_kwargs: pytest.fail("an unavailable fact cannot permit recovery"), raising=False)
@@ -330,7 +331,7 @@ def test_stop_during_preview_and_context_only_selection_do_not_launch(flow, monk
     active = [True]
     def preview(*_):
         active[0] = False
-        return {"state": "launchable", "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
+        return {"state": "launchable", "dispatch_allowed": True, "turn_eligible": True, "acceptance_ready": True, "authority_ready": True}
     monkeypatch.setattr(execution.Delegations, "inspect", preview)
     result = execution.dispatch(root, registry, session=session, turn=turn, request=request, receipt=receipt,
                                 execution_allowed=lambda: active[0])

From e815abfef86c63c4ba2cb727b63dd2f1d4f0d440 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 08:59:00 +0800
Subject: [PATCH 5/9] fix(collaboration): resume saved source handoffs after
 native waits

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            |  77 ++++++----
 .../capabilities/manager_context/execution.py | 109 +++++++++++++-
 .../capabilities/manager_context/roundtrip.py |  21 ++-
 loopx/collaboration_mcp.py                    |   7 +-
 .../control_plane/collaboration/delegation.ts |  41 +++--
 .../collaboration/delegation_validation.py    |   6 +
 .../control_plane/effect_runtime_handlers.ts  |   1 +
 .../turn_driver/decision_capture.py           |  10 +-
 loopx/control_plane/turn_driver/executor.py   |  10 ++
 .../test_delegation_lane_readiness.py         |  36 +++--
 .../test_delegation_source_replay.py          | 141 ++++++++++++++++++
 tests/control_plane_ts/delegation.test.ts     |  33 ++++
 12 files changed, 438 insertions(+), 54 deletions(-)

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 04fce5d671..b4c6d06574 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -153,17 +153,29 @@ matching prepared operations may resume only under the restored current grant,
 and matching terminal operations are read back in place. Historical operator
 inspection remains available through the ordinary read command.
 
-A Chat handoff with a separate source execution grant can recover when its
-requester exits after saving the original operation but before starting the
-worker. Replaying that handoff rechecks the source Turn and exact grant, then
-rejoins only its original `prepared` operation. The worker retains this
-precondition across process startup under the existing execution lock. A busy
-worker is observed; an operation that has already reached `running`, returned
-or stopped is not automatically resumed by this replay. Uncertain Host work
-still needs explicit original-operation reconciliation. Recovery preserves the
-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 Chat handoff with a separate source execution grant saves its original
+`prepared` operation even when an occupied or unknown native lane prevents the
+first submission, provided the other current admission, acceptance and runtime
+checks permit deferral. Saving that intent launches no worker or Host. The
+existing Chat return service also resumes these saved source operations after
+the requester finishes its Chat Turn; no sender retry is needed. It considers
+at most 20 local records and one full native preflight per 30-second probe,
+advancing its cursor past refused records. Known busy or unknown lanes use only
+the scoped liveness observation. Source context, grant, binding, full brief,
+return conversation and current native preflight are rechecked before resuming.
+Restart resets only the host probe cursor and delay, never the operation. A
+slow or failed execution preview cannot block the service's separate return IO;
+shutdown is checked again after preview, before offering a worker resume.
+
+Only the original source Chat host can offer that saved operation. Context-only
+Inbox rows and operator/MCP operations do not enroll; removing the source grant,
+closing its session, stopping work or exhausting quota still prevents execution.
+The worker retains the prepared-source condition under its existing execution
+lock across process startup. A busy worker is observed; `running`, returned,
+stopped or uncertain work is not automatically resumed. Explicit reconciliation
+still owns uncertain Host work. Recovery creates no replacement task, grant,
+queue 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,
@@ -175,12 +187,17 @@ 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
+observation owner keeps it `prepared` only for the native `turn_lane_in_flight`
+refusal or a `minimum_interval_wait` with an explicit unreserved native
+admission, both 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.
+The Turn validates its original capture before interval admission and publishes
+it only after admission; a refused start must not freeze a changed observation
+without a Turn journal. Existing captures remain immutable and fully checked.
+Automatic offers cover only the saved Chat-source `prepared` operations above;
+other operations keep their existing explicit recovery.
 
 中文：Chat 发起的执行保留原 Inbox 请求、固定的对话来源和原 operation。worker 在
 入口、Host 启动前和完成结算前重新核对当前发送者、接收者及精确
@@ -192,13 +209,21 @@ brief、peer request、当前 binding 和原回传 Session/Turn；选择另一
 不能改派既有 operation，身份不符时返回原执行不可用反馈且不新发请求。身份一致的
 prepared operation 可在恢复的当前 grant 下续接原 owner，终态 operation 原位读回。
 
-中文：另有来源执行授权的 Chat 交接，在原 operation 已保存、worker 尚未启动时
-请求方退出，可重放同一交接恢复。重放重新核对来源 Turn 与精确 grant，只续接原
-`prepared` operation；worker 启动后仍在既有执行锁内核对同一前置条件。忙碌 worker
-仅被观察，已进入 `running`、已返回或已停止的执行不因该重放自动恢复；Host 结果
-不确定时仍须显式对账原执行。原 Inbox 请求、binding、对话与受治理 Turn 均保留，
-不创建替代任务、授权或调度器。这仍是本文的精确任务绑定路径，不代表普通 Inbox
-空闲接收者的自动激活已完成。
+中文：另有来源执行授权的 Chat 交接，首次提交因原生 lane 忙碌或未知而受阻、但
+其他当前准入、验收和运行时检查允许延期时，也会保存原 `prepared` operation，
+不启动 worker 或 Host。现有 Chat 返回服务在请求方结束本轮后自动续接该操作，
+无需发送方重试。每 30 秒最多读 20 条本地记录、做一次完整原生预检，游标越过
+受阻记录；已知忙碌或未知 lane 只读其作用域内存活事实。续接前重核来源、grant、
+binding、完整 brief、原回传对话及当前原生预检；重启只重置宿主的探测游标和间隔，
+不更换原操作。执行预检缓慢或失败不阻塞独立的结果回传 IO；预检后、续接 worker
+之前再次检查服务是否关闭。
+
+只有原来源 Chat 宿主可以提交该已保存操作。context-only Inbox 和 operator/MCP
+操作不加入自动续接；撤权、关闭来源会话、停止或额度不足仍阻止执行。worker 启动
+后在既有执行锁内核对 prepared-source 条件。忙碌 worker 仅被观察，`running`、
+已返回、已停止或不确定的执行不自动恢复；Host 结果不确定时仍须显式对账原执行。
+不创建替代任务、授权、队列或调度器。这仍是本文的精确任务绑定路径，不代表普通
+Inbox 空闲接收者的自动激活已完成。
 
 仍为 `prepared` 的重放也执行首次提交所用的当前原生预检。Turn lane 忙碌或未知、
 任务准入拒绝、运行时不可用或当前验收依据缺失时，原 operation 保持 `prepared`，
@@ -206,11 +231,13 @@ prepared operation 可在恢复的当前 grant 下续接原 owner，终态 opera
 prepared-source 启动条件下续接同一 operation；已推进的 operation 直接读回，
 不按新启动重查其 Todo。预检仍只是当时观察，之后的竞争由原生启动 fence 拒绝。
 对精确任务绑定的 CLI、MCP 和 Chat worker，原 operation 尚无 Turn journal，
-且原生 `turn_lane_in_flight` 结果明确证明未调用
-Host、未写 Turn 状态、未扣额度、未 ACK scheduler 时，类型化观察 owner 将其保留为
+且原生 `turn_lane_in_flight` 拒绝或带明确未预留准入的 `minimum_interval_wait`
+结果证明未调用 Host、未写 Turn 状态、未扣额度、未 ACK scheduler 时，类型化观察 owner 将其保留为
 `prepared` 并读回原原因；之后获授权的重试沿用原请求、operation 和租约身份。
 已有 journal、已有副作用或结果未知均沿原恢复流程，不回退执行状态。
-本改动不新增自动重试或唤醒循环。
+Turn 在间隔准入前校验原捕获，通过准入后才发布；未准入的启动不能留下没有
+Turn journal 的旧观察，阻断之后的原身份重试。已有捕获仍不可变且完整核对。
+自动续接只覆盖上述已保存的 Chat-source `prepared` 操作；其他操作保留既有显式恢复。
 
 Member completion uses the ordinary active-Goal continuation, including legacy
 non-hard-lease routes. It does not declare terminal `no_followup` for a
diff --git a/loopx/capabilities/manager_context/execution.py b/loopx/capabilities/manager_context/execution.py
index 0208f5ba3f..990236b2b4 100644
--- a/loopx/capabilities/manager_context/execution.py
+++ b/loopx/capabilities/manager_context/execution.py
@@ -5,8 +5,10 @@
 """
 
 from pathlib import Path
-from collections.abc import Callable
-from typing import Any, TYPE_CHECKING
+from time import monotonic
+from collections.abc import Callable, Iterator
+from typing import Any, TYPE_CHECKING, cast
+import logging
 
 if TYPE_CHECKING:
     from ...chat_store import ChatSessionStore
@@ -14,10 +16,12 @@
 from ...agent_registry import load_goal_from_registry
 from ...collaboration_mcp import Delegations
 from ...control_plane.collaboration import conversation_scope
+from ...control_plane.collaboration.inbox import _root
+from ...control_plane.content_digest import BARE_SHA256_PATTERN
 from ...control_plane.collaboration.source_grant_observation import (
     read_delegation_source_execution, source_execution_bindings,
 )
-from ...control_plane.effect_runtime import EffectRuntimeRejected
+from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ...orchestration import compact_orchestration_policy, normalize_subagent_execution_config
 
 
@@ -109,6 +113,14 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
         # must stop before dispatch. Actual launch and acceptance remain owned
         # by the governed Turn, not this point-in-time preview.
         if preflight.get("dispatch_allowed") is not True:
+            if not replayed and preflight.get("queueable") is True:
+                if not execution_allowed() or grant not in _grants(root, registry, session, turn):
+                    raise ValueError("source execution grant changed before queueing")
+                result = service.start(selected, operation, request["brief"],
+                    conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]},
+                    source_request_id=receipt["request_id"], _defer_worker=True)
+                saved = {"operation_id": operation, "todo_id": binding["todo_id"],
+                         "status": result["status"], "replayed": False}
             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.
@@ -129,6 +141,97 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
         return {"submitted": False, "reason": "execution_binding_or_admission_unavailable"}
 
 
+class QueuedSourcePump:
+    """Bounded recovery inside the existing Chat service, not a work scheduler.
+
+    Only already-persisted Chat execution intents are candidates. A context-only
+    Inbox row, operator/MCP operation or uncertain started Turn cannot enroll.
+    The iterator and retry delay are host IO state; admission/once-only effects
+    remain with the existing source, preflight and native Turn owners.
+    """
+
+    def __init__(self, root: Path, registry: Path, store: "ChatSessionStore") -> None:
+        self.root, self.registry, self.store = root, registry, store
+        self.paths: Iterator[Path] | None = None
+        self.next_probe = 0.0
+
+    def tick(self, *, cancelled: Callable[[], bool] = lambda: False) -> None:
+        if cancelled() or monotonic() < self.next_probe:
+            return
+        self.next_probe = monotonic() + 30
+        if self.paths is None:
+            self.paths = (_root(self.root) / "executions").glob("*/*.json")
+        # Advance the cursor past refused/unavailable rows so one request cannot
+        # starve another. At most 20 local records and one full native preflight.
+        for _ in range(20):
+            if cancelled():
+                break
+            path = next(self.paths, None)
+            if path is None:
+                self.paths = None
+                break
+            try:
+                if self._resume(path, cancelled):
+                    break
+            except (OSError, ValueError, KeyError, TypeError, RuntimeError):
+                # Keep the original intent. Unknown/revoked sources neither
+                # become empty nor create another request or execution.
+                continue
+
+    def _resume(self, path: Path, cancelled: Callable[[], bool]) -> bool:
+        from ...control_plane.collaboration.inbox import _entry, _read
+        from ...control_plane.collaboration.goal_instance_scope import collaboration_goal_scope
+        from ...control_plane.collaboration.source_chat_observation import is_source_chat_host, observe_source_chat
+        from ...control_plane.turn_driver.lane_fence import turn_lane_liveness
+
+        row = _read(path)
+        identity = row.get("identity", {})
+        source_id = identity.get("source_request_id") if isinstance(identity, dict) else None
+        if (row.get("status") != "prepared" or not isinstance(source_id, str)
+                or not BARE_SHA256_PATTERN.fullmatch(source_id) or path.is_symlink()
+                or not BARE_SHA256_PATTERN.fullmatch(path.stem)):
+            return False
+        route = _read(_root(self.root) / "roundtrips" / (source_id + ".json"))
+        if not is_source_chat_host(self.root, route, self.store):
+            return False
+        session, turn = observe_source_chat(self.root, route)
+        target = {key: route[key] for key in ("goal_id", "agent_id")}
+        for grant in _grants(self.root, self.registry, session, turn):
+            if (any(grant[key] != value for key, value in target.items())
+                    or grant["binding_id"] != identity["binding"]["id"]):
+                continue
+            service, binding = _service(self.root, self.registry, grant)
+            operation = "context-" + source_id
+            if service.path(operation) != path or service.read(operation)["worker_active"]:
+                continue
+            with collaboration_goal_scope(self.registry, goal_id=target["goal_id"], agents=(),
+                    caller_goal_ref=route.get("goal_ref")) as scope:
+                original = _entry(self.root, target["goal_id"], target["agent_id"], source_id, scope=scope)
+            conversation = {"session_id": session["session_id"], "turn_id": turn["turn_id"]}
+            def observe() -> dict[str, Any]:
+                return cast(dict[str, Any], read_delegation_source_execution(service, binding["id"], operation,
+                    source_request_id=source_id, brief=original["brief"], conversation=conversation))
+            if observe()["status"] != "prepared":
+                return False
+            # A cheap scoped observation avoids native CLI/model polling while
+            # another Turn still owns this recipient's lane, or it is unknown.
+            lane = turn_lane_liveness(service._lane_target(binding))
+            if effect_runtime_result("collaboration.delegation.lane_readiness", {"lane": {"state": lane["state"]}})["available"] is not True:
+                return False
+            if cancelled():
+                return False
+            try:
+                preflight = service.inspect(binding["id"])
+                if preflight.get("dispatch_allowed") is True and not cancelled() and observe()["status"] == "prepared":
+                    service.resume(operation, prepared_source_request_id=source_id)
+            except (OSError, ValueError, KeyError, TypeError, RuntimeError):
+                # A failed full preview still consumes this probe's one
+                # attempt. Keep its intent and let the next cursor run proceed.
+                logging.getLogger(__name__).warning("Queued source preflight unavailable; retaining original intent")
+            return True
+        return False
+
+
 def handoff_message(
     receipt: dict[str, Any],
     execution: dict[str, Any],
diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py
index 7172a8288a..9308e9ea8b 100644
--- a/loopx/capabilities/manager_context/roundtrip.py
+++ b/loopx/capabilities/manager_context/roundtrip.py
@@ -584,9 +584,13 @@ def __init__(self, root, registry, store, external_sender):
         self.thread = threading.Thread(
             target=self.run, daemon=True, name="loopx-manager-returns"
         )
+        self.execution_thread = threading.Thread(
+            target=self.run_executions, daemon=True, name="loopx-manager-executions"
+        )
 
     def start(self) -> None:
         self.thread.start()
+        self.execution_thread.start()
 
     def run(self):
         try:
@@ -602,7 +606,20 @@ def run(self):
                 )
             self.stop.wait(3)
 
+    def run_executions(self) -> None:
+        # Native previews may wait on a subprocess. Keep result publication
+        # independent, using the same server lifetime and one serial probe.
+        from .execution import QueuedSourcePump
+        execution_pump = QueuedSourcePump(*self.args[:3])
+        while not self.stop.is_set():
+            try:
+                execution_pump.tick(cancelled=self.stop.is_set)
+            except (OSError, ValueError, KeyError, TypeError, RuntimeError):
+                logging.getLogger(__name__).warning("Queued source execution unavailable; retrying")
+            self.stop.wait(3)
+
     def close(self) -> None:
         self.stop.set()
-        if self.thread.ident is not None:
-            self.thread.join(timeout=3)
+        for thread in (self.thread, self.execution_thread):
+            if thread.ident is not None:
+                thread.join(timeout=1.5)
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 56a607049c..f635d57e82 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -705,7 +705,7 @@ def authority_fault(exc: OSError | ValueError) -> dict[str, object]:
     def start(self, binding_id: str, operation_id: str, brief: dict,
               parent_request_id: str | None = None, *, conversation: dict | None = None,
               confirmed_operation_id: str | None = None,
-              source_request_id: str | None = None) -> dict:
+              source_request_id: str | None = None, _defer_worker: bool = False) -> dict:
         """Start or replay one bound operation.
 
         ``conversation`` is supplied only by the trusted Chat host, never by
@@ -714,6 +714,8 @@ def start(self, binding_id: str, operation_id: str, brief: dict,
         that conversation and no other. ``source_request_id`` is also host-only:
         the receiver's existing inbox request, separate from the user brief.
         """
+        if _defer_worker and source_request_id is None:
+            raise ValueError("deferred execution requires the original Chat source")
         binding = self.binding(binding_id, require_active=True)
         require_operation_id(operation_id)
         brief = normalize_request({"goal_id": self.goal_id, "agent_id": binding["agent_id"], "brief": brief})["brief"]
@@ -753,7 +755,8 @@ def start(self, binding_id: str, operation_id: str, brief: dict,
                 prepare_host_process_record(self._host_process_record(path))
                 _write(path, {"identity": identity, "status": "prepared", "created_at": time.time(),
                               **({"conversation": origin} if origin else {})})
-                self._spawn(operation_id)
+                if not _defer_worker:
+                    self._spawn(operation_id)
         return self.read(operation_id)
 
     def _spawn(self, operation_id: str, *, prepared_source_request_id: str | None = None) -> None:
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index f832aa6745..7d41d8f0d4 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -272,6 +272,15 @@ function delegationAcceptanceFacts(todoId: unknown, acceptance: JsonObject | nul
     acceptance_next_action: "review_original_task_acceptance"};
 }
 
+/** Shared interpretation of the native lock owner's read-only liveness facts. */
+export function delegationLaneReadiness(params: JsonObject): JsonObject {
+  const lane = requireJsonObject(params.lane, "Turn lane observation");
+  requireThat(typeof lane.state === "string"
+    && ["live", "released", "dead", "absent", "foreign_host", "unreadable"].includes(lane.state),
+  "invalid Turn lane observation");
+  return {available: ["released", "dead", "absent"].includes(lane.state), state: lane.state};
+}
+
 /** Read the actual dry-run route/profile, never infer readiness from assignment. */
 export function delegationPreflight(params: JsonObject): JsonObject {
   const binding = requireJsonObject(params.binding, "binding identity");
@@ -369,11 +378,8 @@ export function delegationPreflight(params: JsonObject): JsonObject {
   // Reuse the file-lock owner's liveness facts. These describe only the native
   // Turn lane, never the presence/availability of an App or CLI conversation.
   // Omission preserves earlier callers; an explicit unknown fails closed.
-  const lane = params.lane === undefined ? null : requireJsonObject(params.lane, "Turn lane observation");
-  if (lane !== null) requireThat(typeof lane.state === "string"
-    && ["live", "released", "dead", "absent", "foreign_host", "unreadable"].includes(lane.state),
-  "invalid Turn lane observation");
-  const laneBlocked = lane !== null && !["released", "dead", "absent"].includes(String(lane.state));
+  const lane = params.lane === undefined ? null : delegationLaneReadiness(params);
+  const laneBlocked = lane !== null && lane.available !== true;
   const eligible = route.would_invoke_host === true && route.selected_todo_id === binding.todo_id && !laneBlocked;
   const acceptance = params.acceptance === null ? null : requireJsonObject(params.acceptance, "task acceptance");
   const acceptanceFacts = delegationAcceptanceFacts(binding.todo_id, acceptance, params.validation_files_current);
@@ -384,6 +390,10 @@ export function delegationPreflight(params: JsonObject): JsonObject {
   return {
     schema_version: "loopx_delegation_preflight_v0", binding,
     state, dispatch_allowed: state === "launchable" || state === "runtime_unverified",
+    // Only lane contention may defer an otherwise admissible source launch.
+    // This permits saving its original intent, never a Host invocation.
+    queueable: laneBlocked && route.would_invoke_host === true
+      && route.selected_todo_id === binding.todo_id && pinned && executor.available !== false,
     turn_eligible: eligible, turn_route: route.kind,
     ...acceptanceFacts, authority_ready: true, authority_reason: null,
     authority_state: "promoted", authority_next_action: "none",
@@ -522,19 +532,30 @@ 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.
+  // A native lane race or owner interval wait before this operation has a
+  // Turn retains its original intent. Partial/resumed/unknown Turns keep normal
+  // recovery; only an exact, effect-free native refusal can return to prepared.
   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;
+    const admission = result.admission as JsonObject | undefined;
+    const laneWait = result.status === "unavailable" && result.reason === "turn_lane_in_flight"
+      && host?.executable === "not_invoked";
+    const intervalWait = result.status === "interval_wait" && result.reason === "minimum_interval_wait"
+      && admission?.reason === "minimum_interval_wait" && admission.admitted === false
+      && admission.reserved === false && admission.resumed !== true
+      && admission.pre_model_admission === "managed_turn_only"
+      && typeof admission.min_interval_minutes === "number" && Number.isSafeInteger(admission.min_interval_minutes)
+      && admission.min_interval_minutes > 0
+      && typeof admission.next_eligible_at_ms === "number" && Number.isSafeInteger(admission.next_eligible_at_ms)
+      && admission.next_eligible_at_ms > 0;
     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"
+        && (laneWait || intervalWait)
         && result.result_kind === null && result.quota_slot_spend_count === 0
-        && host?.executable === "not_invoked" && effects !== undefined && effects !== null
+        && 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"};
diff --git a/loopx/control_plane/collaboration/delegation_validation.py b/loopx/control_plane/collaboration/delegation_validation.py
index f30145b010..bd9e4c776a 100644
--- a/loopx/control_plane/collaboration/delegation_validation.py
+++ b/loopx/control_plane/collaboration/delegation_validation.py
@@ -151,6 +151,12 @@ def record_turn_result(
         )}
         host = result.get("host")
         observation["host"] = {"executable": host.get("executable")} if isinstance(host, dict) else None
+        admission = result.get("admission")
+        if isinstance(admission, dict):
+            observation["admission"] = {key: admission.get(key) for key in (
+                "admitted", "reserved", "resumed", "reason", "next_eligible_at_ms",
+                "min_interval_minutes", "pre_model_admission",
+            )}
         decision = effect_runtime_result("collaboration.delegation.observe", {
             "from": row["status"], "to": "turn_returned",
             "turn_execution": observation, "unstarted_turn": unstarted_turn,
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 44044e12cb..69ea835de4 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -611,6 +611,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.delegation.source_replay", lazyHandler(() => import("./collaboration/delegation.ts"), ({verifyDelegationSourceReplay}) => verifyDelegationSourceReplay)],
     ["turn.selection.rejection", lazyHandler(() => import("./turn_driver/selection_rejection.ts"), ({projectTurnSelectionRejection}) => projectTurnSelectionRejection)],
     ["collaboration.delegation.preflight", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationPreflight}) => delegationPreflight)],
+    ["collaboration.delegation.lane_readiness", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationLaneReadiness}) => delegationLaneReadiness)],
     ["collaboration.delegation.validation_plan", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationValidationPlan}) => delegationValidationPlan)],
     ["collaboration.delegation.checked_artifacts", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationCheckedArtifacts}) => delegationCheckedArtifacts)],
     ["collaboration.delegation.turn_plan", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationTurnPlanDecision}) => delegationTurnPlanDecision)],
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 1f6f7184fe..da9c913eb0 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -35,9 +35,13 @@ def write_decision_capture(
 
 def prepare_turn_decision_capture(
     plan: Mapping[str, Any], *, journal_path: Path,
-    decision: Mapping[str, Any] | None, existing_journal: bool,
+    decision: Mapping[str, Any] | None, existing_journal: bool, persist: bool = True,
 ) -> dict[str, Any]:
-    """Called under the Turn journal lock before Host/admission effects.
+    """Called under the Turn journal lock before Host effects.
+
+    ``persist=False`` validates the exact source before cadence admission without
+    freezing a refused, unstarted observation. After admission the caller must
+    run the normal publication path before giving that plan to the Host.
 
     Existing journals must retain their original reference. A missing legacy
     observation cannot be reconstructed from today's guard. Already committed
@@ -71,7 +75,7 @@ def prepare_turn_decision_capture(
         # Legacy journals retain their recorded delivery/recovery route.
         reader_argv=current_loopx_cli_argv() if not existing_journal else None,
     )
-    if not path.exists():
+    if persist and not path.exists():
         path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
         write_decision_capture(path, observation)
     return {**plan, "turn_envelope": bound}
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 0514370f88..745301845a 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -1621,6 +1621,7 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
             plan = prepare_turn_decision_capture(
                 original_plan, journal_path=journal_path,
                 decision=decision_capture_payload, existing_journal=journal is not None,
+                persist=admit_start is None,
             )
             # Preserve any session recovery adjustment; only the original
             # observation replaces the fresh request's context.
@@ -1648,6 +1649,15 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 return execution_payload(
                     plan, waiting, execute=True, replayed=False, effects=empty_effects
                 )
+        if require_decision_capture and needs_host and admit_start is not None:
+            # Validate before reserving cadence/settlement identity, but publish
+            # only an admitted observation. Otherwise an interval refusal leaves
+            # a capture without a journal which poisons the same Turn's retry.
+            plan = prepare_turn_decision_capture(
+                original_plan, journal_path=journal_path,
+                decision=decision_capture_payload, existing_journal=journal is not None,
+            )
+            request = {**request, "turn_envelope": plan["turn_envelope"]}
         if journal is not None and recovery_decision is not None:
             journal["recovery_audit"] = build_turn_recovery_audit(
                 recovery_decision,
diff --git a/tests/control_plane/test_delegation_lane_readiness.py b/tests/control_plane/test_delegation_lane_readiness.py
index ac5fa6bf51..741f1c2e8d 100644
--- a/tests/control_plane/test_delegation_lane_readiness.py
+++ b/tests/control_plane/test_delegation_lane_readiness.py
@@ -1,6 +1,7 @@
 """Delegation previews must not mistake an occupied native Turn lane for idle."""
 
 import json
+import time
 from contextlib import ExitStack
 from pathlib import Path
 
@@ -9,7 +10,7 @@
 from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight
 from loopx.file_lock import lock_holder_path
 from loopx.capabilities.manager_context import execution
-from loopx.capabilities.manager_context.roundtrip import drain
+from loopx.capabilities.manager_context.roundtrip import ReturnService, drain
 from test_delegation_cli import cli
 from test_independent_delegation_validation import independent_binding
 from test_local_delegation import demo, service, wait  # noqa: F401
@@ -92,7 +93,7 @@ def test_another_members_lane_does_not_block_this_binding(service):  # noqa: F81
         assert preview["turn_lane"] == {"scope": "native_turn_lane", "state": "absent"}
 
 
-def test_busy_handoff_keeps_original_context_then_returns_once_after_release(service):  # noqa: F811
+def test_busy_handoff_resumes_after_source_completion_without_another_dispatch(service):  # noqa: F811
     root, runner = service
     independent_binding(service)
     data = json.loads(runner.registry.read_text())
@@ -126,13 +127,12 @@ def dispatch():
         for _ in range(2):
             blocked = dispatch()
             assert not blocked["submitted"] and blocked["preflight"]["reason"] == "turn_lane_in_flight"
-        assert not runner.path(operation).exists()
+        assert runner.path(operation).is_file()
+        queued = json.loads(runner.path(operation).read_text())
+        assert queued["status"] == "prepared"
+        assert queued["identity"]["source_request_id"] == receipt["request_id"]
+        assert queued["conversation"] == {"session_id": session["session_id"], "turn_id": turn["turn_id"]}
         assert not (root / "analyst/initial/host-invocations").exists()
-    started = dispatch()
-    worker = execution._service(runner.root, runner.registry, {"goal_id": runner.goal_id,
-        "agent_id": "analyst", "requester_agent_id": "lead", "binding_id": "analysis"})[0]
-    assert started["submitted"] and wait(worker, operation)["status"] == "accepted"
-    assert dispatch()["operation_id"] == operation
     store.update_turn(session["session_id"], turn["turn_id"], status="completing",
         response={"context_handoff_receipt": receipt})
     store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
@@ -142,7 +142,25 @@ def transport(route, *_args, **_kwargs):
         sends.append(route["session_id"])
         return {"message_id": "fixture-return", "reply_verified": True}
 
-    drain(runner.root, runner.registry, store, transport)
+    # The actual existing background service must recover the saved intent.
+    # No sender retry, fresh handoff, new Task or model selection supplies it.
+    background = ReturnService(runner.root, runner.registry, store, transport)
+    try:
+        background.start()
+        worker = execution._service(runner.root, runner.registry, {"goal_id": runner.goal_id,
+            "agent_id": "analyst", "requester_agent_id": "lead", "binding_id": "analysis"})[0]
+        assert wait(worker, operation)["status"] == "accepted"
+        deadline = time.monotonic() + 15
+        while not sends and time.monotonic() < deadline:
+            time.sleep(.1)
+    finally:
+        background.close()
     drain(runner.root, runner.registry, store, transport)
     assert sends == [session["session_id"]]
     assert (root / "analyst/initial/host-invocations").read_text() == "1"
+    completed = json.loads(worker.path(operation).read_text())
+    assert completed["identity"] == queued["identity"]
+    assert completed["conversation"] == queued["conversation"]
+    journals = list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert len(journals) == 1
+    assert json.loads(journals[0].read_text())["completed_phases"].count("quota_spend") == 1
diff --git a/tests/control_plane/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
index b8a7219a7f..360867c3e7 100644
--- a/tests/control_plane/test_delegation_source_replay.py
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -8,6 +8,8 @@
 import json
 from pathlib import Path
 import subprocess
+import threading
+import time
 
 import pytest
 
@@ -27,6 +29,145 @@ def durable_bytes(root):
             if p.is_file() and not p.name.endswith(".lock")}
 
 
+@pytest.mark.parametrize("change", ["revoked", "closed", "other_host", "context_only", "advanced", "cancelled"])
+def test_background_source_recovery_cannot_create_or_retarget_execution(saved_handoff, monkeypatch, change):  # noqa: F811
+    root, worker, binding, (store, session, turn, _, receipt, policy), operation = saved_handoff
+    if change == "revoked":
+        policy["sources"][session["channel_id"]]["execution_bindings"] = []
+        _write(_root(worker.root) / "policy.json", policy)
+    elif change == "closed":
+        store.update_session(session["session_id"], status="closed")
+    elif change == "other_host":
+        from loopx.chat_store import ChatSessionStore
+        store = ChatSessionStore(root / "another-chat-host")
+    elif change == "context_only":
+        worker.path(operation).unlink()  # A saved Inbox alone is never launch intent.
+    elif change == "advanced":
+        row = _read(worker.path(operation))
+        row["status"] = "running"
+        _write(worker.path(operation), row)
+    calls = []
+    monkeypatch.setattr(execution.Delegations, "_spawn", lambda *_args, **kwargs: calls.append(kwargs))
+    before, tasks = durable_bytes(worker.root), demo.canonical_tasks(root)
+    pump = execution.QueuedSourcePump(worker.root, worker.registry, store)
+    pump.tick(cancelled=lambda: change == "cancelled")
+    assert calls == []
+    assert durable_bytes(worker.root) == before
+    assert demo.canonical_tasks(root) == tasks
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+
+
+@pytest.mark.parametrize("preview", ["refused", "error"])
+def test_background_source_recovery_rechecks_current_preflight_and_its_bounded_retry(saved_handoff, monkeypatch, preview):  # noqa: F811
+    root, worker, binding, (store, _, _, _, _, _), operation = saved_handoff
+    calls = []
+    monkeypatch.setattr(execution.Delegations, "_spawn", lambda *_args, **kwargs: calls.append(kwargs))
+    clock = [0.0]
+    monkeypatch.setattr(execution, "monotonic", lambda: clock[0])
+    inspect = execution.Delegations.inspect
+    observations = []
+    def blocked(self, selected):
+        observations.append(selected)
+        if preview == "error":
+            raise RuntimeError("native preview unavailable")
+        return {**inspect(self, selected), "dispatch_allowed": False}
+    monkeypatch.setattr(execution.Delegations, "inspect", blocked)
+    pump = execution.QueuedSourcePump(worker.root, worker.registry, store)
+    # Multiple candidates cannot turn a failed full preview into repeated IO
+    # in one probe. Use the real retained source at every candidate address.
+    pump.paths = iter([worker.path(operation)] * 3)
+    pump.tick()
+    pump.tick()
+    assert len(observations) == 1 and calls == []
+    assert _read(worker.path(operation))["status"] == "prepared"
+    monkeypatch.setattr(execution.Delegations, "inspect", inspect)
+    # A new service after restart may inspect immediately, but recovers only
+    # the original prepared operation and carries its worker-entry CAS.
+    restarted = execution.QueuedSourcePump(worker.root, worker.registry, store)
+    restarted.tick()
+    assert calls == [{"prepared_source_request_id": _read(worker.path(operation))["identity"]["source_request_id"]}]
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+
+
+def test_slow_queue_preview_does_not_block_original_return_or_server_close(saved_handoff, monkeypatch):  # noqa: F811
+    from loopx.capabilities.manager_context.roundtrip import ReturnService, report
+    from loopx.capabilities.manager_context import acknowledge
+
+    _, worker, _, (store, session, turn, _, receipt, _), operation = saved_handoff
+    store.update_turn(session["session_id"], turn["turn_id"], status="completing",
+        response={"context_handoff_receipt": receipt})
+    store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    entered, release, sent = threading.Event(), threading.Event(), threading.Event()
+    launched = []
+    def slow_preview(*_):
+        entered.set()
+        assert release.wait(15)
+        return {"dispatch_allowed": True}
+    monkeypatch.setattr(execution.Delegations, "inspect", slow_preview)
+    monkeypatch.setattr(execution.Delegations, "_spawn", lambda *args, **kwargs: launched.append(args))
+    def transport(*_):
+        sent.set()
+        return {"reply_verified": True, "idempotency_key": "fixture-provider-return"}
+    background = ReturnService(worker.root, worker.registry, store, transport)
+    try:
+        background.start()
+        assert entered.wait(5)
+        acknowledge(worker.root, worker.goal_id, "analyst", receipt["request_id"], "no_change", "Reviewed original")
+        report(worker.root, worker.goal_id, "analyst", receipt["request_id"], "conclusion", "No execution needed")
+        assert sent.wait(8), "An unrelated slow execution preview blocked the completed source return"
+        started = time.monotonic()
+        background.close()
+        assert time.monotonic() - started < 4
+    finally:
+        background.close()
+        release.set()
+        background.execution_thread.join(timeout=5)
+    assert not launched
+    assert not background.thread.is_alive() and not background.execution_thread.is_alive()
+    assert _read(worker.path(operation))["status"] == "prepared"
+
+
+def test_unstarted_native_cadence_wait_retains_source_until_its_original_interval(saved_handoff):  # noqa: F811
+    from loopx.control_plane.effect_runtime import effect_runtime_result
+
+    root, worker, binding, (_, _, _, _, receipt, _), operation = saved_handoff
+    scope = {"runtime_root": str(worker.root), "goal_id": worker.goal_id,
+        "agent_id": binding["agent_id"], "automation_id": None}
+    effect_runtime_result("quota.automation_cadence.manage", {**scope, "operation": "configure",
+        "expected_revision": 0, "min_interval_minutes": 1,
+        "owner_reference": "fixture-owner-cadence", "execute": True})
+    # Seed a prior native reservation in this disposable runtime, retaining a
+    # real interval with enough time for the first worker to observe its wait.
+    anchor = time.time_ns() // 1_000_000 - 40_000
+    previous = {**scope, "request_id": "fixture-preceding-turn", "now_ms": anchor, "trigger_at_ms": anchor}
+    assert effect_runtime_result("quota.automation_cadence.admit", previous)["admitted"]
+    effect_runtime_result("quota.automation_cadence.confirm_start", previous)
+    original, tasks = _read(worker.path(operation)), demo.canonical_tasks(root)
+    worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    waiting = _read(worker.path(operation))
+    assert waiting["turn_result"]["reason"] == "minimum_interval_wait", waiting
+    assert waiting["status"] == "prepared", waiting
+    assert "turn_key" not in waiting
+    assert waiting["identity"] == original["identity"]
+    assert demo.canonical_tasks(root) == tasks
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    # Wait for the real floor; do not rewrite a start receipt or bypass policy.
+    assert not list((worker.root / "goals" / worker.goal_id / "turns" / "decisions").glob("*.json"))
+    due = effect_runtime_result("quota.automation_cadence.manage", {**scope, "operation": "read"})["next_eligible_at_ms"]
+    time.sleep(max(0, (due - time.time_ns() // 1_000_000) / 1000) + .02)
+    worker.execute(operation, prepared_source_request_id=receipt["request_id"])
+    accepted = _read(worker.path(operation))
+    assert accepted["status"] == "accepted", accepted
+    assert accepted["identity"] == original["identity"]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    journals = list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert len(journals) == 1
+    assert json.loads(journals[0].read_text())["completed_phases"].count("quota_spend") == 1
+
+
 @pytest.mark.parametrize("change", ["binding", "brief", "source_request", "peer_request", "conversation"])
 def test_saved_handoff_replay_cannot_substitute_its_original_identity(saved_handoff, change):  # noqa: F811
     _, worker, binding, (_, session, turn, request, receipt, policy), operation = saved_handoff
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index c6dbd49471..d0682686c1 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -248,6 +248,31 @@ test("only an unstarted native lane refusal keeps delegation queued", () => {
   assert.throws(() => transitionDelegationObservation({...input, to: "prepared"}), /transition/);
 });
 
+test("native interval waits retain only an unstarted effect-free operation", () => {
+  const effects = {host_invoked: false, state_written: false, quota_spent: false,
+    scheduler_acknowledged: false};
+  const admission = {admitted: false, reserved: false, resumed: null, reason: "minimum_interval_wait",
+    min_interval_minutes: 3, next_eligible_at_ms: 180_001, pre_model_admission: "managed_turn_only"};
+  const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false,
+    dry_run: false, replayed: false, status: "interval_wait", reason: "minimum_interval_wait",
+    result_kind: null, quota_slot_spend_count: 0, host: {executable: "configured-host"}, effects, admission};
+  const input = {from: "running", to: "turn_returned", unstarted_turn: true, turn_execution: result};
+  assert.deepEqual(transitionDelegationObservation(input), {status: "prepared"});
+  for (const unstarted_turn of [false, undefined])
+    assert.deepEqual(transitionDelegationObservation({...input, unstarted_turn}), {status: "turn_returned"});
+  for (const patch of [
+    {admitted: true}, {admitted: null}, {reserved: true}, {reserved: null}, {resumed: true},
+    {reason: "duplicate_or_stale_trigger"}, {pre_model_admission: "not_qualified"},
+    ...[null, 0, -1, "3", 1.5].map(min_interval_minutes => ({min_interval_minutes})),
+    ...[null, 0, -1, "180001", 1.5].map(next_eligible_at_ms => ({next_eligible_at_ms})),
+  ]) assert.deepEqual(transitionDelegationObservation({...input,
+    turn_execution: {...result, admission: {...admission, ...patch}}}), {status: "turn_returned"});
+  for (const patch of [{admission: null}, {reason: "duplicate_or_stale_trigger"},
+    {replayed: true}, {quota_slot_spend_count: 1}, {effects: {...effects, state_written: true}}])
+    assert.deepEqual(transitionDelegationObservation({...input, turn_execution: {...result, ...patch}}),
+      {status: "turn_returned"});
+});
+
 test("only a conversation-origin accepted result leaves a wake intent for the exact requester", () => {
   // An ordinary CLI/MCP delegation has no conversation. It must keep the
   // transition it always had: no intent, no wake state, no widened readback.
@@ -486,6 +511,7 @@ test("preflight uses scoped native lane facts without inventing App presence", (
   const original = delegationPreflight(params);
   assert.equal(original.state, "runtime_unverified");
   assert.equal(original.dispatch_allowed, true);
+  assert.equal(original.queueable, false);
   assert.equal(original.turn_lane, undefined);
   for (const state of ["absent", "released", "dead"]) {
     const {turn_lane, ...observed} = delegationPreflight({...params, lane: {state, holder: "withheld"}});
@@ -497,6 +523,7 @@ test("preflight uses scoped native lane facts without inventing App presence", (
     assert.equal(observed.state, "turn_blocked");
     assert.equal(observed.turn_eligible, false);
     assert.equal(observed.dispatch_allowed, false);
+    assert.equal(observed.queueable, true);
     assert.equal(observed.reason, state === "live" ? "turn_lane_in_flight" : "turn_lane_state_unknown");
     assert.equal(observed.acceptance_ready, true);
     assert.deepEqual(observed.effects, original.effects);
@@ -504,6 +531,12 @@ test("preflight uses scoped native lane facts without inventing App presence", (
   }
   for (const lane of [null, {}, {state: "idle"}, {state: ["live"]}, {state: true}])
     assert.throws(() => delegationPreflight({...params, lane}));
+  for (const change of [
+    {acceptance: null}, {validation_files_current: false},
+    {preview: {...params.preview, route: {...params.preview.route, selected_todo_id: "other"}}},
+    {preview: {...params.preview, route: {...params.preview.route, would_invoke_host: false}}},
+    {preview: {...params.preview, managed_executor: {...params.preview.managed_executor, available: false}}},
+  ]) assert.equal(delegationPreflight({...params, ...change, lane: {state: "live"}}).queueable, false);
 });
 
 const runtimePreflight = (executor: Record<string, unknown>) => delegationPreflight({

From 640fd0495fcc5e0972fec00fb7d35b4cbd0e68d8 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 14:45:12 +0800
Subject: [PATCH 6/9] refactor(turn): stage immutable decision capture before
 admission

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../turn_driver/decision_capture.py           | 53 ++++++++++++++-----
 loopx/control_plane/turn_driver/executor.py   | 23 +++-----
 .../test_turn_host_decision_capture.py        | 20 ++++++-
 3 files changed, 65 insertions(+), 31 deletions(-)

diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index da9c913eb0..22e8acd77c 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -9,6 +9,7 @@
 import stat
 import tempfile
 from collections.abc import Callable, Mapping
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -33,15 +34,34 @@ def write_decision_capture(
         Path(temporary).unlink(missing_ok=True)
 
 
-def prepare_turn_decision_capture(
-    plan: Mapping[str, Any], *, journal_path: Path,
-    decision: Mapping[str, Any] | None, existing_journal: bool, persist: bool = True,
-) -> dict[str, Any]:
-    """Called under the Turn journal lock before Host effects.
+@dataclass(frozen=True)
+class StagedTurnDecisionCapture:
+    """Validated private IO, published only after the native start is admitted."""
+
+    plan: dict[str, Any]
+    path: Path
+    observation: dict[str, Any]
+    existed: bool
+
+    def publish(self) -> dict[str, Any]:
+        if self.path.is_symlink():
+            raise ValueError("Turn decision capture must not be a symlink")
+        if self.path.exists():
+            if json.loads(self.path.read_text(encoding="utf-8")) != self.observation:
+                raise ValueError("Turn decision capture changed before publication")
+        elif self.existed:
+            raise ValueError("Original Turn decision capture is missing; restore the original observation")
+        else:
+            self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+            write_decision_capture(self.path, self.observation)
+        return self.plan
+
 
-    ``persist=False`` validates the exact source before cadence admission without
-    freezing a refused, unstarted observation. After admission the caller must
-    run the normal publication path before giving that plan to the Host.
+def stage_turn_decision_capture(
+    plan: Mapping[str, Any], *, journal_path: Path,
+    decision: Mapping[str, Any] | None, existing_journal: bool,
+) -> StagedTurnDecisionCapture:
+    """Validate under the journal lock before admission, without publishing IO.
 
     Existing journals must retain their original reference. A missing legacy
     observation cannot be reconstructed from today's guard. Already committed
@@ -56,7 +76,8 @@ def prepare_turn_decision_capture(
                          "recover its pending effects before starting a fresh Turn")
     if path.is_symlink():
         raise ValueError("Turn decision capture must not be a symlink")
-    if path.exists():
+    existed = path.exists()
+    if existed:
         observation = json.loads(path.read_text(encoding="utf-8"))
     elif existing_journal or capture:
         raise ValueError("Original Turn decision capture is missing; restore the original "
@@ -75,10 +96,16 @@ def prepare_turn_decision_capture(
         # Legacy journals retain their recorded delivery/recovery route.
         reader_argv=current_loopx_cli_argv() if not existing_journal else None,
     )
-    if persist and not path.exists():
-        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        write_decision_capture(path, observation)
-    return {**plan, "turn_envelope": bound}
+    return StagedTurnDecisionCapture({**plan, "turn_envelope": bound}, path, observation, existed)
+
+
+def prepare_turn_decision_capture(
+    plan: Mapping[str, Any], *, journal_path: Path,
+    decision: Mapping[str, Any] | None, existing_journal: bool,
+) -> dict[str, Any]:
+    """Retain the existing immediate publication API for original-capture callers."""
+    return stage_turn_decision_capture(plan, journal_path=journal_path,
+        decision=decision, existing_journal=existing_journal).publish()
 
 
 def validate_handoff_socket(value: str) -> Path:
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 745301845a..28a760a087 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -37,6 +37,7 @@
     reward_memory_reflection_digest,
 )
 from .driver import selected_turn_todo
+from .decision_capture import stage_turn_decision_capture
 from .execution_readback import execution_payload
 from .host_process_transport import run_host_process
 from .host_binding import (
@@ -1614,18 +1615,12 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
         )
         if needs_host and goal_admission is not None:
             goal_admission.require_current()
+        capture = None
         if require_decision_capture and needs_host:
-            from .decision_capture import prepare_turn_decision_capture
-
-            original_plan = journal["plan"] if journal is not None else plan
-            plan = prepare_turn_decision_capture(
-                original_plan, journal_path=journal_path,
+            capture = stage_turn_decision_capture(
+                journal["plan"] if journal is not None else plan, journal_path=journal_path,
                 decision=decision_capture_payload, existing_journal=journal is not None,
-                persist=admit_start is None,
             )
-            # Preserve any session recovery adjustment; only the original
-            # observation replaces the fresh request's context.
-            request = {**request, "turn_envelope": plan["turn_envelope"]}
         admission = None
         if needs_host and admit_start is not None:
             admission = admit_start(
@@ -1649,14 +1644,8 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 return execution_payload(
                     plan, waiting, execute=True, replayed=False, effects=empty_effects
                 )
-        if require_decision_capture and needs_host and admit_start is not None:
-            # Validate before reserving cadence/settlement identity, but publish
-            # only an admitted observation. Otherwise an interval refusal leaves
-            # a capture without a journal which poisons the same Turn's retry.
-            plan = prepare_turn_decision_capture(
-                original_plan, journal_path=journal_path,
-                decision=decision_capture_payload, existing_journal=journal is not None,
-            )
+        if capture is not None:
+            plan = capture.publish()
             request = {**request, "turn_envelope": plan["turn_envelope"]}
         if journal is not None and recovery_decision is not None:
             journal["recovery_audit"] = build_turn_recovery_audit(
diff --git a/tests/control_plane/test_turn_host_decision_capture.py b/tests/control_plane/test_turn_host_decision_capture.py
index f0d87b05f5..2c3a7da99d 100644
--- a/tests/control_plane/test_turn_host_decision_capture.py
+++ b/tests/control_plane/test_turn_host_decision_capture.py
@@ -11,7 +11,9 @@
 import pytest
 
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
-from loopx.control_plane.turn_driver.decision_capture import prepare_turn_decision_capture, write_decision_capture
+from loopx.control_plane.turn_driver.decision_capture import (
+    prepare_turn_decision_capture, stage_turn_decision_capture, write_decision_capture,
+)
 from loopx.control_plane.turn_driver.journal_store import load_loopx_turn_plan_from_journal, turn_journal_path
 from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
@@ -144,6 +146,22 @@ def test_real_host_reads_original_capture_after_source_change(
     assert not any(repeated["effects"].values())
     assert f._quota_spend_count(runtime) == 1
     assert path.read_bytes() == original
+    # Admission can run between source validation and publication. It cannot
+    # replace or silently recreate an original capture changed in that window.
+    staged = stage_turn_decision_capture(plan, journal_path=journal_path,
+                                        decision=None, existing_journal=True)
+    path.write_text(json.dumps({**source_a, "reason": "changed during admission"}))
+    changed = path.read_bytes()
+    with pytest.raises(ValueError, match="changed before publication"):
+        staged.publish()
+    assert path.read_bytes() == changed
+    path.unlink()
+    with pytest.raises(ValueError, match="missing"):
+        staged.publish()
+    assert not path.exists()
+    path.write_bytes(original)
+    assert staged.publish()["turn_envelope"]["detail_ref"] == detail
+    assert path.read_bytes() == original
     # Corrupt/missing evidence cannot be replaced by the caller's new guard.
     path.write_text(json.dumps({**source_a, "reason": "wrong original"}))
     with pytest.raises((ValueError, RuntimeError), match="original observation"):

From 2e7d7a39803f67e4d47c1046ee7f8e6363902e00 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 14:49:54 +0800
Subject: [PATCH 7/9] fix(collaboration): integrate current native host owners

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/turn_driver/executor.py          | 10 ++++------
 tests/control_plane/test_operation_wake_preflight.py |  3 ++-
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 28a760a087..ad0f412862 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -1615,12 +1615,10 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
         )
         if needs_host and goal_admission is not None:
             goal_admission.require_current()
-        capture = None
-        if require_decision_capture and needs_host:
-            capture = stage_turn_decision_capture(
-                journal["plan"] if journal is not None else plan, journal_path=journal_path,
-                decision=decision_capture_payload, existing_journal=journal is not None,
-            )
+        capture = stage_turn_decision_capture(
+            journal["plan"] if journal is not None else plan, journal_path=journal_path,
+            decision=decision_capture_payload, existing_journal=journal is not None,
+        ) if require_decision_capture and needs_host else None
         admission = None
         if needs_host and admit_start is not None:
             admission = admit_start(
diff --git a/tests/control_plane/test_operation_wake_preflight.py b/tests/control_plane/test_operation_wake_preflight.py
index 8883738e85..93138799f6 100644
--- a/tests/control_plane/test_operation_wake_preflight.py
+++ b/tests/control_plane/test_operation_wake_preflight.py
@@ -14,7 +14,8 @@
 from loopx.cli import build_parser
 from loopx.control_plane.collaboration.operation_wake import dispatch_confirmed_operation_wake
 from loopx.control_plane.turn_driver.codex_cli import load_codex_cli_session
-from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler, run_codex_operation_host
+from loopx.control_plane.turn_driver.codex_operation_host import run_codex_operation_host
+from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler
 from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight
 from loopx.file_lock import lock_holder_path
 from test_codex_operation_host import FAKE_SERVER

From 17a97a0e3e278ddbc2e85e4abcdcbe026fae1cb7 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 15:47:17 +0800
Subject: [PATCH 8/9] fix(turn): preserve JSON types across capture publication

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/turn_driver/decision_capture.py   |  6 +++++-
 .../control_plane/test_turn_host_decision_capture.py  | 11 +++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 22e8acd77c..8964e03824 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -47,7 +47,11 @@ def publish(self) -> dict[str, Any]:
         if self.path.is_symlink():
             raise ValueError("Turn decision capture must not be a symlink")
         if self.path.exists():
-            if json.loads(self.path.read_text(encoding="utf-8")) != self.observation:
+            # Python considers True == 1; original JSON evidence must retain
+            # its value types as well as its content. Object key order is IO.
+            current = json.loads(self.path.read_text(encoding="utf-8"))
+            if (json.dumps(current, sort_keys=True, allow_nan=False)
+                    != json.dumps(self.observation, sort_keys=True, allow_nan=False)):
                 raise ValueError("Turn decision capture changed before publication")
         elif self.existed:
             raise ValueError("Original Turn decision capture is missing; restore the original observation")
diff --git a/tests/control_plane/test_turn_host_decision_capture.py b/tests/control_plane/test_turn_host_decision_capture.py
index 2c3a7da99d..804ee99515 100644
--- a/tests/control_plane/test_turn_host_decision_capture.py
+++ b/tests/control_plane/test_turn_host_decision_capture.py
@@ -150,6 +150,17 @@ def test_real_host_reads_original_capture_after_source_change(
     # replace or silently recreate an original capture changed in that window.
     staged = stage_turn_decision_capture(plan, journal_path=journal_path,
                                         decision=None, existing_journal=True)
+    # Numeric lookalikes compare equal to booleans in Python, but change the
+    # original JSON observation and its typed source hash.
+    for numeric in (1, 1.0):
+        path.write_text(json.dumps({**source_a, "ok": numeric}))
+        changed = path.read_bytes()
+        with pytest.raises(ValueError, match="changed before publication"):
+            staged.publish()
+        assert path.read_bytes() == changed
+    path.write_text(json.dumps(dict(reversed(source_a.items()))))
+    assert staged.publish()["turn_envelope"]["detail_ref"] == detail
+    path.write_bytes(original)
     path.write_text(json.dumps({**source_a, "reason": "changed during admission"}))
     changed = path.read_bytes()
     with pytest.raises(ValueError, match="changed before publication"):

From 05a739abdda299600f6997ef78947c378c0c2fdd Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 28 Oct 2026 16:41:44 +0800
Subject: [PATCH 9/9] fix(delegation): verify original source at worker effect
 boundaries

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            | 12 ++-
 loopx/collaboration_mcp.py                    | 12 ++-
 .../control_plane/collaboration/delegation.ts |  3 +-
 .../collaboration/source_grant_observation.py | 24 ++++--
 .../test_delegation_source_authority.py       | 82 +++++++++++++++++++
 tests/control_plane_ts/delegation.test.ts     |  6 +-
 6 files changed, 123 insertions(+), 16 deletions(-)

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index b4c6d06574..3141e837ea 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -146,7 +146,12 @@ operations without a Chat source retain their existing binding authority.
 
 Chat retry also verifies the saved operation against the original Inbox request,
 complete brief, peer request, current binding and original return Session/Turn
-before reporting its status. Selecting another authorized binding does not
+before reporting its status. The worker uses this same typed comparison before
+launch and completion/settlement, even without a Chat retry. Its return identity
+must match the observed original source conversation; consistent-looking saved
+records cannot supply a replacement conversation. Missing or changed provenance
+retains the original failure/Turn for recovery without accepting new work.
+Selecting another authorized binding does not
 retarget an existing operation. A mismatch returns the existing unavailable
 execution response without publishing a peer request or resubmitting the worker;
 matching prepared operations may resume only under the restored current grant,
@@ -202,7 +207,10 @@ other operations keep their existing explicit recovery.
 中文：Chat 发起的执行保留原 Inbox 请求、固定的对话来源和原 operation。worker 在
 入口、Host 启动前和完成结算前重新核对当前发送者、接收者及精确
 Goal/Agent/requester/binding grant；撤权、来源不可读或身份变化会拒绝在原
-operation 上的新执行。最终来源观察、规范完成和原 Turn 结算共享来源策略锁，
+operation 上的新执行。同一个 typed owner 还核对完整来源/peer 简报、原 peer
+operation 和实际来源对话，不依赖 Chat 再次重放；保存的返回地址不能替换实际
+来源对话。来源缺失或变化保留原失败和 Turn，恢复后沿原身份继续。
+最终来源观察、规范完成和原 Turn 结算共享来源策略锁，
 撤权只能发生在观察之前或两项副作用完成之后；历史仍可检查。Host 已返回时保留原 Turn 和未结算效果；
 恢复授权后只结算该 operation，不重跑 Host。重试在报告状态前核对原 Inbox、完整
 brief、peer request、当前 binding 和原回传 Session/Turn；选择另一个合法 binding
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index f635d57e82..e6f2b4db0c 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -66,7 +66,7 @@
 )
 from .control_plane.collaboration import delegation_results, delegation_stop_lease, delegation_validation
 from .control_plane.collaboration.source_grant_observation import (
-    read_delegation_source_execution, require_delegation_source_execution,
+    read_delegation_source_execution, verify_delegation_source_execution,
 )
 from .control_plane.collaboration.source_policy_fence import source_policy_fence
 from .control_plane.collaboration.peers import (
@@ -868,9 +868,13 @@ def _bound(
         if row["identity"]["binding"] != binding:
             raise ValueError("delegation binding changed; reconcile original execution")
         if require_active and "source_request_id" in row["identity"]:
-            require_delegation_source_execution(self.root, self.registry, goal_id=self.goal_id,
-                requester_agent_id=self.agent_id, source_request_id=row["identity"]["source_request_id"], binding=binding,
-                caller_goal_ref=self._caller_goal_ref(), policy_locked=policy_locked)
+            try:
+                verify_delegation_source_execution(self, binding["id"], row["identity"]["operation_id"],
+                    source_request_id=row["identity"]["source_request_id"],
+                    brief=delegation_results.operation_brief(self, row), conversation=row.get("conversation", {}),
+                    policy_locked=policy_locked)
+            except OSError as exc:
+                raise ValueError("original source execution provenance unavailable") from exc
         return binding
 
     @staticmethod
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 7d41d8f0d4..9a98aa6601 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -188,7 +188,8 @@ export function verifyDelegationSourceReplay(params: JsonObject): JsonObject {
   const brief = normalizeCollaborationBrief(params.brief);
   requireThat(same(source.brief, brief) && same(peer.brief, brief), "delegation replay brief changed");
   requireThat(Object.keys(conversation).length === 2 && text(conversation.session_id) && text(conversation.turn_id)
-    && same(record.conversation, conversation), "delegation replay return conversation changed");
+    && same(record.conversation, conversation) && same(params.source_conversation, conversation),
+  "delegation replay return conversation changed");
   return {matched: true};
 }
 
diff --git a/loopx/control_plane/collaboration/source_grant_observation.py b/loopx/control_plane/collaboration/source_grant_observation.py
index a78bdb2a00..aa5bd40c78 100644
--- a/loopx/control_plane/collaboration/source_grant_observation.py
+++ b/loopx/control_plane/collaboration/source_grant_observation.py
@@ -178,7 +178,7 @@ def source_execution_bindings(
 def require_delegation_source_execution(runtime_root: Path, registry_path: Path, *,
         goal_id: str, requester_agent_id: str, source_request_id: str,
         binding: dict[str, Any], caller_goal_ref: dict[str, Any] | None,
-        policy_locked: bool = False) -> None:
+        policy_locked: bool = False) -> dict[str, str]:
     """Reobserve the original Chat source; the existing TS owner decides its grant.
 
     A return address only locates frozen evidence. It never grants execution,
@@ -211,13 +211,15 @@ def require_delegation_source_execution(runtime_root: Path, registry_path: Path,
                 raise ValueError("original source execution context is no longer authorized")
             source_execution_bindings(runtime_root, registry_path, session, turn,
                 required_binding={**target, "requester_agent_id": requester_agent_id, "binding_id": binding["id"]})
+            return {"session_id": session["session_id"], "turn_id": turn["turn_id"]}
     except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRejected) as exc:
         raise ValueError("original source execution authority is unavailable or revoked") from exc
 
 
-def read_delegation_source_execution(service, binding_id: str, operation_id: str, *,
-        source_request_id: str, brief: dict, conversation: dict) -> dict:
-    """Observe the existing exact-Goal sources for the typed replay comparison.
+def verify_delegation_source_execution(service, binding_id: str, operation_id: str, *,
+        source_request_id: str, brief: dict, conversation: dict,
+        policy_locked: bool = False) -> None:
+    """Reobserve exact source/peer work before replay or worker-owned effects.
 
     This performs no request publication, worker submission or result adoption.
     Legacy sources retain their legacy lifetime profile; they are not promoted.
@@ -231,9 +233,9 @@ def read_delegation_source_execution(service, binding_id: str, operation_id: str
     goal_id, requester_agent_id = service.goal_id, service.agent_id
     caller_goal_ref = service._caller_goal_ref()
     binding = service.binding(binding_id, require_active=True)
-    require_delegation_source_execution(runtime_root, registry_path, goal_id=goal_id,
+    source_conversation = require_delegation_source_execution(runtime_root, registry_path, goal_id=goal_id,
         requester_agent_id=requester_agent_id, source_request_id=source_request_id,
-        binding=binding, caller_goal_ref=caller_goal_ref)
+        binding=binding, caller_goal_ref=caller_goal_ref, policy_locked=policy_locked)
     record = _read(service.path(operation_id))
     with collaboration_goal_scope(registry_path, goal_id=goal_id,
             agents=(requester_agent_id, binding["agent_id"]),
@@ -249,10 +251,16 @@ def read_delegation_source_execution(service, binding_id: str, operation_id: str
                 "goal_id": goal_id, "goal_ref": scope.record_identity().get("goal_ref"), "record": record,
                 "operation_id": operation_id, "source_request_id": source_request_id,
                 "source_request": original, "peer_request": peer, "peer_operation": _read(operation_path),
-                "brief": brief, "conversation": conversation,
+                "brief": brief, "conversation": conversation, "source_conversation": source_conversation,
             })
         except EffectRuntimeRejected as exc:
-            raise ValueError("delegation operation identity conflict") from exc
+            raise ValueError("delegation operation identity conflict in original source execution") from exc
+
+
+def read_delegation_source_execution(service, binding_id: str, operation_id: str, *,
+        source_request_id: str, brief: dict, conversation: dict) -> dict:
+    verify_delegation_source_execution(service, binding_id, operation_id,
+        source_request_id=source_request_id, brief=brief, conversation=conversation)
     return service.read(operation_id)
 
 
diff --git a/tests/control_plane/test_delegation_source_authority.py b/tests/control_plane/test_delegation_source_authority.py
index e6f65246d5..53a9248fdc 100644
--- a/tests/control_plane/test_delegation_source_authority.py
+++ b/tests/control_plane/test_delegation_source_authority.py
@@ -21,6 +21,8 @@
 from loopx.control_plane.collaboration import delegation_results, delegation_validation
 from loopx.control_plane.collaboration.inbox import _read
 from loopx.control_plane.collaboration.source_policy_fence import source_policy_fence
+from loopx.control_plane.collaboration.goal_instance_scope import collaboration_goal_scope
+from loopx.control_plane.collaboration.peers import _operation_address
 from test_independent_delegation_validation import independent_binding
 from test_local_delegation import demo, service as delegation_service  # noqa: F401
 from test_manager_context_execution import source
@@ -61,6 +63,86 @@ def run_worker(worker, operation):
     assert completed.returncode == 0, completed.stderr
 
 
+def changed_source_work(worker, binding, operation, source_id, change):
+    """Corrupt only disposable provenance; the real worker must not adopt it."""
+    row = _read(worker.path(operation))
+    if change == "conversation":
+        row["conversation"] = {"session_id": "another-session", "turn_id": "another-turn"}
+        _write(worker.path(operation), row)
+        return
+    if change in {"peer_operation", "operation_missing"}:
+        with collaboration_goal_scope(worker.registry, goal_id=worker.goal_id, agents=(),
+                caller_goal_ref=worker._caller_goal_ref()) as scope:
+            _, path = _operation_address(worker.root, scope, worker.agent_id, operation)
+    else:
+        request_id = source_id if change == "source_brief" else row["identity"]["request_id"]
+        path = next((_root(worker.root) / "entries").glob("*/" + request_id + ".json"))
+    original = path.read_bytes()
+    if change == "operation_missing":
+        path.unlink()
+        return path, original
+    entry = _read(path)
+    entry["brief"]["constraints"].append("Changed after the original dispatch")
+    _write(path, entry)
+    return path, original
+
+
+@pytest.mark.parametrize("change", ["source_brief", "peer_brief", "peer_operation", "operation_missing", "conversation"])
+def test_worker_checks_original_work_without_a_chat_replay(saved_handoff, change):
+    root, worker, binding, (_, _, _, _, receipt, _), operation = saved_handoff
+    tasks_before = demo.canonical_tasks(root)
+    changed_source_work(worker, binding, operation, receipt["request_id"], change)
+    run_worker(worker, operation)
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert demo.canonical_tasks(root) == tasks_before
+    row = _read(worker.path(operation))
+    assert row["status"] == "rejected"
+    assert "source execution" in row["error"]
+    assert worker.read(operation)["operation_id"] == operation
+
+
+@pytest.mark.parametrize("change", ["source_brief", "peer_brief", "peer_operation"])
+def test_worker_rechecks_original_work_after_host_before_acceptance(saved_handoff, monkeypatch, change):
+    root, worker, binding, (_, _, _, _, receipt, _), operation = saved_handoff
+    tasks_before = demo.canonical_tasks(root)
+    cli = worker._cli
+    changed = False
+    original_sources = []
+
+    def change_after_host(current_binding, *args, **kwargs):
+        nonlocal changed
+        result = cli(current_binding, *args, **kwargs)
+        if not changed and args[:2] == ("turn", "run-once") and "--execute" in args:
+            changed = True
+            original_sources.append(changed_source_work(worker, binding, operation, receipt["request_id"], change))
+        return result
+
+    monkeypatch.setattr(worker, "_cli", change_after_host)
+    worker.execute(operation)
+    assert changed
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    assert demo.canonical_tasks(root) == tasks_before
+    journals = list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert len(journals) == 1
+    journal = _read(journals[0])
+    assert "quota_spend" not in journal["completed_phases"]
+    assert journal["task_validation"]["ok"] is True
+    assert _read(worker.path(operation))["status"] != "accepted"
+    # Restore only the original disposable source. The saved validated Turn,
+    # not a second Host, must finish its original completion and settlement.
+    for path, original in original_sources:
+        path.write_bytes(original)
+    worker.execute(operation)
+    settled = _read(worker.path(operation))
+    assert settled["status"] == "accepted", settled
+    assert settled["turn_key"] == journal["turn_key"]
+    assert demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+    assert _read(journals[0])["completed_phases"].count("quota_spend") == 1
+    worker.execute(operation)
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
 @pytest.mark.parametrize("change", ["revoke", "sender", "blocked", "requester", "binding",
                                   "policy_missing", "route_missing", "source_missing", "body", "closed"])
 def test_changed_queued_source_cannot_reach_the_real_worker_host(saved_handoff, change):
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index d0682686c1..689589c705 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -22,7 +22,7 @@ test("source replay requires the original binding, complete brief, peer and retu
     source_request_id: source.request_id};
   const input = {...params, goal_id: "goal", goal_ref: goalRef, operation_id: identity.operation_id,
     source_request_id: source.request_id, source_request: source, peer_request: peer, peer_operation: peer,
-    brief, conversation, record: {identity, conversation, status: "prepared"}};
+    brief, conversation, source_conversation: conversation, record: {identity, conversation, status: "prepared"}};
   assert.deepEqual(verifyDelegationSourceReplay(input), {matched: true});
   for (const record of [{...input.record, status: "accepted"}, {...input.record, status: "stopped"}])
     assert.deepEqual(verifyDelegationSourceReplay({...input, record}), {matched: true});
@@ -31,6 +31,10 @@ test("source replay requires the original binding, complete brief, peer and retu
     {source_request_id: "c".repeat(64)}, {operation_id: "other-operation"},
     {brief: {...brief, constraints: ["Different scope"]}},
     {conversation: {...conversation, turn_id: "other-turn"}},
+    {source_conversation: {...conversation, turn_id: "other-turn"}},
+    {source_conversation: null},
+    {conversation: {session_id: "other-session", turn_id: "other-turn"},
+      record: {...input.record, conversation: {session_id: "other-session", turn_id: "other-turn"}}},
     {source_request: {...source, execution_binding_id: "other"}},
     {peer_request: {...peer, source_agent_id: "another-requester"}},
     {peer_request: {...peer, brief: {...brief, acceptance: ["Different work"]}}},