fix(collaboration): bind queued handoffs and replay to their original source

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

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

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")

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

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):

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()

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)
                     )