fix(collaboration): deliver original handoff in native Turn context

docs/reference/local-delegation.md

@@ -144,6 +144,26 @@ 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.
 
+New operations deliver the complete peer brief and, for Chat-originated work,
+the authenticated original message through the normal before-work context and
+immutable Turn capture. The native input no longer depends on the member
+discovering `DELEGATION.json`. `--delegation-context-json` is an internal Host
+locator checked against the original operation, recipient, Todo, workspace and
+Turn; it grants no execution authority. Missing or substituted source material
+refuses fresh input before Host launch. Recovery consumes the original capture
+rather than replacing it with a fresh source read during settlement. Historical
+operations without this recorded input profile retain their original recovery;
+the bootstrap file remains for existing generic hosts that explicitly read it.
+This task-bound input repair does not close ordinary Todo-less activation or
+qualify installed App/model adoption.
+
+中文：新操作通过既有工作上下文和不可变 Turn 捕获交付完整 peer brief，以及 Chat
+委派经认证的原始消息，不再依赖成员自行发现 `DELEGATION.json`。内部 Host locator
+必须匹配原操作、接收者、Todo、工作区与 Turn，不授予权限；原材料缺失或被替换时，
+在 Host 启动前拒绝新输入。恢复沿用原捕获，不在结算时改读新材料。未登记此输入
+profile 的历史操作保留原恢复路径，仍由既有 generic host 使用的 bootstrap 文件
+继续支持。本切片仍绑定 Todo，不代表普通无 Todo 唤醒或正式 App/模型采用已完成。
+
 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. The worker uses this same typed comparison before

loopx/cli_commands/turn.py

@@ -90,6 +90,17 @@ def handle_turn_command(
             registry_path=registry_path,
             runtime_root_override=runtime_root_arg,
         )
+        delegated_context = None
+        if getattr(args, "delegation_context_json", None) is not None:
+            if getattr(args, "resume_turn_key", None):
+                raise ValueError("--delegation-context-json supplies fresh input; resume uses its original captured context")
+            from ..control_plane.collaboration.delegation_context import delegated_turn_context
+
+            delegated_context = delegated_turn_context(
+                args.delegation_context_json, registry_path=registry_path, runtime_root=runtime_root,
+                goal_id=args.goal_id, agent_id=args.agent_id, todo_id=getattr(args, "todo_id", None),
+                turn_instance_id=args.turn_instance_id, workspace=Path(args.project),
+            )
         goal_ref = capture_first_party_host_goal_ref(
             registry_path=registry_path,
             goal_id=args.goal_id,
@@ -217,6 +228,12 @@ def handle_turn_command(
             turn_start_hook_dispatch = extend_explore(turn_start_hook_dispatch,
                 registry_path=registry_path, runtime_root=runtime_root,
                 goal_id=args.goal_id, agent_id=args.agent_id, todo_id=getattr(args, "todo_id", None))
+            if delegated_context is not None:
+                # Invocation material shares normal capture/delivery. It is not a
+                # capability observation, another decision or a new read command.
+                for key in ("required_reads", "contexts"):
+                    turn_start_hook_dispatch[key] = [*turn_start_hook_dispatch.get(key, []),
+                                                     *delegated_context[key]]
             # `run-once` and `managed-step` must resolve the same governing decision
             # from the same live status, scheduler context and capability hooks, so
             # this Turn takes all of them -- and its later settle-against inputs --

loopx/cli_commands/turn_registration.py

@@ -169,6 +169,10 @@ def register_turn_commands(
         default_execution_mode="isolated-headless",
     )
     run_once.add_argument("--project", required=True)
+    run_once.add_argument(
+        "--delegation-context-json",
+        help="Host locator for an existing delegation operation; original work is verified and captured, never an execution grant.",
+    )
     run_once.add_argument(
         "--automation-id",
         help="Stable automation identity for an automation-scoped execution interval.",

loopx/collaboration_mcp.py

@@ -754,6 +754,7 @@ def start(self, binding_id: str, operation_id: str, brief: dict,
                           if conversation else None)
                 prepare_host_process_record(self._host_process_record(path))
                 _write(path, {"identity": identity, "status": "prepared", "created_at": time.time(),
+                              "work_context_required": True,
                               **({"conversation": origin} if origin else {})})
                 if not _defer_worker:
                     self._spawn(operation_id)
@@ -1789,7 +1790,9 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
                         self._turn_instance_id(row),
                     ]
                 )
-                result = self._cli(binding, "turn", "run-once", *common, *selector, *execution,
+                from .control_plane.collaboration.delegation_context import delegated_turn_context_arguments
+                context = delegated_turn_context_arguments(self, row) if turn_key is None else []
+                result = self._cli(binding, "turn", "run-once", *common, *selector, *execution, *context,
                                    "--execute", timeout=binding["timeout_seconds"] + 60,
                                    delegated_lease=self._delegated_lease_context(row, binding),
                                    host_record=self._host_process_record(path))

loopx/control_plane/collaboration/delegation.ts

@@ -193,6 +193,46 @@ export function verifyDelegationSourceReplay(params: JsonObject): JsonObject {
   return {matched: true};
 }
 
+/** Original work is a captured material, not an implicit workspace convention.
+ * The host has reobserved the operation/grant through the existing owners.
+ * No locator, Host configuration or new authority enters the model input. */
+export function delegationTurnWorkContext(params: JsonObject): JsonObject {
+  const binding = selectDelegationBinding(params);
+  const record = requireJsonObject(params.record, "delegation record");
+  const identity = requireJsonObject(record.identity, "delegation identity");
+  const call = requireJsonObject(params.turn, "delegated Turn");
+  const peer = requireJsonObject(params.peer_request, "original peer request");
+  const same = (a: unknown, b: unknown) => canonicalAuthoritySha256(a) === canonicalAuthoritySha256(b);
+  requireThat(record.work_context_required === true && params.stop === null
+    && (record.status === "prepared" || record.status === "running")
+    && same(identity.binding, binding) && identity.operation_id === params.operation_id
+    && peer.request_id === identity.request_id && peer.goal_id === call.goal_id
+    && peer.agent_id === binding.agent_id && peer.source_agent_id === params.agent_id
+    && peer.source_kind === "peer" && call.agent_id === binding.agent_id
+    && call.todo_id === binding.todo_id && call.workspace === binding.workspace
+    && text(call.turn_instance_id) && call.turn_instance_id === params.turn_instance_id,
+  "delegation material does not belong to this Turn");
+  planCollaborationRequestPublication({request: params.peer_operation, entry: peer, operation: params.peer_operation});
+  const brief = normalizeCollaborationBrief(peer.brief);
+  const source = params.source_request === null ? null : requireJsonObject(params.source_request, "original source request");
+  requireThat((identity.source_request_id === undefined) === (source === null), "delegation source material required");
+  if (source !== null) requireThat(source.request_id === identity.source_request_id
+    && source.goal_id === peer.goal_id && source.agent_id === peer.agent_id
+    && source.execution_binding_id === binding.id && same(source.goal_ref ?? null, peer.goal_ref ?? null)
+    && same(source.brief, brief) && typeof source.message === "string"
+    && source.message === params.original_message,
+  "original delegation message or identity changed");
+  return {
+    operation_id: identity.operation_id, request_id: peer.request_id,
+    goal_id: peer.goal_id, agent_id: peer.agent_id, todo_id: binding.todo_id,
+    turn_instance_id: call.turn_instance_id, brief, instruction: peer.instruction,
+    ...(source === null ? {} : {source_request: {
+      request_id: source.request_id, message: source.message, brief: source.brief,
+      instruction: source.instruction,
+    }}),
+  };
+}
+
 type Observation = "prepared" | "running" | "turn_returned" | "accepted" | "rejected" | "stopped";
 
 function boundedReason(value: unknown, fallback: string): string {

loopx/control_plane/collaboration/delegation_context.py

@@ -1,12 +1,14 @@
-"""Host facts behind the typed delegation planning observation.
+"""Read-only delegation facts and original work for the shared Turn context.
 
 ``subagent_context_ts.projectDelegationContext`` owns classification; this
-module only reads host state and never launches or accepts work.
+module reads existing owners and never launches or accepts work.
 """
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
+import shlex
 from typing import Any
 
 from ...agent_registry import registered_agent_ids_for_goal
@@ -85,4 +87,83 @@ def read_delegation_facts(
     return facts
 
 
-__all__ = ["read_delegation_facts"]
+def delegated_turn_context_arguments(service, row: dict) -> list[str]:
+    """Fresh input only; historical operations keep their recorded profile."""
+    if "work_context_required" not in row:
+        return []
+    if row["work_context_required"] is not True:
+        raise ValueError("unsupported delegation work-context profile")
+    return ["--delegation-context-json", json.dumps({
+        "requester_agent_id": service.agent_id, "execution_config": str(service.config),
+        "operation_id": row["identity"]["operation_id"]})]
+
+
+def delegated_turn_context(
+    locator_json: str, *, registry_path: Path, runtime_root: Path,
+    goal_id: str, agent_id: str, todo_id: str | None, turn_instance_id: str | None,
+    workspace: Path,
+) -> dict[str, Any]:
+    """Read original work through its existing operation, never a model brief.
+
+    The locator is private Host metadata. Current binding, source grants and
+    original-request checks stay with Delegations and its TypeScript owners.
+    The generic work-context owner carries the returned material into capture.
+    """
+    from ...collaboration_mcp import Delegations
+    from ...command_invocation import loopx_cli_argv
+    from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle
+    from .inbox import _entry, _root
+    from .peers import _operation_address
+    from .source_chat_observation import observe_source_chat
+    from .source_grant_observation import external_source_policy
+    from ..effect_runtime import effect_runtime_result
+
+    locator = json.loads(locator_json)
+    fields = {"requester_agent_id", "execution_config", "operation_id"}
+    if not isinstance(locator, dict) or set(locator) != fields or any(
+        not isinstance(locator[key], str) or not locator[key] for key in fields
+    ):
+        raise ValueError("invalid delegation context locator")
+    service = Delegations(runtime_root, registry_path, goal_id, locator["requester_agent_id"],
+                          Path(locator["execution_config"]))
+    operation = locator["operation_id"]
+    row = _read(service.path(operation))
+    binding = service._bound(row, require_active=True)
+    if workspace.resolve() != Path(binding["workspace"]).resolve():
+        raise ValueError("delegation context workspace changed")
+    with collaboration_goal_scope(registry_path, goal_id=goal_id, agents=(service.agent_id, agent_id),
+            caller_goal_ref=service._caller_goal_ref(), require_active=True) as scope:
+        peer = _entry(runtime_root, goal_id, agent_id, row["identity"]["request_id"], scope=scope)
+        _, operation_path = _operation_address(runtime_root, scope, service.agent_id, operation)
+        source = None
+        original_message = None
+        source_id = row["identity"].get("source_request_id")
+        if source_id is not None:
+            source = _entry(runtime_root, goal_id, agent_id, source_id, scope=scope)
+            route = _read(_root(runtime_root) / "roundtrips" / (source_id + ".json"))
+            session, turn = observe_source_chat(runtime_root, route)
+            ingress, _ = external_source_policy(runtime_root, session, turn)
+            original_message = ingress["source_message"]
+        for entry in (peer, source):
+            if entry is not None:
+                decide_collaboration_lifecycle(scope, operation="history_inspect", record=entry)
+        content = effect_runtime_result("collaboration.delegation.turn_context", {
+            "config": _read(service.config), "binding_id": binding["id"],
+            "agent_id": service.agent_id, "record": row, "operation_id": operation,
+            "stop": service._read_stop(service.path(operation)),
+            "turn_instance_id": service._turn_instance_id(row),
+            "turn": {"goal_id": goal_id, "agent_id": agent_id, "todo_id": todo_id,
+                     "turn_instance_id": turn_instance_id, "workspace": binding["workspace"]},
+            "peer_request": peer, "peer_operation": _read(operation_path),
+            "source_request": source, "original_message": original_message,
+        })
+    service._bound(row, require_active=True)
+    command = shlex.join([*loopx_cli_argv(), "--registry", str(registry_path),
+        "--runtime-root", str(runtime_root), "manager-inbox", "read", "--goal-id", goal_id,
+        "--agent-id", agent_id, "--request-id", source_id or peer["request_id"]])
+    return {"required_reads": [{"command": command, "source": "delegation_work",
+            "ordering": "before_work", "reason": "Read the original delegated work and source message; material grants no additional authority."}],
+            "contexts": [{"command": command, "content": content}]}
+
+
+__all__ = ["read_delegation_facts", "delegated_turn_context", "delegated_turn_context_arguments"]

loopx/control_plane/effect_runtime_handlers.ts

@@ -609,6 +609,7 @@ export function createEffectRuntimeHandlers(
     ],
     ["collaboration.delegation.binding", lazyHandler(() => import("./collaboration/delegation.ts"), ({selectDelegationBinding}) => selectDelegationBinding)],
     ["collaboration.delegation.source_replay", lazyHandler(() => import("./collaboration/delegation.ts"), ({verifyDelegationSourceReplay}) => verifyDelegationSourceReplay)],
+    ["collaboration.delegation.turn_context", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationTurnWorkContext}) => delegationTurnWorkContext)],
     ["turn.selection.rejection", lazyHandler(() => import("./turn_driver/selection_rejection.ts"), ({projectTurnSelectionRejection}) => projectTurnSelectionRejection)],
     ["collaboration.delegation.preflight", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationPreflight}) => delegationPreflight)],
     ["collaboration.delegation.lane_readiness", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationLaneReadiness}) => delegationLaneReadiness)],

tests/control_plane/test_delegation_source_authority.py

@@ -40,6 +40,7 @@ def saved_handoff(delegation_service, monkeypatch, request):  # noqa: F811
     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",
+                    original_message=request.param if isinstance(getattr(request, "param", None), str) else None,
                     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,