diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 3141e837ea..3c23ba0746 100644
--- a/docs/reference/local-delegation.md
+++ b/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
diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index aa61b7740c..3b4c750b0e 100644
--- a/loopx/cli_commands/turn.py
+++ b/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 --
diff --git a/loopx/cli_commands/turn_registration.py b/loopx/cli_commands/turn_registration.py
index ad80c4add8..1d24d09486 100644
--- a/loopx/cli_commands/turn_registration.py
+++ b/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.",
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index e6f2b4db0c..0e0787cef9 100644
--- a/loopx/collaboration_mcp.py
+++ b/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))
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 9a98aa6601..505198a4c9 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/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 {
diff --git a/loopx/control_plane/collaboration/delegation_context.py b/loopx/control_plane/collaboration/delegation_context.py
index db8dab79e9..43eafafecc 100644
--- a/loopx/control_plane/collaboration/delegation_context.py
+++ b/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"]
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 69ea835de4..d98eb1db88 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/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)],
diff --git a/tests/control_plane/test_delegation_source_authority.py b/tests/control_plane/test_delegation_source_authority.py
index 53a9248fdc..074ec75cc0 100644
--- a/tests/control_plane/test_delegation_source_authority.py
+++ b/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,
diff --git a/tests/control_plane/test_delegation_turn_context.py b/tests/control_plane/test_delegation_turn_context.py
new file mode 100644
index 0000000000..3dbebed8f6
--- /dev/null
+++ b/tests/control_plane/test_delegation_turn_context.py
@@ -0,0 +1,138 @@
+"""Original delegation material reaches the actual native Host input.
+
+Real File/SQLite and disposable subprocesses; no provider/model or live App.
+The input oracle is the original sender material, independent of the serializer.
+"""
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.collaboration.inbox import _read, _root, _write
+from loopx.control_plane.turn_driver.codex_cli import run_codex_cli_host
+from test_delegation_source_authority import saved_handoff  # noqa: F401
+from test_local_delegation import demo, service as delegation_service  # noqa: F401
+from test_loopx_turn_codex_cli import _fake_codex
+
+
+ORIGINAL = "原始委派：保留全部约束、负结果与原文。\n" * 700 + 'END: "quoted" \\ unicode 🧭'
+
+
+def preview(worker, binding, operation, *, include_context=True, **override):
+    row = _read(worker.path(operation))
+    selector = {"goal-id": worker.goal_id, "agent-id": binding["agent_id"],
+                "todo-id": binding["todo_id"], "turn-instance-id": worker._turn_instance_id(row)}
+    selector.update(override)
+    args = [value for key, item in selector.items() for value in ("--" + key, item)]
+    context = []
+    if include_context:
+        from loopx.control_plane.collaboration.delegation_context import delegated_turn_context_arguments
+        context = delegated_turn_context_arguments(worker, row)
+    return worker._cli(binding, "turn", "run-once", *args,
+        *worker._execution_arguments(binding, operation), *context)
+
+
+def original_context(envelope):
+    return next(source["content"] for source in envelope["work_context"]["sources"]
+                if source.get("source") == "delegation_work")
+
+
+@pytest.mark.parametrize("saved_handoff", [ORIGINAL], indirect=True, ids=["long-mixed-source"])
+def test_original_material_reaches_capture_and_actual_codex_stdin(saved_handoff, monkeypatch):  # noqa: F811
+    root, worker, binding, (_, _, _, request, receipt, _), operation = saved_handoff
+    tasks = demo.canonical_tasks(root)
+    plan = preview(worker, binding, operation, include_context=False)
+    assert plan["route"]["would_invoke_host"], plan
+    assert not (Path(binding["workspace"]) / "DELEGATION.json").exists()
+    assert demo.canonical_tasks(root) == tasks
+    assert not list((worker.root / "goals" / worker.goal_id / "turns").glob("*.json"))
+    host = root / "fixture-host.py"
+    host.write_text(host.read_text().replace("root = Path(sys.argv[1])",
+        "root = Path(sys.argv[1])\n(root / 'received-request.json').write_text(json.dumps(request))"))
+    worker.execute(operation)
+    assert worker.read(operation)["status"] == "accepted"
+    host_request = json.loads((root / "received-request.json").read_text())
+    content = original_context(host_request["turn_envelope"])
+    assert content["source_request"]["message"] == ORIGINAL
+    assert content["source_request"]["request_id"] == receipt["request_id"]
+    assert content["brief"] == content["source_request"]["brief"] == request["brief"]
+    assert content["request_id"] == _read(worker.path(operation))["identity"]["request_id"]
+    assert str(worker.config) not in json.dumps(content)
+    assert not (Path(binding["workspace"]) / "DELEGATION.json").exists()
+    tasks = demo.canonical_tasks(root)
+
+    # Exercise the production Codex renderer and native process transport with
+    # that real plan. This is stdin qualification, not another governed business
+    # Turn, a paid model or a claim that a model adopted the material.
+    executable, log = _fake_codex(root)
+    prompt_file = root / "native-prompt.txt"
+    executable.write_text(executable.read_text().replace("prompt = sys.stdin.read()",
+        "prompt = sys.stdin.read()\npathlib.Path(os.environ['FAKE_CODEX_PROMPT']).write_text(prompt)"))
+    monkeypatch.setenv("FAKE_CODEX_LOG", str(log))
+    monkeypatch.setenv("FAKE_CODEX_PROMPT", str(prompt_file))
+    result = run_codex_cli_host(host_request, runtime_root=worker.root,
+        project=Path(binding["workspace"]), codex_bin=str(executable), registry_path=worker.registry,
+        timeout_seconds=10)
+    assert result["result_kind"] == "validated_progress"
+    prompt = prompt_file.read_text()
+    assert json.dumps(ORIGINAL, ensure_ascii=False)[1:-1] in prompt
+    assert json.dumps(request["brief"], ensure_ascii=False, sort_keys=True, separators=(",", ":")) in prompt
+    assert demo.canonical_tasks(root) == tasks
+
+
+def test_historical_operation_retains_its_recorded_input_profile(saved_handoff):  # noqa: F811
+    from loopx.control_plane.collaboration.delegation_context import delegated_turn_context_arguments
+    root, worker, binding, _, operation = saved_handoff
+    row = _read(worker.path(operation))
+    row.pop("work_context_required")
+    _write(worker.path(operation), row)
+    assert not delegated_turn_context_arguments(worker, row)
+    worker.execute(operation)
+    assert worker.read(operation)["status"] == "accepted"
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+def test_worker_consumes_captured_work_without_reading_compatibility_bootstrap(saved_handoff):  # noqa: F811
+    root, worker, binding, (_, _, _, request, receipt, _), operation = saved_handoff
+    host = root / "fixture-host.py"
+    host.write_text(host.read_text().replace(
+        "delegation = json.loads((workspace / 'DELEGATION.json').read_text())",
+        "delegation = next(s['content'] for s in envelope['work_context']['sources'] if s.get('source') == 'delegation_work')\n"
+        "(root / 'received-work.json').write_text(json.dumps(delegation))"))
+    worker.execute(operation)
+    assert worker.read(operation)["status"] == "accepted"
+    content = json.loads((root / "received-work.json").read_text())
+    assert content["brief"] == request["brief"]
+    assert content["source_request"]["request_id"] == receipt["request_id"]
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+    assert not (Path(binding["workspace"]) / "DELEGATION.json").exists()
+    worker.execute(operation)  # Original capture/receipt replay, no second Host.
+    assert (Path(binding["workspace"]) / "host-invocations").read_text() == "1"
+
+
+@pytest.mark.parametrize("field", ["agent-id", "todo-id", "turn-instance-id"])
+def test_material_locator_cannot_retarget_work(saved_handoff, field):  # noqa: F811
+    root, worker, binding, _, operation = saved_handoff
+    tasks = demo.canonical_tasks(root)
+    result = preview(worker, binding, operation, **{field: "wrong-original-identity"})
+    assert result.get("ok") is False
+    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) == tasks
+
+
+@pytest.mark.parametrize("change", ["missing", "message"])
+def test_missing_or_substituted_original_message_prevents_host(saved_handoff, change):  # noqa: F811
+    root, worker, binding, (_, _, _, _, receipt, _), operation = saved_handoff
+    tasks = demo.canonical_tasks(root)
+    path = next((_root(worker.root) / "entries").glob("*/" + receipt["request_id"] + ".json"))
+    if change == "missing":
+        path.unlink()
+    else:
+        entry = _read(path)
+        entry["message"] = "A substituted request with the same brief is not the original message."
+        _write(path, entry)
+    worker.execute(operation)
+    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) == tasks
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index 689589c705..301e796f03 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, verifyDelegationSourceReplay} from "../../loopx/control_plane/collaboration/delegation.ts";
+import {recordDelegationAdoption, decideDelegationStop, decideDelegationWakeObservation, delegationCheckedArtifacts, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, delegationTurnWorkContext, 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,42 @@ 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("captured delegation work retains original material without Host configuration or surrogate authority", () => {
+  const brief = {schema_version: "collaboration_brief_v0", purpose: "Review", context: "Full original scope",
+    constraints: ["No external writes"], inputs: [], acceptance: ["Check evidence"], return_requirement: "Return here"};
+  const peer = {request_id: "b".repeat(64), goal_id: "goal", agent_id: binding.agent_id,
+    source_kind: "peer", source_agent_id: "coordinator", brief, instruction: "Read before work."};
+  const source = {request_id: "a".repeat(64), goal_id: "goal", agent_id: binding.agent_id,
+    execution_binding_id: binding.id, message: "原始内容\n完整约束 🧭", brief, instruction: "Context grants no additional authority."};
+  const turn = {goal_id: "goal", agent_id: binding.agent_id, todo_id: binding.todo_id,
+    workspace: binding.workspace, turn_instance_id: "original-turn"};
+  const identity = {binding, operation_id: "original-operation", request_id: peer.request_id,
+    source_request_id: source.request_id};
+  const input = {...params, operation_id: identity.operation_id, turn, turn_instance_id: turn.turn_instance_id,
+    record: {identity, status: "prepared", work_context_required: true}, stop: null, peer_request: peer, peer_operation: peer,
+    source_request: source, original_message: source.message};
+  const context = delegationTurnWorkContext(input);
+  assert.deepEqual(context.brief, brief);
+  assert.deepEqual(context.source_request, {request_id: source.request_id, message: source.message, brief,
+    instruction: source.instruction});
+  assert.equal(context.turn_instance_id, turn.turn_instance_id);
+  for (const field of ["config", "binding", "workspace", "host_args", "requesters"])
+    assert.equal(Object.hasOwn(context, field), false);
+  for (const change of [
+    {turn: {...turn, todo_id: "other"}}, {turn: {...turn, agent_id: "other"}},
+    {turn: {...turn, workspace: "/other"}}, {turn: {...turn, turn_instance_id: "other"}},
+    {operation_id: "other"}, {source_request: null}, {original_message: "substituted"},
+    {peer_request: {...peer, source_agent_id: "other"}},
+    {peer_operation: {...peer, brief: {...brief, context: "Changed"}}},
+    {record: {identity}}, {record: {...input.record, status: "stopped"}}, {stop: {}},
+  ]) assert.throws(() => delegationTurnWorkContext({...input, ...change}));
+  const {source_request_id: _source, ...peerIdentity} = identity;
+  const operator = delegationTurnWorkContext({...input, record: {...input.record, identity: peerIdentity},
+    source_request: null, original_message: null});
+  assert.deepEqual(operator.brief, brief);
+  assert.equal(Object.hasOwn(operator, "source_request"), false);
+});
+
 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"};
diff --git a/tests/test_manager_context_execution.py b/tests/test_manager_context_execution.py
index 866b0b5b41..6a94020467 100644
--- a/tests/test_manager_context_execution.py
+++ b/tests/test_manager_context_execution.py
@@ -21,7 +21,7 @@
 from test_independent_delegation_validation import independent_binding
 
 
-def source(root, registry, *, goal_id, agent_id, requester, binding, source_id="lark:exact-message", semantic_brief=None, chat_root=None):
+def source(root, registry, *, goal_id, agent_id, requester, binding, source_id="lark:exact-message", semantic_brief=None, chat_root=None, original_message=None):
     store = ChatSessionStore(root if chat_root is None else chat_root)
     session = store.create_session(goal_id="loopx-manager", agent_id="codex",
                                    adapter_kind="codex_app_server", upstream_thread_id="original",
@@ -35,7 +35,7 @@ def source(root, registry, *, goal_id, agent_id, requester, binding, source_id="
     _write(_root(root) / "policy.json", policy)
     register_ingress(root, session_id=session["session_id"], client_turn_id=turn["client_turn_id"],
                      channel=session["channel_id"], sender_id="owner", message=turn["message"],
-                     source_id=source_id)
+                     source_id=source_id, source_message=original_message)
     request = {"goal_id": goal_id, "agent_id": agent_id, "execution_binding_id": binding, "brief": brief() if semantic_brief is None else semantic_brief}
     receipt = deliver(root, registry, session=session, turn=turn, request=request)
     return store, session, turn, request, receipt, policy