From 0aa4cab18f83b3c833351fee01593080db94ef51 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 22 Oct 2026 05:42:16 +0800
Subject: [PATCH 1/4] fix(collaboration): recheck original source grants in
 workers

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            |  18 ++
 loopx/collaboration_mcp.py                    |  25 +--
 .../collaboration/source_grant_observation.py |  41 ++++-
 .../collaboration/source_grants.ts            |  52 +++---
 .../test_delegation_source_authority.py       | 165 ++++++++++++++++++
 tests/control_plane_ts/source_grants.test.ts  |  10 ++
 6 files changed, 279 insertions(+), 32 deletions(-)
 create mode 100644 tests/control_plane/test_delegation_source_authority.py

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 21cfd8d9e6..e7170aee4d 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -53,6 +53,17 @@ work or configures owner acceptance. Disable by removing the exact binding
 from the operator configuration; existing operations retain their history and
 cannot re-execute or return accepted evidence under a revoked grant.
 
+Chat-originated execution also retains its original Inbox request and frozen
+conversation provenance. The worker rechecks that source's current sender,
+recipient and exact Goal/Agent/requester/binding grant at entry, before Host
+launch, and before completion. Missing provenance or a revoked source grant
+rejects queued work on the original operation before a Host or Turn starts.
+Historical inspection remains available. If the Host already returned, its
+original Turn and unsettled effects remain visible; restoring authority permits
+only that operation's settlement without repeating the Host. Completing the
+requester's Chat Turn does not cancel independently delegated work. Operator/MCP
+operations without a Chat source retain their existing binding authority.
+
 Member completion uses the ordinary active-Goal continuation, including legacy
 non-hard-lease routes. It does not declare terminal `no_followup` for a
 requester-owned synthesis. This lets controller validation finish the Todo
@@ -68,6 +79,13 @@ Goal 验收的任务，必须通过既有 `todo add --validation-command-json` 
 不配置 owner 验收；移除原配置中的精确 binding 即撤销 grant，保留历史但拒绝重新
 执行或返回已撤权任务的有效结果。
 
+Chat 发起的执行还保留原 Inbox 请求与固定的对话来源。worker 在入口、Host 启动前和
+完成前，重新核对来源当前的发送者、接收者及精确 Goal/Agent/requester/binding grant。
+来源不可读或 grant 被撤销时，排队工作在原 operation 上拒绝，不启动 Host 或 Turn；
+历史记录仍可查看。Host 已返回时保留原 Turn 与未结算效果，恢复授权后仅续接该
+operation 的结算，不重跑 Host。请求方 Chat Turn 完成不取消独立委派的工作。
+没有 Chat 来源的 operator/MCP 操作继续沿用原 binding 授权。
+
 成员完成沿用普通 active-Goal 继续状态，旧的非 hard-lease 路径也如此；不会为仍由
 请求方负责的汇总声明 terminal `no_followup`。因此可以先通过 controller 校验完成
 Todo，再仅恢复原 Turn 的结算，不重跑 host。显式终结仍须具备匹配的写回和扣额回执。
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index a850904a07..4b1635be2f 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -63,6 +63,7 @@
     decide_collaboration_lifecycle,
 )
 from .control_plane.collaboration import delegation_results, delegation_stop_lease, delegation_validation
+from .control_plane.collaboration.source_grant_observation import require_delegation_source_execution
 from .control_plane.collaboration.peers import (
     _goal,
     consume_return,
@@ -761,9 +762,7 @@ def resume(self, operation_id: str) -> dict:
                         raise ValueError(DELEGATION_STOPPED_MESSAGE) from None
                 should_spawn = row["status"] not in DELEGATION_TERMINAL_STATUSES
                 if should_spawn:
-                    self.binding(
-                        row["identity"]["binding"]["id"], require_active=True
-                    )
+                    self._bound(row, require_active=True)
         except LockAcquireTimeoutError:
             # A live worker already owns the operation lock.  Observation is
             # sufficient; spawning another process cannot advance settlement.
@@ -823,6 +822,10 @@ def _bound(self, row: dict, *, require_active: bool = False) -> dict:
         binding = self.binding(row["identity"]["binding"]["id"], require_active=require_active)
         if row["identity"]["binding"] != binding:
             raise ValueError("delegation binding changed; reconcile original execution")
+        if require_active and "source_request_id" in row["identity"]:
+            require_delegation_source_execution(self.root, self.registry, goal_id=self.goal_id,
+                requester_agent_id=self.agent_id, source_request_id=row["identity"]["source_request_id"], binding=binding,
+                caller_goal_ref=self._caller_goal_ref())
         return binding
 
     @staticmethod
@@ -1370,13 +1373,12 @@ def execute(self, operation_id: str) -> None:
                 # Different request ids cannot run the same assigned task concurrently.
                 task_lock = _root(self.root) / "execution-slots" / _hash([self.goal_id, binding["todo_id"]])
                 with exclusive_file_lock(task_lock, policy=LockAcquisitionPolicy.SINGLE_FLIGHT):
-                    try:
-                        self._execute(path, row, binding)
-                    except (ValueError, KeyError, subprocess.TimeoutExpired, EffectRuntimeRemoteError) as exc:
-                        row["error"] = str(exc)[:180] if isinstance(exc, (ValueError, EffectRuntimeRemoteError)) else type(exc).__name__
-                        self._fenced_write(path, row)
-                        if row["status"] == "prepared":
-                            self._observe(path, row, "rejected")
+                    self._execute(path, row, binding)
+            except (ValueError, KeyError, subprocess.TimeoutExpired, EffectRuntimeRemoteError) as exc:
+                row["error"] = str(exc)[:180] if isinstance(exc, (ValueError, EffectRuntimeRemoteError)) else type(exc).__name__
+                self._fenced_write(path, row)
+                if row["status"] == "prepared":
+                    self._observe(path, row, "rejected")
             except DelegationStopRequested as stop:
                 # SIGTERM, a checkpoint or a fenced write acknowledges intent.
                 # Each Host supervisor may still be cleaning its own group;
@@ -1724,6 +1726,7 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                 acceptance = delegation_validation.capture(self, binding)
                 if acceptance["plan"]["state"] != "ready" or not acceptance["files_current"]:
                     raise ValueError("delegation task acceptance rejected before host launch")
+                self._bound(row, require_active=True)
                 row["turn_instance_id"] = self._turn_instance_id(row)
                 self._write_delegation_bootstrap(row, binding)
                 self._acquire_delegation_lease(path, row, binding)
@@ -1735,6 +1738,7 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                 self._acquire_delegation_lease(path, row, binding)
             if row["status"] == "running":
                 self._raise_if_stop_requested(path)
+                self._bound(row, require_active=True)
                 turn_key = delegation_validation.matching_turn_key(self, row, binding)
                 selector = (
                     ["--resume-turn-key", turn_key]
@@ -1787,6 +1791,7 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
             # fence through settlement, result publication and acceptance.
             with exclusive_file_lock(self._dispatch_lock(path)):
                 self._raise_if_stop_requested(path)
+                self._bound(row, require_active=True)
                 self._complete_delegated_todo(row, binding)
                 if needs_settlement:
                     result = self._cli(
diff --git a/loopx/control_plane/collaboration/source_grant_observation.py b/loopx/control_plane/collaboration/source_grant_observation.py
index f80adc55da..0818679d20 100644
--- a/loopx/control_plane/collaboration/source_grant_observation.py
+++ b/loopx/control_plane/collaboration/source_grant_observation.py
@@ -156,19 +156,58 @@ def _source_registry_recipients(registry_path: Path | None, *, context_only: boo
 
 
 def source_execution_bindings(
-    runtime_root: Path, registry_path: Path | None, session: dict[str, Any], turn: dict[str, Any]
+    runtime_root: Path, registry_path: Path | None, session: dict[str, Any], turn: dict[str, Any],
+    *, required_binding: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Observe one compatible registry; TS owns sender and exact binding grants."""
     if conversation_scope(session, origin=turn.get("origin", "unknown"))["kind"] != "external_audience":
+        if required_binding is not None:
+            raise ValueError("original source execution grant requires verified external provenance")
         return {"bindings": []}
     observed = _source_registry_recipients(registry_path)
     ingress, source = external_source_policy(runtime_root, session, turn)
     result: dict[str, Any] = effect_runtime_result("collaboration.source.execution_bindings", {
         "source": source, "sender_id": ingress["sender_id"], "available": observed["available"],
+        **({"required_binding": required_binding} if required_binding is not None else {}),
     })
     return result
 
 
+def require_delegation_source_execution(runtime_root: Path, registry_path: Path, *,
+        goal_id: str, requester_agent_id: str, source_request_id: str,
+        binding: dict[str, Any], caller_goal_ref: dict[str, Any] | None) -> None:
+    """Reobserve the original Chat source; the existing TS owner decides its grant.
+
+    A return address only locates frozen evidence. It never grants execution,
+    and observing it must not recover or mutate the original Chat store.
+    """
+    from ...file_lock import exclusive_file_lock
+    from .goal_instance_scope import collaboration_goal_scope
+    from .inbox import _entry
+    from .source_chat_observation import observe_source_chat
+
+    try:
+        with exclusive_file_lock((_root(runtime_root) / "policy.json").with_suffix(".lock")), collaboration_goal_scope(
+            registry_path, goal_id=goal_id, agents=(requester_agent_id, binding["agent_id"]),
+            caller_goal_ref=caller_goal_ref, require_active=True,
+        ) as scope:
+            original = _entry(runtime_root, goal_id, binding["agent_id"], source_request_id, scope=scope)
+            route = _read(_root(runtime_root) / "roundtrips" / (source_request_id + ".json"))
+            if (original.get("execution_binding_id") != binding["id"]
+                    or any(route.get(key) != original.get(key) for key in
+                           ("request_id", "goal_id", "agent_id", "source_id", "goal_ref"))):
+                raise ValueError("original source execution identity changed")
+            session, turn = observe_source_chat(runtime_root, route)
+            target = {"goal_id": goal_id, "agent_id": binding["agent_id"]}
+            grant = source_context_target_authority(runtime_root, session, turn, target)
+            if grant.get("source_id") != original["source_id"] or target not in grant.get("targets", []):
+                raise ValueError("original source execution context is no longer authorized")
+            source_execution_bindings(runtime_root, registry_path, session, turn,
+                required_binding={**target, "requester_agent_id": requester_agent_id, "binding_id": binding["id"]})
+    except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRejected) as exc:
+        raise ValueError("original source execution authority is unavailable or revoked") from exc
+
+
 def source_context_authority(
     runtime_root: Path, registry_path: Path, session: dict[str, Any], turn: dict[str, Any]
 ) -> dict[str, Any]:
diff --git a/loopx/control_plane/collaboration/source_grants.ts b/loopx/control_plane/collaboration/source_grants.ts
index 3db213221f..50d4e15278 100644
--- a/loopx/control_plane/collaboration/source_grants.ts
+++ b/loopx/control_plane/collaboration/source_grants.ts
@@ -196,8 +196,29 @@ export function validateSourceIngress(params: JsonObject): JsonObject {
   return {verified: true};
 }
 
+function executionBinding(raw: unknown): JsonObject {
+  const row = requireJsonObject(raw, "source execution binding");
+  const keys = ["goal_id", "agent_id", "requester_agent_id", "binding_id"];
+  if (Object.keys(row).some(key => !keys.includes(key)) || keys.some(key => !Object.hasOwn(row, key))) {
+    throw new EffectRuntimeRequestError("exact source execution binding fields required");
+  }
+  const binding: JsonObject = {};
+  for (const key of keys) {
+    const id = requireNonEmptyString(row[key], key);
+    if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,159}$/.test(id)) {
+      throw new EffectRuntimeRequestError("invalid source execution binding identity");
+    }
+    binding[key] = id;
+  }
+  if (binding.requester_agent_id === binding.agent_id) {
+    throw new EffectRuntimeRequestError("execution requester must be an independent registered Agent");
+  }
+  return binding;
+}
+
 /** A separate, exact operator grant for existing governed work. Context delivery
  * alone never authorizes launch; neither models nor sources choose host profiles.
+ * required_binding lets an executing worker recheck its original exact grant.
  */
 export function sourceExecutionBindings(params: JsonObject): JsonObject {
   const source = requireJsonObject(params.source, "source policy");
@@ -207,32 +228,21 @@ export function sourceExecutionBindings(params: JsonObject): JsonObject {
   if (!Array.isArray(rows) || rows.length > 100) {
     throw new EffectRuntimeRequestError("bounded source execution bindings required");
   }
-  const bindings = rows.map(raw => {
-    const row = requireJsonObject(raw, "source execution binding");
-    const keys = ["goal_id", "agent_id", "requester_agent_id", "binding_id"];
-    if (Object.keys(row).some(key => !keys.includes(key)) || keys.some(key => !Object.hasOwn(row, key))) {
-      throw new EffectRuntimeRequestError("exact source execution binding fields required");
-    }
-    const binding: JsonObject = {};
-    for (const key of keys) {
-      const id = requireNonEmptyString(row[key], key);
-      if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,159}$/.test(id)) {
-        throw new EffectRuntimeRequestError("invalid source execution binding identity");
-      }
-      binding[key] = id;
-    }
-    if (binding.requester_agent_id === binding.agent_id) {
-      throw new EffectRuntimeRequestError("execution requester must be an independent registered Agent");
-    }
-    return binding;
-  });
+  const bindings = rows.map(executionBinding);
   const identities = bindings.map(row => JSON.stringify([row.goal_id, row.agent_id, row.binding_id]));
   if (new Set(identities).size !== identities.length) {
     throw new EffectRuntimeRequestError("ambiguous source execution binding");
   }
-  return {bindings: bindings.filter(row =>
+  const permitted = bindings.filter(row =>
     authorized.some(target => target.goal_id === row.goal_id && target.agent_id === row.agent_id)
-    && available.some(target => target.goal_id === row.goal_id && target.agent_id === row.requester_agent_id))};
+    && available.some(target => target.goal_id === row.goal_id && target.agent_id === row.requester_agent_id));
+  if (Object.hasOwn(params, "required_binding")) {
+    const required = executionBinding(params.required_binding);
+    if (!permitted.some(row => Object.keys(required).every(key => row[key] === required[key]))) {
+      throw new EffectRuntimeRequestError("original source execution binding is no longer authorized");
+    }
+  }
+  return {bindings: permitted};
 }
 
 /** Plan one trusted-local configuration change; the adapter owns locking and IO.
diff --git a/tests/control_plane/test_delegation_source_authority.py b/tests/control_plane/test_delegation_source_authority.py
new file mode 100644
index 0000000000..f0b76e64be
--- /dev/null
+++ b/tests/control_plane/test_delegation_source_authority.py
@@ -0,0 +1,165 @@
+"""A persisted Chat handoff cannot outlive its original execution grant.
+
+Real File/SQLite authority and a real worker use the existing fixture Host.
+This checks source revocation, not live App or provider qualification.
+"""
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from loopx.capabilities.manager_context import _root, _write, deliver, execution
+from loopx.control_plane.collaboration import delegation_results, delegation_validation
+from loopx.control_plane.collaboration.inbox import _read
+from test_independent_delegation_validation import independent_binding
+from test_local_delegation import demo, service as delegation_service  # noqa: F401
+from test_manager_context_execution import source
+
+
+@pytest.fixture
+def saved_handoff(delegation_service, monkeypatch, request):  # noqa: F811
+    root, service = delegation_service
+    independent_binding(delegation_service)
+    data = json.loads(service.registry.read_text())
+    config = Path(data["goals"][0]["repo"]) / ".loopx/config/delegations.json"
+    config.parent.mkdir(parents=True, exist_ok=True)
+    config.write_bytes(service.config.read_bytes())
+    data["goals"][0]["spawn_policy"] = {"execution_config": ".loopx/config/delegations.json"}
+    service.registry.write_text(json.dumps(data))
+    values = source(service.root, service.registry, goal_id=service.goal_id,
+                    agent_id="analyst", requester="lead", binding="analysis",
+                    chat_root=root / "source-chat" if getattr(request, "param", False) else None)
+    store, session, turn, request, receipt, _ = values
+    deliver(service.root, service.registry, session=session, turn=turn,
+            request=request, source_store=store)
+    monkeypatch.setattr(execution.Delegations, "_spawn", lambda *_: None)
+    submitted = execution.dispatch(service.root, service.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert submitted["submitted"] and submitted["status"] == "prepared", submitted
+    worker, binding = execution._service(service.root, service.registry, {
+        "goal_id": service.goal_id, "agent_id": "analyst",
+        "requester_agent_id": "lead", "binding_id": "analysis"})
+    return root, worker, binding, values, submitted["operation_id"]
+
+
+def run_worker(worker, operation):
+    completed = subprocess.run([sys.executable, "-m", "loopx.collaboration_mcp",
+        "--delegation-action", "worker", "--runtime-root", str(worker.root),
+        "--registry", str(worker.registry), "--goal-id", worker.goal_id,
+        "--agent-id", worker.agent_id, "--execution-config", str(worker.config),
+        "--operation-id", operation], capture_output=True, text=True, timeout=60)
+    assert completed.returncode == 0, completed.stderr
+
+
+@pytest.mark.parametrize("change", ["revoke", "sender", "blocked", "requester", "binding",
+                                  "policy_missing", "route_missing", "source_missing", "body", "closed"])
+def test_changed_queued_source_cannot_reach_the_real_worker_host(saved_handoff, change):
+    root, worker, binding, (store, session, turn, _, receipt, policy), operation = saved_handoff
+    before = demo.canonical_tasks(root)
+    grant = policy["sources"][session["channel_id"]]
+    if change == "revoke":
+        grant["execution_bindings"] = []
+    elif change == "sender":
+        grant["sender_ids"] = ["another-person"]
+    elif change == "blocked":
+        grant["blocked_targets"] = [{"goal_id": worker.goal_id, "agent_id": binding["agent_id"]}]
+    elif change in {"requester", "binding"}:
+        grant["execution_bindings"][0]["requester_agent_id" if change == "requester" else "binding_id"] = "reviewer"
+    elif change == "route_missing":
+        (_root(worker.root) / "roundtrips" / (receipt["request_id"] + ".json")).unlink()
+    elif change == "source_missing":
+        store._turn_path(session["session_id"], turn["turn_id"]).unlink()
+    elif change == "body":
+        # Simulate corrupted immutable provenance in this disposable Chat store.
+        _write(store._turn_path(session["session_id"], turn["turn_id"]),
+               {**turn, "message": "Different source request"})
+    elif change == "closed":
+        store.update_session(session["session_id"], status="closed")
+    _write(_root(worker.root) / "policy.json", policy)
+    if change == "policy_missing":
+        (_root(worker.root) / "policy.json").unlink()
+    chat_before = {p.relative_to(store.root): p.read_bytes() for p in store.root.rglob("*") if p.is_file()}
+    run_worker(worker, operation)
+    assert {p.relative_to(store.root): p.read_bytes() for p in store.root.rglob("*") if p.is_file()} == chat_before
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    assert demo.canonical_tasks(root) == before
+    row = _read(worker.path(operation))
+    assert row["status"] == "rejected"
+    assert "source execution" in row["error"]
+    # Revocation blocks effects, not inspection of the historical operation.
+    assert worker.read(operation)["operation_id"] == operation
+
+
+@pytest.mark.parametrize("saved_handoff", [False, True], indirect=True)
+def test_completed_source_and_separate_chat_store_keep_authorized_execution(saved_handoff):
+    root, worker, binding, (store, session, turn, _, receipt, _), operation = saved_handoff
+    store.update_turn(session["session_id"], turn["turn_id"], status="completing",
+                      response={"context_handoff_receipt": receipt})
+    store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"])
+    assert store.load_turn(session["session_id"], turn["turn_id"])["status"] == "completed"
+    run_worker(worker, operation)
+    row = worker.read(operation)
+    assert row["status"] == "accepted", row
+    assert demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    run_worker(worker, operation)
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+def revoke(worker, session, policy):
+    policy["sources"][session["channel_id"]]["execution_bindings"] = []
+    _write(_root(worker.root) / "policy.json", policy)
+
+
+def test_revocation_during_real_acceptance_capture_is_rechecked_before_launch(saved_handoff, monkeypatch):
+    _, worker, binding, (_, session, _, _, _, policy), operation = saved_handoff
+    capture = delegation_validation.capture
+    def changed(*args):
+        observed = capture(*args)
+        revoke(worker, session, policy)
+        return observed
+    monkeypatch.setattr(delegation_validation, "capture", changed)
+    worker.execute(operation)
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert _read(worker.path(operation))["status"] == "rejected"
+
+
+@pytest.mark.parametrize("checkpoint", ["host_return", "completion_dependencies"])
+def test_revocation_after_host_preserves_original_turn_for_authorized_settlement(saved_handoff, monkeypatch, checkpoint):
+    root, worker, binding, (_, session, _, _, _, policy), operation = saved_handoff
+    original_policy = json.loads(json.dumps(policy))
+    cli, dependencies = worker._cli, delegation_results.require_dependencies
+    def returned(*args, **kwargs):
+        result = cli(*args, **kwargs)
+        if "run-once" in args:
+            revoke(worker, session, policy)
+        return result
+    def checked(*args):
+        dependencies(*args)
+        revoke(worker, session, policy)
+    with monkeypatch.context() as boundary:
+        if checkpoint == "host_return":
+            boundary.setattr(worker, "_cli", returned)
+        else:
+            boundary.setattr(delegation_results, "require_dependencies", checked)
+        worker.execute(operation)
+    held = _read(worker.path(operation))
+    assert held["status"] == "turn_returned", held
+    assert "source execution" in held["error"]
+    assert not demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+    assert delegation_validation.validated_turn_journal(worker, held, binding) is not None
+    counter = Path(binding["workspace"]) / "host-invocations"
+    assert counter.read_text() == "1"
+    with pytest.raises(ValueError, match="source execution"):
+        worker.resume(operation)
+    _write(_root(worker.root) / "policy.json", original_policy)
+    worker.execute(operation)
+    settled = worker.read(operation)
+    assert settled["status"] == "accepted", settled
+    assert _read(worker.path(operation))["turn_key"] == held["turn_key"]
+    assert counter.read_text() == "1"
+    assert demo.canonical_tasks(root)[binding["todo_id"]]["done"]
diff --git a/tests/control_plane_ts/source_grants.test.ts b/tests/control_plane_ts/source_grants.test.ts
index a6a4249fdd..c971ba9f4e 100644
--- a/tests/control_plane_ts/source_grants.test.ts
+++ b/tests/control_plane_ts/source_grants.test.ts
@@ -63,6 +63,16 @@ test("execution needs an independent exact grant, current sender and both regist
   assert.deepEqual(sourceExecutionBindings(params), {bindings: []});
   const granted = {...source, execution_bindings: [binding]};
   assert.deepEqual(sourceExecutionBindings({...params, source: granted}), {bindings: [binding]});
+  const required = {...params, source: granted, required_binding: binding};
+  assert.deepEqual(sourceExecutionBindings(required), {bindings: [binding]});
+  for (const field of ["goal_id", "agent_id", "requester_agent_id", "binding_id"]) {
+    assert.throws(() => sourceExecutionBindings({...required,
+      required_binding: {...binding, [field]: "different"}}));
+  }
+  for (const change of [{source}, {available: [worker]},
+    {source: {...granted, blocked_targets: [worker]}}, {required_binding: null}]) {
+    assert.throws(() => sourceExecutionBindings({...required, ...change}));
+  }
   assert.deepEqual(sourceExecutionBindings({...params, source: granted, available: [worker]}), {bindings: []});
   assert.deepEqual(sourceExecutionBindings({...params, source: {...granted, blocked_targets: [worker]}}), {bindings: []});
   assert.throws(() => sourceExecutionBindings({...params, source: granted, sender_id: "other"}));

From 809d9f1811d5010483728d74e428951a472d2e25 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:08:21 +0800
Subject: [PATCH 2/4] fix(collaboration): verify the original handoff on replay

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/local-delegation.md            | 13 +++
 .../capabilities/manager_context/execution.py |  6 +-
 loopx/collaboration_mcp.py                    |  7 +-
 .../control_plane/collaboration/delegation.ts | 34 ++++++++
 loopx/control_plane/collaboration/peers.py    | 23 +++---
 .../collaboration/source_grant_observation.py | 41 ++++++++++
 .../control_plane/effect_runtime_handlers.ts  |  1 +
 .../test_delegation_source_replay.py          | 81 +++++++++++++++++++
 tests/control_plane_ts/delegation.test.ts     | 34 +++++++-
 tests/test_manager_context_execution.py       |  7 ++
 10 files changed, 231 insertions(+), 16 deletions(-)
 create mode 100644 tests/control_plane/test_delegation_source_replay.py

diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index e7170aee4d..52ddcb7b4b 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -64,6 +64,14 @@ only that operation's settlement without repeating the Host. Completing the
 requester's Chat Turn does not cancel independently delegated work. Operator/MCP
 operations without a Chat source retain their existing binding authority.
 
+Chat retry also verifies the saved operation against the original Inbox request,
+complete brief, peer request, current binding and original return Session/Turn
+before reporting its status. Selecting another authorized binding does not
+retarget an existing operation. A mismatch returns the existing unavailable
+execution response without publishing a peer request or resubmitting the worker;
+matching prepared or terminal operations are read back in place. Historical
+operator inspection remains available through the ordinary read command.
+
 Member completion uses the ordinary active-Goal continuation, including legacy
 non-hard-lease routes. It does not declare terminal `no_followup` for a
 requester-owned synthesis. This lets controller validation finish the Todo
@@ -86,6 +94,11 @@ Chat 发起的执行还保留原 Inbox 请求与固定的对话来源。worker 
 operation 的结算，不重跑 Host。请求方 Chat Turn 完成不取消独立委派的工作。
 没有 Chat 来源的 operator/MCP 操作继续沿用原 binding 授权。
 
+Chat 重试在报告状态前，也核对已保存 operation 的原 Inbox 请求、完整 brief、peer
+请求、当前 binding 及原回传 Session/Turn。选择另一个获准 binding 不会改派既有
+operation；身份不符时返回原执行不可用反馈，不新发 peer 请求或重新提交 worker。
+身份一致的 prepared 或终态操作原位读回；普通 read 命令仍可查看历史。
+
 成员完成沿用普通 active-Goal 继续状态，旧的非 hard-lease 路径也如此；不会为仍由
 请求方负责的汇总声明 terminal `no_followup`。因此可以先通过 controller 校验完成
 Todo，再仅恢复原 Turn 的结算，不重跑 host。显式终结仍须具备匹配的写回和扣额回执。
diff --git a/loopx/capabilities/manager_context/execution.py b/loopx/capabilities/manager_context/execution.py
index 42c1e4b371..59387f9d7a 100644
--- a/loopx/capabilities/manager_context/execution.py
+++ b/loopx/capabilities/manager_context/execution.py
@@ -15,7 +15,7 @@
 from ...collaboration_mcp import Delegations
 from ...control_plane.collaboration import conversation_scope
 from ...control_plane.collaboration.source_grant_observation import (
-    source_execution_bindings,
+    read_delegation_source_execution, source_execution_bindings,
 )
 from ...control_plane.effect_runtime import EffectRuntimeRejected
 from ...orchestration import compact_orchestration_policy, normalize_subagent_execution_config
@@ -89,7 +89,9 @@ def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[
         # Recover an existing operation instead of probing a now-completed Todo
         # and misreporting an accepted result as a refused new launch.
         if service.path(operation).exists():
-            row = service.read(operation)
+            row = read_delegation_source_execution(service, selected, operation,
+                source_request_id=receipt["request_id"], brief=request["brief"],
+                conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]})
             return {"submitted": True, "operation_id": operation, "todo_id": binding["todo_id"],
                     "status": row["status"], "replayed": True}
         preflight = service.inspect(selected)
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 4b1635be2f..8319ff03e6 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -63,7 +63,9 @@
     decide_collaboration_lifecycle,
 )
 from .control_plane.collaboration import delegation_results, delegation_stop_lease, delegation_validation
-from .control_plane.collaboration.source_grant_observation import require_delegation_source_execution
+from .control_plane.collaboration.source_grant_observation import (
+    read_delegation_source_execution, require_delegation_source_execution,
+)
 from .control_plane.collaboration.peers import (
     _goal,
     consume_return,
@@ -708,6 +710,9 @@ def start(self, binding_id: str, operation_id: str, brief: dict,
         path = self.path(operation_id)
         with exclusive_file_lock(path.with_suffix(".dispatch")):
             exists = path.exists()
+            if exists and source_request_id is not None:
+                return read_delegation_source_execution(self, binding_id, operation_id,
+                    source_request_id=source_request_id, brief=brief, conversation=conversation or {})
             if not exists:
                 delegation_results.require_dependencies(self, binding, brief)
             delivered = request(self.root, self.registry, self.goal_id, self.agent_id,
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 718fb929c3..f7997e15df 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -9,6 +9,7 @@ import {normalizeTodoCompletionValidationDeclaration} from "../todos/completion_
 import {readTurnSelectionRejection, turnSelectionRejectionState} from "../turn_driver/selection_rejection.ts";
 import { BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 import {PROJECTION_ENVELOPE_SEAL_REQUEST, sealProjectionEnvelope} from "../projection_envelope.ts";
+import {normalizeCollaborationBrief, planCollaborationRequestPublication} from "./semantic_request.ts";
 
 function requireThat(ok: unknown, message: string): asserts ok {
   if (!ok) throw new EffectRuntimeRequestError(message);
@@ -158,6 +159,39 @@ export function selectDelegationBinding(params: JsonObject): JsonObject {
   return binding;
 }
 
+/** An operation locator is not proof that a new selection owns its result.
+ * Recover only the original source, peer work, binding and return conversation.
+ * The host observes these records through the existing exact-Goal Inbox owner.
+ */
+export function verifyDelegationSourceReplay(params: JsonObject): JsonObject {
+  const binding = selectDelegationBinding(params);
+  const record = requireJsonObject(params.record, "saved delegation");
+  const identity = requireJsonObject(record.identity, "saved delegation identity");
+  const source = requireJsonObject(params.source_request, "original source request");
+  const peer = requireJsonObject(params.peer_request, "original peer request");
+  const operation = requireJsonObject(params.peer_operation, "original peer operation");
+  planCollaborationRequestPublication({request: operation, entry: peer, operation});
+  const conversation = requireJsonObject(params.conversation, "original conversation");
+  const same = (a: unknown, b: unknown) => canonicalAuthoritySha256(a) === canonicalAuthoritySha256(b);
+  requireThat(typeof params.source_request_id === "string" && BARE_SHA256_PATTERN.test(params.source_request_id)
+    && source.request_id === params.source_request_id && source.execution_binding_id === binding.id
+    && peer.source_kind === "peer" && peer.source_agent_id === params.agent_id
+    && source.goal_id === params.goal_id && peer.goal_id === params.goal_id
+    && source.agent_id === binding.agent_id && peer.agent_id === binding.agent_id
+    && same(source.goal_ref ?? null, peer.goal_ref ?? null)
+    && same(source.goal_ref ?? null, params.goal_ref ?? null), "delegation replay source identity mismatch");
+  requireThat(text(params.operation_id) && typeof peer.request_id === "string"
+    && BARE_SHA256_PATTERN.test(peer.request_id)
+    && same(identity, {binding, request_id: peer.request_id,
+      operation_id: params.operation_id, source_request_id: params.source_request_id}),
+  "delegation replay operation identity mismatch");
+  const brief = normalizeCollaborationBrief(params.brief);
+  requireThat(same(source.brief, brief) && same(peer.brief, brief), "delegation replay brief changed");
+  requireThat(Object.keys(conversation).length === 2 && text(conversation.session_id) && text(conversation.turn_id)
+    && same(record.conversation, conversation), "delegation replay return conversation changed");
+  return {matched: true};
+}
+
 type Observation = "prepared" | "running" | "turn_returned" | "accepted" | "rejected" | "stopped";
 
 function boundedReason(value: unknown, fallback: string): string {
diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py
index d79725156a..36e0ea616a 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -73,6 +73,16 @@ def require_operation_id(value: str) -> str:
     return value
 
 
+def _operation_address(root: Path, scope, source_agent_id: str, operation_id: str) -> tuple[str, Path]:
+    """The existing immutable peer-operation address, shared with replay reads."""
+    source_id = "peer:" + _hash([
+        scope.caller_goal_ref if scope.exact else scope.goal_id,
+        source_agent_id, require_operation_id(operation_id),
+    ])
+    return source_id, (_root(root) / "peer-operations"
+                       / _hash(scope.target(source_agent_id)) / (source_id[5:] + ".json"))
+
+
 def request(
     root,
     registry,
@@ -143,12 +153,7 @@ def request(
                 "message": parent["message"],
                 **({"brief": parent["brief"]} if "brief" in parent else {}),
             }
-        source_identity = (
-            [goal_scope.caller_goal_ref, source_agent_id, operation_id]
-            if goal_scope.exact
-            else [goal_id, source_agent_id, operation_id]
-        )
-        source_id = "peer:" + _hash(source_identity)
+        source_id, operation_path = _operation_address(root, goal_scope, source_agent_id, operation_id)
         request_id = _hash(
             [source_id, _target(goal_id, target_agent_id, goal_scope)]
         )
@@ -172,12 +177,6 @@ def request(
                 "host_surface": binding["host_surface"],
                 "thread_id": binding["thread_id"],
             }
-        operation_path = (
-            _root(root)
-            / "peer-operations"
-            / _hash(_target(goal_id, source_agent_id, goal_scope))
-            / (source_id[5:] + ".json")
-        )
         path = (
             _root(root)
             / "entries"
diff --git a/loopx/control_plane/collaboration/source_grant_observation.py b/loopx/control_plane/collaboration/source_grant_observation.py
index 0818679d20..260bba53a1 100644
--- a/loopx/control_plane/collaboration/source_grant_observation.py
+++ b/loopx/control_plane/collaboration/source_grant_observation.py
@@ -208,6 +208,47 @@ def require_delegation_source_execution(runtime_root: Path, registry_path: Path,
         raise ValueError("original source execution authority is unavailable or revoked") from exc
 
 
+def read_delegation_source_execution(service, binding_id: str, operation_id: str, *,
+        source_request_id: str, brief: dict, conversation: dict) -> dict:
+    """Observe the existing exact-Goal sources for the typed replay comparison.
+
+    This performs no request publication, worker submission or result adoption.
+    Legacy sources retain their legacy lifetime profile; they are not promoted.
+    """
+    from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle
+    from .inbox import _entry
+    from .peers import _operation_address, require_operation_id
+
+    require_operation_id(operation_id)
+    runtime_root, registry_path = service.root, service.registry
+    goal_id, requester_agent_id = service.goal_id, service.agent_id
+    caller_goal_ref = service._caller_goal_ref()
+    binding = service.binding(binding_id, require_active=True)
+    require_delegation_source_execution(runtime_root, registry_path, goal_id=goal_id,
+        requester_agent_id=requester_agent_id, source_request_id=source_request_id,
+        binding=binding, caller_goal_ref=caller_goal_ref)
+    record = _read(service.path(operation_id))
+    with collaboration_goal_scope(registry_path, goal_id=goal_id,
+            agents=(requester_agent_id, binding["agent_id"]),
+            caller_goal_ref=caller_goal_ref, require_active=True) as scope:
+        original = _entry(runtime_root, goal_id, binding["agent_id"], source_request_id, scope=scope)
+        peer = _entry(runtime_root, goal_id, binding["agent_id"], record["identity"]["request_id"], scope=scope)
+        _, operation_path = _operation_address(runtime_root, scope, requester_agent_id, operation_id)
+        for entry in (original, peer):
+            decide_collaboration_lifecycle(scope, operation="history_inspect", record=entry)
+        try:
+            effect_runtime_result("collaboration.delegation.source_replay", {
+                "config": _read(service.config), "binding_id": binding["id"], "agent_id": requester_agent_id,
+                "goal_id": goal_id, "goal_ref": scope.record_identity().get("goal_ref"), "record": record,
+                "operation_id": operation_id, "source_request_id": source_request_id,
+                "source_request": original, "peer_request": peer, "peer_operation": _read(operation_path),
+                "brief": brief, "conversation": conversation,
+            })
+        except EffectRuntimeRejected as exc:
+            raise ValueError("delegation operation identity conflict") from exc
+    return service.read(operation_id)
+
+
 def source_context_authority(
     runtime_root: Path, registry_path: Path, session: dict[str, Any], turn: dict[str, Any]
 ) -> dict[str, Any]:
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 9825c80a64..7d7d18e587 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -605,6 +605,7 @@ export function createEffectRuntimeHandlers(
       lazyHandler(() => import("./post_writeback_hook_transaction.ts"), ({evaluatePostWritebackHookTransaction}) => evaluatePostWritebackHookTransaction),
     ],
     ["collaboration.delegation.binding", lazyHandler(() => import("./collaboration/delegation.ts"), ({selectDelegationBinding}) => selectDelegationBinding)],
+    ["collaboration.delegation.source_replay", lazyHandler(() => import("./collaboration/delegation.ts"), ({verifyDelegationSourceReplay}) => verifyDelegationSourceReplay)],
     ["turn.selection.rejection", lazyHandler(() => import("./turn_driver/selection_rejection.ts"), ({projectTurnSelectionRejection}) => projectTurnSelectionRejection)],
     ["collaboration.delegation.preflight", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationPreflight}) => delegationPreflight)],
     ["collaboration.delegation.validation_plan", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationValidationPlan}) => delegationValidationPlan)],
diff --git a/tests/control_plane/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
new file mode 100644
index 0000000000..cd091197d6
--- /dev/null
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -0,0 +1,81 @@
+"""Replaying a Chat handoff must recover that exact original operation.
+
+Use real Inbox, Chat and File/SQLite owners. A saved operation id alone cannot
+identify a newly selected binding, substitute a brief or redirect the return.
+"""
+
+import copy
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.capabilities.manager_context import _root, _write, execution
+from loopx.control_plane.collaboration.inbox import _read
+from loopx.control_plane.collaboration.peers import request as peer_request
+from loopx.control_plane.collaboration.source_grant_observation import read_delegation_source_execution
+from test_delegation_source_authority import saved_handoff  # noqa: F401
+from test_local_delegation import service as delegation_service  # noqa: F401
+
+
+def durable_bytes(root):
+    # Scoped read locks update diagnostic holder timestamps, not business state.
+    return {p.relative_to(root): p.read_bytes() for p in root.rglob("*")
+            if p.is_file() and not p.name.endswith(".lock")}
+
+
+@pytest.mark.parametrize("change", ["binding", "brief", "source_request", "peer_request", "conversation"])
+def test_saved_handoff_replay_cannot_substitute_its_original_identity(saved_handoff, change):  # noqa: F811
+    _, worker, binding, (_, session, turn, request, receipt, policy), operation = saved_handoff
+    request = copy.deepcopy(request)
+    if change == "binding":
+        config = json.loads(worker.config.read_text())
+        replacement = {**binding, "id": "other-review"}
+        config["bindings"].append(replacement)
+        worker.config.write_text(json.dumps(config))
+        policy["sources"][session["channel_id"]]["execution_bindings"].append({
+            "goal_id": worker.goal_id, "agent_id": binding["agent_id"],
+            "requester_agent_id": worker.agent_id, "binding_id": replacement["id"],
+        })
+        _write(_root(worker.root) / "policy.json", policy)
+        request["execution_binding_id"] = replacement["id"]
+    elif change == "brief":
+        request["brief"]["constraints"].append("A changed request is not the original work.")
+    else:
+        row = _read(worker.path(operation))
+        if change == "source_request":
+            row["identity"]["source_request_id"] = "f" * 64
+        elif change == "peer_request":
+            other = peer_request(worker.root, worker.registry, worker.goal_id, worker.agent_id,
+                binding["agent_id"], "another-operation", request["brief"])
+            row["identity"]["request_id"] = other["request_id"]
+        else:
+            row["conversation"] = {"session_id": "another-session", "turn_id": "another-turn"}
+        _write(worker.path(operation), row)
+    before = durable_bytes(worker.root)
+    result = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert result["submitted"] is False, result
+    assert "status" not in result and "replayed" not in result
+    assert durable_bytes(worker.root) == before
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+
+
+def test_exact_prepared_replay_is_read_only_and_keeps_the_original_return(saved_handoff, monkeypatch):  # noqa: F811
+    _, worker, binding, (_, session, turn, request, receipt, _), operation = saved_handoff
+    monkeypatch.setattr(execution.Delegations, "_spawn",
+                        lambda *_: pytest.fail("readback cannot resubmit a prepared operation"))
+    before = durable_bytes(worker.root)
+    assert read_delegation_source_execution(worker, binding["id"], operation,
+        source_request_id=receipt["request_id"], brief=request["brief"],
+        conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]})["status"] == "prepared"
+    for _ in range(2):
+        result = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+            request=request, receipt=receipt, execution_allowed=lambda: True)
+        assert result["submitted"] and result["replayed"] and result["status"] == "prepared"
+        assert result["operation_id"] == operation
+    assert durable_bytes(worker.root) == before
+    assert _read(worker.path(operation))["conversation"] == {
+        "session_id": session["session_id"], "turn_id": turn["turn_id"],
+    }
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index fb399c3418..ff5ff279f2 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, decideDelegationStop, decideDelegationWakeObservation, delegationCheckedArtifacts, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, delegationValidationPlan, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "../../loopx/control_plane/collaboration/delegation.ts";
+import {recordDelegationAdoption, decideDelegationStop, decideDelegationWakeObservation, delegationCheckedArtifacts, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, delegationValidationPlan, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation, verifyDelegationSourceReplay} from "../../loopx/control_plane/collaboration/delegation.ts";
 import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
 import {projectTurnSelectionRejection} from "../../loopx/control_plane/turn_driver/selection_rejection.ts";
 
@@ -9,6 +9,38 @@ const binding = {id: "review", agent_id: "reviewer", todo_id: "todo_review", wor
 const params = {agent_id: "coordinator", binding_id: "review",
   config: {schema_version: "loopx_local_delegation_v0", bindings: [binding]}};
 
+test("source replay requires the original binding, complete brief, peer and return identity", () => {
+  const brief = {schema_version: "collaboration_brief_v0", purpose: "Review", context: "Original scope",
+    constraints: ["No external writes"], inputs: [], acceptance: ["Return findings"], return_requirement: "Reply here"};
+  const goalRef = {goal_id: "goal", goal_instance_id: "original-instance"};
+  const source = {request_id: "a".repeat(64), goal_id: "goal", goal_ref: goalRef,
+    agent_id: "reviewer", execution_binding_id: binding.id, brief};
+  const peer = {request_id: "b".repeat(64), goal_id: "goal", goal_ref: goalRef,
+    agent_id: "reviewer", source_kind: "peer", source_agent_id: "coordinator", brief};
+  const conversation = {session_id: "original-session", turn_id: "original-turn"};
+  const identity = {binding, request_id: peer.request_id, operation_id: "original-operation",
+    source_request_id: source.request_id};
+  const input = {...params, goal_id: "goal", goal_ref: goalRef, operation_id: identity.operation_id,
+    source_request_id: source.request_id, source_request: source, peer_request: peer, peer_operation: peer,
+    brief, conversation, record: {identity, conversation, status: "prepared"}};
+  assert.deepEqual(verifyDelegationSourceReplay(input), {matched: true});
+  for (const record of [{...input.record, status: "accepted"}, {...input.record, status: "stopped"}])
+    assert.deepEqual(verifyDelegationSourceReplay({...input, record}), {matched: true});
+  for (const change of [
+    {binding_id: "other"}, {goal_id: "other"}, {goal_ref: {...goalRef, goal_instance_id: "replacement"}},
+    {source_request_id: "c".repeat(64)}, {operation_id: "other-operation"},
+    {brief: {...brief, constraints: ["Different scope"]}},
+    {conversation: {...conversation, turn_id: "other-turn"}},
+    {source_request: {...source, execution_binding_id: "other"}},
+    {peer_request: {...peer, source_agent_id: "another-requester"}},
+    {peer_request: {...peer, brief: {...brief, acceptance: ["Different work"]}}},
+    {peer_request: {...peer, goal_ref: {...goalRef, goal_instance_id: "replacement"}}},
+    {record: {...input.record, identity: {...identity, binding: {...binding, workspace: "/other"}}}},
+    {record: {...input.record, identity: {...identity, source_request_id: "c".repeat(64)}}},
+    {record: {...input.record, identity: {...identity, confirmed_operation_id: "another-effect"}}},
+  ]) assert.throws(() => verifyDelegationSourceReplay({...input, ...change}));
+});
+
 test("wake observation preserves conversation and current Goal-reference scope", () => {
   const goalRef = {goal_id: "goal", goal_instance_id: "original-instance"};
   const intent = {requester: {goal_id: "goal", agent_id: "lead", goal_ref: goalRef},
diff --git a/tests/test_manager_context_execution.py b/tests/test_manager_context_execution.py
index c2030c68fb..d4736ccade 100644
--- a/tests/test_manager_context_execution.py
+++ b/tests/test_manager_context_execution.py
@@ -75,6 +75,13 @@ def read(self, operation):
             return {"status": "accepted"}
 
     monkeypatch.setattr(execution, "Delegations", BoundService)
+    def replay(service, binding_id, operation, **kwargs):
+        assert binding_id == "review"
+        assert started[0][3]["source_request_id"] == kwargs["source_request_id"]
+        assert started[0][3]["conversation"] == kwargs["conversation"]
+        assert started[0][2] == kwargs["brief"]
+        return service.read(operation)
+    monkeypatch.setattr(execution, "read_delegation_source_execution", replay)
     values = source(tmp_path, registry, goal_id="research", agent_id="worker", requester="lead", binding="review")
     return tmp_path, registry, values, started
 

From d776f4e701c540f666c59402e7a4495aeec5974e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:16:33 +0800
Subject: [PATCH 3/4] test(collaboration): cover the Chat replay error response

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_delegation_source_replay.py           | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/tests/control_plane/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
index cd091197d6..601d60b976 100644
--- a/tests/control_plane/test_delegation_source_replay.py
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -79,3 +79,21 @@ def test_exact_prepared_replay_is_read_only_and_keeps_the_original_return(saved_
         "session_id": session["session_id"], "turn_id": turn["turn_id"],
     }
     assert not (Path(binding["workspace"]) / "host-invocations").exists()
+
+
+def test_chat_handoff_replay_reports_an_original_return_mismatch(saved_handoff):  # noqa: F811
+    _, worker, binding, (store, session, turn, request, _, _), operation = saved_handoff
+    row = _read(worker.path(operation))
+    row["conversation"] = {"session_id": "another-session", "turn_id": "another-turn"}
+    _write(worker.path(operation), row)
+    before = worker.path(operation).read_bytes()
+    answer = "The original scope remains in effect."
+    response = execution.handoff_response(worker.root, worker.registry,
+        session=session, turn=turn, response={"context_handoff": request, "message": answer},
+        source_authorized=lambda: True, execution_allowed=lambda: True, source_store=store)
+    assert response["context_handoff_receipt"]["status"] == "delivered"
+    assert response["context_execution"]["submitted"] is False
+    assert "status" not in response["context_execution"]
+    assert response["message"].startswith(answer + "\n\n")
+    assert worker.path(operation).read_bytes() == before
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()

From 077b2b25c715352f74e1986296701b1009ea8c6e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 22 Oct 2026 12:44:00 +0800
Subject: [PATCH 4/4] test: record collaboration source fence ceiling

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/canary/module_metric_baseline.json | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index 8112dec08b..d1b107545d 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/loopx/canary/module_metric_baseline.json
@@ -39,6 +39,9 @@
       "any_count": 25,
       "dict_any_count": 0
     },
+    "loopx/collaboration_mcp.py": {
+      "lines": 2032
+    },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,
       "dict_any_count": 0