diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md
index 908fa52640..4a2cc8e21f 100644
--- a/docs/reference/protocols/turn-envelope-v0.md
+++ b/docs/reference/protocols/turn-envelope-v0.md
@@ -617,7 +617,7 @@ capsule witnesses. Packet-free v0 envelopes remain supported. The envelope keeps
 - validation/writeback and quota-spend policy;
 - the current scheduler action and cadence acknowledgement command.
 
-The envelope also carries a bounded `contract_capsule` for interaction mode,
+The envelope also carries a `contract_capsule` for interaction mode,
 work-lane and execution obligations, successor/replan duties, automation
 liveness, vision/handoff state, and actionable warning references. A canonical
 `action_signature` is independently built from the full decision and from the
@@ -625,6 +625,84 @@ envelope; matching hashes prove the covered action dimensions agree for that
 projection. They do not prove that every possible quota state has test
 coverage.
 
+The five execution-contract objects `work_lane_contract`,
+`execution_obligation`, `autonomous_replan_scope`, `agent_scope_frontier`, and
+`automation_liveness` retain their complete source JSON, including nested data,
+ordered lists, empty values and unrecognized metadata. Their clauses are not
+subject to display truncation. Both quota-envelope emission and direct Turn
+composition use this shared producer. The shared structured Host prompt now
+forwards the remaining signed capsule for ordinary execution as well as staged
+replanning; previously the ordinary path omitted it. Retaining metadata does
+not interpret it as a new execution permission or support a new required
+semantic. Existing schema/signature coverage and historical captured envelopes
+remain readable without rewriting them. Re-projecting an older raw decision
+with the corrected producer can produce a different signature; validate an
+existing envelope against its saved content and digest. Raw quota output and admission,
+scheduler and settlement decisions are unchanged. Larger capsule bytes remain
+visible in the existing diagnostic budget warning; this correction does not
+claim lower token cost or complete controller-contract unification.
+
+中文：上述五份执行契约完整保留源 JSON，包括嵌套数据、有序列表、空值和未知
+元数据，不按展示预算截断条款。quota 短包与直接 Turn 共用该生产者；共享结构化
+Host 在普通执行和分阶段重规划中都传递剩余的已签名 capsule，修复普通执行
+原先遗漏的问题。保留元数据不表示解释新权限或支持新的必需语义。既有 schema、
+签名覆盖与历史捕获内容继续可读，不改写旧记录。用修正后的生产者重新投影旧
+原始决定可能得到不同签名；既有 envelope 应按保存内容与摘要校验。原始 quota
+输出、准入、调度和
+结算决定不变。增长的字节量仍通过既有预算诊断可见；这项修正不证明 token
+成本下降，也不表示 controller 契约已完全统一。
+
+Live Turn planning, native Codex prompt construction and structured Host
+consumption share the TypeScript owner's version gate. The current consumer
+accepts `loopx_turn_envelope_v0`, action-signature schema v0 and the registered
+v0–v5 coverage versions. An explicitly unknown or null version is rejected
+before Host launch, even when its unchanged action digest still matches. The
+historical digest document does not sign these version declarations, so digest
+equality alone cannot establish consumer compatibility. Existing v0 callers
+that omit signature-version metadata retain their prior behavior; digest,
+lineage, current authority and settlement checks remain separate. Saved capture
+bytes and old signature documents are not rewritten. This gate does not qualify
+a future unified schema or interpret unknown required nested semantics.
+
+中文：实时 Turn 规划、原生 Codex 提示构造和结构化 Host 消费共用 TypeScript
+owner 的版本门。当前 consumer 接受 envelope v0、action-signature schema v0
+及已注册的 v0–v5 coverage；显式未知或 null 版本即使仍携带匹配的旧行动摘要，
+也会在 Host 启动前拒绝。历史摘要文档不签署这些版本声明，摘要相等不能证明
+消费者兼容。省略签名版本元数据的既有 v0 调用方保持原行为；摘要、谱系、当前
+权限和结算校验仍各自负责。既有捕获字节和历史签名文档不改写。此门不表示未来
+统一 schema 已获得资格，也不负责解释未知的必需嵌套语义。
+
+Ordinary structured Host prompts also retain the signed `user`,
+`execution_policy`, `scheduler`, `response_plan`, and
+`task_orchestration_contract` sections, plus the existing captured-decision
+reference when present. These sections are no longer conditional on a staged
+replan reader. Notification policy does not grant or cancel execution authority;
+carrying surrounding Turn plans does not delegate their effects. The prompt
+keeps state writeback, quota and scheduler operations with the outer Turn
+executor and asks the work segment to return its semantic result.
+
+中文：普通结构化 Host 还保留已签名的用户通知、执行策略、调度、响应计划、
+任务编排五个区段，以及已有的捕获决定引用，不再以分阶段 replan reader
+是否存在为条件。通知策略不授予或取消执行权限；传递外围 Turn 计划不等于
+委派其副作用。提示明确由外围 Turn executor 负责状态写回、quota 和调度操作，
+工作段返回语义结果。
+
+The structured Host also receives signed `agent_id`, `decision`, `should_run`,
+`effective_action` and `state`. Its `action` object retains execution flags,
+action choices and capability intent while the existing task/planning carriers
+keep their established locations. `recommended_action` remains a display
+fallback, omitted from this Host projection because it can retain an older
+recommendation than the authoritative `primary_action`. Neither a suggested
+alternative nor a transported intent overrides admission, scope or effect
+ownership. This is input preservation, not a new admission decision or the
+unified-contract writer transition.
+
+中文：结构化 Host 同时收到已签名的 Agent 身份、决定、是否运行、有效行动和
+状态；`action` 保留执行标记、行动选项及能力意图，任务和规划字段沿用已有位置。
+`recommended_action` 是可能落后于权威 `primary_action` 的展示回退，仍不放入
+此 Host 投影。建议选项和传递的意图不覆盖准入、作用域或副作用归属。这是输入
+保真修复，不增加准入决策，也不表示统一契约 writer 已切换。
+
 Required reads use the authoritative `interaction_contract.agent_channel` list,
 including an explicit empty list. Historical interaction-root and payload lists
 remain fallbacks when that carrier is absent. Commands retain their exact bytes;
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index 4179d8b806..01de9f1b73 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -47,6 +47,11 @@ def _turn_envelope_result(
     return dict(result)
 
 
+def assert_turn_envelope_consumer_version(envelope: Mapping[str, Any]) -> None:
+    """Use the typed wire-version gate before any Host consumes this packet."""
+    _turn_envelope_result("consumer_version", envelope=dict(envelope))
+
+
 def turn_envelope_action_signature_document(
     envelope: Mapping[str, Any],
 ) -> dict[str, Any]:
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index 6373bf9fb3..6441f5c62d 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -108,32 +108,10 @@ const ACTIONABLE_WARNING_FIELDS = [
 
 const CONTRACT_CAPSULE_FIELDS: Readonly<Record<string, readonly string[]>> = {
   interaction_contract: ["schema_version", "mode"],
-  work_lane_contract: [
-    "schema_version", "lane", "monitor_kind", "next_lane", "obligation",
-    "must_attempt_work", "reason_codes", "monitor_policy", "selected_todo_id",
-    "selected_next_due_at", "non_runnable_non_monitor_count",
-    "material_transition", "action",
-  ],
-  execution_obligation: [
-    "kind", "contract", "contract_obligation", "must_attempt_work",
-    "notify_is_execution_gate", "delivery_allowed", "reason",
-  ],
   goal_route_hint: [
     "schema_version", "kind", "route_decision", "preserves_goal_next_action",
     "goal_next_action_mutation",
   ],
-  autonomous_replan_scope: [
-    "schema_version", "required", "applies", "scope", "owner_agent_ids",
-    "selected_peer_agent",
-  ],
-  agent_scope_frontier: [
-    "schema_version", "action", "effective_action", "blocks_delivery",
-    "quiet_noop_allowed", "requires_replan", "recommended_action", "spend_policy",
-  ],
-  automation_liveness: [
-    "schema_version", "keep_active", "pause_allowed", "pause_policy",
-    "automation_action",
-  ],
   handoff_readiness: [
     "ready", "codex_ready", "handoff_status", "post_handoff_run_seen",
   ],
@@ -143,13 +121,6 @@ const CONTRACT_CAPSULE_FIELDS: Readonly<Record<string, readonly string[]>> = {
   ],
 };
 
-const CONTRACT_CAPSULE_TEXT_LIMITS: Readonly<Record<string, Readonly<Record<string, number>>>> = {
-  work_lane_contract: { action: 300, material_transition: 240 },
-  execution_obligation: { reason: 240 },
-  agent_scope_frontier: { recommended_action: 320, spend_policy: 220 },
-  automation_liveness: { pause_policy: 260 },
-};
-
 function object(value: unknown): JsonObject {
   return typeof value === "object" && value !== null && !Array.isArray(value)
     ? { ...(value as JsonObject) }
@@ -750,10 +721,21 @@ function contractCapsule(payload: JsonObject): JsonObject {
   };
   for (const [sourceKey, fields] of Object.entries(CONTRACT_CAPSULE_FIELDS)) {
     const compact = compactFields(
-      payload[sourceKey], fields, CONTRACT_CAPSULE_TEXT_LIMITS[sourceKey] || {},
+      payload[sourceKey], fields, {},
     );
     if (Object.keys(compact).length > 0) capsule[sourceKey] = compact;
   }
+  // These source owners provide execution contracts, not display summaries.
+  // Preserve clauses and nested/unknown data without interpreting new fields.
+  // The existing signature covers the complete retained JSON; budget overflow
+  // remains diagnostic rather than permission to truncate an obligation.
+  for (const field of [
+    "work_lane_contract", "execution_obligation", "autonomous_replan_scope",
+    "agent_scope_frontier", "automation_liveness",
+  ]) {
+    const contract = object(payload[field]);
+    if (Object.keys(contract).length > 0) capsule[field] = structuredClone(contract);
+  }
   const unavailableContext = projectTurnStartUnavailableContext(payload.turn_start_capability_hook_dispatch);
   if (unavailableContext) capsule.unavailable_context = unavailableContext;
   // Work qualification is behavioral context, including its outcome and
@@ -797,17 +779,6 @@ function contractCapsule(payload: JsonObject): JsonObject {
   }
   const taskScope = text(payload.task_scope, 80);
   if (taskScope) capsule.task_scope = taskScope;
-  const workLane = object(payload.work_lane_contract);
-  const outcome = object(workLane.outcome_followthrough);
-  if (Object.keys(outcome).length > 0) {
-    const compactLane = object(capsule.work_lane_contract);
-    compactLane.outcome_followthrough = compactFields(
-      outcome,
-      ["required", "obligation", "accepted_resolution_kinds", "spend_policy"],
-      { spend_policy: 220 },
-    );
-    capsule.work_lane_contract = compactLane;
-  }
   const warnings = ACTIONABLE_WARNING_FIELDS.filter((field) => Boolean(payload[field]));
   if (warnings.length > 0) capsule.actionable_warning_refs = warnings;
   return capsule;
@@ -1071,6 +1042,28 @@ function signatureCoverage(envelope: JsonObject, responsePlanValue: unknown): st
   return ACTION_SIGNATURE_COVERAGE_V0;
 }
 
+/** Live consumers must understand the declared wire versions before dispatch.
+ * Historical v0 callers may omit signature metadata; digest verification stays
+ * with its existing owner. Explicit unknown/null declarations are never v0.
+ * This does not make the bounded envelope a complete future work contract.
+ */
+export function assertTurnEnvelopeConsumerVersion(envelope: JsonObject): void {
+  assertSupportedQuotaPacket(envelope);
+  if (envelope.schema_version !== TURN_ENVELOPE_SCHEMA_VERSION) {
+    throw new EffectRuntimeRequestError("unsupported TurnEnvelope schema_version; use a compatible consumer");
+  }
+  const signature = object(envelope.action_signature);
+  if (Object.hasOwn(signature, "schema_version") && signature.schema_version !== ACTION_SIGNATURE_SCHEMA_VERSION) {
+    throw new EffectRuntimeRequestError("unsupported TurnEnvelope action signature schema_version; use a compatible consumer");
+  }
+  if (Object.hasOwn(signature, "coverage") && (typeof signature.coverage !== "string" || ![
+    ACTION_SIGNATURE_COVERAGE_V0, ACTION_SIGNATURE_COVERAGE_V1, ACTION_SIGNATURE_COVERAGE_V2,
+    ACTION_SIGNATURE_COVERAGE_V3, ACTION_SIGNATURE_COVERAGE_V4, ACTION_SIGNATURE_COVERAGE_V5,
+  ].includes(signature.coverage))) {
+    throw new EffectRuntimeRequestError("unsupported TurnEnvelope action signature coverage; use a compatible consumer");
+  }
+}
+
 export function turnEnvelopeActionSignatureDocument(value: unknown): JsonObject {
   const envelope = requireJsonObject(value, "turn_envelope");
   assertSupportedQuotaPacket(envelope);
@@ -1275,6 +1268,10 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
       autonomous_replan_obligation: {...obligation, context_read: reads.review},
       replan_action_packet: replanActionPacket(payload)};
   }
+  if (operation === "consumer_version") {
+    assertTurnEnvelopeConsumerVersion(requireJsonObject(request.envelope, "turn_envelope"));
+    return {ok: true};
+  }
   if (operation === "build") return buildTurnEnvelope(request);
   if (operation === "context_delivery") return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
   if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request);
diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py
index bd88af8dce..b36c8d2437 100644
--- a/loopx/control_plane/turn_driver/codex_cli.py
+++ b/loopx/control_plane/turn_driver/codex_cli.py
@@ -307,7 +307,10 @@ def codex_cli_result_schema(
 
 
 def _prompt(request: Mapping[str, Any]) -> str:
+    from ..quota.turn_envelope import assert_turn_envelope_consumer_version
+
     envelope = _mapping(request.get("turn_envelope"))
+    assert_turn_envelope_consumer_version(envelope)
     staged = _mapping(_mapping(envelope.get("detail_ref")).get("replan_action_packet")).get("review")
     model_request = request
     if staged:
diff --git a/loopx/control_plane/turn_driver/driver.py b/loopx/control_plane/turn_driver/driver.py
index 95591f2394..e80fbf381a 100644
--- a/loopx/control_plane/turn_driver/driver.py
+++ b/loopx/control_plane/turn_driver/driver.py
@@ -19,7 +19,7 @@
     build_subagent_execution_topology,
 )
 from ..quota.turn_envelope import (
-    TURN_ENVELOPE_SCHEMA_VERSION,
+    assert_turn_envelope_consumer_version,
     turn_envelope_action_signature_document,
     turn_envelope_replan_settlement_identity,
 )
@@ -138,8 +138,6 @@ def _mapping(value: Any) -> dict[str, Any]:
 
 
 def _typed_route(envelope: Mapping[str, Any]) -> LoopXTurnRoute:
-    if envelope.get("schema_version") != TURN_ENVELOPE_SCHEMA_VERSION:
-        return LoopXTurnRoute.CONTRACT_ERROR
     signature = _mapping(envelope.get("action_signature"))
     source_hash = str(signature.get("source_hash") or "")
     envelope_hash = str(signature.get("envelope_hash") or "")
@@ -509,9 +507,14 @@ def build_loopx_turn_plan(
         scheduler_owner=scheduler_owner,
     )
     envelope = dict(turn_envelope)
+    contract_error = None
+    try:
+        assert_turn_envelope_consumer_version(envelope)
+    except (ValueError, RuntimeError) as exc:
+        contract_error = str(exc)
     route = (
         _typed_route(envelope)
-        if execution_context.ok
+        if execution_context.ok and contract_error is None
         else LoopXTurnRoute.CONTRACT_ERROR
     )
     selected_todo = selected_turn_todo(envelope)
@@ -609,8 +612,8 @@ def build_loopx_turn_plan(
             "opaque_session_handle_omitted": True,
         },
         **(
-            {"error": session_error or "; ".join(execution_context.errors)}
-            if session_error or not execution_context.ok
+            {"error": contract_error or session_error or "; ".join(execution_context.errors)}
+            if contract_error or session_error or not execution_context.ok
             else {}
         ),
     }
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 422440c36b..bc827590b5 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -156,7 +156,7 @@ def build_loopx_turn_host_request(plan: Mapping[str, Any]) -> dict[str, Any]:
         raise ValueError("LoopX Turn plan has no valid transaction turn_key")
     route = plan.get("route") if isinstance(plan.get("route"), dict) else {}
     if route.get("would_invoke_host") is not True:
-        raise ValueError("LoopX Turn route is not host executable")
+        raise ValueError(str(plan.get("error") or "LoopX Turn route is not host executable"))
     request = {
         "schema_version": LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION,
         "turn_key": turn_key,
diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py
index d01a976c49..efdfbd7de4 100644
--- a/loopx/control_plane/turn_driver/host_candidate.py
+++ b/loopx/control_plane/turn_driver/host_candidate.py
@@ -11,7 +11,7 @@
 import json
 from typing import Any
 
-from ..quota.turn_envelope import turn_envelope_action_signature_document
+from ..quota.turn_envelope import assert_turn_envelope_consumer_version, turn_envelope_action_signature_document
 from ..work_items.delivery_batch_scale import DELIVERY_BATCH_SCALE_CHOICES
 from ..work_items.delivery_outcome import DELIVERY_OUTCOME_CHOICES
 from .driver import selected_turn_todo
@@ -74,6 +74,7 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
     """Return the signed action and safety boundary exactly as projected."""
 
     envelope = _mapping(request.get("turn_envelope"))
+    assert_turn_envelope_consumer_version(envelope)
     signature = _mapping(envelope.get("action_signature"))
     source_hash = str(signature.get("source_hash") or "")
     envelope_hash = str(signature.get("envelope_hash") or "")
@@ -104,6 +105,18 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
         "write_scope": list(write_scope) if isinstance(write_scope, list) else [],
         "workspace_guard": _mapping(boundary.get("workspace_guard")),
     }
+    # Preserve signed decision facts, including explicit false/null values.
+    # They describe the admitted Turn; this adapter never re-decides admission.
+    for field in ("agent_id", "decision", "should_run", "effective_action", "state"):
+        if field in envelope:
+            authority[field] = envelope[field]
+    # Keep decision flags and registered choices alongside the relocated task.
+    # recommended_action is a bounded display fallback and can retain an older
+    # recommendation; it must not compete with the signed primary_action.
+    remaining_action = {key: value for key, value in action.items()
+        if key not in {"primary_action", "recommended_action", "selected_todo", "core_goal", "planning_horizon", "acceptance_index"}}
+    if remaining_action:
+        authority["action"] = remaining_action
     # Keep the established scope/workspace aliases once, and transport every
     # remaining signed restriction rather than rebuilding policy in this IO adapter.
     remaining_boundary = {
@@ -112,10 +125,12 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
     }
     if remaining_boundary:
         authority["boundary"] = remaining_boundary
-    if staged:
-        for field in ("user", "execution_policy", "scheduler", "response_plan", "task_orchestration_contract"):
-            if envelope.get(field) is not None:
-                authority[field] = envelope[field]
+    # These signed obligations apply to ordinary work as well as staged replan.
+    # Forward the existing projection; transport must not decide which policy
+    # is relevant from the presence of a replan reader.
+    for field in ("user", "execution_policy", "scheduler", "response_plan", "task_orchestration_contract"):
+        if envelope.get(field) is not None:
+            authority[field] = envelope[field]
     if isinstance(envelope.get("work_context"), Mapping):
         authority["work_context"] = dict(envelope["work_context"])
     selected = selected_turn_todo(envelope)
@@ -131,8 +146,10 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
         value = _mapping(capsule.get(field))
         if value:
             authority[field] = value
+    # Ordinary execution needs the same work/continuation obligations as replan.
+    # Keep their signed source data; this adapter must not classify or execute it.
     remaining_capsule = {key: value for key, value in capsule.items() if key not in authority}
-    if staged and remaining_capsule:
+    if remaining_capsule:
         authority["contract_capsule"] = remaining_capsule
     context = _mapping(envelope.get("agent_context"))
     if context:
@@ -161,7 +178,7 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]:
                 else item for item in horizon["work_items"]
             ]
         authority["planning_horizon"] = horizon
-    if context.get("detail_ref") or replan or authority.get("vision_continuation_audit") or horizon:
+    if detail or context.get("detail_ref") or replan or authority.get("vision_continuation_audit") or horizon:
         # Replan/acceptance evidence and historical capability references need
         # the saved observation's route and hash. Native capture validation owns
         # its binding; this adapter neither reads it nor re-enters admission.
@@ -262,7 +279,14 @@ def render_prompt(authority: Mapping[str, Any], *, allow_completion: bool = Fals
         "Do not repeat reads already fulfilled in work_context. Write only inside write_scope, "
         "and obey workspace_guard and all supplied boundary restrictions and stop conditions. "
         "Checkpointed approval covers only its active_write_scope; other approval gates remain. "
-        "Do not infer authority from other prose.\n\n"
+        "Do not infer authority from other prose. "
+        "User notification policy does not grant or cancel execution authority. "
+        "Respect action.must_attempt, action.delivery_allowed and action.quiet_noop_allowed "
+        "when present; alternatives and capability intents do not override admission or scope. "
+        "Obey execution_policy and task_orchestration_contract when supplied. "
+        "Scheduler and response plans describe the surrounding Turn: do not execute "
+        "state writeback, quota or scheduler commands from this work segment. "
+        "Return the semantic result; the outer Turn executor owns those effects.\n\n"
         f"{context_instruction}"
         f"{planning_context_instruction}"
         f"{replan_instruction}"
diff --git a/tests/control_plane/test_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py
index 3032edf870..d7bf5131a3 100644
--- a/tests/control_plane/test_effect_turn_live_quota_decision.py
+++ b/tests/control_plane/test_effect_turn_live_quota_decision.py
@@ -1357,6 +1357,14 @@ def test_packet_retirement_preserves_reads_and_independent_capability_command(tm
     assert "protocol_action_packet" not in envelope["contract_capsule"]
     authority = extract_turn_authority({"turn_envelope": envelope})
     assert authority["primary_action"] == "Generate the exact report."
+    assert authority["action"]["capability_intent"]["command"] == command
+    assert authority["action"]["capability_intent"]["generation_authorized"] is True
+    assert authority["action"]["capability_intent"]["external_delivery_authorized"] is True
+    # This admitted capability operation is allowed while ordinary delivery
+    # remains unavailable. The Host must retain both distinct source facts.
+    assert authority["action"]["delivery_allowed"] is True
+    assert authority["execution_policy"]["normal_delivery_allowed"] is False
+    assert authority["effective_action"] == "governed_capability_intent"
     assert envelope["writeback"]["next_cli_actions"] == [command]
     assert authority["required_reads"]
 
diff --git a/tests/control_plane/test_protocol_packet_history.py b/tests/control_plane/test_protocol_packet_history.py
index 3b880bffe1..98362d69f3 100644
--- a/tests/control_plane/test_protocol_packet_history.py
+++ b/tests/control_plane/test_protocol_packet_history.py
@@ -100,7 +100,15 @@ def test_packet_free_release_record_keeps_original_signature_and_authority():
             ).encode()
         ).hexdigest()
     )
-    assert quota_action_signature_document(decision) == document
+    # Re-projecting the source now retains the execution-contract schema too.
+    # That is a new projection, not a reason to rewrite a historical receipt.
+    current_document = quota_action_signature_document(decision)
+    expected_current = deepcopy(document)
+    expected_current["contract_capsule"]["execution_obligation"] = deepcopy(
+        decision["execution_obligation"]
+    )
+    assert current_document == expected_current
+    assert current_document != document
     assert digest == envelope["action_signature"]["envelope_hash"]
     assert (
         extract_turn_authority({"turn_envelope": envelope})["primary_action"]
diff --git a/tests/control_plane/test_quota_action_portfolio.py b/tests/control_plane/test_quota_action_portfolio.py
index 33ccd6bf94..b649089581 100644
--- a/tests/control_plane/test_quota_action_portfolio.py
+++ b/tests/control_plane/test_quota_action_portfolio.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from copy import deepcopy
+import json
 
 import pytest
 
@@ -47,6 +48,14 @@ def test_accepted_replan_promotes_the_new_route_and_keeps_explicit_choice() -> N
         envelope["action"]["action_portfolio"]["primary"]["todo_id"]
         == "todo_replan_successor"
     )
+    from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority, render_prompt
+
+    authority = extract_turn_authority({"turn_envelope": envelope})
+    prompt = render_prompt(authority)
+    delivered, _ = json.JSONDecoder().raw_decode(prompt.split("Turn authority JSON:\n", 1)[1])
+    assert delivered["action"]["action_portfolio"] == portfolio
+    assert delivered["action"]["action_portfolio"]["selection_policy"]["recommendation_role"] == "default_not_binding"
+    assert delivered["selected_todo"]["todo_id"] == "todo_replan_successor"
     chosen = build_quota_should_run(
         status,
         goal_id=REPLAN_GOAL,
diff --git a/tests/control_plane/test_turn_host_decision_capture.py b/tests/control_plane/test_turn_host_decision_capture.py
index 5aa845e942..f0d87b05f5 100644
--- a/tests/control_plane/test_turn_host_decision_capture.py
+++ b/tests/control_plane/test_turn_host_decision_capture.py
@@ -92,6 +92,11 @@ def test_real_host_reads_original_capture_after_source_change(
     source_a = json.loads(original)
     assert source_a["goal_id"] == f.GOAL_ID
     assert source_a["agent_identity"]["agent_id"] == f.AGENT_ID
+    # Direct `turn` bypasses quota CLI emission. Check its actual subprocess
+    # prompt against the same original contracts, including recovery below.
+    for field in ("work_lane_contract", "execution_obligation", "automation_liveness"):
+        assert source_a[field]
+        assert request["turn_envelope"]["contract_capsule"][field] == source_a[field]
     # Change the real provider through its native CAS CLI, not the projection.
     prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"]
     code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID, "--todo-id", f.TODO_ID])
diff --git a/tests/control_plane/test_work_context_host_delivery.py b/tests/control_plane/test_work_context_host_delivery.py
new file mode 100644
index 0000000000..c913e2eef1
--- /dev/null
+++ b/tests/control_plane/test_work_context_host_delivery.py
@@ -0,0 +1,194 @@
+"""Authored requirements must reach the Host, not just agree after projection."""
+import copy
+import json
+import os
+import shlex
+import subprocess
+from urllib.parse import quote
+
+import pytest
+
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority, render_prompt
+from tests.control_plane.canonical_authority_fixture import (
+    initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos,
+)
+from tests.control_plane.test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _configure_read_only_todo,
+    _heartbeat_receipt_count, _run_cli, _spend_run_count, _write_fixture,
+)
+
+
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
+@pytest.mark.parametrize('repetitions', [1, 40])
+def test_authored_requirements_survive_capture_projection_and_host(tmp_path, monkeypatch, provider, repetitions):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = _configure_read_only_todo(project)
+    task = ('Compare the retained evidence / 保留原始证据. ' * repetitions
+            + 'Stop before publication; return unresolved discrepancies.')
+    goal = ('Keep acceptance separate from task status.\n' * 12
+            + 'Missing comparison evidence remains unknown, not unchanged.')
+    note = 'A complete task body does not replace this separate review requirement.'
+    authored = state.read_text().replace(
+        '- [ ] [P1] Validate and settle the selected delivery.', f'- [ ] [P1] {task}',
+    ).replace('## Next Action', f'## Acceptance\n\n{goal}\n\n## Next Action')
+    # Author only disposable source input, before creating canonical authority.
+    state.write_text(authored.replace(' -->', f' note={quote(note, safe="")} -->'), encoding='utf-8')
+    todos = read_fixture_todos(state)
+    source_task = next(item for item in todos if item['todo_id'] == TODO_ID)
+    # The fixture helper returns an attention view. Seed the canonical record
+    # from the authored full body, never its display truncation.
+    source_task.update(text=f'[P1] {task}', note=note)
+    projection = build_todo_runtime_shadow_projection(
+        goal_id=GOAL_ID, handoff_mode='soft_claim', todos=todos,
+    )
+    initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider)
+    args = ['quota', 'should-run', '--goal-id', GOAL_ID, '--agent-id', AGENT_ID,
+            '--todo-id', TODO_ID, '--turn-instance-id', TURN_ID, '--codex-app',
+            '--available-capability', 'shell', '--available-capability', 'filesystem_read']
+    capture = tmp_path / 'original-context'
+    code, envelope = _run_cli(registry, runtime, *args, '--turn-envelope',
+                              '--decision-output-dir', str(capture))
+    assert code == 0, envelope
+    saved = capture / 'decision.json'
+    original_bytes = saved.read_bytes()
+    full = json.loads(original_bytes)
+    source_context = full['interaction_contract']['agent_channel']['work_context']
+    assert source_context['complete'] is True
+    assert source_context['instruction']  # Non-vacuous instruction transport.
+    authority = extract_turn_authority({'turn_envelope': envelope})
+    # The raw heartbeat and the structured Host have different effect owners,
+    # but ordinary execution must not lose notification, bypass or scheduler
+    # restrictions just because no staged replan reader is present.
+    assert 'replan_action_packet' not in authority
+    assert authority['agent_id'] == AGENT_ID
+    for field in ('must_attempt', 'delivery_allowed', 'quiet_noop_allowed'):
+        assert authority['action'][field] == full['interaction_contract']['agent_channel'][field]
+    assert authority['should_run'] == full['should_run']
+    assert authority['effective_action'] == full['effective_action']
+    assert full['safe_bypass_allowed'] is False
+    assert authority['execution_policy']['safe_bypass_allowed'] is False
+    assert authority['user']['notify'] == full['interaction_contract']['user_channel']['notify']
+    assert authority['scheduler'] == envelope['scheduler']
+    assert authority['detail_ref']['captured_decision'] == envelope['detail_ref']['captured_decision']
+    # These are execution obligations, not display summaries. The ordinary
+    # structured Host must receive the same contracts as the raw quota caller.
+    for field in ('work_lane_contract', 'execution_obligation', 'automation_liveness'):
+        assert full[field]
+        assert envelope['contract_capsule'][field] == full[field]
+        assert authority['contract_capsule'][field] == full[field]
+    context = authority['work_context']
+    assert context['instruction'] == source_context['instruction']
+    assert authority['selected_todo']['text_source_verified'] is True
+    source_by_kind = {source['source']: source['content'] for source in context['sources']}
+    task_detail = source_by_kind['selected_todo']['todo']
+    assert task_detail['note'] == note
+    delivered_task = task_detail.get('text', authority['selected_todo']['text'])
+    assert delivered_task == source_task['text']
+    # This fixture has an unregistered Goal narrative: pending read must remain
+    # explicit even when all available inline sources were loaded successfully.
+    goal_read = next(item for item in authority['required_reads'] if item['source'] == 'goal_state')
+    assert goal_read['ordering'] == 'before_work'
+    if os.name != 'nt':
+        readback = subprocess.run(shlex.split(goal_read['command']), capture_output=True,
+                                  text=True, check=True)
+        assert goal in readback.stdout
+    else:
+        assert goal in state.read_text(encoding='utf-8')
+    # The actual Host prompt must contain the independent source requirements,
+    # not merely match an action signature computed by the same projector.
+    prompt = render_prompt(authority)
+    transported, _ = json.JSONDecoder().raw_decode(prompt.split('Turn authority JSON:\n', 1)[1])
+    assert transported['selected_todo'] == authority['selected_todo']
+    assert transported['required_reads'] == authority['required_reads']
+    assert transported['work_context'] == context
+    assert transported['contract_capsule'] == authority['contract_capsule']
+    assert transported['action'] == authority['action']
+    for field in ('user', 'execution_policy', 'scheduler'):
+        assert transported[field] == envelope[field]
+    assert 'the outer Turn executor owns those effects' in prompt
+    # A compromised carrier cannot silently remove either the instruction or a
+    # separate obligation while retaining the signed action's original digest.
+    for carrier, changed_field in (('work_context', 'instruction'),
+                                   ('work_context', 'sources'),
+                                   ('contract_capsule', 'execution_obligation'),
+                                   ('execution_policy', 'safe_bypass_allowed')):
+        altered = copy.deepcopy(envelope)
+        del altered[carrier][changed_field]
+        with pytest.raises(ValueError, match='action signature'):
+            extract_turn_authority({'turn_envelope': altered})
+    # A matching v0 digest does not authorize interpreting a future wire version.
+    # Version declarations are outside the historical action digest document.
+    from loopx.control_plane.effect_runtime import EffectRuntimeRejected
+    from loopx.control_plane.turn_driver.codex_cli import _prompt
+    from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
+    for path in (('schema_version',), ('action_signature', 'schema_version'),
+                 ('action_signature', 'coverage')):
+        for unsupported in ('future_required_semantics', None):
+            altered = copy.deepcopy(envelope)
+            carrier = altered if len(path) == 1 else altered[path[0]]
+            carrier[path[-1]] = unsupported
+            for consumer in (extract_turn_authority, _prompt):
+                with pytest.raises(EffectRuntimeRejected, match='unsupported TurnEnvelope'):
+                    consumer({'turn_envelope': altered})
+            plan = build_loopx_turn_plan(altered, host='generic-cli', execution_mode='isolated-headless')
+            assert plan['route']['kind'] == 'contract_error'
+            assert plan['route']['would_invoke_host'] is False
+    receipts = _heartbeat_receipt_count(runtime, TURN_ID)
+    code, ordinary = _run_cli(registry, runtime, *args)
+    assert code == 0, ordinary
+    ordinary_context = ordinary['interaction_contract']['agent_channel']['work_context']
+    assert ordinary_context['instruction'] == source_context['instruction']
+    ordinary_task = next(source['content']['todo'] for source in ordinary_context['sources']
+                         if source['source'] == 'selected_todo')
+    assert ordinary_task.get('text', ordinary['selected_todo']['text']) == source_task['text']
+    assert ordinary_task['note'] == note
+    assert ordinary['interaction_contract']['cli_channel']['settlement_plan']['identity'] == full[
+        'interaction_contract']['cli_channel']['settlement_plan']['identity']
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == receipts
+    assert _spend_run_count(runtime) == 0
+    assert saved.read_bytes() == original_bytes
+
+
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
+@pytest.mark.parametrize('declaration', ['envelope', 'signature', 'coverage'])
+def test_real_turn_refuses_unsupported_consumer_version_before_launch(tmp_path, monkeypatch, provider, declaration):
+    import contextlib
+    import io
+    import sys
+    from loopx.cli import main
+    from loopx.cli_commands import turn
+    from tests.test_loopx_turn_driver import _write_live_fixture
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_live_fixture(tmp_path, provider=provider)
+    original_builder = turn.build_turn_envelope
+    observed = []
+
+    def incompatible_writer(*args, **kwargs):
+        packet = original_builder(*args, **kwargs)
+        if declaration == 'envelope':
+            packet['schema_version'] = 'unsupported_work_contract'
+        else:
+            packet['action_signature']['schema_version' if declaration == 'signature' else 'coverage'] = 'unsupported_required_semantics'
+        observed.append(copy.deepcopy(packet))
+        return packet
+
+    monkeypatch.setattr(turn, 'build_turn_envelope', incompatible_writer)
+    marker = project / 'host-started'
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        code = main(['--registry', str(registry), '--runtime-root', str(runtime), '--format', 'json',
+            'turn', 'run-once', '--host', 'generic-cli', '--goal-id', 'loopx-turn-fixture',
+            '--agent-id', 'codex-fixture', '--project', str(project), '--host-command-json',
+            json.dumps([sys.executable, '-c', f'from pathlib import Path; Path({str(marker)!r}).touch()']),
+            '--no-global-sync', '--execute'])
+    result = json.loads(output.getvalue())
+    assert observed and code != 0, result
+    assert not marker.exists()
+    assert 'unsupported TurnEnvelope' in result['error'], result
+    assert 'compatible consumer' in result['error'], result
+    # Generic CLI error effects are unknown; the marker and ledger independently
+    # prove this pre-dispatch refusal, without interpreting null as false.
+    assert _spend_run_count(runtime) == 0
diff --git a/tests/control_plane_ts/turn_envelope.test.ts b/tests/control_plane_ts/turn_envelope.test.ts
index 99115157b6..f40b64b7b4 100644
--- a/tests/control_plane_ts/turn_envelope.test.ts
+++ b/tests/control_plane_ts/turn_envelope.test.ts
@@ -5,6 +5,7 @@ import {
   ACTION_SIGNATURE_COVERAGE_V0,
   ACTION_SIGNATURE_COVERAGE_V3,
   buildTurnEnvelope,
+  assertTurnEnvelopeConsumerVersion,
   evaluateTurnEnvelope,
   quotaActionSignatureDocument,
   turnEnvelopeActionSignatureDocument,
@@ -86,6 +87,44 @@ function payload(): Record<string, unknown> {
   };
 }
 
+test("execution contracts retain complete clauses, nested data and empty values", () => {
+  const source = payload();
+  const clause = "Preserve original identity / 保留原身份. ".repeat(40) + "Stop on source mismatch.";
+  const contracts = {
+    work_lane_contract: {schema_version: "work_lane_contract_v1", action: clause,
+      outcome_followthrough: {required: true, spend_policy: clause,
+        accepted_resolution_kinds: Array.from({length: 12}, (_, i) => `resolution-${i}`),
+        evidence: {ordered: ["artifact", "validation"], missing: null}}},
+    execution_obligation: {must_attempt_work: true, reason: clause, notify_is_execution_gate: false},
+    automation_liveness: {keep_active: true, pause_allowed: false, pause_policy: clause,
+      spend_policy: "Only after validated writeback", reason: "", observation: null},
+    agent_scope_frontier: {recommended_action: clause, spend_policy: clause, blocks_delivery: false},
+    autonomous_replan_scope: {required: false, applies: false, owner_agent_ids: [], context: null},
+  };
+  Object.assign(source, structuredClone(contracts));
+  const envelope = buildTurnEnvelope({payload: source, scheduler_execution_args: ""});
+  const capsule = envelope.contract_capsule as JsonObject;
+  for (const [field, expected] of Object.entries(contracts)) {
+    assert.deepEqual(capsule[field], expected);
+    const changed = structuredClone(envelope);
+    delete (changed.contract_capsule as JsonObject)[field];
+    assert.notDeepEqual(turnEnvelopeActionSignatureDocument(changed),
+      turnEnvelopeActionSignatureDocument(envelope));
+  }
+  // Retained future metadata is data, not a newly interpreted execution rule.
+  assert.equal(envelope.should_run, true);
+  assert.equal((envelope.compaction as JsonObject).within_budget, false);
+  // Existing historical envelopes are checked from their captured content;
+  // signature verification never rebuilds an old capsule through the producer.
+  const historical = structuredClone(envelope);
+  historical.contract_capsule = {schema_version: "loopx_contract_capsule_v0",
+    source: "full_quota_decision", work_lane_contract: {must_attempt_work: true}};
+  const signatureBefore = turnEnvelopeActionSignatureDocument(historical);
+  (source.work_lane_contract as JsonObject).action = "changed later";
+  assert.deepEqual(turnEnvelopeActionSignatureDocument(historical), signatureBefore);
+  assert.deepEqual(capsule.work_lane_contract, contracts.work_lane_contract);
+});
+
 test("a Todo-less planning identity must match the original signed quota binding", () => {
   const identity = {
     schema_version: "quota_settlement_identity_v1", goal_id: "goal-turn-envelope", agent_id: "agent-ts",
@@ -1164,3 +1203,25 @@ test("native Turn capture binds the original observation without inventing admis
       EffectRuntimeRequestError);
   }
 });
+
+
+test("consumer version gate rejects explicit future/null declarations without rewriting v0", () => {
+  const envelope = buildTurnEnvelope({payload: payload(), scheduler_execution_args: ""});
+  const before = structuredClone(envelope);
+  assertTurnEnvelopeConsumerVersion(envelope);
+  for (const field of ["envelope", "signature", "coverage"]) {
+    for (const unsupported of [null, "future-required-semantics", "", 1]) {
+      const changed = structuredClone(envelope);
+      const carrier = field === "envelope" ? changed : changed.action_signature as JsonObject;
+      carrier[field === "coverage" ? "coverage" : "schema_version"] = unsupported;
+      assert.throws(() => assertTurnEnvelopeConsumerVersion(changed), /unsupported TurnEnvelope/);
+    }
+  }
+  // Preserve the existing v0 planning transport; this gate neither upgrades
+  // historical objects nor substitutes for the caller's digest verification.
+  const legacy = structuredClone(envelope);
+  delete (legacy.action_signature as JsonObject).schema_version;
+  delete (legacy.action_signature as JsonObject).coverage;
+  assertTurnEnvelopeConsumerVersion(legacy);
+  assert.deepEqual(envelope, before);
+});
diff --git a/tests/test_loopx_turn_codex_cli.py b/tests/test_loopx_turn_codex_cli.py
index 4d65be5a45..4846eb346c 100644
--- a/tests/test_loopx_turn_codex_cli.py
+++ b/tests/test_loopx_turn_codex_cli.py
@@ -1690,3 +1690,24 @@ def stop_source_before_dispatch(request):
     assert not _session_files(runtime)
     index = runtime / "goals/loopx-turn-fixture/runs/index.jsonl"
     assert not index.exists() or "quota_slot_spent" not in index.read_text()
+
+
+@pytest.mark.parametrize("declaration", ["envelope", "signature", "coverage"])
+def test_codex_host_rejects_unknown_contract_version_before_process(tmp_path, monkeypatch, declaration):
+    from loopx.control_plane.effect_runtime import EffectRuntimeRejected
+
+    executable, log = _fake_codex(tmp_path)
+    monkeypatch.setenv("FAKE_CODEX_LOG", str(log))
+    request = _request()
+    envelope = request["turn_envelope"]
+    if declaration == "envelope":
+        envelope["schema_version"] = "unsupported_work_contract"
+    else:
+        envelope["action_signature"] = {
+            "schema_version" if declaration == "signature" else "coverage": "unsupported_required_semantics",
+        }
+    with pytest.raises(EffectRuntimeRejected, match="unsupported TurnEnvelope"):
+        run_codex_cli_host(request, runtime_root=tmp_path / "runtime", project=tmp_path,
+                           codex_bin=str(executable), timeout_seconds=5)
+    assert not log.exists()
+    assert not _session_files(tmp_path / "runtime")
diff --git a/tests/test_turn_envelope.py b/tests/test_turn_envelope.py
index 99c1f374dc..7934a25aef 100644
--- a/tests/test_turn_envelope.py
+++ b/tests/test_turn_envelope.py
@@ -902,6 +902,58 @@ def test_modern_decision_through_real_readers() -> None:
     assert "protocol_action_packet" not in capsule
 
 
+def test_ordinary_host_preserves_signed_surrounding_turn_policies() -> None:
+    from loopx.control_plane.turn_driver.host_candidate import render_prompt
+
+    source = _full_decision()
+    source.update(should_run=False, normal_delivery_allowed=False, decision='skip',
+                  effective_action='quota_skip', action_required=True, open_count=1)
+    source['execution_obligation']['must_attempt_work'] = False
+    source['work_lane_contract']['must_attempt_work'] = False
+    source['interaction_contract']['cli_channel'].update(
+        next_cli_actions=[], spend_allowed_now=False, spend_after_validation=False,
+    )
+    source['interaction_contract']['user_channel'].update(action_required=True, notify='NOTIFY')
+    source['interaction_contract']['agent_channel'].update(
+        must_attempt=False, delivery_allowed=False, quiet_noop_allowed=False,
+    )
+    source['interaction_contract']['response_plan'] = {
+        'schema_version': 'interaction_response_plan_v0',
+        'kind': 'surface_user_gate', 'decision': 'ask_user', 'action_sequence': ['notify', 'wait'],
+        'silent_wait_allowed': False,
+    }
+    source['task_orchestration_contract'] = {
+        'schema_version': 'task_orchestration_contract_v1',
+        'mode': 'task_scoped_peer', 'activation_allowed': False,
+        'writeback_owner': 'task_coordinator',
+    }
+    envelope = build_turn_envelope(source)
+    assert not envelope.get('replan_action_packet')
+    authority = extract_turn_authority({'turn_envelope': envelope})
+    prompt = render_prompt(authority)
+    delivered, _ = json.JSONDecoder().raw_decode(prompt.split('Turn authority JSON:\n', 1)[1])
+    assert delivered['user']['notify'] == 'NOTIFY'
+    assert delivered['action']['must_attempt'] is False
+    assert delivered['action']['delivery_allowed'] is False
+    assert delivered['action']['quiet_noop_allowed'] is False
+    assert delivered['should_run'] is False
+    assert delivered['effective_action'] == 'quota_skip'
+    assert delivered['execution_policy']['normal_delivery_allowed'] is False
+    assert delivered['execution_policy']['safe_bypass_allowed'] is False
+    assert delivered['task_orchestration_contract']['activation_allowed'] is False
+    assert delivered['response_plan']['silent_wait_allowed'] is False
+    for field in ('user', 'execution_policy', 'scheduler', 'response_plan', 'task_orchestration_contract'):
+        assert delivered[field] == envelope[field]
+        altered = deepcopy(envelope)
+        del altered[field]
+        with pytest.raises(ValueError, match='action signature'):
+            extract_turn_authority({'turn_envelope': altered})
+    # Carrying policy is not delegating its effects to the model. The native
+    # executor still applies settlement and scheduler operations after return.
+    assert 'do not execute state writeback, quota or scheduler commands' in prompt
+    assert 'writeback' not in authority
+
+
 @pytest.mark.parametrize("user_required", [False, True], ids=["quiet-wait", "user-gate"])
 def test_missing_protocol_packet_cannot_reopen_a_non_delivery_decision(user_required: bool) -> None:
     source = _full_decision()