fix(context): deliver complete Goal source before host dispatch

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.

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 材料读取调用仍按原有阶段与工作限制执行。

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.

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

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,

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

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,

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.

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:

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