diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 9aae084042..1c5e78a3d9 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -130,6 +130,29 @@ 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.
+The final source observation, canonical completion and original-Turn settlement
+share the source-policy fence, so a grant revocation linearizes before the
+observation or after both effects.
+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.
+
+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 operations may resume only under the restored current grant,
+and matching terminal operations are read back in place. Historical operator
+inspection remains available through the ordinary read command.
+
 A Chat handoff with a separate source execution grant can recover when its
 requester exits after saving the original operation but before starting the
 worker. Replaying that handoff rechecks the source Turn and exact grant, then
@@ -142,6 +165,16 @@ original Inbox request, binding, conversation and governed Turn; it creates no
 replacement task, grant or scheduler. This remains the explicit task-bound
 route described here, not general automatic activation of an idle Inbox owner.
 
+中文：Chat 发起的执行保留原 Inbox 请求、固定的对话来源和原 operation。worker 在
+入口、Host 启动前和完成结算前重新核对当前发送者、接收者及精确
+Goal/Agent/requester/binding grant；撤权、来源不可读或身份变化会拒绝在原
+operation 上的新执行。最终来源观察、规范完成和原 Turn 结算共享来源策略锁，
+撤权只能发生在观察之前或两项副作用完成之后；历史仍可检查。Host 已返回时保留原 Turn 和未结算效果；
+恢复授权后只结算该 operation，不重跑 Host。重试在报告状态前核对原 Inbox、完整
+brief、peer request、当前 binding 和原回传 Session/Turn；选择另一个合法 binding
+不能改派既有 operation，身份不符时返回原执行不可用反馈且不新发请求。身份一致的
+prepared operation 可在恢复的当前 grant 下续接原 owner，终态 operation 原位读回。
+
 中文：另有来源执行授权的 Chat 交接，在原 operation 已保存、worker 尚未启动时
 请求方退出，可重放同一交接恢复。重放重新核对来源 Turn 与精确 grant，只续接原
 `prepared` operation；worker 启动后仍在既有执行锁内核对同一前置条件。忙碌 worker
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
diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py
index 1305d76597..1f912f3b2a 100644
--- a/loopx/capabilities/manager_context/__init__.py
+++ b/loopx/capabilities/manager_context/__init__.py
@@ -15,6 +15,7 @@
     source_context_authority,
     source_context_target_authority,
 )
+from ...control_plane.collaboration.source_policy_fence import source_policy_fence
 from ...control_plane.collaboration.goal_instance_scope import (
     collaboration_goal_scope,
     decide_collaboration_lifecycle,
@@ -120,7 +121,7 @@ def deliver(
     # Order: source policy -> Goal lifetime -> request; local owner stays independent.
     scope = conversation_scope(session, origin=turn.get("origin", "unknown"))
     with (
-        exclusive_file_lock((_root(runtime_root) / "policy.json").with_suffix(".lock"))
+        source_policy_fence(_root(runtime_root) / "policy.json")
         if scope["kind"] == "external_audience" else nullcontext()
     ), collaboration_goal_scope(
         registry_path,
@@ -354,7 +355,7 @@ def revoke_conversation_sources(runtime_root: Path, binding_id: str) -> None:
         path.stat()
     except FileNotFoundError:
         return
-    with exclusive_file_lock(path.with_suffix(".lock")):
+    with source_policy_fence(path):
         try:
             policy = _read(path)
         except FileNotFoundError:
@@ -397,7 +398,7 @@ def update(apply: bool) -> dict:
 
     if execute:
         path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        with exclusive_file_lock(path.with_suffix(".lock")):
+        with source_policy_fence(path):
             result = update(True)
             if update(False)["would_change"]:
                 raise ValueError("read scope verification failed")
@@ -456,7 +457,7 @@ def update(*, apply: bool) -> dict:
 
     if not execute:
         return update(apply=False)
-    with exclusive_file_lock(path.with_suffix(".lock")):
+    with source_policy_fence(path):
         result = update(apply=True)
         if update(apply=False)["granted_before"] != grant:
             raise ValueError("delivery target verification failed")
@@ -495,7 +496,7 @@ def update(apply: bool) -> dict:
         return planned
 
     if execute:
-        with exclusive_file_lock(path.with_suffix(".lock")):
+        with source_policy_fence(path):
             result = update(True)
             if update(False)["would_change"]:
                 raise ValueError("delivery scope verification failed")
diff --git a/loopx/capabilities/manager_context/execution.py b/loopx/capabilities/manager_context/execution.py
index 5650a4580e..3b68433dd6 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"]})
             if row["status"] == "prepared":
                 # The requester may have died after persisting the operation
                 # but before starting its worker. Rejoin that same owner under
diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py
index d18889a854..fc7c35d660 100644
--- a/loopx/capabilities/manager_context/roundtrip.py
+++ b/loopx/capabilities/manager_context/roundtrip.py
@@ -24,6 +24,7 @@
     collaboration_goal_scope,
     decide_collaboration_lifecycle,
 )
+from ...control_plane.collaboration.source_policy_fence import source_policy_fence
 from ...control_plane.projects.registry_codec import (
     SOURCE_SESSION_PROFILE_ID,
     load_project_registry,
@@ -346,9 +347,7 @@ def _hold_return_source_fence(root, route, effects):
     # Match the grant writer through transcript/provider publication. Owner-only
     # returns stay independent of source policy and create no policy storage.
     if conversation_scope(route)["kind"] == "external_audience":
-        effects.enter_context(
-            exclusive_file_lock((_root(root) / "policy.json").with_suffix(".lock"))
-        )
+        effects.enter_context(source_policy_fence(_root(root) / "policy.json"))
 
 
 def _exact_return_context(root, registry, store, path, state_path, now, effects):
diff --git a/loopx/capabilities/manager_context/ssh_evidence.py b/loopx/capabilities/manager_context/ssh_evidence.py
index d3056a6f5d..de9ba74b3b 100644
--- a/loopx/capabilities/manager_context/ssh_evidence.py
+++ b/loopx/capabilities/manager_context/ssh_evidence.py
@@ -13,7 +13,7 @@
 from typing import Any
 
 from . import POLICY_SCHEMA, _root, _read, _write
-from ...file_lock import exclusive_file_lock
+from ...control_plane.collaboration.source_policy_fence import source_policy_fence
 from ...control_plane.status.ssh_host_catalog import configured_ssh_host_aliases
 
 # --- The bounded turn-time source read ---------------------------------------
@@ -66,7 +66,7 @@ def configure(root, *, channel, host, goal_ids, execute=False, config_path=None)
     ids = sorted(set(goal_ids))
     if execute:
         path = _root(root) / "policy.json"
-        with exclusive_file_lock(path.with_suffix(".lock")):
+        with source_policy_fence(path):
             policy = (
                 _read(path)
                 if path.exists()
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index d3adc844f0..8aa4cf24ff 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -18,6 +18,7 @@
 import subprocess
 import sys
 import time
+from contextlib import nullcontext
 from functools import lru_cache
 from pathlib import Path
 from threading import Lock
@@ -63,6 +64,10 @@
     decide_collaboration_lifecycle,
 )
 from .control_plane.collaboration import delegation_results, delegation_stop_lease, delegation_validation
+from .control_plane.collaboration.source_grant_observation import (
+    read_delegation_source_execution, require_delegation_source_execution,
+)
+from .control_plane.collaboration.source_policy_fence import source_policy_fence
 from .control_plane.collaboration.peers import (
     _goal,
     consume_return,
@@ -704,6 +709,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,
@@ -766,9 +774,7 @@ def resume(self, operation_id: str, *, prepared_source_request_id: str | None =
                 should_spawn = (row["status"] == "prepared" if prepared_replay
                                 else 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.
@@ -828,10 +834,16 @@ def revalidate(self, operation_id: str) -> dict:
             self._spawn(operation_id)
         return self.read(operation_id)
 
-    def _bound(self, row: dict, *, require_active: bool = False) -> dict:
+    def _bound(
+        self, row: dict, *, require_active: bool = False, policy_locked: 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(), policy_locked=policy_locked)
         return binding
 
     @staticmethod
@@ -1383,13 +1395,12 @@ def execute(self, operation_id: str, *, prepared_source_request_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;
@@ -1737,6 +1748,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)
@@ -1748,6 +1760,11 @@ 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)
+                # A prelaunch source refusal clears this private Host input in
+                # finally. Recreate the same original-operation input before a
+                # later authorized resume, without creating a new operation.
+                self._write_delegation_bootstrap(row, binding)
                 turn_key = delegation_validation.matching_turn_key(self, row, binding)
                 selector = (
                     ["--resume-turn-key", turn_key]
@@ -1798,8 +1815,23 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
             # for a member whose Todo and reply had already landed. Recovery of
             # a validated journal uses this same completion entry, retaining the
             # fence through settlement, result publication and acceptance.
-            with exclusive_file_lock(self._dispatch_lock(path)):
+            source_request_id = row["identity"].get("source_request_id")
+            source_guard = (
+                source_policy_fence(_root(self.root) / "policy.json")
+                if source_request_id
+                else nullcontext()
+            )
+            # Hold the same policy fence used by source revocation until both
+            # canonical completion and original-Turn settlement commit. A
+            # revocation therefore linearizes either before this check or
+            # after these effects, never between the last check and effects.
+            with exclusive_file_lock(self._dispatch_lock(path)), source_guard:
                 self._raise_if_stop_requested(path)
+                self._bound(
+                    row,
+                    require_active=True,
+                    policy_locked=source_request_id is not None,
+                )
                 self._complete_delegated_todo(row, binding)
                 if needs_settlement:
                     result = self._cli(
@@ -1822,7 +1854,11 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                         row["error"] = "delegation receiver did not adopt the request"
                         self._observe(path, row, "rejected", already_locked=True)
                         return
-                    self._bound(row, require_active=True)
+                    self._bound(
+                        row,
+                        require_active=True,
+                        policy_locked=source_request_id is not None,
+                    )
                     delegation_results.require_dependencies(
                         self, binding, delegation_results.operation_brief(self, row)
                     )
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 4f0d1f7df8..3a5c016a82 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..1fa082cb6c 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -28,13 +28,13 @@
     plan_request_publication,
 )
 from .peer_context_observation import require_parent_context_access
+from .source_policy_fence import source_policy_fence
 from .goal_instance_scope import (
     collaboration_goal_scope,
     decide_collaboration_lifecycle,
     goal_accepts_collaboration,
 )
 from ...agent_registry import registered_agent_ids_for_goal
-from ...file_lock import exclusive_file_lock
 from ...thread_agent_binding import resolve_thread_agent_binding
 from ..projects.registry_codec import load_project_registry
 from ..runtime.file_paths import windows_extended_path
@@ -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,
@@ -95,7 +105,7 @@ def request(
     # An inherited source grant must remain current until its request is stored.
     # Take the policy fence before Goal/request locks, as the source writer does.
     with (
-        exclusive_file_lock((_root(root) / "policy.json").with_suffix(".lock"))
+        source_policy_fence(_root(root) / "policy.json")
         if parent_request_id else nullcontext()
     ), collaboration_goal_scope(
         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 f80adc55da..a78bdb2a00 100644
--- a/loopx/control_plane/collaboration/source_grant_observation.py
+++ b/loopx/control_plane/collaboration/source_grant_observation.py
@@ -1,5 +1,6 @@
 """Provider/store observation for the typed context-recipient policy owner."""
 
+from contextlib import nullcontext
 from pathlib import Path
 from typing import Any, cast
 
@@ -13,6 +14,7 @@
 )
 from . import conversation_scope
 from .inbox import _hash, _read, _root
+from .source_policy_fence import source_policy_fence
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 
 POLICY_SCHEMA = "loopx_manager_context_policy_v1"
@@ -156,19 +158,104 @@ 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,
+        policy_locked: bool = False) -> 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 .goal_instance_scope import collaboration_goal_scope
+    from .inbox import _entry
+    from .source_chat_observation import observe_source_chat
+
+    try:
+        policy_guard = (
+            nullcontext()
+            if policy_locked
+            else source_policy_fence(_root(runtime_root) / "policy.json")
+        )
+        with policy_guard, 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 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/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/loopx/control_plane/collaboration/source_policy_fence.py b/loopx/control_plane/collaboration/source_policy_fence.py
new file mode 100644
index 0000000000..c101d01dd1
--- /dev/null
+++ b/loopx/control_plane/collaboration/source_policy_fence.py
@@ -0,0 +1,28 @@
+"""Serialize source-policy readers and writers across threads and processes."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from pathlib import Path
+from threading import Lock, RLock
+from typing import Iterator
+
+from ...file_lock import exclusive_file_lock
+
+
+_LOCKS_GUARD = Lock()
+_THREAD_LOCKS: dict[str, RLock] = {}
+
+
+def _thread_lock(path: Path) -> RLock:
+    key = str(path.expanduser().resolve())
+    with _LOCKS_GUARD:
+        return _THREAD_LOCKS.setdefault(key, RLock())
+
+
+@contextmanager
+def source_policy_fence(policy_path: Path) -> Iterator[None]:
+    """Hold the in-process and cross-process fences used by source writers."""
+
+    with _thread_lock(policy_path), exclusive_file_lock(policy_path.with_suffix(".lock")):
+        yield
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 12d4d321d7..268ddf2ae7 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -611,6 +611,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_authority.py b/tests/control_plane/test_delegation_source_authority.py
new file mode 100644
index 0000000000..e6f65246d5
--- /dev/null
+++ b/tests/control_plane/test_delegation_source_authority.py
@@ -0,0 +1,257 @@
+"""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 threading
+
+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 loopx.control_plane.collaboration.source_policy_fence import source_policy_fence
+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("saved_handoff", [True], indirect=True)
+def test_final_source_revocation_cannot_split_completion_and_settlement(
+    saved_handoff, monkeypatch,
+):
+    root, worker, binding, (_, session, _, _, _, _), operation = saved_handoff
+    complete = worker._complete_delegated_todo
+    revocation_started = threading.Event()
+    revocation_finished = threading.Event()
+    revocation_errors = []
+    revocation_threads = []
+    completed_before_revocation = []
+
+    def revoke_from_owner():
+        revocation_started.set()
+        try:
+            policy_path = _root(worker.root) / "policy.json"
+            with source_policy_fence(policy_path):
+                current = _read(policy_path)
+                current["sources"][session["channel_id"]]["execution_bindings"] = []
+                _write(policy_path, current)
+        except Exception as exc:  # pragma: no cover - asserted below
+            revocation_errors.append(exc)
+        finally:
+            revocation_finished.set()
+
+    def complete_with_concurrent_revocation(row, current_binding):
+        thread = threading.Thread(target=revoke_from_owner, daemon=True)
+        thread.start()
+        assert revocation_started.wait(5)
+        complete(row, current_binding)
+        completed_before_revocation.append(not revocation_finished.is_set())
+        revocation_threads.append(thread)
+
+    monkeypatch.setattr(worker, "_complete_delegated_todo", complete_with_concurrent_revocation)
+    worker.execute(operation)
+    [thread] = revocation_threads
+    thread.join(10)
+    assert not revocation_errors
+    assert completed_before_revocation == [True]
+    assert revocation_finished.is_set()
+    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"
+    worker.execute(operation)
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+@pytest.mark.parametrize("saved_handoff", [True], indirect=True)
+def test_prelaunch_refusal_recreates_original_bootstrap_on_authorized_resume(
+    saved_handoff, monkeypatch,
+):
+    root, worker, binding, (_, session, _, _, _, policy), operation = saved_handoff
+    original_policy = json.loads(json.dumps(policy))
+    acquire = worker._acquire_delegation_lease
+    revoked = False
+
+    def revoke_after_original_lease(path, row, current_binding):
+        nonlocal revoked
+        result = acquire(path, row, current_binding)
+        if not revoked:
+            revoked = True
+            policy["sources"][session["channel_id"]]["execution_bindings"] = []
+            _write(_root(worker.root) / "policy.json", policy)
+        return result
+
+    monkeypatch.setattr(worker, "_acquire_delegation_lease", revoke_after_original_lease)
+    worker.execute(operation)
+    refused = _read(worker.path(operation))
+    assert refused["status"] == "running", refused
+    assert "source execution" in refused["error"]
+    assert not (Path(binding["workspace"]) / "host-invocations").exists()
+    assert not demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+    assert not (Path(binding["workspace"]) / "DELEGATION.json").exists()
+
+    _write(_root(worker.root) / "policy.json", original_policy)
+    worker.execute(operation)
+    resumed = worker.read(operation)
+    assert resumed["status"] == "accepted", resumed
+    assert demo.canonical_tasks(root)[binding["todo_id"]]["done"]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    worker.execute(operation)
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+@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/test_delegation_source_replay.py b/tests/control_plane/test_delegation_source_replay.py
new file mode 100644
index 0000000000..1ec5caf6c5
--- /dev/null
+++ b/tests/control_plane/test_delegation_source_replay.py
@@ -0,0 +1,104 @@
+"""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_rejoins_only_the_original_operation(saved_handoff, monkeypatch):  # noqa: F811
+    _, worker, binding, (_, session, turn, request, receipt, _), operation = saved_handoff
+    resumed = []
+    monkeypatch.setattr(
+        execution.Delegations,
+        "_spawn",
+        lambda _self, operation_id, **kwargs: resumed.append((operation_id, kwargs)),
+    )
+    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"
+    result = execution.dispatch(worker.root, worker.registry, session=session, turn=turn,
+        request=request, receipt=receipt, execution_allowed=lambda: True)
+    assert result["submitted"], result
+    assert result["replayed"] and result["status"] == "prepared"
+    assert result["operation_id"] == operation
+    assert resumed == [(operation, {"prepared_source_request_id": receipt["request_id"]})]
+    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()
+
+
+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()
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index feb11f6ff1..9ff0c2dabd 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/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"}));
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