From ebb7c6e076e17ac36f7ece64a57e840e2b7fa69d Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 15:05:55 +0800
Subject: [PATCH 1/6] fix(control-plane): harden promoted goal acceptance

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../todos/machine_section_projection.py       | 26 +++--
 loopx/todos.py                                |  6 +-
 .../test_runtime_shadow_writer_capture.py     | 42 +++++++-
 .../test_todo_machine_section_projection.py   | 98 +++++++++++++++++++
 4 files changed, 160 insertions(+), 12 deletions(-)

diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index 4e1f871e83..bf167d20ee 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -425,16 +425,28 @@ def render_canonical_todo_sections(
     except ValueError as error:
         raise TodoSectionProjectionError(str(error)) from error
     private_source = _private_validation_metadata(source_document)
+    canonical_validation_digests = {
+        str(record.get("todo_id") or ""): record.get("completion_validation_sha256")
+        for record in canonical
+        if record.get("completion_validation_required") is True
+    }
     for todo_id, declaration in (private_validation_declarations or {}).items():
         external = _private_validation_entry(declaration)
         existing = private_source.get(todo_id)
-        if existing is not None and (
-            completion_validation_declaration_sha256(existing[0])
-            != completion_validation_declaration_sha256(external[0])
-        ):
-            raise TodoSectionProjectionError(
-                f"Todo {todo_id!r} has divergent private validation declarations"
-            )
+        external_digest = completion_validation_declaration_sha256(external[0])
+        if existing is not None:
+            existing_digest = completion_validation_declaration_sha256(existing[0])
+            if existing_digest != external_digest:
+                # A successful CAS revision updates provider authority and the
+                # private declaration sidecar before the readable Markdown can
+                # be regenerated.  In that bounded state the old Markdown is
+                # expected to disagree.  Only the declaration selected by the
+                # canonical record may replace it; every other divergence is
+                # still rejected closed.
+                if canonical_validation_digests.get(todo_id) != external_digest:
+                    raise TodoSectionProjectionError(
+                        f"Todo {todo_id!r} has divergent private validation declarations"
+                    )
         private_source[todo_id] = external
     private_validation: dict[str, Mapping[str, object]] = {}
     for record in canonical:
diff --git a/loopx/todos.py b/loopx/todos.py
index a6db16df66..ddeb92bf6b 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -1308,15 +1308,16 @@ def update_goal_todo(
         validation_failure = completion_validation_gate.get("failure")
         if validation_failure is not None:
             return validation_failure
+    write_class = "todo_claim" if claim_only else "todo_update"
     with legacy_todo_write_transaction(
-        registry_path, goal_id, resolved_state_file, agent_id or claimed_by, "todo_update", dry_run,
+        registry_path, goal_id, resolved_state_file, agent_id or claimed_by, write_class, dry_run,
         runtime_root=shadow_runtime_root,
     ), ExitStack() as handoff_gate_stack:
         original = resolved_state_file.read_text(encoding="utf-8")
         shadow_capture = begin_todo_runtime_shadow_capture(
             registry_path=registry_path, runtime_root=shadow_runtime_root,
             goal_id=goal_id, state_path=resolved_state_file,
-            write_class="todo_update", original_text=original,
+            write_class=write_class, original_text=original,
         )
         lines = original.splitlines()
         updated_at = now_local()
@@ -1507,7 +1508,6 @@ def update_goal_todo(
         if changed and not dry_run:
             write_captured_todo_state(shadow_capture, runtime_root=shadow_runtime_root, goal_id=goal_id,
                 state_path=resolved_state_file, text=new_text)
-    write_class = "todo_claim" if claim_only else "todo_update"
     payload = {
         "ok": True,
         "dry_run": dry_run,
diff --git a/tests/control_plane/test_runtime_shadow_writer_capture.py b/tests/control_plane/test_runtime_shadow_writer_capture.py
index 10f1035dc2..c8f1e42f68 100644
--- a/tests/control_plane/test_runtime_shadow_writer_capture.py
+++ b/tests/control_plane/test_runtime_shadow_writer_capture.py
@@ -18,14 +18,19 @@
 GOAL_ID = "runtime-shadow-writer"
 
 
-def _fixture(tmp_path: Path, *, enabled: bool) -> tuple[Path, Path, Path]:
+def _fixture(
+    tmp_path: Path,
+    *,
+    enabled: bool,
+    handoff_mode: str = "hard_lease",
+) -> tuple[Path, Path, Path]:
     repo = tmp_path / "repo"
     repo.mkdir()
     state = repo / "ACTIVE_GOAL_STATE.md"
     state.write_text(
         "---\n"
         f"goal_id: {GOAL_ID}\n"
-        "handoff_mode: hard_lease\n"
+        f"handoff_mode: {handoff_mode}\n"
         "updated_at: 2026-09-04T00:00:00+00:00\n"
         "---\n\n## Agent Todo\n\n",
         encoding="utf-8",
@@ -107,6 +112,39 @@ def test_runtime_shadow_todo_writer_captures_full_records_and_reuses_one_store(
     assert not (runtime_root / "authority-shadow" / "file" / GOAL_ID).exists()
 
 
+def test_runtime_shadow_todo_claim_preserves_claim_write_class(tmp_path: Path) -> None:
+    registry, _state, runtime_root = _fixture(
+        tmp_path,
+        enabled=True,
+        handoff_mode="legacy",
+    )
+    added = add_goal_todo(
+        registry_path=registry,
+        goal_id=GOAL_ID,
+        role="agent",
+        text="Qualify the legacy claim path.",
+        task_class="advancement_task",
+    )
+
+    claimed = update_goal_todo(
+        registry_path=registry,
+        goal_id=GOAL_ID,
+        todo_id=str(added["todo_id"]),
+        claimed_by="agent-a",
+        agent_id="agent-a",
+        claim_only=True,
+    )
+
+    assert claimed["coordination_runtime_shadow"]["outcome"] == "delivered"
+    view = adapter.read_local_authority_shadow(
+        runtime_root=runtime_root,
+        goal_id=GOAL_ID,
+        scan_limit=10,
+    )
+    receipt = view["proof"]["transactions"][-1]["receipts"][0]
+    assert receipt["write_class"] == "todo_claim"
+
+
 def test_runtime_shadow_todo_writer_is_zero_effect_by_default(tmp_path: Path) -> None:
     registry, _state, runtime_root = _fixture(tmp_path, enabled=False)
 
diff --git a/tests/control_plane/test_todo_machine_section_projection.py b/tests/control_plane/test_todo_machine_section_projection.py
index 96d45252ad..c6a1e43895 100644
--- a/tests/control_plane/test_todo_machine_section_projection.py
+++ b/tests/control_plane/test_todo_machine_section_projection.py
@@ -253,6 +253,104 @@ def test_projection_keeps_validation_revision_receipts_provider_only() -> None:
     assert replay.changed is False
 
 
+def test_projection_converges_stale_markdown_after_validator_revision() -> None:
+    old_declaration = {
+        "validation_command": None,
+        "validation_command_argv": [
+            "python",
+            "-m",
+            "pytest",
+            "-q",
+            "tests/test_goal_topic_runtime.py",
+        ],
+        "validation_label": "manager route reconciliation tests",
+        "validation_timeout_seconds": None,
+    }
+    new_declaration = {
+        **old_declaration,
+        "validation_command_argv": [
+            "python",
+            "-m",
+            "pytest",
+            "-q",
+            "tests/extensions/test_lark_goal_topic_runtime.py",
+        ],
+    }
+    record = deepcopy(_records()[0])
+    record.update(
+        completion_validation_required=True,
+        completion_validation_sha256=completion_validation_declaration_sha256(
+            old_declaration
+        ),
+    )
+    old_projection = render_canonical_todo_sections(
+        SOURCE,
+        [record],
+        provider_revision="validation-before-revision",
+        private_validation_declarations={"todo_agent": old_declaration},
+    )
+
+    record.update(
+        completion_validation_sha256=completion_validation_declaration_sha256(
+            new_declaration
+        ),
+        completion_validation_revision=1,
+    )
+    revised = render_canonical_todo_sections(
+        old_projection.markdown,
+        [record],
+        provider_revision="validation-after-revision",
+        private_validation_declarations={"todo_agent": new_declaration},
+    )
+
+    assert revised.changed is True
+    assert "tests%2Fextensions%2Ftest_lark_goal_topic_runtime.py" in revised.markdown
+    assert "tests%2Ftest_goal_topic_runtime.py" not in revised.markdown
+    replay = render_canonical_todo_sections(
+        revised.markdown,
+        [record],
+        provider_revision="validation-after-revision",
+    )
+    assert replay.changed is False
+
+
+def test_projection_rejects_divergent_sidecar_not_selected_by_authority() -> None:
+    authority_declaration = {
+        "validation_command": "python3 -c 'raise SystemExit(0)'",
+        "validation_command_argv": None,
+        "validation_label": "authority validator",
+        "validation_timeout_seconds": 5,
+    }
+    divergent_declaration = {
+        **authority_declaration,
+        "validation_command": "python3 -c 'raise SystemExit(1)'",
+    }
+    record = deepcopy(_records()[0])
+    record.update(
+        completion_validation_required=True,
+        completion_validation_sha256=completion_validation_declaration_sha256(
+            authority_declaration
+        ),
+    )
+    authority_projection = render_canonical_todo_sections(
+        SOURCE,
+        [record],
+        provider_revision="validation-authority",
+        private_validation_declarations={"todo_agent": authority_declaration},
+    )
+
+    with pytest.raises(
+        TodoSectionProjectionError,
+        match="divergent private validation declarations",
+    ):
+        render_canonical_todo_sections(
+            authority_projection.markdown,
+            [record],
+            provider_revision="validation-authority",
+            private_validation_declarations={"todo_agent": divergent_declaration},
+        )
+
+
 def test_projection_renders_native_archive_with_role_and_replays() -> None:
     source = SOURCE + "\n## Completed Work Archive\n\n- [x] stale archive\n"
     records = _records()

From cd64f38813a34325510791830f1c6e60456fa4f0 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 15:40:41 +0800
Subject: [PATCH 2/6] fix(coordination): reopen promoted unleased claims

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../coordination/todo_update_admission.ts     | 13 ++++++--
 tests/control_plane_ts/todo_update.test.ts    | 32 +++++++++++++++++++
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/loopx/control_plane/coordination/todo_update_admission.ts b/loopx/control_plane/coordination/todo_update_admission.ts
index db70cee450..3d1c0100f1 100644
--- a/loopx/control_plane/coordination/todo_update_admission.ts
+++ b/loopx/control_plane/coordination/todo_update_admission.ts
@@ -77,6 +77,15 @@ export function todoUpdateAdmissionRejection(
       : "Todo update is outside the actor's registered owner/binding scope";
     return reject(code, reason);
   }
+  // Promotion deliberately preserves legacy claims without inventing leases.
+  // An explicitly granted controller must be able to repair planning state on
+  // that unleased claim; otherwise a stale blocked/deferred status can never
+  // become eligible enough for the real owner to acquire its first lease.
+  // Once any lease lineage exists, the ordinary holder/CAS fence still wins.
+  const delegatedUnleasedOverride =
+    authorityDecision.authority_mode === "delegated_orchestration_override" &&
+    mode === "hard_lease" && lease === undefined &&
+    input.lease_idempotency_key == null && input.lease_expected_version == null;
   // Preserve the single-agent compatibility path only for genuinely
   // unowned work. An empty registry is not evidence that an arbitrary actor
   // may rewrite an already-owned Todo.
@@ -102,8 +111,8 @@ export function todoUpdateAdmissionRejection(
     }
     return null;
   }
-  if (lease !== undefined || mode === "hard_lease" ||
-      input.lease_idempotency_key != null || input.lease_expected_version != null) {
+  if (!delegatedUnleasedOverride && (lease !== undefined || mode === "hard_lease" ||
+      input.lease_idempotency_key != null || input.lease_expected_version != null)) {
     try {
       const fence = evaluateCanonicalTaskLeaseProof({todo, lease, handoff_mode: mode,
         registered_agents: input.registered_agents, actor_agent_id: input.actor_agent_id,
diff --git a/tests/control_plane_ts/todo_update.test.ts b/tests/control_plane_ts/todo_update.test.ts
index 38cfa9f584..fd62790201 100644
--- a/tests/control_plane_ts/todo_update.test.ts
+++ b/tests/control_plane_ts/todo_update.test.ts
@@ -79,6 +79,38 @@ test("delegated reassign cannot smuggle a copy edit or override exclusion/bindin
   }
 });
 
+test("delegated controller reopens a promoted legacy claim before its first lease", async () => {
+  const {store, request} = await seeded({status: "blocked", done: false, claimed_by: "agent-a"});
+  const head = await store.loadAuthority();
+  assert.equal(head.status, "loaded");
+  if (head.status !== "loaded") return;
+  await store.commitAuthority({operation_id: "promote-without-inventing-lease",
+    expected_provider_revision: head.provider_revision, events: [], receipts: [],
+    next_projection: {...head.head, handoff_mode: "hard_lease", leases: []}});
+  const edit = {...request, operation_id: "reviewed-reopen", actor_agent_id: "agent-b",
+    patch: {note: "Promotion resolved the stale blocker"}, clear_fields: [],
+    planning_intent: {status: "open", reason: "Canonical authority is promoted"},
+    authority_reason: "Reviewed controller recovery after promotion",
+    lifecycle_grants: [{agent_id: "agent-b", actions: ["update"], requires_reason: true}]};
+  assert.equal((await executeCoordinationTodoUpdate(store, edit)).status, "applied");
+  const after = await store.loadAuthority();
+  assert.equal(after.status, "loaded");
+  if (after.status !== "loaded") return;
+  const updated = (after.head.todos as Record<string, unknown>[])[0]!;
+  assert.equal(updated.status, "open");
+  assert.equal(updated.claimed_by, "agent-a");
+  assert.equal(updated.note, "Promotion resolved the stale blocker");
+  assert.deepEqual(after.head.leases, []);
+  await store.commitAuthority({operation_id: "worker-acquired-first-lease",
+    expected_provider_revision: after.provider_revision, events: [], receipts: [],
+    next_projection: {...after.head, leases: [{todo_id: "todo_a", owner: "agent-a",
+      status: "active", expires_at: "2026-09-06T00:00:00Z", idempotency_key: "execution-a",
+      version: 1, lease_epoch: 1, write_scopes: []}]}});
+  assert.equal((await executeCoordinationTodoUpdate(store, {...edit,
+    operation_id: "controller-cannot-cross-live-lease",
+    patch: {note: "Must not cross active execution"}})).reason_code, "lease_fence_required");
+});
+
 test("native planning edit commits nonterminal state and clears its wait atomically", async () => {
   const {store, request} = await seeded({task_class: "advancement_task"});
   const edit = {...request, patch: {text: "Old text"}, clear_fields: [], planning_intent: {

From 071f5bb3f19d8c98762be34f4221215135a31edd Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 15:57:51 +0800
Subject: [PATCH 3/6] fix(delegation): pin managed workers to release

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/collaboration_mcp.py                    | 47 ++++++++++++++-----
 .../coordination/local_authority.py           |  6 +++
 .../test_local_coordination_authority.py      | 29 +++++++-----
 tests/test_delegation_preflight.py            | 21 ++++++++-
 4 files changed, 80 insertions(+), 23 deletions(-)

diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 9c7ce8465b..8d6ef77334 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -44,6 +44,32 @@
 )
 
 
+def _pinned_release_environment() -> dict[str, str]:
+    """Keep managed children on the release that admitted the delegation.
+
+    A delegated workspace may itself be a LoopX checkout.  Plain
+    ``python -m loopx...`` prepends that workspace to ``sys.path`` and can
+    silently run an older control plane than the parent process.  Safe-path
+    mode removes the current directory while an explicit release-root
+    ``PYTHONPATH`` keeps source checkouts and installed releases deterministic.
+    """
+
+    environment = os.environ.copy()
+    release_root = str(Path(__file__).resolve().parent.parent)
+    inherited = [
+        entry
+        for entry in environment.get("PYTHONPATH", "").split(os.pathsep)
+        if entry and Path(entry).resolve(strict=False) != Path(release_root)
+    ]
+    environment["PYTHONPATH"] = os.pathsep.join([release_root, *inherited])
+    environment["PYTHONSAFEPATH"] = "1"
+    return environment
+
+
+def _python_module_command(module: str) -> list[str]:
+    return [sys.executable, "-P", "-m", module]
+
+
 def create_server(
     root: Path, registry: Path, goal_id: str, agent_id: str, workspace: Path,
     execution_config: Path | None = None,
@@ -276,12 +302,12 @@ def _spawn(self, operation_id: str) -> None:
         # No inherited stdio pipes: closing the conversation cannot cancel or
         # hang this bounded execution. The worker owns a kernel single-flight lock.
         operation_id = require_operation_id(operation_id)
-        subprocess.Popen([
-            sys.executable, "-m", "loopx.collaboration_mcp", "--delegation-action", "worker", "--runtime-root", str(self.root),
+        subprocess.Popen([*_python_module_command("loopx.collaboration_mcp"),
+            "--delegation-action", "worker", "--runtime-root", str(self.root),
             "--registry", str(self.registry), "--goal-id", self.goal_id,
             "--agent-id", self.agent_id, "--execution-config", str(self.config), "--operation-id=" + operation_id,
         ], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            start_new_session=True, close_fds=True)
+            start_new_session=True, close_fds=True, env=_pinned_release_environment())
 
     def resume(self, operation_id: str) -> dict:
         row = _read(self.path(operation_id))
@@ -349,10 +375,11 @@ def _observe(self, path: Path, row: dict, status: str, **facts) -> None:
         _write(path, row)
 
     def _cli(self, binding: dict, *args: str, timeout: int = 60) -> dict:
-        completed = subprocess.run([
-            sys.executable, "-m", "loopx.cli", "--registry", str(self.registry),
+        completed = subprocess.run([*_python_module_command("loopx.cli"),
+            "--registry", str(self.registry),
             "--runtime-root", str(self.root), "--format", "json", *args,
-        ], cwd=binding["workspace"], capture_output=True, text=True, encoding="utf-8", timeout=timeout)
+        ], cwd=binding["workspace"], capture_output=True, text=True, encoding="utf-8",
+            timeout=timeout, env=_pinned_release_environment())
         try:
             value = json.loads(completed.stdout)
         except ValueError as exc:
@@ -424,7 +451,8 @@ def execute(self, operation_id: str) -> None:
     def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
         """Exactly the same profile, workspace and validation arguments for preview/run."""
         # Preserve the journaled validator argv so existing Turns retain their resume identity.
-        validator = [sys.executable, "-m", "loopx.collaboration_mcp", "--delegation-action", "validate", "--runtime-root", str(self.root),
+        validator = [*_python_module_command("loopx.collaboration_mcp"),
+                     "--delegation-action", "validate", "--runtime-root", str(self.root),
                      "--registry", str(self.registry), "--goal-id", self.goal_id,
                      "--agent-id", self.agent_id, "--execution-config", str(self.config),
                      "--workspace", binding["workspace"], "--operation-id", operation_id]
@@ -433,10 +461,7 @@ def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
             mcp_server = {
                 "schema_version": "codex_stdio_mcp_server_v0",
                 "name": "loopx_delegation",
-                "command": [
-                    sys.executable,
-                    "-m",
-                    "loopx.collaboration_mcp",
+                "command": [*_python_module_command("loopx.collaboration_mcp"),
                     "--runtime-root",
                     str(self.root),
                     "--registry",
diff --git a/loopx/control_plane/coordination/local_authority.py b/loopx/control_plane/coordination/local_authority.py
index 2d75ca4ff5..6dd85e6fa7 100644
--- a/loopx/control_plane/coordination/local_authority.py
+++ b/loopx/control_plane/coordination/local_authority.py
@@ -29,6 +29,7 @@
 
 
 LOCAL_COORDINATION_TODO_LIST_METHOD = "coordination.local_authority.todo_list"
+LOCAL_COORDINATION_TODO_LIST_TIMEOUT_SECONDS = 15.0
 LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA = (
     "loopx_local_coordination_todo_claim_request_v1"
 )
@@ -205,6 +206,11 @@ def read_canonical_todos_if_promoted(
             **({"include_leases": True} if include_leases else {}),
             **({"projection_readback": dict(projection_readback)} if projection_readback is not None else {}),
         },
+        # Promoted goals can carry hundreds of preserved Todos.  Keep the
+        # generic Effect request budget strict, but give this known bounded
+        # canonical scan the same cold-start allowance as the neighbouring
+        # shadow/lease authority reads.
+        timeout=LOCAL_COORDINATION_TODO_LIST_TIMEOUT_SECONDS,
     )
     if not isinstance(result, Mapping):
         raise LocalCoordinationAuthorityUnavailable(
diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py
index 0f96922fb6..75c6d38190 100644
--- a/tests/control_plane/test_local_coordination_authority.py
+++ b/tests/control_plane/test_local_coordination_authority.py
@@ -146,9 +146,11 @@ def test_engaged_fence_reads_typescript_provider_result(
     tmp_path: Path,
 ) -> None:
     _engage_fence(tmp_path)
-    monkeypatch.setattr(
-        "loopx.control_plane.coordination.local_authority.effect_runtime_result",
-        lambda method, params: {
+    calls: list[tuple[str, float]] = []
+
+    def _read(method: str, _params: object, *, timeout: float) -> dict[str, object]:
+        calls.append((method, timeout))
+        return {
             "status": "loaded",
             "todos": [{"todo_id": "todo_a", "role": "agent", "status": "open"}],
             "todo_read_model": _todo_read_model(1),
@@ -157,7 +159,11 @@ def test_engaged_fence_reads_typescript_provider_result(
             "source_authority": "file_v0",
             "decision_read_from_provider": True,
             "legacy_fallback_used": False,
-        },
+        }
+
+    monkeypatch.setattr(
+        "loopx.control_plane.coordination.local_authority.effect_runtime_result",
+        _read,
     )
     result = read_canonical_todos_if_promoted(
         runtime_root=tmp_path,
@@ -165,6 +171,7 @@ def test_engaged_fence_reads_typescript_provider_result(
     )
     assert result is not None
     assert result["todos"][0]["todo_id"] == "todo_a"
+    assert calls == [("coordination.local_authority.todo_list", 15.0)]
 
 
 def test_promoted_claim_adapter_invokes_typescript_without_markdown_fallback(
@@ -908,7 +915,7 @@ def test_engaged_fence_never_falls_back_when_provider_is_missing(
     _engage_fence(tmp_path)
     monkeypatch.setattr(
         "loopx.control_plane.coordination.local_authority.effect_runtime_result",
-        lambda method, params: {
+        lambda method, params, **_kwargs: {
             "status": "missing",
             "source_authority": "file_v0",
             "decision_read_from_provider": True,
@@ -990,7 +997,7 @@ def test_promoted_claim_rejection_preserves_legacy_valueerror_contract(
     _engage_fence(tmp_path)
     monkeypatch.setattr(
         "loopx.control_plane.coordination.local_authority.effect_runtime_result",
-        lambda method, params: {
+        lambda method, params, **_kwargs: {
             "status": "failed",
             "failure_kind": "decision_rejection",
             "reason_code": "todo_not_open",
@@ -1142,7 +1149,7 @@ def test_todo_list_uses_provider_after_cutover_even_when_markdown_disagrees(
     state_file.unlink()
     monkeypatch.setattr(
         "loopx.control_plane.coordination.local_authority.effect_runtime_result",
-        lambda method, params: {
+        lambda method, params, **_kwargs: {
             "status": "loaded",
             "todos": [
                 {
@@ -1462,10 +1469,10 @@ def count_runtime_call(method: str, params: dict[str, object]) -> object:
         return result
 
     def count_authority_runtime_call(
-        method: str, params: dict[str, object]
+        method: str, params: dict[str, object], **kwargs: object
     ) -> object:
         runtime_calls.append(method)
-        return original_authority_runtime_result(method, params)
+        return original_authority_runtime_result(method, params, **kwargs)
 
     monkeypatch.setattr(
         provider_terminal_lifecycle,
@@ -1991,10 +1998,10 @@ def count_runtime_call(method: str, params: dict[str, object]) -> object:
         return original_effect_runtime_result(method, params)
 
     def count_authority_runtime_call(
-        method: str, params: dict[str, object]
+        method: str, params: dict[str, object], **kwargs: object
     ) -> object:
         runtime_calls.append(method)
-        return original_authority_runtime_result(method, params)
+        return original_authority_runtime_result(method, params, **kwargs)
 
     monkeypatch.setattr(
         provider_terminal_lifecycle,
diff --git a/tests/test_delegation_preflight.py b/tests/test_delegation_preflight.py
index 9ced3837b5..5ab417fa87 100644
--- a/tests/test_delegation_preflight.py
+++ b/tests/test_delegation_preflight.py
@@ -2,6 +2,7 @@
 
 import json
 import sys
+from pathlib import Path
 
 import pytest
 
@@ -205,13 +206,31 @@ def test_selected_codex_managed_agent_profile_is_projected_exactly(service):
     assert native["schema_version"] == "codex_stdio_mcp_server_v0"
     assert native["name"] == "loopx_delegation"
     command = native["command"]
-    assert command[:3] == [sys.executable, "-m", "loopx.collaboration_mcp"]
+    assert command[:4] == [sys.executable, "-P", "-m", "loopx.collaboration_mcp"]
     assert command[command.index("--agent-id") + 1] == "analyst"
     assert command[command.index("--workspace") + 1] == binding["workspace"]
     assert command[command.index("--execution-config") + 1] == str(runner.config)
     assert "lead" not in command
 
 
+def test_preflight_ignores_stale_loopx_checkout_in_worker_workspace(service):
+    root, runner = service
+    workspace = Path(runner.binding("analysis", require_active=True)["workspace"])
+    shadow = workspace / "loopx"
+    shadow.mkdir()
+    (shadow / "__init__.py").write_text("", encoding="utf-8")
+    (shadow / "cli.py").write_text(
+        "raise RuntimeError('stale workspace LoopX must not be imported')\n",
+        encoding="utf-8",
+    )
+
+    status, result = cli(runner, "inspect", "--binding-id", "analysis")
+
+    assert status == 0, result
+    assert result["turn_eligible"] is True
+    assert not any(result["effects"].values())
+
+
 def test_preflight_does_not_call_an_invalidated_acceptance_ready(service):
     root, runner = service
     from loopx.agent_registry import load_goal_from_registry

From 2da5b969745775b01975ad55a0dbbe7303b82eaf Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:24:18 +0800
Subject: [PATCH 4/6] fix(delegation): launch native MCP from managed runtime

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/collaboration_mcp.py         | 91 +++++++++++++++++++++++++++++-
 tests/test_delegation_preflight.py | 27 ++++++++-
 2 files changed, 115 insertions(+), 3 deletions(-)

diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 8d6ef77334..7b7519fb56 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -18,6 +18,7 @@
 import subprocess
 import sys
 import time
+from functools import lru_cache
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal
 
@@ -44,6 +45,19 @@
 )
 
 
+_PINNED_MODULE_LAUNCHER = (
+    "import runpy,sys;"
+    "release_root,module=sys.argv[1:3];"
+    "sys.path.insert(0,release_root);"
+    "sys.argv=[module,*sys.argv[3:]];"
+    "runpy.run_module(module,run_name='__main__')"
+)
+
+
+def _release_root() -> Path:
+    return Path(__file__).resolve().parent.parent
+
+
 def _pinned_release_environment() -> dict[str, str]:
     """Keep managed children on the release that admitted the delegation.
 
@@ -55,7 +69,7 @@ def _pinned_release_environment() -> dict[str, str]:
     """
 
     environment = os.environ.copy()
-    release_root = str(Path(__file__).resolve().parent.parent)
+    release_root = str(_release_root())
     inherited = [
         entry
         for entry in environment.get("PYTHONPATH", "").split(os.pathsep)
@@ -70,6 +84,79 @@ def _python_module_command(module: str) -> list[str]:
     return [sys.executable, "-P", "-m", module]
 
 
+def _mcp_python_candidates() -> tuple[Path, ...]:
+    configured = os.environ.get("LOOPX_MCP_PYTHON")
+    managed = (
+        Path.home()
+        / ".local"
+        / "share"
+        / "loopx"
+        / "mcp-venv"
+        / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python")
+    )
+    values: list[Path] = []
+    if configured:
+        selected = Path(configured).expanduser()
+        if not selected.is_absolute():
+            raise ValueError("LOOPX_MCP_PYTHON must be an absolute path")
+        values.append(selected)
+    values.extend([Path(sys.executable), managed])
+    # Do not resolve interpreter symlinks: a venv's ``python`` commonly points
+    # at the base executable, but its original path is what selects the venv
+    # site-packages containing FastMCP.
+    return tuple(dict.fromkeys(values))
+
+
+@lru_cache(maxsize=1)
+def _mcp_python_executable() -> str:
+    """Resolve a Python that can actually serve the required stdio MCP.
+
+    LoopX itself intentionally has no mandatory third-party dependencies.  Its
+    installers provision the shared MCP venv separately, so a managed Codex
+    child must not assume that the control-plane interpreter also has FastMCP.
+    """
+
+    for candidate in _mcp_python_candidates():
+        if not candidate.is_file():
+            continue
+        try:
+            probe = subprocess.run(
+                [
+                    str(candidate),
+                    "-P",
+                    "-c",
+                    "from mcp.server.fastmcp import FastMCP",
+                ],
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                env=_pinned_release_environment(),
+                timeout=10,
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            continue
+        if probe.returncode == 0:
+            return str(candidate)
+    raise ValueError(
+        "LoopX collaboration MCP runtime is unavailable; provision the shared "
+        "LoopX MCP venv or set LOOPX_MCP_PYTHON to a compatible interpreter"
+    )
+
+
+def _mcp_module_command() -> list[str]:
+    # Codex intentionally sanitizes PYTHONPATH for stdio MCP children.  Carry
+    # the admitted release root in argv and rebuild sys.path inside the child
+    # rather than trusting ambient process state.
+    return [
+        _mcp_python_executable(),
+        "-P",
+        "-c",
+        _PINNED_MODULE_LAUNCHER,
+        str(_release_root()),
+        "loopx.collaboration_mcp",
+    ]
+
+
 def create_server(
     root: Path, registry: Path, goal_id: str, agent_id: str, workspace: Path,
     execution_config: Path | None = None,
@@ -461,7 +548,7 @@ def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
             mcp_server = {
                 "schema_version": "codex_stdio_mcp_server_v0",
                 "name": "loopx_delegation",
-                "command": [*_python_module_command("loopx.collaboration_mcp"),
+                "command": [*_mcp_module_command(),
                     "--runtime-root",
                     str(self.root),
                     "--registry",
diff --git a/tests/test_delegation_preflight.py b/tests/test_delegation_preflight.py
index 5ab417fa87..2e7153ed70 100644
--- a/tests/test_delegation_preflight.py
+++ b/tests/test_delegation_preflight.py
@@ -1,6 +1,8 @@
 """A binding inspection must use the actual Turn without launching or spending."""
 
 import json
+import os
+import subprocess
 import sys
 from pathlib import Path
 
@@ -206,12 +208,35 @@ def test_selected_codex_managed_agent_profile_is_projected_exactly(service):
     assert native["schema_version"] == "codex_stdio_mcp_server_v0"
     assert native["name"] == "loopx_delegation"
     command = native["command"]
-    assert command[:4] == [sys.executable, "-P", "-m", "loopx.collaboration_mcp"]
+    assert command[:3] == [sys.executable, "-P", "-c"]
+    assert "loopx.collaboration_mcp" in command
+    assert str(Path(__file__).resolve().parents[1]) in command
     assert command[command.index("--agent-id") + 1] == "analyst"
     assert command[command.index("--workspace") + 1] == binding["workspace"]
     assert command[command.index("--execution-config") + 1] == str(runner.config)
     assert "lead" not in command
 
+    shadow = Path(binding["workspace"]) / "loopx"
+    shadow.mkdir()
+    (shadow / "__init__.py").write_text("", encoding="utf-8")
+    (shadow / "collaboration_mcp.py").write_text(
+        "raise RuntimeError('stale workspace MCP must not be imported')\n",
+        encoding="utf-8",
+    )
+    clean_environment = os.environ.copy()
+    clean_environment.pop("PYTHONPATH", None)
+    completed = subprocess.run(
+        command,
+        cwd=binding["workspace"],
+        input="",
+        text=True,
+        capture_output=True,
+        env=clean_environment,
+        timeout=10,
+    )
+    assert completed.returncode == 0, completed.stderr
+    assert "stale workspace MCP" not in completed.stderr
+
 
 def test_preflight_ignores_stale_loopx_checkout_in_worker_workspace(service):
     root, runner = service

From 99e6b362105a404f7af1079733a3492fd4891822 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 17:41:32 +0800
Subject: [PATCH 5/6] fix(delegation): recover validated managed settlements

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/cli_commands/turn.py                    |  17 +
 loopx/collaboration_mcp.py                    | 383 +++++++++++++++---
 .../control_plane/collaboration/delegation.ts |  27 ++
 .../control_plane/effect_runtime_handlers.ts  |   3 +-
 .../turn_driver/journal_store.py              |  63 +++
 tests/control_plane_ts/delegation.test.ts     |  26 +-
 tests/test_delegation_preflight.py            | 171 +++++++-
 tests/test_loopx_turn_executor.py             |  49 +++
 8 files changed, 684 insertions(+), 55 deletions(-)

diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index f618d6d24e..43368f9b87 100644
--- a/loopx/cli_commands/turn.py
+++ b/loopx/cli_commands/turn.py
@@ -112,6 +112,7 @@ def handle_turn_command(
             args, registry_path=registry_path, runtime_root_arg=runtime_root_arg,
             output_format=output_format, print_payload=print_payload,
         )
+    payload: dict[str, Any] = {}
     try:
         if getattr(args, "todo_id", None) is not None and (
             getattr(args, "resume_turn_key", None)
@@ -1066,6 +1067,12 @@ def post_settlement_reward_memory(
         else:
             raise ValueError("turn requires the `plan` or `run-once` subcommand")
     except Exception as exc:  # noqa: BLE001 - CLI boundary renders typed JSON failure
+        planned_transaction = (
+            payload.get("transaction")
+            if isinstance(payload.get("transaction"), Mapping)
+            else {}
+        )
+        planned_turn_key = str(planned_transaction.get("turn_key") or "")
         payload = {
             **({"error_code": exc.code, **getattr(exc, "payload", {})} if isinstance(getattr(exc, "code", None), str) else {}),
             "ok": False,
@@ -1082,6 +1089,16 @@ def post_settlement_reward_memory(
                 "scheduler_acknowledged": False,
                 "quota_spent": False,
             },
+            **(
+                {
+                    "resume_turn_key": planned_turn_key,
+                    "journal_ref": (
+                        f"turn:{planned_turn_key.removeprefix('sha256:')[:16]}"
+                    ),
+                }
+                if args.turn_command == "run-once" and planned_turn_key
+                else {}
+            ),
             **(
                 {"recovery_decision": exc.decision}
                 if isinstance(exc, TurnRecoveryBlockedError)
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 7b7519fb56..a057876f78 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -30,7 +30,12 @@
 from .control_plane.effect_runtime import effect_runtime_result, EffectRuntimeRemoteError
 from .control_plane.goals.acceptance import inspect_goal_acceptance, validate_goal_task_acceptance, goal_task_validation_files_current
 from .control_plane.coordination.local_authority import local_authority_is_promoted
-from .control_plane.turn_driver.journal_store import turn_journal_path
+from .control_plane.todos.handoff_mode import show_goal_handoff_mode
+from .control_plane.turn_driver.journal_store import (
+    find_loopx_turn_key_by_settlement_identity,
+    load_turn_journal,
+    turn_journal_path,
+)
 from .control_plane.turn_driver.host_binding import turn_host_arg_option
 from .control_plane.collaboration.inbox import _hash, _read, _write, _root, _receipt
 from .control_plane.collaboration.peers import return_result
@@ -66,6 +71,8 @@ def _pinned_release_environment() -> dict[str, str]:
     silently run an older control plane than the parent process.  Safe-path
     mode removes the current directory while an explicit release-root
     ``PYTHONPATH`` keeps source checkouts and installed releases deterministic.
+    Safe-path is selected on LoopX's own argv instead of exported globally:
+    host and acceptance scripts may legitimately import sibling modules.
     """
 
     environment = os.environ.copy()
@@ -76,7 +83,7 @@ def _pinned_release_environment() -> dict[str, str]:
         if entry and Path(entry).resolve(strict=False) != Path(release_root)
     ]
     environment["PYTHONPATH"] = os.pathsep.join([release_root, *inherited])
-    environment["PYTHONSAFEPATH"] = "1"
+    environment.pop("PYTHONSAFEPATH", None)
     return environment
 
 
@@ -84,6 +91,19 @@ def _python_module_command(module: str) -> list[str]:
     return [sys.executable, "-P", "-m", module]
 
 
+def _pinned_module_command(module: str, *, interpreter: str | None = None) -> list[str]:
+    """Build a self-contained module argv for hosts that sanitize env vars."""
+
+    return [
+        interpreter or sys.executable,
+        "-P",
+        "-c",
+        _PINNED_MODULE_LAUNCHER,
+        str(_release_root()),
+        module,
+    ]
+
+
 def _mcp_python_candidates() -> tuple[Path, ...]:
     configured = os.environ.get("LOOPX_MCP_PYTHON")
     managed = (
@@ -147,14 +167,10 @@ def _mcp_module_command() -> list[str]:
     # Codex intentionally sanitizes PYTHONPATH for stdio MCP children.  Carry
     # the admitted release root in argv and rebuild sys.path inside the child
     # rather than trusting ambient process state.
-    return [
-        _mcp_python_executable(),
-        "-P",
-        "-c",
-        _PINNED_MODULE_LAUNCHER,
-        str(_release_root()),
+    return _pinned_module_command(
         "loopx.collaboration_mcp",
-    ]
+        interpreter=_mcp_python_executable(),
+    )
 
 
 def create_server(
@@ -397,10 +413,25 @@ def _spawn(self, operation_id: str) -> None:
             start_new_session=True, close_fds=True, env=_pinned_release_environment())
 
     def resume(self, operation_id: str) -> dict:
-        row = _read(self.path(operation_id))
-        self._bound(row)
-        if row["status"] not in {"accepted", "rejected"}:
-            self.binding(row["identity"]["binding"]["id"], require_active=True)
+        path = self.path(operation_id)
+        try:
+            with exclusive_file_lock(
+                path, policy=LockAcquisitionPolicy.SINGLE_FLIGHT
+            ):
+                row = _read(path)
+                binding = self._bound(row)
+                if row["status"] == "rejected":
+                    self._recover_validated_settlement(path, row, binding)
+                should_spawn = row["status"] not in {"accepted", "rejected"}
+                if should_spawn:
+                    self.binding(
+                        row["identity"]["binding"]["id"], require_active=True
+                    )
+        except LockAcquireTimeoutError:
+            # A live worker already owns the operation lock.  Observation is
+            # sufficient; spawning another process cannot advance settlement.
+            return self.read(operation_id)
+        if should_spawn:
             self._spawn(operation_id)
         return self.read(operation_id)
 
@@ -419,6 +450,85 @@ def _bound(self, row: dict, *, require_active: bool = False) -> dict:
             raise ValueError("delegation binding changed; reconcile original execution")
         return binding
 
+    @staticmethod
+    def _turn_instance_id(row: dict) -> str:
+        return str(
+            row.get("turn_instance_id")
+            or "delegation-" + row["identity"]["request_id"][:32]
+        )
+
+    def _matching_turn_key(self, row: dict, binding: dict) -> str | None:
+        """Find only the journal bound to this operation's settlement identity."""
+
+        return find_loopx_turn_key_by_settlement_identity(
+            self.root,
+            goal_id=self.goal_id,
+            agent_id=binding["agent_id"],
+            todo_id=binding["todo_id"],
+            turn_instance_id=self._turn_instance_id(row),
+        )
+
+    def _validated_turn_journal(self, row: dict, binding: dict) -> dict | None:
+        turn_key = self._matching_turn_key(row, binding)
+        if turn_key is None:
+            return None
+        journal = load_turn_journal(
+            turn_journal_path(self.root, goal_id=self.goal_id, turn_key=turn_key)
+        )
+        if journal is None:
+            return None
+        phases = journal.get("completed_phases")
+        validation = journal.get("task_validation")
+        host_result = journal.get("host_result")
+        if (
+            journal.get("status") != "in_progress"
+            or journal.get("result_kind") != "validated_progress"
+            or phases != ["host_execute", "typed_result", "validation"]
+            or not isinstance(validation, dict)
+            or validation.get("ok") is not True
+            or not isinstance(host_result, dict)
+            or host_result.get("turn_key") != turn_key
+            or host_result.get("result_kind") != "validated_progress"
+        ):
+            return None
+        row["turn_key"] = turn_key
+        return journal
+
+    def _recover_validated_settlement(
+        self, path: Path, row: dict, binding: dict
+    ) -> bool:
+        """Reopen only an exact, independently validated settlement boundary."""
+
+        journal = self._validated_turn_journal(row, binding)
+        if journal is None:
+            return False
+        decision = effect_runtime_result(
+            "collaboration.delegation.recover_validated_settlement",
+            {
+                "from": row["status"],
+                "identity_matched": True,
+                "journal_status": journal.get("status"),
+                "result_kind": journal.get("result_kind"),
+                "completed_phases": journal.get("completed_phases"),
+                "task_validation_passed": (
+                    isinstance(journal.get("task_validation"), dict)
+                    and journal["task_validation"].get("ok") is True
+                ),
+            },
+        )
+        row["status"] = decision["status"]
+        row["turn_result"] = {
+            "status": journal.get("status"),
+            "result_kind": journal.get("result_kind"),
+            "resume_turn_key": row["turn_key"],
+            "reason": "validated Turn settlement requires same-operation recovery",
+            "host_failure": None,
+            "error": None,
+        }
+        row.pop("error", None)
+        _write(path, row)
+        return True
+
     def adopt_result(self, operation_id: str, consumer_operation_id: str) -> dict:
         return delegation_results.adopt_result(self, operation_id, consumer_operation_id)
 
@@ -472,7 +582,14 @@ def _cli(self, binding: dict, *args: str, timeout: int = 60) -> dict:
         except ValueError as exc:
             raise ValueError("delegation CLI returned no structured result") from exc
         if completed.returncode and "turn" not in args:
-            raise ValueError("delegation canonical command rejected")
+            raise ValueError(
+                str(
+                    value.get("error")
+                    or value.get("reason")
+                    or value.get("reason_code")
+                    or "delegation canonical command rejected"
+                )
+            )
         return value
 
     def _validate(self, binding: dict) -> None:
@@ -538,7 +655,11 @@ def execute(self, operation_id: str) -> None:
     def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
         """Exactly the same profile, workspace and validation arguments for preview/run."""
         # Preserve the journaled validator argv so existing Turns retain their resume identity.
-        validator = [*_python_module_command("loopx.collaboration_mcp"),
+        # Turn validation intentionally runs with a reduced environment.  Carry
+        # the admitted release root in argv just like the native MCP command;
+        # PYTHONPATH pinning on the parent CLI is not a durable validator
+        # identity and may be removed by the host boundary.
+        validator = [*_pinned_module_command("loopx.collaboration_mcp"),
                      "--delegation-action", "validate", "--runtime-root", str(self.root),
                      "--registry", str(self.registry), "--goal-id", self.goal_id,
                      "--agent-id", self.agent_id, "--execution-config", str(self.config),
@@ -571,57 +692,227 @@ def _execution_arguments(self, binding: dict, operation_id: str) -> list[str]:
                      "--validation-failure-kind", "repair_required", *native_tools,
                      *binding["host_args"]]
 
+    def _record_turn_result(
+        self, path: Path, row: dict, result: dict, *, publish: bool = True
+    ) -> None:
+        turn_key = result.get("resume_turn_key")
+        if turn_key:
+            # Validate the public shape before persisting an address supplied
+            # by the CLI boundary.
+            turn_journal_path(self.root, goal_id=self.goal_id, turn_key=turn_key)
+            row["turn_key"] = turn_key
+        row["turn_result"] = {
+            key: result.get(key)
+            for key in (
+                "status",
+                "result_kind",
+                "resume_turn_key",
+                "reason",
+                "host_failure",
+                "error",
+            )
+        }
+        if publish:
+            self._observe(path, row, "turn_returned")
+        else:
+            _write(path, row)
+
+    def _receiver_adopted(self, row: dict, binding: dict) -> bool:
+        request_id = row["identity"]["request_id"]
+        decision, error = _receipt(
+            self.root,
+            "decisions",
+            _entry(self.root, self.goal_id, binding["agent_id"], request_id),
+        )
+        return not error and bool(decision) and decision["decision"] == "adopt"
+
+    def _acquire_delegation_lease(
+        self, path: Path, row: dict, binding: dict
+    ) -> dict:
+        """Acquire the promoted hard lease before worker or completion effects."""
+
+        if not local_authority_is_promoted(
+            runtime_root=self.root,
+            goal_id=self.goal_id,
+        ):
+            row["task_lease"] = {"required": False, "handoff_mode": "legacy"}
+            _write(path, row)
+            return row["task_lease"]
+        handoff_mode = show_goal_handoff_mode(
+            registry_path=self.registry,
+            runtime_root_arg=str(self.root),
+            goal_id=self.goal_id,
+        )["handoff_mode"]
+        if handoff_mode != "hard_lease":
+            row["task_lease"] = {
+                "required": False,
+                "handoff_mode": handoff_mode,
+            }
+            _write(path, row)
+            return row["task_lease"]
+        lease_key = self._turn_instance_id(row)
+        result = self._cli(
+            binding,
+            "todo",
+            "claim",
+            "--goal-id",
+            self.goal_id,
+            "--todo-id",
+            binding["todo_id"],
+            "--claimed-by",
+            binding["agent_id"],
+            "--agent-id",
+            binding["agent_id"],
+            "--claim-operation-id",
+            "delegation-claim-" + row["identity"]["request_id"][:32],
+            "--task-lease-idempotency-key",
+            lease_key,
+        )
+        lease = result.get("lease")
+        if (
+            result.get("ok") is not True
+            or not isinstance(lease, dict)
+            or lease.get("owner") != binding["agent_id"]
+            or lease.get("idempotency_key") != lease_key
+            or lease.get("status") != "active"
+            or not isinstance(lease.get("version"), int)
+        ):
+            raise ValueError(
+                str(
+                    result.get("error")
+                    or result.get("reason")
+                    or "delegation task lease acquisition rejected"
+                )
+            )
+        row["task_lease"] = {
+            "required": True,
+            "handoff_mode": "hard_lease",
+            "idempotency_key": lease_key,
+            "version": lease["version"],
+        }
+        _write(path, row)
+        return row["task_lease"]
+
+    def _complete_delegated_todo(self, row: dict, binding: dict) -> None:
+        lease = row.get("task_lease")
+        if not isinstance(lease, dict):
+            raise ValueError("delegation Todo completion requires its acquired task lease")
+        arguments = [
+            "todo",
+            "complete",
+            "--goal-id",
+            self.goal_id,
+            "--agent-id",
+            binding["agent_id"],
+            "--todo-id",
+            binding["todo_id"],
+            "--claimed-by",
+            binding["agent_id"],
+            "--turn-instance-id",
+            self._turn_instance_id(row),
+            "--note",
+            "Bounded delegated work; requester owns synthesis.",
+        ]
+        if lease.get("required") is True:
+            arguments += [
+                "--task-lease-idempotency-key",
+                str(lease["idempotency_key"]),
+                "--task-lease-expected-version",
+                str(lease["version"]),
+            ]
+        else:
+            arguments.append("--no-follow-up")
+        result = self._cli(binding, *arguments)
+        if result.get("ok") is not True:
+            raise ValueError(
+                str(result.get("error") or result.get("reason") or "delegation Todo completion rejected")
+            )
+
     def _execute(self, path: Path, row: dict, binding: dict) -> None:
         request_id = row["identity"]["request_id"]
         common = ["--goal-id", self.goal_id, "--agent-id", binding["agent_id"]]
-        host = binding["host_args"]
         execution = self._execution_arguments(binding, row["identity"]["operation_id"])
         if row["status"] == "prepared":
-            selected_host = turn_host_arg_option(host, "--host")
-            if not selected_host:
-                raise ValueError("delegation host_args require --host")
-            iteration_context = (
-                turn_host_arg_option(host, "--iteration-context")
-                or "resume-if-available"
-            )
+            row["turn_instance_id"] = self._turn_instance_id(row)
             _write(Path(binding["workspace"]) / "DELEGATION.json", {
                 "request_id": request_id, "brief": _entry(self.root, self.goal_id, binding["agent_id"], request_id)["brief"],
-                "instruction": "Read context and assess this request independently before working. Return results through the bound tools.",
+                "instruction": (
+                    "Use the loopx_delegation tools to read_context and call assess_request "
+                    "for this request before working. If you adopt it, call return_result "
+                    "with the evidence-backed conclusion after validation. Final-answer prose "
+                    "alone is not an adoption or return receipt."
+                ),
             })
-            plan = self._cli(binding, "turn", "plan", *common, "--todo-id", binding["todo_id"],
-                             "--turn-instance-id", "delegation-" + request_id[:32],
-                             "--execution-mode", "isolated-headless", "--scan-root", binding["workspace"],
-                             "--host", selected_host,
-                             "--iteration-context", iteration_context,
-                             "--include-transaction-detail")
-            decision = effect_runtime_result("collaboration.delegation.turn_plan", {"plan": plan})
-            if decision["state"] != "planned":
-                raise ValueError(f"delegation Turn plan rejected: {decision['reason']}")
-            row["turn_key"] = decision["turn_key"]
+            self._acquire_delegation_lease(path, row, binding)
             self._observe(path, row, "running")
         try:
+            todo_completed_for_settlement = False
             if row["status"] == "running":
-                journal = turn_journal_path(self.root, goal_id=self.goal_id, turn_key=row["turn_key"])
-                selector = (["--resume-turn-key", row["turn_key"]] if journal.exists() else
-                            ["--todo-id", binding["todo_id"], "--turn-instance-id", "delegation-" + request_id[:32]])
+                turn_key = self._matching_turn_key(row, binding)
+                selector = (
+                    ["--resume-turn-key", turn_key]
+                    if turn_key
+                    else [
+                        "--todo-id",
+                        binding["todo_id"],
+                        "--turn-instance-id",
+                        self._turn_instance_id(row),
+                    ]
+                )
                 result = self._cli(binding, "turn", "run-once", *common, *selector, *execution,
                                    "--execute", timeout=binding["timeout_seconds"] + 60)
-                row["turn_result"] = {key: result.get(key) for key in ("status", "result_kind", "resume_turn_key", "reason", "host_failure", "error")}
-                self._observe(path, row, "turn_returned")
+                self._record_turn_result(path, row, result)
             result = row["turn_result"]
             if result.get("status") != "committed" or result.get("result_kind") != "validated_progress":
-                row["error"] = "delegation Turn rejected; inspect the original Turn before retrying"
-                self._observe(path, row, "rejected")
-                return
-            decision, error = _receipt(self.root, "decisions", _entry(self.root, self.goal_id, binding["agent_id"], request_id))
-            if error or not decision or decision["decision"] != "adopt":
+                journal = self._validated_turn_journal(row, binding)
+                if journal is None:
+                    row["error"] = str(
+                        result.get("error")
+                        or result.get("reason")
+                        or "delegation Turn rejected; inspect the original Turn before retrying"
+                    )[:180]
+                    self._observe(path, row, "rejected")
+                    return
+                if not self._receiver_adopted(row, binding):
+                    row["error"] = "delegation receiver did not adopt the request"
+                    self._observe(path, row, "rejected")
+                    return
+                self._bound(row, require_active=True)
+                delegation_results.require_dependencies(
+                    self, binding, delegation_results.operation_brief(self, row)
+                )
+                if not isinstance(row.get("task_lease"), dict):
+                    self._acquire_delegation_lease(path, row, binding)
+                self._complete_delegated_todo(row, binding)
+                todo_completed_for_settlement = True
+                result = self._cli(
+                    binding,
+                    "turn",
+                    "run-once",
+                    *common,
+                    "--resume-turn-key",
+                    row["turn_key"],
+                    *execution,
+                    "--execute",
+                    timeout=binding["timeout_seconds"] + 60,
+                )
+                self._record_turn_result(path, row, result, publish=False)
+            if result.get("status") != "committed" or result.get("result_kind") != "validated_progress":
+                raise ValueError(
+                    str(
+                        result.get("error")
+                        or result.get("reason")
+                        or "validated delegation settlement remains incomplete"
+                    )
+                )
+            if not self._receiver_adopted(row, binding):
                 row["error"] = "delegation receiver did not adopt the request"
                 self._observe(path, row, "rejected")
                 return
             self._bound(row, require_active=True)  # revocation or rebinding while the model ran
             delegation_results.require_dependencies(self, binding, delegation_results.operation_brief(self, row))
-            self._cli(binding, "todo", "complete", *common, "--todo-id", binding["todo_id"],
-                      "--no-follow-up", "--note", "Bounded delegated work; requester owns synthesis.")
+            if not todo_completed_for_settlement:
+                self._complete_delegated_todo(row, binding)
             row["artifacts"] = self._accepted(binding)
             if not (_root(self.root) / "replies" / request_id / "conclusion.json").exists():
                 return_result(self.root, self.goal_id, binding["agent_id"], request_id,
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 779f5c7162..d0f8a326ee 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -187,6 +187,33 @@ export function transitionDelegationObservation(params: JsonObject): JsonObject
   return {status: to};
 }
 
+/** Repair only a false terminal observation after the exact Turn validated.
+ *
+ * This does not retry model work.  The host boundary must prove that the
+ * caller-stable settlement identity resolved one canonical journal and that
+ * the journal already completed independent validation.  Recovery returns to
+ * ``turn_returned`` so the existing settlement path can complete durably.
+ */
+export function recoverValidatedDelegationSettlement(params: JsonObject): JsonObject {
+  requireThat(params.from === "rejected", "delegation recovery requires a rejected observation");
+  requireThat(params.identity_matched === true, "delegation recovery requires the exact Turn identity");
+  requireThat(params.journal_status === "in_progress",
+    "delegation recovery requires an unsettled Turn journal");
+  requireThat(params.result_kind === "validated_progress",
+    "delegation recovery requires validated progress");
+  requireThat(params.task_validation_passed === true,
+    "delegation recovery requires independent task validation");
+  requireThat(Array.isArray(params.completed_phases)
+    && JSON.stringify(params.completed_phases) === JSON.stringify([
+      "host_execute", "typed_result", "validation",
+    ]), "delegation recovery requires the validated settlement boundary");
+  return {
+    status: "turn_returned",
+    recovery_kind: "settlement_only",
+    host_reexecution_allowed: false,
+  };
+}
+
 /** Explicit requester decision backed by two current accepted executions. */
 export function recordDelegationAdoption(params: JsonObject): JsonObject {
   const source = requireJsonObject(params.source, "source execution");
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 0882c6941d..cbe7d2bb65 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -7,7 +7,7 @@ import {evaluateTodoPriority} from "./todos/priority.ts";
 import {evaluateUserCompletion} from "./todos/user_completion.ts";
 import {projectTodoSuccession, projectTodoClosure} from "./todos/succession.ts";
 import {projectTodoSummaryLanes, projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts";
-import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts";
+import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts";
 import {planChatMode} from "./collaboration/chat_mode.ts";
 import {resolveConversationScope} from "./collaboration/conversation_scope.ts";
 import {previewTeamPlan, planTeamTransaction, teamTransactionIdentity} from "./work_items/team_plan.ts";
@@ -676,6 +676,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.chat_mode", planChatMode],
     ["collaboration.conversation.scope", resolveConversationScope],
     ["collaboration.delegation.observe", transitionDelegationObservation],
+    ["collaboration.delegation.recover_validated_settlement", recoverValidatedDelegationSettlement],
     ["collaboration.delegation.adoption", recordDelegationAdoption],
     [
       "collaboration.request.normalize",
diff --git a/loopx/control_plane/turn_driver/journal_store.py b/loopx/control_plane/turn_driver/journal_store.py
index b65edc3fc0..ccc0725095 100644
--- a/loopx/control_plane/turn_driver/journal_store.py
+++ b/loopx/control_plane/turn_driver/journal_store.py
@@ -93,6 +93,69 @@ def load_loopx_turn_plan_from_journal(
     return dict(plan)
 
 
+def find_loopx_turn_key_by_settlement_identity(
+    runtime_root: Path,
+    *,
+    goal_id: str,
+    agent_id: str,
+    todo_id: str,
+    turn_instance_id: str,
+) -> str | None:
+    """Resolve one journal from its durable settlement identity.
+
+    A host process can finish after its caller loses the command reply, so the
+    caller-stable ``turn_instance_id`` is the recovery address.  Never select a
+    journal by recency or by a partial Goal/Agent match: the transaction's
+    typed settlement identity must equal every supplied field, and ambiguity
+    fails closed.
+    """
+
+    expected = SettlementIdentity(
+        goal_id=goal_id,
+        agent_id=agent_id,
+        todo_id=todo_id,
+        turn_instance_id=turn_instance_id,
+    )
+    turns_dir = runtime_root / "goals" / goal_id / "turns"
+    if not turns_dir.is_dir():
+        return None
+    matches: list[str] = []
+    for path in sorted(turns_dir.glob("*.json")):
+        turn_key = f"sha256:{path.stem}"
+        if not TURN_KEY_RE.fullmatch(turn_key):
+            continue
+        try:
+            plan = load_loopx_turn_plan_from_journal(
+                runtime_root,
+                goal_id=goal_id,
+                turn_key=turn_key,
+            )
+        except (OSError, TypeError, ValueError, json.JSONDecodeError):
+            continue
+        if _journal_plan_turn_instance_id(plan) != turn_instance_id:
+            continue
+        transaction = plan.get("transaction")
+        settlement = (
+            transaction.get("settlement_plan")
+            if isinstance(transaction, Mapping)
+            else None
+        )
+        identity = (
+            settlement.get("identity") if isinstance(settlement, Mapping) else None
+        )
+        if not isinstance(identity, Mapping):
+            continue
+        try:
+            actual = SettlementIdentity.from_runtime_payload(identity)
+        except RuntimeError:
+            continue
+        if _identity_binding_tuple(actual) == _identity_binding_tuple(expected):
+            matches.append(turn_key)
+    if len(matches) > 1:
+        raise ValueError("LoopX Turn settlement identity matched multiple journals")
+    return matches[0] if matches else None
+
+
 def _journal_plan_turn_instance_id(plan: Mapping[str, Any]) -> str | None:
     transaction = plan.get("transaction")
     if isinstance(transaction, Mapping):
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index 74880b7e92..866689a5b6 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -1,6 +1,6 @@
 import test from "node:test";
 import assert from "node:assert/strict";
-import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, selectDelegationBinding, transitionDelegationObservation} from "../../loopx/control_plane/collaboration/delegation.ts";
+import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "../../loopx/control_plane/collaboration/delegation.ts";
 
 const binding = {id: "review", agent_id: "reviewer", todo_id: "todo_review", workspace: "/fixture",
   requesters: ["coordinator", "analyst"], host_args: ["--host", "dsh"], timeout_seconds: 60, output_refs: ["output.json"]};
@@ -31,6 +31,30 @@ test("message receipt and model return do not imply accepted work", () => {
     canonical_done: true, acceptance_ready: true, artifacts_current: true}), {status: "accepted"});
 });
 
+test("a false rejection can reopen only for exact validated settlement recovery", () => {
+  const evidence = {
+    from: "rejected",
+    identity_matched: true,
+    journal_status: "in_progress",
+    result_kind: "validated_progress",
+    task_validation_passed: true,
+    completed_phases: ["host_execute", "typed_result", "validation"],
+  };
+  assert.deepEqual(recoverValidatedDelegationSettlement(evidence), {
+    status: "turn_returned",
+    recovery_kind: "settlement_only",
+    host_reexecution_allowed: false,
+  });
+  for (const patch of [
+    {from: "turn_returned"},
+    {identity_matched: false},
+    {journal_status: "committed"},
+    {result_kind: "host_failure"},
+    {task_validation_passed: false},
+    {completed_phases: ["host_execute", "typed_result"]},
+  ]) assert.throws(() => recoverValidatedDelegationSettlement({...evidence, ...patch}));
+});
+
 test("inventory paging is bounded and never interprets a missing result as accepted", () => {
   assert.deepEqual(delegationInventoryQuery({}), {limit: 20, cursor: null});
   for (const limit of [0, 51, true, "2"]) assert.throws(() => delegationInventoryQuery({limit}));
diff --git a/tests/test_delegation_preflight.py b/tests/test_delegation_preflight.py
index 2e7153ed70..ac5c09e2fa 100644
--- a/tests/test_delegation_preflight.py
+++ b/tests/test_delegation_preflight.py
@@ -8,6 +8,11 @@
 
 import pytest
 
+from loopx.control_plane.turn_driver import build_loopx_turn_plan
+from loopx.control_plane.turn_driver.executor import (
+    LOOPX_TURN_JOURNAL_SCHEMA_VERSION,
+)
+from loopx.control_plane.turn_driver.journal_store import turn_journal_path
 from test_delegation_cli import cli
 from test_local_delegation import service as delegation_service
 
@@ -111,7 +116,7 @@ def unavailable(**_kwargs):
     assert not (root / "host-started").exists()
 
 
-def test_dispatch_preserves_turn_plan_rejection_before_transaction_read(
+def test_dispatch_preserves_run_once_rejection_before_host_launch(
     service, monkeypatch
 ):
     root, runner = service
@@ -127,30 +132,177 @@ def test_dispatch_preserves_turn_plan_rejection_before_transaction_read(
     })
     calls = []
 
-    def rejected_plan(_binding, *args, **_kwargs):
+    def rejected_turn(_binding, *args, **_kwargs):
         calls.append(args)
         return {
             "ok": False,
-            "schema_version": "loopx_turn_plan_v0",
-            "mode": "plan",
+            "schema_version": "loopx_turn_execution_v0",
+            "mode": "run_once",
             "error": "Requested Turn Todo is not accepted by canonical authority",
             "effects": {"host_invoked": False, "state_written": False,
                         "scheduler_acknowledged": False, "quota_spent": False},
         }
 
-    monkeypatch.setattr(runner, "_cli", rejected_plan)
+    monkeypatch.setattr(runner, "_cli", rejected_turn)
     runner.execute("rejected-plan")
     result = runner.read("rejected-plan")
     assert result["status"] == "rejected"
     assert result["error"] == (
-        "delegation Turn plan rejected: "
         "Requested Turn Todo is not accepted by canonical authority"
     )
-    assert len(calls) == 1 and calls[0][:2] == ("turn", "plan")
+    assert len(calls) == 1 and calls[0][:2] == ("turn", "run-once")
+    assert "--execute" in calls[0]
     assert "turn_key" not in result
     assert not (root / "host-started").exists()
 
 
+def test_hard_lease_delegation_claims_before_host_launch(service, monkeypatch):
+    from loopx import collaboration_mcp as delegation
+
+    _, runner = service
+    monkeypatch.setattr(runner, "_spawn", lambda _: None)
+    monkeypatch.setattr(
+        delegation,
+        "show_goal_handoff_mode",
+        lambda **_kwargs: {"handoff_mode": "hard_lease"},
+    )
+    runner.start("analysis", "leased-dispatch", brief={
+        "schema_version": "collaboration_brief_v0",
+        "purpose": "Exercise an atomic hard-lease dispatch",
+        "context": "The lease must precede the managed host.",
+        "constraints": ["No external actions"],
+        "inputs": [],
+        "acceptance": ["Preserve canonical lease identity"],
+        "return_requirement": "Return no model result",
+    })
+    calls = []
+
+    def canonical(_binding, *args, **_kwargs):
+        calls.append(args)
+        if args[:2] == ("todo", "claim"):
+            key = args[args.index("--task-lease-idempotency-key") + 1]
+            return {
+                "ok": True,
+                "lease": {
+                    "owner": "analyst",
+                    "idempotency_key": key,
+                    "status": "active",
+                    "version": 7,
+                },
+            }
+        return {
+            "ok": False,
+            "schema_version": "loopx_turn_execution_v0",
+            "mode": "run_once",
+            "error": "stop after lease evidence",
+            "effects": {
+                "host_invoked": False,
+                "state_written": False,
+                "scheduler_acknowledged": False,
+                "quota_spent": False,
+            },
+        }
+
+    monkeypatch.setattr(runner, "_cli", canonical)
+    runner.execute("leased-dispatch")
+    result = runner.read("leased-dispatch")
+    row = json.loads(runner.path("leased-dispatch").read_text())
+
+    assert [call[:2] for call in calls] == [("todo", "claim"), ("turn", "run-once")]
+    assert row["task_lease"] == {
+        "required": True,
+        "handoff_mode": "hard_lease",
+        "idempotency_key": row["turn_instance_id"],
+        "version": 7,
+    }
+    assert result["status"] == "rejected"
+    assert result["error"] == "stop after lease evidence"
+
+
+def test_exact_validated_turn_can_reopen_a_false_terminal_observation(
+    service, monkeypatch
+):
+    _, runner = service
+    monkeypatch.setattr(runner, "_spawn", lambda _: None)
+    runner.start("analysis", "recover-settlement", {
+        "schema_version": "collaboration_brief_v0",
+        "purpose": "Recover one validated settlement",
+        "context": "The model result and independent validation already exist.",
+        "constraints": ["Never rerun model work"],
+        "inputs": [],
+        "acceptance": ["Resume only the exact Turn"],
+        "return_requirement": "Return the validated artifact",
+    })
+    path = runner.path("recover-settlement")
+    row = json.loads(path.read_text())
+    turn_instance_id = "delegation-" + row["identity"]["request_id"][:32]
+    row.update(
+        status="rejected",
+        turn_instance_id=turn_instance_id,
+        error="legacy false terminal observation",
+    )
+    path.write_text(json.dumps(row))
+    plan = build_loopx_turn_plan(
+        {
+            "ok": True,
+            "schema_version": "loopx_turn_envelope_v0",
+            "goal_id": runner.goal_id,
+            "agent_id": "analyst",
+            "should_run": True,
+            "effective_action": "normal_run",
+            "action": {
+                "must_attempt": True,
+                "delivery_allowed": True,
+                "quiet_noop_allowed": False,
+                "selected_todo": {"todo_id": "todo_analyst-initial"},
+            },
+            "user": {"action_required": False, "open_count": 0},
+            "writeback": {"spend_after_validation": True},
+            "scheduler": {"action": "run_now"},
+            "action_signature": {
+                "matches": True,
+                "source_hash": "sha256:fixture",
+                "envelope_hash": "sha256:fixture",
+            },
+            "compaction": {"within_budget": True},
+        },
+        host="generic-cli",
+        execution_mode="isolated-headless",
+        turn_instance_id=turn_instance_id,
+        iteration_context_policy="fresh",
+    )
+    transaction = plan["transaction"]
+    turn_key = transaction["turn_key"]
+    journal_path = turn_journal_path(
+        runner.root, goal_id=runner.goal_id, turn_key=turn_key
+    )
+    journal_path.parent.mkdir(parents=True, exist_ok=True)
+    host_result = {
+        "schema_version": "loopx_turn_result_v0",
+        "turn_key": turn_key,
+        "result_kind": "validated_progress",
+        "completed_phases": ["host_execute", "typed_result"],
+    }
+    journal_path.write_text(json.dumps({
+        "schema_version": LOOPX_TURN_JOURNAL_SCHEMA_VERSION,
+        "goal_id": runner.goal_id,
+        "turn_key": turn_key,
+        "status": "in_progress",
+        "result_kind": "validated_progress",
+        "completed_phases": ["host_execute", "typed_result", "validation"],
+        "plan": plan,
+        "host_result": host_result,
+        "task_validation": {"ok": True, "status": "passed"},
+    }))
+
+    binding = runner.binding("analysis", require_active=True)
+    assert runner._recover_validated_settlement(path, row, binding) is True
+    recovered = json.loads(path.read_text())
+    assert recovered["status"] == "turn_returned"
+    assert recovered["turn_key"] == turn_key
+    assert recovered["turn_result"]["resume_turn_key"] == turn_key
+
+
 def test_selected_dsh_profile_is_not_replaced_by_the_default(service):
     root, runner = service
     config = json.loads(runner.config.read_text())
@@ -216,6 +368,11 @@ def test_selected_codex_managed_agent_profile_is_projected_exactly(service):
     assert command[command.index("--execution-config") + 1] == str(runner.config)
     assert "lead" not in command
 
+    validator = json.loads(execution[execution.index("--validation-command-json") + 1])
+    assert validator[:3] == [sys.executable, "-P", "-c"]
+    assert "loopx.collaboration_mcp" in validator
+    assert str(Path(__file__).resolve().parents[1]) in validator
+
     shadow = Path(binding["workspace"]) / "loopx"
     shadow.mkdir()
     (shadow / "__init__.py").write_text("", encoding="utf-8")
diff --git a/tests/test_loopx_turn_executor.py b/tests/test_loopx_turn_executor.py
index 028708d43b..958f4f5536 100644
--- a/tests/test_loopx_turn_executor.py
+++ b/tests/test_loopx_turn_executor.py
@@ -16,6 +16,9 @@
     run_loopx_turn_once,
     validate_loopx_turn_host_result,
 )
+from loopx.control_plane.turn_driver.journal_store import (
+    find_loopx_turn_key_by_settlement_identity,
+)
 from loopx.control_plane.turn_driver.subagent_execution_topology import (
     OPAQUE_REF_PATTERN,
     child_execution_receipts_json_schema,
@@ -98,6 +101,52 @@ def _managed_plan(*, runtime_available: bool) -> dict[str, object]:
     return managed
 
 
+def test_turn_journal_resolves_only_from_exact_settlement_identity(
+    tmp_path: Path,
+) -> None:
+    plan = _plan()
+    transaction = plan["transaction"]
+    assert isinstance(transaction, dict)
+    turn_key = str(transaction["turn_key"])
+    settlement = transaction["settlement_plan"]
+    assert isinstance(settlement, dict)
+    identity = settlement["identity"]
+    assert isinstance(identity, dict)
+    runtime_root = tmp_path / "runtime"
+    path = turn_journal_path(
+        runtime_root, goal_id="fixture-goal", turn_key=turn_key
+    )
+    path.parent.mkdir(parents=True)
+    path.write_text(
+        json.dumps(
+            {
+                "schema_version": LOOPX_TURN_JOURNAL_SCHEMA_VERSION,
+                "goal_id": "fixture-goal",
+                "turn_key": turn_key,
+                "status": "in_progress",
+                "completed_phases": [],
+                "plan": plan,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    assert find_loopx_turn_key_by_settlement_identity(
+        runtime_root,
+        goal_id="fixture-goal",
+        agent_id="codex-fixture",
+        todo_id="todo_fixture0001",
+        turn_instance_id=str(identity["turn_instance_id"]),
+    ) == turn_key
+    assert find_loopx_turn_key_by_settlement_identity(
+        runtime_root,
+        goal_id="fixture-goal",
+        agent_id="other-agent",
+        todo_id="todo_fixture0001",
+        turn_instance_id=str(identity["turn_instance_id"]),
+    ) is None
+
+
 def _adaptive_observation_plan(
     *,
     required_write_scopes: list[str] | None = None,

From f0b6a4a8f5fce647d815df1264a7ef2af368b407 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 19:07:30 +0800
Subject: [PATCH 6/6] fix(delegation): align managed recovery with quota guards

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/cli_commands/turn.py                    | 14 ++++
 loopx/collaboration_mcp.py                    | 73 ++++++++++++----
 .../collaboration/delegation_results.py       | 12 ++-
 loopx/control_plane/collaboration/peers.py    | 30 ++++---
 .../control_plane/quota/heartbeat_receipt.py  | 52 +++++++++++-
 tests/control_plane/test_quota_settlement.py  | 83 +++++++++++++++++++
 tests/test_local_delegation.py                | 18 ++++
 tests/test_loopx_turn_driver.py               | 28 +++++++
 tests/test_peer_collaboration.py              | 10 +++
 9 files changed, 293 insertions(+), 27 deletions(-)

diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index 43368f9b87..a43eff6d57 100644
--- a/loopx/cli_commands/turn.py
+++ b/loopx/cli_commands/turn.py
@@ -30,6 +30,9 @@
     read_heartbeat_settlement,
 )
 from ..control_plane.quota.turn_envelope import build_turn_envelope
+from ..control_plane.work_items.autonomous_replan_obligation import (
+    replan_obligation_id_from_packet,
+)
 from ..control_plane.runtime.status_projection_cache import (
     resolve_status_projection_cache_runtime_root,
 )
@@ -171,6 +174,7 @@ def handle_turn_command(
             args.turn_command == "run-once"
             and args.host == "codex-cli"
             and not resume_requested
+            and not args.resume_turn_key
             and turn_envelope.get("effective_action") != EffectiveAction.GOVERNED_CAPABILITY_INTENT.value
         ):
             session_binding = codex_cli_session_binding(runtime_root, turn_envelope)
@@ -395,6 +399,16 @@ def handle_turn_command(
                 ensure_turn_heartbeat_settlement_receipt(
                     runtime_root,
                     settlement_identity,
+                    semantic_replan_guard_scoped=(
+                        "replan_action_packet" in envelope
+                    ),
+                    semantic_replan_obligation_id=(
+                        replan_obligation_id_from_packet(
+                            envelope.get("replan_action_packet")
+                        )
+                        if "replan_action_packet" in envelope
+                        else None
+                    ),
                 )
 
             def require_effect_ref(
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index a057876f78..c302dc9132 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -726,6 +726,53 @@ def _receiver_adopted(self, row: dict, binding: dict) -> bool:
         )
         return not error and bool(decision) and decision["decision"] == "adopt"
 
+    def _delegation_bootstrap(self, row: dict, binding: dict) -> dict:
+        request_id = row["identity"]["request_id"]
+        return {
+            "request_id": request_id,
+            "brief": _entry(
+                self.root, self.goal_id, binding["agent_id"], request_id
+            )["brief"],
+            "instruction": (
+                "Use the loopx_delegation tools to read_context and call "
+                "assess_request for this request before working. If you adopt "
+                "it, call return_result with the evidence-backed conclusion "
+                "after validation. Final-answer prose alone is not an adoption "
+                "or return receipt."
+            ),
+        }
+
+    def _write_delegation_bootstrap(self, row: dict, binding: dict) -> None:
+        """Expose the compatibility file only while the delegated host runs.
+
+        Some generic hosts still read ``DELEGATION.json`` directly.  It is a
+        host input, not a delivery artifact, so retaining the untracked file
+        after the host exits would make the completion workspace fail its own
+        clean-worktree guard.  Never overwrite an unrelated caller file.
+        """
+
+        path = Path(binding["workspace"]) / "DELEGATION.json"
+        expected = self._delegation_bootstrap(row, binding)
+        if path.exists():
+            if _read(path) != expected:
+                raise ValueError(
+                    "delegation bootstrap path is occupied by another request"
+                )
+            return
+        _write(path, expected)
+
+    def _clear_delegation_bootstrap(self, row: dict, binding: dict) -> None:
+        """Remove only this operation's host input before workspace validation."""
+
+        path = Path(binding["workspace"]) / "DELEGATION.json"
+        if not path.exists():
+            return
+        if _read(path) != self._delegation_bootstrap(row, binding):
+            raise ValueError(
+                "delegation bootstrap changed while the delegated host was running"
+            )
+        path.unlink()
+
     def _acquire_delegation_lease(
         self, path: Path, row: dict, binding: dict
     ) -> dict:
@@ -832,21 +879,12 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
         request_id = row["identity"]["request_id"]
         common = ["--goal-id", self.goal_id, "--agent-id", binding["agent_id"]]
         execution = self._execution_arguments(binding, row["identity"]["operation_id"])
-        if row["status"] == "prepared":
-            row["turn_instance_id"] = self._turn_instance_id(row)
-            _write(Path(binding["workspace"]) / "DELEGATION.json", {
-                "request_id": request_id, "brief": _entry(self.root, self.goal_id, binding["agent_id"], request_id)["brief"],
-                "instruction": (
-                    "Use the loopx_delegation tools to read_context and call assess_request "
-                    "for this request before working. If you adopt it, call return_result "
-                    "with the evidence-backed conclusion after validation. Final-answer prose "
-                    "alone is not an adoption or return receipt."
-                ),
-            })
-            self._acquire_delegation_lease(path, row, binding)
-            self._observe(path, row, "running")
         try:
-            todo_completed_for_settlement = False
+            if row["status"] == "prepared":
+                row["turn_instance_id"] = self._turn_instance_id(row)
+                self._write_delegation_bootstrap(row, binding)
+                self._acquire_delegation_lease(path, row, binding)
+                self._observe(path, row, "running")
             if row["status"] == "running":
                 turn_key = self._matching_turn_key(row, binding)
                 selector = (
@@ -862,6 +900,13 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                 result = self._cli(binding, "turn", "run-once", *common, *selector, *execution,
                                    "--execute", timeout=binding["timeout_seconds"] + 60)
                 self._record_turn_result(path, row, result)
+        finally:
+            # The compatibility bootstrap is private host input.  Keeping it
+            # after the host returns (including an exception or timeout) makes
+            # an otherwise clean Git worktree fail canonical validation.
+            self._clear_delegation_bootstrap(row, binding)
+        try:
+            todo_completed_for_settlement = False
             result = row["turn_result"]
             if result.get("status") != "committed" or result.get("result_kind") != "validated_progress":
                 journal = self._validated_turn_journal(row, binding)
diff --git a/loopx/control_plane/collaboration/delegation_results.py b/loopx/control_plane/collaboration/delegation_results.py
index a3e95095c7..81275ec2ac 100644
--- a/loopx/control_plane/collaboration/delegation_results.py
+++ b/loopx/control_plane/collaboration/delegation_results.py
@@ -20,7 +20,17 @@ def dependencies(service, binding, brief):
     inputs = [item for item in brief["inputs"] if "delegation" in item]
     if not inputs:
         return []
-    materials = input_readiness(service.registry, service.goal_id, {"inputs": inputs}, workspace=binding["workspace"])
+    # The operator-owned binding already authorizes the managed worker's
+    # absolute workspace.  It may intentionally live in another repository,
+    # unlike an ambient peer-inbox workspace which still requires a canonical
+    # project alias before it can replace the Goal root.
+    materials = input_readiness(
+        service.registry,
+        service.goal_id,
+        {"inputs": inputs},
+        workspace=binding["workspace"],
+        configured_workspace=True,
+    )
     result = []
     for item, material in zip(inputs, materials, strict=True):
         link = item["delegation"]
diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py
index 76560c3502..021d28586e 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -285,22 +285,32 @@ def _request_id(value):
     return value
 
 
-def input_readiness(registry, goal_id, brief, *, workspace=None):
+def input_readiness(
+    registry,
+    goal_id,
+    brief,
+    *,
+    workspace=None,
+    configured_workspace: bool = False,
+):
     """Check local input versions, without fetching or claiming agent comprehension."""
     goal = _goal(registry, goal_id)
     goal_workspace = Path(goal["repo"]).resolve()
     selected = goal_workspace
     if workspace is not None and Path(workspace).resolve() != goal_workspace:
-        from ...project_alias import resolve_canonical_project_alias
-
-        alias = resolve_canonical_project_alias(
-            Path(workspace), goal_id=goal_id, global_registry=registry
-        )
-        if (
-            alias.get("applied")
-            and Path(alias["canonical_project"]).resolve() == goal_workspace
-        ):
+        if configured_workspace:
             selected = Path(workspace).resolve()
+        else:
+            from ...project_alias import resolve_canonical_project_alias
+
+            alias = resolve_canonical_project_alias(
+                Path(workspace), goal_id=goal_id, global_registry=registry
+            )
+            if (
+                alias.get("applied")
+                and Path(alias["canonical_project"]).resolve() == goal_workspace
+            ):
+                selected = Path(workspace).resolve()
     workspace = selected
     result = []
     for item in brief.get("inputs", []):
diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py
index 8fe2b37b63..9a3002825f 100644
--- a/loopx/control_plane/quota/heartbeat_receipt.py
+++ b/loopx/control_plane/quota/heartbeat_receipt.py
@@ -170,8 +170,26 @@ def find_heartbeat_receipt(
 def ensure_turn_heartbeat_settlement_receipt(
     runtime_root: Path,
     identity: SettlementIdentity,
+    *,
+    semantic_replan_guard_scoped: bool,
+    semantic_replan_obligation_id: str | None,
 ) -> dict[str, object]:
-    """Idempotently bind a Turn-created quota guard to its settlement identity."""
+    """Idempotently bind a Turn-created quota guard to its settlement identity.
+
+    Current Turn envelopes always carry ``replan_action_packet`` even when no
+    obligation was selected.  Persist that explicit empty selection so an
+    obligation opened while the host is running cannot retroactively reject
+    the admitted Turn.  Old envelopes without the field stay legacy-unscoped.
+    """
+
+    normalized_semantic_replan_obligation_id = (
+        normalize_todo_replan_obligation_id(semantic_replan_obligation_id)
+    )
+    if (
+        semantic_replan_obligation_id is not None
+        and normalized_semantic_replan_obligation_id is None
+    ):
+        raise ValueError("Turn semantic replan obligation id is malformed")
 
     log_path = rollout_event_log_path(runtime_root, identity.goal_id)
     log_path.parent.mkdir(parents=True, exist_ok=True)
@@ -196,7 +214,33 @@ def ensure_turn_heartbeat_settlement_receipt(
                     raise HeartbeatReceiptIdentityConflictError(
                         "Turn heartbeat receipt belongs to another settlement identity"
                     )
-                return effective
+                effective_details = effective.get("details")
+                effective_details = (
+                    effective_details
+                    if isinstance(effective_details, Mapping)
+                    else {}
+                )
+                if not semantic_replan_guard_scoped:
+                    return effective
+                if "semantic_replan_obligation_id" in effective_details:
+                    raw_existing_guard = effective_details.get(
+                        "semantic_replan_obligation_id"
+                    )
+                    existing_guard = normalize_todo_replan_obligation_id(
+                        raw_existing_guard
+                    )
+                    if (
+                        str(raw_existing_guard or "").strip()
+                        and existing_guard is None
+                    ):
+                        raise HeartbeatReceiptIdentityConflictError(
+                            "Turn heartbeat receipt has a malformed semantic replan guard"
+                        )
+                    if existing_guard != normalized_semantic_replan_obligation_id:
+                        raise HeartbeatReceiptIdentityConflictError(
+                            "Turn heartbeat receipt belongs to another semantic replan guard"
+                        )
+                    return effective
 
         details = {
             "turn_instance_id": identity.turn_instance_id,
@@ -206,6 +250,10 @@ def ensure_turn_heartbeat_settlement_receipt(
             "stall_observation": "not_applicable",
             "source": "loopx_turn_run_once",
         }
+        if semantic_replan_guard_scoped:
+            details["semantic_replan_obligation_id"] = (
+                normalized_semantic_replan_obligation_id or ""
+            )
         source_event_id = (
             str(effective.get("event_id") or "").strip()
             if effective is not None
diff --git a/tests/control_plane/test_quota_settlement.py b/tests/control_plane/test_quota_settlement.py
index 28428c40fa..a32f41b76c 100644
--- a/tests/control_plane/test_quota_settlement.py
+++ b/tests/control_plane/test_quota_settlement.py
@@ -20,6 +20,7 @@
 from loopx.control_plane.quota import effect_program as quota_effect_program
 from loopx.control_plane.quota import settlement as quota_settlement
 from loopx.control_plane.quota.heartbeat_receipt import (
+    ensure_turn_heartbeat_settlement_receipt,
     heartbeat_receipt_settlement_replan_obligation_id,
     heartbeat_receipt_settlement_todo_id,
 )
@@ -32,6 +33,9 @@
     quota_rollout_replan_obligation_id,
     quota_rollout_todo_id,
 )
+from loopx.control_plane.quota.error_codes import (
+    HeartbeatReceiptIdentityConflictError,
+)
 from loopx.control_plane.quota.turn_envelope import quota_action_signature_document
 from loopx.control_plane.scheduler.execution_context import (
     SchedulerRuntimeProfile,
@@ -241,6 +245,85 @@ def _append_run_index_record(runtime_root: Path, record: dict) -> None:
         handle.write(json.dumps(record) + "\n")
 
 
+def test_turn_guard_upgrades_matching_legacy_receipt_to_explicit_empty_scope(
+    tmp_path: Path,
+) -> None:
+    runtime_root = tmp_path / "runtime"
+    identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID)
+
+    ensure_turn_heartbeat_settlement_receipt(
+        runtime_root,
+        identity,
+        semantic_replan_guard_scoped=False,
+        semantic_replan_obligation_id=None,
+    )
+    upgraded = ensure_turn_heartbeat_settlement_receipt(
+        runtime_root,
+        identity,
+        semantic_replan_guard_scoped=True,
+        semantic_replan_obligation_id=None,
+    )
+    replayed = ensure_turn_heartbeat_settlement_receipt(
+        runtime_root,
+        identity,
+        semantic_replan_guard_scoped=True,
+        semantic_replan_obligation_id=None,
+    )
+
+    assert upgraded["details"]["semantic_replan_obligation_id"] is None
+    assert replayed == upgraded
+    events = [
+        json.loads(line)
+        for line in rollout_event_log_path(runtime_root, GOAL_ID)
+        .read_text(encoding="utf-8")
+        .splitlines()
+    ]
+    assert len(events) == 2
+    assert events[1]["causality"] == {
+        "caused_by": events[0]["event_id"],
+        "source_event_id": events[0]["event_id"],
+    }
+
+    readback = read_heartbeat_settlement(
+        runtime_root,
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
+        todo_id=TODO_ID,
+        turn_instance_id=TURN_ID,
+    )
+    assert readback is not None
+    assert readback.semantic_replan_guard == {
+        "schema_version": "semantic_replan_guard_v0",
+        "scope": "turn_guard",
+        "selected_obligation_id": None,
+    }
+
+
+def test_turn_guard_refuses_to_change_an_existing_semantic_selection(
+    tmp_path: Path,
+) -> None:
+    runtime_root = tmp_path / "runtime"
+    identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID)
+    obligation_id = "replan-0000000000000001"
+    ensure_turn_heartbeat_settlement_receipt(
+        runtime_root,
+        identity,
+        semantic_replan_guard_scoped=True,
+        semantic_replan_obligation_id=obligation_id,
+    )
+
+    with pytest.raises(
+        HeartbeatReceiptIdentityConflictError,
+        match="another semantic replan guard",
+    ):
+        ensure_turn_heartbeat_settlement_receipt(
+            runtime_root,
+            identity,
+            semantic_replan_guard_scoped=True,
+            semantic_replan_obligation_id=None,
+        )
+
+
 def test_quota_settlement_readback_returns_the_complete_typed_chain(
     tmp_path: Path,
 ) -> None:
diff --git a/tests/test_local_delegation.py b/tests/test_local_delegation.py
index 90166ef191..3326324140 100644
--- a/tests/test_local_delegation.py
+++ b/tests/test_local_delegation.py
@@ -2,6 +2,7 @@
 import json
 import asyncio
 from pathlib import Path
+import subprocess
 import sys
 import time
 from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
@@ -167,6 +168,7 @@ async def disconnect_requester():
     result = wait(reconnected)
     assert result["status"] == "accepted", result
     assert (root / "analyst" / "initial" / "host-invocations").read_text() == "1"
+    assert not (root / "analyst" / "initial" / "DELEGATION.json").exists()
     assert demo.canonical_tasks(root)["todo_analyst-initial"]["done"]
     returned = returns(original.root, original.goal_id, "lead")["items"]
     assert len(returned) == 1
@@ -193,6 +195,22 @@ async def disconnect_requester():
         reconnected.read("analysis-1")
 
 
+def test_host_timeout_removes_private_delegation_bootstrap(service, monkeypatch):
+    root, runner = service
+    monkeypatch.setattr(runner, "_spawn", lambda _operation_id: None)
+    runner.start("analysis", "analysis-timeout", brief())
+
+    def timeout(*_args, **_kwargs):
+        raise subprocess.TimeoutExpired("loopx turn", 1)
+
+    monkeypatch.setattr(runner, "_cli", timeout)
+    runner.execute("analysis-timeout")
+
+    workspace = root / "analyst" / "initial"
+    assert not (workspace / "DELEGATION.json").exists()
+    assert runner.read("analysis-timeout")["error"] == "TimeoutExpired"
+
+
 def test_model_success_without_receiver_adoption_cannot_complete(service):
     root, runner = service
     (root / "skip-adoption").touch()
diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py
index 555baed935..17ef6c1618 100644
--- a/tests/test_loopx_turn_driver.py
+++ b/tests/test_loopx_turn_driver.py
@@ -2114,6 +2114,27 @@ def test_turn_run_once_cli_completes_selected_todo_after_validation(
     assert "todo_id=todo_fixture0001 status=done" in state
     assert "LoopX%20Turn%20validated%20completion" in state
     assert f"completion_turn_key={payload['resume_turn_key']}" in state
+    guard_events = [
+        event
+        for event in (
+            json.loads(line)
+            for line in (
+                runtime
+                / "goals"
+                / "loopx-turn-fixture"
+                / "rollout-event-log.jsonl"
+            )
+            .read_text(encoding="utf-8")
+            .splitlines()
+        )
+        if event.get("event_kind") == "quota_should_run"
+        and event.get("run_id") == payload["resume_turn_key"]
+    ]
+    assert len(guard_events) == 1
+    # Managed Turn uses the same explicit semantic-replan guard as ordinary
+    # quota should-run.  Null means this admitted Turn selected no obligation;
+    # an obligation opened while its host runs belongs to the next Turn.
+    assert guard_events[0]["details"]["semantic_replan_obligation_id"] is None
 
     next_plan_output = io.StringIO()
     with contextlib.redirect_stdout(next_plan_output):
@@ -3309,11 +3330,14 @@ def test_turn_run_once_cli_resumes_session_from_recoverable_failed_turn(
     project, runtime, registry = _write_live_fixture(tmp_path)
     session_available = False
     session_actions: list[str] = []
+    session_binding_calls = 0
 
     def fake_session_binding(
         _runtime_root: Path,
         _turn_envelope: dict[str, object],
     ) -> dict[str, str] | None:
+        nonlocal session_binding_calls
+        session_binding_calls += 1
         if not session_available:
             return None
         return {
@@ -3399,3 +3423,7 @@ def fake_codex_host(
     assert recovered["status"] == "stopped"
     assert recovered["quota_slot_spend_count"] == 0
     assert session_actions == ["start_new", "resume"]
+    # A journal resume resolves the saved Turn before consulting Codex session
+    # state.  Rebuilding a binding from the fresh decision can fail after the
+    # selected Todo has already completed and disappeared from the live route.
+    assert session_binding_calls == 2
diff --git a/tests/test_peer_collaboration.py b/tests/test_peer_collaboration.py
index 0f9ef202f4..c1920b0611 100644
--- a/tests/test_peer_collaboration.py
+++ b/tests/test_peer_collaboration.py
@@ -396,6 +396,16 @@ def git(*args):
     outside.mkdir()
     result = input_readiness(registry, "delivery", brief, workspace=outside)[0]
     assert result["status"] == "available" and result["basis"] == "goal_workspace"
+    (outside / "inputs").mkdir()
+    (outside / "inputs/demand.csv").write_text("configured delegation version")
+    result = input_readiness(
+        registry,
+        "delivery",
+        brief,
+        workspace=outside,
+        configured_workspace=True,
+    )[0]
+    assert result["status"] == "changed" and result["basis"] == "receiver_worktree"
 
 
 def test_unrelated_damaged_return_route_does_not_break_legacy_inbox(scenario):