diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md
index ca113799c5..94232d06c7 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md
@@ -1826,3 +1826,23 @@ The directory is the index. A planning/source-inspection entry does not establis
 runtime delivery; live priorities and claims remain in the roadmap and canonical
 work. Keep historical entries append-only and refine the current normative plan
 here rather than treating a dated audit as a permanent execution queue.
+
+
+### Local registered Goal material checkpoint
+
+The complete Turn writer and existing native context-delivery opt-in now request
+full registered Goal material before capture. The source adapter returns original
+UTF-8 text for unmarked or mixed files; the existing TypeScript context owner
+checks the full-file revision and fulfills that read. Existing marked-region
+lineage and canonical Todo authority remain. Ordinary quota with delivery off
+keeps progressive reads. The bridge requires a `complete_goal_v1` acknowledgement;
+an older runtime cannot silently ignore the request.
+
+Automatic business-work Host context uses `host_dispatch_context`, which refuses pending or
+invalid Goal material before provider loading. Historical `host_context` metadata
+reads remain available for original-version interpretation and necessary result
+finalization; stored captures are not rewritten. This is a bounded material
+delivery prerequisite. General capability-read interlocks, current-condition
+revalidation, packaged adoption, real-model adherence and complete costs still
+need their own evidence. Replan material-reading invocations retain their staged
+reader and work restrictions. C1 remains unqualified.
diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md
index 22a81bd2a5..fd55dd9239 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md
@@ -1347,3 +1347,18 @@ C1/C2 组合/capture/最终输入工作与这些精简同属一个候选前沿
 目录即索引。规划/源码检查记录不证明 runtime 交付；当前优先级与认领仍在 roadmap 和
 canonical 工作中维护。历史记录保持 append-only，当前规范性计划在本文修订，不能把
 某日审计当作永久执行队列。
+
+
+### 本地注册 Goal 材料交付检查点
+
+完整 Turn writer 和现有 native context-delivery 显式启用路径，现在在 capture 前
+请求完整的注册 Goal 材料。源 adapter 为未标记或混合文件返回原始 UTF-8 全文；
+现有 TypeScript context owner 核对完整文件 revision 后完成该项读取。已标记区域
+的 lineage 和 canonical Todo 权威保留。普通 quota 在 delivery 关闭时继续渐进读取。
+桥接要求 `complete_goal_v1` 回执；旧 runtime 不能静默忽略这个要求。
+
+自动业务执行的 Host 上下文使用 `host_dispatch_context`，在加载 provider 前拒绝未完成或无效的
+Goal 材料。历史 `host_context` 元数据读取继续用于按原版本解释与必要结果收尾；
+已保存的 capture 不被重写。本项是范围有限的材料交付前置。通用 capability 必读
+强制、当前条件复核、安装采用、真实模型遵守和完整成本，仍各需独立证据。C1 未获
+完整资格。Replan 材料读取调用仍按原有阶段与工作限制执行。
diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 086068ea85..8cbd763dca 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -57,10 +57,29 @@ narrative is never classified by headings or clipped to a size budget. Read all
 delivered context and remaining reads before dependent work; changed sources need
 a fresh guard. Enabled scoped acceptance does not replace original Goal intent.
 
+The complete Turn writer and explicitly enabled native context delivery request
+complete Goal material before capture. For an unmarked or mixed Goal, the
+registered-source adapter supplies the entire original UTF-8 text with its
+full-file revision and `full_source: true`. The existing TypeScript context owner
+checks that revision before fulfilling the Goal read. Marked sources retain
+their existing excluded-region lineage and canonical Todo authority. Ordinary
+quota with context delivery off retains the progressive behavior above.
+
+The internal `complete_goal_v1` acknowledgement prevents an older runtime from
+silently ignoring this requirement. Automatic complete-contract business-work Host dispatch
+uses `host_dispatch_context`: pending or invalid Goal material rejects before
+loading the Host. The `host_context` metadata read remains available for
+historical interpretation and result finalization. Saved contracts are not
+rewritten; a historical pending Goal read cannot become fulfilled merely by
+restoring its capture. This change neither grants execution permission nor
+establishes enforcement for other pending capability reads. Replan material-reading
+invocations retain their existing staged reader and work restrictions.
+
 The Goal availability adapter still reads and UTF-8-validates the complete file
-and hashes all bytes for its revision. For unmarked or mixed ownership, only its path and revision cross the
-internal context-projection bridge, preserving the complete progressive read
-without an unused body copy. Explicitly owned sources send the complete
+and hashes all bytes for its revision. In ordinary progressive delivery with
+context delivery off, unmarked or mixed sources send only their path and revision
+across the internal context-projection bridge, without an unused body copy.
+Explicitly owned sources send the complete
 non-generated narrative and region lineage to the typed consumer. Missing files
 and invalid UTF-8 still block dependent delivery. Byte reduction alone does not
 establish model-token, latency or outcome gains.
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index bffd5002e8..11520f9030 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -9,6 +9,7 @@
 from .quota_capture import (
     bind_capture_selection_transport, capture_decision, handoff_decision,
     handle_native_context_command, handle_replan_view, prepare_quota_capture, native_context_capture_view,
+    native_context_activation,
 )
 from ..control_plane.turn_driver.decision_capture import (
     context_capture_read_command,
@@ -371,8 +372,6 @@ def _emit_quota_result(
             turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id),
             started_at=usage_quota_started,
         )
-    from .quota_capture import native_context_activation
-
     context_delivery_enabled = payload.get("ok") is True and native_context_activation(args)["enabled"]
     bind_capture_selection_transport(payload, args)
     delivery_request = None
@@ -617,6 +616,7 @@ def handle_quota_command(
                 status_payload,
                 goal_id=args.goal_id,
                 agent_id=args.agent_id,
+                inline_goal_source=native_context_activation(args)["enabled"],
                 available_capabilities=args.available_capabilities,
                 include_scheduler_detail="scheduler" in detail_sections,
                 include_agent_todo_detail=(
diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py
index 08c762a063..9169420968 100644
--- a/loopx/cli_commands/turn_decision.py
+++ b/loopx/cli_commands/turn_decision.py
@@ -170,6 +170,7 @@ def build_turn_decision(
     ) -> dict[str, Any]:
         decision = build_live_quota_should_run_decision(
             status_payload,
+            inline_goal_source=True,
             goal_id=args.goal_id,
             agent_id=args.agent_id,
             available_capabilities=args.available_capabilities,
diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py
index 8518f03ad5..982c5695d1 100644
--- a/loopx/control_plane/quota/live_decision.py
+++ b/loopx/control_plane/quota/live_decision.py
@@ -547,6 +547,7 @@ def build_live_quota_should_run_decision(
     turn_start_hook_dispatch: Mapping[str, Any] | None = None,
     workspace_path: Path | None = None,
     goal_ref: Mapping[str, object] | None = None,
+    inline_goal_source: bool = False,
 ) -> dict[str, Any]:
     """Build one live CLI decision while keeping host observation injectable."""
     resolved_context = resolve_scheduler_execution_context(scheduler_execution_context)
@@ -824,5 +825,5 @@ def build_live_quota_should_run_decision(
     from ..work_items.context_readback import attach_work_context
 
     attach_work_context(payload, registry_path=registry_path, runtime_root=runtime_root,
-        hook_dispatch=turn_start_hook_dispatch)
+        hook_dispatch=turn_start_hook_dispatch, inline_goal_source=inline_goal_source)
     return payload
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index 73a6a7220e..c5c14f651f 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -21,7 +21,7 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { projectTurnStartUnavailableContext } from "../capability_hooks.ts";
 import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral } from "../runtime_decode.ts";
 import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts";
-import { capturedTodoTextSha256, decodeInteractionContract, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, scopeInteractionOutcomeReadGuidance, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
+import { assertGoalContextDelivered, capturedTodoTextSha256, decodeInteractionContract, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, scopeInteractionOutcomeReadGuidance, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
 import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts";
 export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
 
@@ -2076,8 +2076,11 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
       agent_id: scalarString(object(decision.agent_identity).agent_id, "Turn Agent identity").trim(),
       todo_id: (selectedWorkTodoId(decision) ?? "").trim()};
   }
-  if (operation === "host_context") {
+  if (operation === "host_context" || operation === "host_dispatch_context") {
     const {decision, projection} = completeHostSource(request);
+    if (operation === "host_dispatch_context" && Object.keys(object(projection.replan_action_packet)).length === 0) {
+      assertGoalContextDelivered(object(object(decision.interaction_contract).agent_channel));
+    }
     // The caller already holds this exact checked source. Return only the
     // resolved facts it needs, not another full copy over the transport.
     return {primary_action: object(projection.action).primary_action ?? null,
diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py
index d8d114b47d..c6cd7ffb24 100644
--- a/loopx/control_plane/turn_driver/codex_cli.py
+++ b/loopx/control_plane/turn_driver/codex_cli.py
@@ -324,7 +324,7 @@ def _checked_prompt_context(request: Mapping[str, Any]) -> dict[str, Any] | None
     # The complete owner checks the adapter version, source and lineage in one
     # call. Keep the standalone version gate only for historical requests.
     try:
-        checked = _complete_host_source(request, operation="host_context")
+        checked = _complete_host_source(request, operation="host_dispatch_context")
     except RuntimeError:
         # Preserve the historical diagnostic precedence for doubly malformed
         # input without imposing a second bridge call on valid requests.
diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py
index e695b60d81..04be995533 100644
--- a/loopx/control_plane/turn_driver/host_candidate.py
+++ b/loopx/control_plane/turn_driver/host_candidate.py
@@ -127,7 +127,7 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
     """Use the complete source; decode the old layout for historical input only."""
 
     try:
-        checked = _complete_host_source(request, operation="host_context")
+        checked = _complete_host_source(request, operation="host_dispatch_context")
     except RuntimeError as exc:
         raise ValueError("complete Turn contract rejected") from exc
     if checked is not None:
diff --git a/loopx/control_plane/work_items/context_readback.py b/loopx/control_plane/work_items/context_readback.py
index f338e6f9da..ebfee8f4b0 100644
--- a/loopx/control_plane/work_items/context_readback.py
+++ b/loopx/control_plane/work_items/context_readback.py
@@ -20,7 +20,8 @@
 from ..todos.summary_item import todo_text_content_revision
 
 
-def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agent_id=None):
+def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agent_id=None,
+                    inline_goal_source=False):
     source = read.get("source")
     if source == "goal_state":
         _goal, _project, state_file = resolve_goal_state(
@@ -32,8 +33,8 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agen
             "revision": "sha256:" + sha256(source_bytes).hexdigest()}
         # Only the writer's explicitly marked regions have generated ownership.
         # Retain every other byte, except identity-bound historical Monitor
-        # receipt lines owned by their codec. Legacy/unmarked input stays a full
-        # progressive read; malformed ownership fails rather than hiding text.
+        # receipt lines owned by their codec. Progressive callers keep legacy
+        # reads; complete-host delivery retains the entire unmarked source.
         lines = text.splitlines(keepends=True)
         visible = visible_markdown_lines(lines)
         regions = find_todo_regions(lines, visible=visible)
@@ -63,6 +64,8 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agen
                 # grows the current prompt without adding recovery information.
                 content["excluded_monitor_receipts"] = {"count": receipt_count,
                     "line_ranges": receipt_ranges}
+        if inline_goal_source and "text" not in content:
+            content.update(text=text, full_source=True)
         return content
     if source == "goal_acceptance":
         # The typed acceptance owner scopes task-binding observations while
@@ -127,7 +130,8 @@ def _user_context(*, registry_path, runtime_root, goal_id, agent_id):
 
 
 def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
-                        runtime_root: Path, hook_dispatch: dict[str, Any] | None) -> None:
+                        runtime_root: Path, hook_dispatch: dict[str, Any] | None,
+                        inline_goal_source: bool = False) -> None:
     packet_selected = payload.get("selected_todo")
     selected_text_snapshot = None
     if isinstance(packet_selected, dict):
@@ -152,7 +156,8 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
         result = {"command": read["command"]}
         try:
             result["content"] = _source_content(read, registry_path=registry_path,
-                runtime_root=runtime_root, goal_id=goal_id, todo_id=plan.get("todo_id"), agent_id=agent_id)
+                runtime_root=runtime_root, goal_id=goal_id, todo_id=plan.get("todo_id"),
+                agent_id=agent_id, inline_goal_source=inline_goal_source)
         except (OSError, ValueError, RuntimeError):
             result["error_code"] = "context_source_unavailable"
         results.append(result)
@@ -166,8 +171,12 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
     projected = effect_runtime_result("work_item.context.project", {
         "required_reads": reads, "source_results": results,
         "selected_todo": plan.get("selected_todo"), "user_todos": users,
-        "hook_dispatch": hook_dispatch,
+        "hook_dispatch": hook_dispatch, "require_full_goal_source": inline_goal_source,
     }, large_local_snapshot=True)
+    # An older runtime must not silently ignore the required delivery profile.
+    profile = projected.pop("source_delivery_profile", None)
+    if inline_goal_source and profile != "complete_goal_v1":
+        raise RuntimeError("Complete Goal source delivery is unsupported by this runtime")
     channel.update(projected)
     selected_todo = payload.get("selected_todo")
     if isinstance(selected_todo, dict):
diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts
index 83c51b931d..aff9ed3052 100644
--- a/loopx/control_plane/work_items/interaction_contract.ts
+++ b/loopx/control_plane/work_items/interaction_contract.ts
@@ -883,9 +883,17 @@ export function planInteractionWorkContext(request: JsonObject): JsonObject {
 
 /** The registered file adapter reuses the writer's strict region parser.
  * No heading/prose classification here: its exact complement is Goal context,
- * and canonical Todo readers retain task authority. Unknown/legacy input keeps
- * the complete progressive read rather than pretending to fulfill it. */
+ * and canonical Todo readers retain task authority. A complete original file
+ * needs no exclusions; metadata alone retains the progressive read. */
 function completeGoalNarrative(content: JsonObject): boolean {
+  if (Object.hasOwn(content, "read_complete") && content.read_complete !== true) return false;
+  if (Object.hasOwn(content, "full_source")) {
+    return content.full_source === true && typeof content.text === "string"
+      && typeof content.state_file === "string" && content.state_file.length > 0
+      && content.revision === `sha256:${createHash("sha256").update(content.text, "utf8").digest("hex")}`
+      && !Object.hasOwn(content, "excluded_todo_regions")
+      && !Object.hasOwn(content, "excluded_monitor_receipts");
+  }
   if (!Array.isArray(content.excluded_todo_regions) || !content.excluded_todo_regions.length
       || typeof content.text !== "string" || typeof content.state_file !== "string" || !content.state_file
       || typeof content.revision !== "string" || !ENVELOPED_SHA256_PATTERN.test(content.revision)) return false;
@@ -935,6 +943,24 @@ function completeGoalNarrative(content: JsonObject): boolean {
   }) && Number.isSafeInteger(count) && count === expectedCount;
 }
 
+export function assertGoalContextDelivered(channel: JsonObject): void {
+  const reads = Array.isArray(channel.required_reads) ? channel.required_reads : [];
+  if (reads.some(read => jsonObject(read)?.source === "goal_state")) {
+    throw new EffectRuntimeRequestError("Complete Goal source is required before host dispatch");
+  }
+  const context = channel.work_context === undefined ? {}
+    : requireJsonObject(channel.work_context, "Goal work context");
+  if (context.sources !== undefined && !Array.isArray(context.sources)) {
+    throw new EffectRuntimeRequestError("Goal source records must be an array");
+  }
+  for (const value of Array.isArray(context.sources) ? context.sources : []) {
+    const source = jsonObject(value);
+    if (source?.source === "goal_state" && !completeGoalNarrative(jsonObject(source.content) ?? {})) {
+      throw new EffectRuntimeRequestError("Goal source content or revision is invalid");
+    }
+  }
+}
+
 /** Completion binds to the admitted body, including whitespace, never a live
  * reread or display prefix. Both complete contracts and historical journals
  * use this resolver; absence of the old witness retains receipt recovery. */
@@ -968,6 +994,9 @@ export function capturedTodoTextSha256(
 /** Deliver full registered sources once. Unowned mixed Goal documents remain
  * progressive reads; failed/changed sources forbid dependent work. */
 export function projectInteractionWorkContext(request: JsonObject): JsonObject {
+  if (request.require_full_goal_source !== undefined && typeof request.require_full_goal_source !== "boolean") {
+    throw new EffectRuntimeRequestError("Goal source delivery requirement must be boolean");
+  }
   if (!Array.isArray(request.required_reads) || !Array.isArray(request.source_results)) {
     throw new EffectRuntimeRequestError("context reads and source results must be arrays");
   }
@@ -1023,11 +1052,13 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
       pending.push(read);
     } else if (read.source === "goal_state") {
       if (completeGoalNarrative(content!)) sources.push({...read,
-        reason: `${String(read.reason ?? "")} This registered source delivers all text outside explicitly generated Todo regions inline, with full-file revision and excluded-region lineage; exact canonical Todo reads own task state/claims.${content!.excluded_monitor_receipts ? " Recognized historical Monitor receipt lines retain source lineage instead of repeating their encoded bodies; the original file remains available for receipt recovery." : ""}`,
+        reason: content!.full_source === true
+          ? `${String(read.reason ?? "")} This registered source delivers the complete original Goal file inline, bound to its full-file revision; canonical Todo reads retain task authority.`
+          : `${String(read.reason ?? "")} This registered source delivers all text outside explicitly generated Todo regions inline, with full-file revision and excluded-region lineage; exact canonical Todo reads own task state/claims.${content!.excluded_monitor_receipts ? " Recognized historical Monitor receipt lines retain source lineage instead of repeating their encoded bodies; the original file remains available for receipt recovery." : ""}`,
         content: content!});
       else {
         pending.push(read);
-        if (Object.hasOwn(content!, "excluded_todo_regions")) failures.push({
+        if (Object.hasOwn(content!, "excluded_todo_regions") || Object.hasOwn(content!, "full_source")) failures.push({
           source: read.source, command: read.command, error_code: "context_source_unavailable",
           instruction: "Recover complete Goal source and rerun the guard; malformed region lineage cannot fulfill it.",
         });
@@ -1077,6 +1108,9 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
       }
     }
   }
+  if (request.require_full_goal_source === true) {
+    assertGoalContextDelivered({required_reads: pending, work_context: {sources}});
+  }
   const dispatch = jsonObject(request.hook_dispatch);
   const unavailable = projectTurnStartUnavailableContext(dispatch);
   const users = jsonObject(request.user_todos);
@@ -1084,7 +1118,8 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
     instruction: "Recover current User obligations and rerun the guard."});
   const includeUserTodos = users && !users.error_code
     && (!Array.isArray(users.todos) || users.todos.length > 0 || users.authority_read != null);
-  return {required_reads: pending, work_context: {complete: failures.length === 0,
+  return {...(request.require_full_goal_source === true ? {source_delivery_profile: "complete_goal_v1"} : {}),
+    required_reads: pending, work_context: {complete: failures.length === 0,
     ...(selectedTodoTextVerified ? {selected_todo_text_verified: true} : {}),
     ...(selectedTodoRef ? {selected_todo_ref: "selected_todo"} : {}),
     ...(selectedTodoAuthority ? {selected_todo_authority: selectedTodoAuthority} : {}),
diff --git a/tests/control_plane/test_complete_turn_decision.py b/tests/control_plane/test_complete_turn_decision.py
index cf0cab3770..7c399f68f5 100644
--- a/tests/control_plane/test_complete_turn_decision.py
+++ b/tests/control_plane/test_complete_turn_decision.py
@@ -104,6 +104,30 @@ def test_complete_source_rejects_tampering_and_binding_substitution():
         compose_turn_decision({**source, "schema_version": "unknown"})
 
 
+def test_pending_goal_capture_remains_readable_but_cannot_dispatch_business_work():
+    from loopx.control_plane.quota.turn_envelope import build_turn_controller_input
+    from loopx.control_plane.turn_driver.codex_cli import prepare_codex_cli_input
+    from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
+    from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority, execution_profile_present
+    from tests.test_turn_envelope import _full_decision
+
+    source = _full_decision()
+    source["interaction_contract"]["agent_channel"]["required_reads"] = [
+        {"source": "goal_state", "command": "cat -- original-goal.md", "ordering": "before_work"}]
+    contract = compose_turn_decision(source)
+    before = copy.deepcopy(contract)
+    plan = build_loopx_turn_plan(build_turn_controller_input(contract).envelope(), work_contract=contract,
+        host="dsh", execution_mode="isolated-headless", turn_instance_id="original-turn")
+    request = build_loopx_turn_host_request(plan)
+    assert isinstance(execution_profile_present(request), bool)
+    with pytest.raises(RuntimeError, match="Goal source"):
+        prepare_codex_cli_input(request)
+    with pytest.raises(ValueError, match="complete Turn contract rejected") as rejected:
+        extract_turn_authority(request)
+    assert "Goal source" in str(rejected.value.__cause__)
+    assert contract == before
+
+
 @pytest.mark.parametrize("action", ["future_execution", "", None, 0])
 def test_complete_source_refuses_unknown_required_action_before_capture(action):
     from tests.control_plane.test_replan_host_context_projection import _quota_payload
diff --git a/tests/control_plane/test_goal_state_context_transport.py b/tests/control_plane/test_goal_state_context_transport.py
index ed8e293553..2e91239fb9 100644
--- a/tests/control_plane/test_goal_state_context_transport.py
+++ b/tests/control_plane/test_goal_state_context_transport.py
@@ -1,4 +1,4 @@
-"""Full Goal availability checks do not transport an unconsumed source body."""
+"""Progressive quota and complete host delivery retain their distinct IO contracts."""
 from __future__ import annotations
 
 import contextlib
@@ -37,13 +37,17 @@ def test_goal_availability_transport_preserves_full_read_and_recovery(tmp_path,
     source_args = dict(registry_path=registry, runtime_root=runtime, goal_id=goal, todo_id=todo_id)
     original_effect = context_readback.effect_runtime_result
     requests, plans = [], []
+    omit_delivery_profile = [False]
 
     def observe(operation, request, **kwargs):
         if operation == "work_item.context.plan":
             plans.append(deepcopy(request["packet"]))
         if operation == "work_item.context.project":
             requests.append(deepcopy(request))
-        return original_effect(operation, request, **kwargs)
+        result = original_effect(operation, request, **kwargs)
+        if operation == "work_item.context.project" and omit_delivery_profile[0]:
+            result.pop("source_delivery_profile", None)
+        return result
 
     monkeypatch.setattr(context_readback, "effect_runtime_result", observe)
 
@@ -77,6 +81,16 @@ def guard():
     assert len(json.dumps(small_request).encode()) == len(json.dumps(large_request).encode())
     content = context_readback._source_content({"source": "goal_state"}, **source_args)
     assert content == {"state_file": str(state), "revision": "sha256:" + sha256(large).hexdigest()}
+    full = context_readback._source_content(
+        {"source": "goal_state"}, **source_args, inline_goal_source=True)
+    assert full == {**content, "text": large.decode("utf-8"), "full_source": True}
+    omit_delivery_profile[0] = True
+    try:
+        with pytest.raises(RuntimeError, match="unsupported by this runtime"):
+            context_readback.attach_work_context(deepcopy(captured_plan), registry_path=registry,
+                runtime_root=runtime, hook_dispatch=None, inline_goal_source=True)
+    finally:
+        omit_delivery_profile[0] = False
     # Independently compare the current typed projector with the former full
     # source input. Removing the transport must not weaken its obligations.
     legacy_request = deepcopy(large_request)
@@ -111,5 +125,8 @@ def guard():
         assert channel["delivery_allowed"] is False
         assert channel["work_context"]["complete"] is False
         assert any(row.get("source") == "goal_state" for row in channel["required_reads"])
+        with pytest.raises(RuntimeError, match="Goal source"):
+            context_readback.attach_work_context(deepcopy(captured_plan), registry_path=registry,
+                runtime_root=runtime, hook_dispatch=None, inline_goal_source=True)
         state.write_bytes(large)
         assert guard() == large_channel
diff --git a/tests/control_plane/test_turn_host_context_delivery.py b/tests/control_plane/test_turn_host_context_delivery.py
index 772e44375a..3ff470579e 100644
--- a/tests/control_plane/test_turn_host_context_delivery.py
+++ b/tests/control_plane/test_turn_host_context_delivery.py
@@ -3,6 +3,7 @@
 
 from copy import deepcopy
 import json
+from pathlib import Path
 import shlex
 import subprocess
 import sys
@@ -14,6 +15,7 @@
 from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
 from loopx.control_plane.turn_driver.executor import build_loopx_turn_host_request
 from loopx.control_plane.turn_driver.host_candidate import _canonical_hash
+from loopx.control_plane.turn_driver import codex_context
 from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
 from tests.control_plane.test_agent_context import POLICY
 from tests.control_plane.test_quota_settlement_cli import (
@@ -41,15 +43,28 @@ def test_captured_context_reaches_actual_host_without_readmission(tmp_path, monk
     _initialize_fixture_authority(project, runtime, provider=provider)
 
     capture = tmp_path / "capture"
-    code, envelope = _run_cli(registry, runtime, "quota", "should-run",
+    if mode != "legacy_reference":
+        native_directory = tmp_path / "context"
+        native_directory.mkdir(mode=0o700)
+        codex_context.codex_context_hook(str(native_directory), {
+            "session_id": "host-context-fixture", "cwd": str(project),
+            "hook_event_name": "SessionStart", "source": "startup",
+        })
+    delivery = (["--context-delivery", "on", "--codex-context-dir", str(tmp_path / "context")]
+        if mode != "legacy_reference" else [])
+    binary = Path(sys.executable).with_name("loopx.exe" if sys.platform == "win32" else "loopx")
+    guard = [str(binary), "--registry", str(registry),
+        "--runtime-root", str(runtime), "--format", "json", "quota", "should-run",
         "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
         "--turn-instance-id", "host-context-turn", "--codex-app",
         "--available-capability", "filesystem_read", "--available-capability", "subagent_spawn",
-        "--turn-envelope", "--decision-output-dir", str(capture), cwd=project)
-    assert code == 0, envelope
+        "--turn-envelope", "--decision-output-dir", str(capture), *delivery]
+    produced = subprocess.run(guard, cwd=project, text=True, capture_output=True, timeout=30)
+    assert produced.returncode == 0, produced.stdout + produced.stderr
     original = (capture / "decision.json").read_bytes()
     source = json.loads(original)
-    assert envelope == source
+    if mode == "legacy_reference":
+        assert json.loads(produced.stdout) == source
     assert source["schema_version"] == "loopx_turn_envelope_v1"
     envelope = build_turn_envelope(source, captured_decision_path=str(capture / "decision.json"))
     context = source["interaction_contract"].get("agent_context")
@@ -126,9 +141,12 @@ def test_captured_context_reaches_actual_host_without_readmission(tmp_path, monk
             rejected = subprocess.run(command, input=json.dumps(invalid), text=True, capture_output=True, timeout=30)
             assert rejected.returncode != 0 and "complete Turn contract rejected" in rejected.stderr
             assert prompt_path.read_text() == prompt  # Refusal precedes the provider seam.
-        assert source["interaction_contract"]["agent_channel"]["required_reads"]
+        channel = source["interaction_contract"]["agent_channel"]
+        assert not any(read.get("source") == "goal_state" for read in channel["required_reads"])
+        assert any(row.get("source") == "goal_state"
+            for row in channel["work_context"]["sources"])
         invalid = deepcopy(request)
-        invalid["work_contract"]["interaction_contract"]["agent_channel"]["required_reads"] = []
+        invalid["work_contract"]["interaction_contract"]["agent_channel"]["work_context"]["sources"] = []
         rejected = subprocess.run(command, input=json.dumps(invalid), text=True, capture_output=True, timeout=30)
         assert rejected.returncode != 0 and "complete Turn contract rejected" in rejected.stderr
         assert prompt_path.read_text() == prompt
@@ -138,6 +156,8 @@ def test_captured_context_reaches_actual_host_without_readmission(tmp_path, monk
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_bound_replan_reaches_actual_host_with_original_evidence(tmp_path, monkeypatch, provider):
+    from loopx.control_plane.turn_driver.codex_cli import prepare_codex_cli_input
+
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry = _write_fixture(tmp_path)
     _configure_selected_todo_replan_fixture(project, registry)
@@ -187,6 +207,9 @@ def test_bound_replan_reaches_actual_host_with_original_evidence(tmp_path, monke
     command = [sys.executable, "-m", "loopx.dsh_goal_mode", "--workspace", str(project),
                "--dsh-home", str(tmp_path / "host-home"), "--dsh-runner", str(runner)]
     request = build_loopx_turn_host_request({**plan, "work_contract": source})
+    codex_prompt, _ = prepare_codex_cli_input(request)
+    assert "Read detail_ref.replan_action_packet.review" in codex_prompt
+    assert "before reading detail_ref.replan_action_packet.writeback" in codex_prompt
     before = {p: p.read_bytes() for p in runtime.rglob("*") if p.is_file()}
     result = subprocess.run(command, input=json.dumps(request), text=True, capture_output=True, timeout=30)
     assert result.returncode == 0, result.stdout + result.stderr
diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts
index 6c401c3414..be1d8012fc 100644
--- a/tests/control_plane_ts/interaction_contract.test.ts
+++ b/tests/control_plane_ts/interaction_contract.test.ts
@@ -13,6 +13,7 @@ import {
   INTERACTION_EXECUTION_FACTS_SCHEMA_VERSION,
   projectInteractionRequiredReads,
   projectInteractionWorkContext,
+  assertGoalContextDelivered,
   scopeInteractionOutcomeReadGuidance,
   projectBlockedSuccessorWaitObservation,
   type AgentInteractionChannel,
@@ -668,6 +669,31 @@ test("scoped presentation keeps the original objective, refinement and stop obli
 });
 
 
+test("full original Goal delivery fulfills its read without excluding legacy text", () => {
+  const text = "Original intent\nUnknown section\n停止条件：不得发布。\n";
+  const read = {source: "goal_state", command: "cat -- goal.md", ordering: "before_work"};
+  const content = {state_file: "goal.md", text, full_source: true,
+    revision: `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}`};
+  const project = (value: JsonObject | null) => projectInteractionWorkContext({
+    required_reads: [read], require_full_goal_source: true,
+    source_results: value ? [{command: read.command, content: value}] : [],
+  });
+  const fulfilled = project(content);
+  assert.equal(fulfilled.source_delivery_profile, "complete_goal_v1");
+  assert.deepEqual(fulfilled.required_reads, []);
+  assert.equal((fulfilled.work_context as JsonObject).complete, true);
+  const [source] = (fulfilled.work_context as JsonObject).sources as JsonObject[];
+  assert.deepEqual(source.content, content);
+  for (const invalid of [null, {state_file: "goal.md", revision: content.revision},
+    {...content, text: "Original intent\n"}, {...content, full_source: false},
+    {...content, excluded_todo_regions: []}, {...content, read_complete: false}]) {
+    assert.throws(() => project(invalid), /Goal source/);
+  }
+  assert.throws(() => assertGoalContextDelivered({required_reads: [read]}), /Goal source/);
+  assert.throws(() => assertGoalContextDelivered({required_reads: [],
+    work_context: {sources: [{source: "goal_state", content: {...content, read_complete: false}}]}}), /Goal source/);
+});
+
 test("Goal narrative fulfillment needs complete source and ordered region lineage", () => {
   const read = {source: "goal_state", command: "cat -- goal.md"};
   const region = {role: "agent", start_line: 4, end_line: 8, revision: `sha256:${"b".repeat(64)}`};