From 7e2d590aa65fd003dad2d99e9059dfff0e2691f3 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 8 Nov 2026 23:50:58 +0800
Subject: [PATCH 1/3] Bind optional Host context to the complete Turn source

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../contract-and-context.md                   |  10 ++
 .../contract-and-context.zh-CN.md             |   7 +
 loopx/control_plane/capability_hooks.py       |  27 +++-
 loopx/control_plane/quota/turn_envelope.py    |   2 +
 loopx/control_plane/quota/turn_envelope.ts    |  73 +++++----
 .../turn_driver/capability_context.ts         |  24 ++-
 loopx/control_plane/turn_driver/executor.py   |  49 +++---
 .../test_complete_host_request.py             | 141 ++++++++++++++++++
 .../test_complete_session_binding.py          |  40 ++++-
 9 files changed, 320 insertions(+), 53 deletions(-)

diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
index efe6de3e5a..a71e12ea01 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
@@ -630,6 +630,16 @@ claim label alone is insufficient, and another execution's lease remains
 unavailable. Source checks precede the independent task argv; rejection and
 original-result recovery retain the original session and once-only effects.
 
+Complete Turn capability advice and its optional Host result fields bind to the
+checked source content hash and Turn key. Changing a selected Todo's required
+capabilities invalidates that advice even when the legacy action projection
+matches. Publishing the same decision under a capture address preserves the
+binding. On retry, Session resolution and Host input both read the journaled
+plan. A journal with an already validated result can still read its original
+bounded result schema under the historical binding. A new Host invocation
+requires the complete binding to deliver that advice. Ordinary quota and
+historical v0 retain their existing decoders.
+
 Modern Codex and structured Host instructions share the existing audit read
 rule: read a declared audit before judging Goal acceptance or proposing
 closeout, including when the Goal stays open. Notification policy does not grant
diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
index 237fd35a57..db98586727 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
@@ -462,6 +462,13 @@ capture 只使用完整来源。
 复用该 proof。claim 标签本身不足，不能借用另一次执行的 lease。来源检查先于独立
 Task argv；拒绝和原结果恢复保留原会话与一次性效果。
 
+完整 Turn 的 capability 建议及其可选 Host 结果字段绑定到经校验的来源内容哈希和
+Turn key。所选 Todo 的 required capabilities 一旦变化，即使旧 action 投影相同，
+建议也会失效。为同一决策发布 capture 地址时保留绑定。重试时，Session 解析和
+Host 输入都读取日志里的原计划。日志里已校验的结果仍能按
+历史绑定读取原有受限结果 schema。新调用 Host 时，建议须通过完整绑定校验。
+普通 quota 与历史 v0 继续使用各自已有的 decoder。
+
 现代 Codex 与结构化 Host 指令共用既有 audit 读取规则：在判断 Goal acceptance 或提出
 closeout 前读取已声明 audit，即使 Goal 保持开放。通知策略不授予或取消执行权；所供
 delivery、attempt、quiet-noop、execution 与 orchestration 约束仍须遵守。状态、quota
diff --git a/loopx/control_plane/capability_hooks.py b/loopx/control_plane/capability_hooks.py
index 5ec9539c5e..500153872e 100644
--- a/loopx/control_plane/capability_hooks.py
+++ b/loopx/control_plane/capability_hooks.py
@@ -224,6 +224,25 @@ def turn_context_source(payload: Mapping[str, Any]) -> Mapping[str, Any]:
     return {"goal_id": payload.get("goal_id"), "agent_id": agent_id}
 
 
+def _turn_context_binding_source(
+    payload: Mapping[str, Any], *, from_journal: bool = False,
+) -> Mapping[str, Any]:
+    from .quota.turn_envelope import (
+        COMPLETE_TURN_PLAN_SCHEMA_VERSION,
+        complete_turn_host_source,
+        turn_plan_host_source,
+    )
+
+    if payload.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION:
+        source = complete_turn_host_source(
+            turn_plan_host_source(payload), operation="host_context_binding",
+            recorded_capability_result=from_journal,
+        )
+        assert source is not None
+        return source
+    return turn_context_source(payload)
+
+
 def dispatch_turn_context_hooks(
     payload: dict[str, Any],
     registrations: Sequence[TurnContextHookRegistration],
@@ -249,7 +268,7 @@ def dispatch_turn_context_hooks(
             if any(key in payload for key in registration.readback_slots):
                 raise ValueError("context readback slot already contributed")
             result = registration.producer()
-            source = turn_context_source(payload)
+            source = _turn_context_binding_source(payload)
             readback_only = (
                 result.get("context_fields") == {}
                 and result.get("instructions") == []
@@ -281,7 +300,9 @@ def dispatch_turn_context_hooks(
         payload["turn_context_hook_failures"] = failures
 
 
-def project_turn_context(payload: Mapping[str, Any]) -> dict[str, Any]:
+def project_turn_context(
+    payload: Mapping[str, Any], *, from_journal: bool = False,
+) -> dict[str, Any]:
     context: dict[str, Any] = {}
     instructions: list[str] = []
     result_fields: dict[str, Any] = {}
@@ -295,7 +316,7 @@ def project_turn_context(payload: Mapping[str, Any]) -> dict[str, Any]:
                 "capability_hook.turn_context.project",
                 {
                     "contribution": item,
-                    "source": turn_context_source(payload),
+                    "source": _turn_context_binding_source(payload, from_journal=from_journal),
                     "readback": {key: payload.get(key) for key in slots},
                 },
             )
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index 8e1318a32c..7706090d29 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -178,6 +178,7 @@ def assert_turn_host_request_version(request: Mapping[str, Any]) -> None:
 def complete_turn_host_source(
     request: Mapping[str, Any], *, operation: str = "host_context",
     include_session_lineage: bool = False,
+    recorded_capability_result: bool = False,
 ) -> dict[str, Any] | None:
     if "schema_version" in request:
         assert_turn_host_request_version(request)
@@ -188,6 +189,7 @@ def complete_turn_host_source(
     ) if key in request}} if request.get("schema_version") == COMPLETE_TURN_HOST_REQUEST_SCHEMA else {})
     return _turn_envelope_result(operation, payload=request["work_contract"],
                                  envelope=_wire_mapping(request.get("turn_envelope")), **header,
+                                 **({"recorded_capability_result": True} if recorded_capability_result else {}),
                                  **({"include_session_lineage": True} if include_session_lineage else {}))
 
 
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index 868ea997cc..b44277a05d 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1736,39 +1736,44 @@ export function requireCompleteTurnDecision(value: unknown, numberTokens?: unkno
   return decision;
 }
 
+function requireCompleteHostHeader(request: JsonObject, decision: JsonObject): JsonObject {
+  // The versioned model wire has one complete source. The outer executor has
+  // already checked its private effect adapter; a standalone reader validates
+  // the source and original capture header without recreating that adapter.
+  const header = requireJsonObject(request.complete_host_request, "complete Host request");
+  if (header.schema_version !== "loopx_turn_host_request_v1" ||
+      Object.keys(object(request.envelope)).length !== 0 ||
+      typeof header.turn_key !== "string" || !ENVELOPED_SHA256_PATTERN.test(header.turn_key)) {
+    throw new EffectRuntimeRequestError("unsupported or invalid complete Host request");
+  }
+  if (Object.hasOwn(header, "source_decision_hash") && header.source_decision_hash !== decisionSourceHash(decision)) {
+    throw new EffectRuntimeRequestError("complete Host request differs from captured source");
+  }
+  const detail = Object.hasOwn(header, "detail_ref") ? requireJsonObject(header.detail_ref, "complete Host detail reference") : {};
+  const hasCapture = Object.hasOwn(detail, "captured_decision");
+  const captured = hasCapture ? requireJsonObject(detail.captured_decision, "complete Host captured decision") : {};
+  // A direct quota capture predates the Host journal and has no Turn key.
+  // Preserve its source observation rather than inventing a journal binding.
+  // A provided journal key keeps the existing exact-key rule.
+  const captureTurnMatches = Object.hasOwn(captured, "turn_key")
+    ? captured.turn_key === header.turn_key
+    : captured.turn_instance_id === (object(decision.heartbeat_receipt).turn_instance_id ?? null);
+  if (hasCapture && (typeof captured.path !== "string" || !captured.path.trim() || captured.path.includes("\0") ||
+      !captureTurnMatches ||
+      captured.goal_id !== decision.goal_id || captured.agent_id !== object(decision.agent_identity).agent_id ||
+      captured.source_hash_ref !== "$.source_decision_hash" || !Object.hasOwn(header, "source_decision_hash"))) {
+    throw new EffectRuntimeRequestError("complete Host request capture binding differs");
+  }
+  return header;
+}
+
 /** One checked source for model input and transport session identity. The
  * historical effect adapter may remain in the request during migration, but it
  * cannot redirect the complete contract to another Goal, Agent or work item. */
 export function completeHostSource(request: JsonObject): {decision: JsonObject; projection: JsonObject} {
   const decision = requireCompleteTurnDecision(request.payload, request.source_number_tokens);
-  // The versioned model wire has one complete source. The outer executor has
-  // already checked its private effect adapter; a standalone reader validates
-  // the source and original capture header without recreating that adapter.
   if (Object.hasOwn(request, "complete_host_request")) {
-    const header = requireJsonObject(request.complete_host_request, "complete Host request");
-    if (header.schema_version !== "loopx_turn_host_request_v1" ||
-        Object.keys(object(request.envelope)).length !== 0 ||
-        typeof header.turn_key !== "string" || !ENVELOPED_SHA256_PATTERN.test(header.turn_key)) {
-      throw new EffectRuntimeRequestError("unsupported or invalid complete Host request");
-    }
-    if (Object.hasOwn(header, "source_decision_hash") && header.source_decision_hash !== decisionSourceHash(decision)) {
-      throw new EffectRuntimeRequestError("complete Host request differs from captured source");
-    }
-    const detail = Object.hasOwn(header, "detail_ref") ? requireJsonObject(header.detail_ref, "complete Host detail reference") : {};
-    const hasCapture = Object.hasOwn(detail, "captured_decision");
-    const captured = hasCapture ? requireJsonObject(detail.captured_decision, "complete Host captured decision") : {};
-    // A direct quota capture predates the Host journal and has no Turn key.
-    // Preserve its source observation rather than inventing a journal binding.
-    // A provided journal key keeps the existing exact-key rule.
-    const captureTurnMatches = Object.hasOwn(captured, "turn_key")
-      ? captured.turn_key === header.turn_key
-      : captured.turn_instance_id === (object(decision.heartbeat_receipt).turn_instance_id ?? null);
-    if (hasCapture && (typeof captured.path !== "string" || !captured.path.trim() || captured.path.includes("\0") ||
-        !captureTurnMatches ||
-        captured.goal_id !== decision.goal_id || captured.agent_id !== object(decision.agent_identity).agent_id ||
-        captured.source_hash_ref !== "$.source_decision_hash" || !Object.hasOwn(header, "source_decision_hash"))) {
-      throw new EffectRuntimeRequestError("complete Host request capture binding differs");
-    }
+    requireCompleteHostHeader(request, decision);
     return {decision, projection: turnActionProjection(decision)};
   }
   const envelope = requireJsonObject(request.envelope, "complete Turn executor envelope");
@@ -2349,6 +2354,20 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     const {decision, projection} = completeHostSource(request);
     return {goal_id: decision.goal_id ?? null, ...projection};
   }
+  if (operation === "host_context_binding") {
+    const decision = requireCompleteTurnDecision(request.payload, request.source_number_tokens);
+    const header = requireCompleteHostHeader(request, decision);
+    if (!Object.hasOwn(header, "source_decision_hash")) {
+      throw new EffectRuntimeRequestError("complete Turn plan requires a source binding");
+    }
+    // Reader addresses are transport metadata. A recorded typed result may
+    // still need the previous basis to recover its optional result schema.
+    return {schema_version: header.schema_version, turn_key: header.turn_key,
+      source_decision_hash: decisionSourceHash(decision),
+      ...(request.recorded_capability_result === true ? {
+        recorded_projection: {goal_id: decision.goal_id ?? null, ...turnActionProjection(decision)},
+      } : {})};
+  }
   if (operation === "replan_view") return replanView(request);
   if (operation === "context_select") return capturedDecisionSelection(request);
   if (operation === "replan_cli_view") {
diff --git a/loopx/control_plane/turn_driver/capability_context.ts b/loopx/control_plane/turn_driver/capability_context.ts
index b4826ce0b0..166475dc8b 100644
--- a/loopx/control_plane/turn_driver/capability_context.ts
+++ b/loopx/control_plane/turn_driver/capability_context.ts
@@ -1,4 +1,5 @@
 import type {JsonObject} from "../effect_program.ts";
+import {ENVELOPED_SHA256_PATTERN} from "../content_digest.ts";
 import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
 import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts";
 import {quotaActionSignatureDocument, turnEnvelopeActionSignatureDocument} from "../quota/turn_envelope.ts";
@@ -22,6 +23,16 @@ const FIELD = /^[a-z][a-z0-9_]{2,95}$/;
 
 function contextSource(value: unknown, bindReaderLocation = false): JsonObject {
   const source = requireJsonObject(value, "context source");
+  if (source.schema_version === "loopx_turn_host_request_v1") {
+    // These identities come from the complete-source validator. The basis
+    // does not include the reader location or regenerate a legacy signature.
+    const turnKey = requireNonEmptyString(source.turn_key, "context Turn key");
+    const sourceHash = requireNonEmptyString(source.source_decision_hash, "context source hash");
+    if (!ENVELOPED_SHA256_PATTERN.test(turnKey) || !ENVELOPED_SHA256_PATTERN.test(sourceHash)) {
+      throw new Error("invalid complete context binding");
+    }
+    return {turn_key: turnKey, source_decision_hash: sourceHash};
+  }
   // The replan reader generates commands that carry the capture's own hash.
   // Bind the governing action and settlement rather than these read addresses.
   const {replan_read_path: _readerPath, ...beforeReaderBinding} = source;
@@ -95,11 +106,20 @@ export function projectTurnContextContribution(input: JsonObject): JsonObject {
       result_fields: contribution.result_fields, readback}});
   if (contribution.basis_digest !== normalized.basis_digest) {
     const source = requireJsonObject(input.source, "context source");
+    const {basis_digest: _currentBasis, ...value} = normalized;
+    if (source.schema_version === "loopx_turn_host_request_v1" && Object.hasOwn(source, "recorded_projection")) {
+      const legacyBasis = canonicalAuthoritySha256({source: contextSource(source.recorded_projection), ...value});
+      if (contribution.basis_digest === legacyBasis) {
+        // An already validated result can recover its original bounded slots.
+        // The historical basis is insufficient to deliver advice to a Host.
+        return {...normalized, context_fields: {}, instructions: [], basis_digest: legacyBasis};
+      }
+    }
     // An existing unversioned quota observation may already have been bound
     // to its exact reader. Preserve that basis only on an exact legacy match.
-    if ("heartbeat_receipt" in source && !Object.hasOwn(source, "contract_binding") &&
+    if (source.schema_version !== "loopx_turn_host_request_v1" &&
+        "heartbeat_receipt" in source && !Object.hasOwn(source, "contract_binding") &&
         typeof source.replan_read_path === "string") {
-      const {basis_digest: _currentBasis, ...value} = normalized;
       const legacyBasis = canonicalAuthoritySha256({source: contextSource(source, true), ...value});
       if (contribution.basis_digest === legacyBasis) return {...normalized, basis_digest: legacyBasis};
     }
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 21004cb544..9766ec1d58 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -381,7 +381,7 @@ def validate_loopx_turn_host_result(
 ) -> dict[str, Any]:
     result = dict(value)
     errors: list[str] = []
-    optional_fields = capability_result_fields(project_turn_context(plan))
+    optional_fields = capability_result_fields(project_turn_context(plan, from_journal=from_journal))
     if (
         from_journal
         and "turn_context_contributions" not in plan
@@ -1038,8 +1038,9 @@ def _host_result_stage(
         result = dict(host_observation["value"])
         completed_phases = list(TRANSACTION_PHASES[:2])
 
+    recorded_plan = journal.get("plan")
     validation = validate_loopx_turn_host_result(
-        plan,
+        recorded_plan if from_journal and isinstance(recorded_plan, Mapping) else plan,
         result or {},
         completion_lifecycle_configured=completion_lifecycle_configured,
         from_journal=from_journal,
@@ -1473,6 +1474,26 @@ def _prepare_host_invocation(
     return None, {"executable": "built-in", "kind": str(planned_host.get("kind") or "codex-cli")}
 
 
+def _host_attempt_requirements(
+    journal: Mapping[str, Any] | None,
+) -> tuple[bool, bool, Mapping[str, Any] | None]:
+    """Choose the Host attempt, result retry and recorded complete input."""
+    if journal is None:
+        return True, False, None
+    receipt = journal.get("receipt")
+    retry_validation = (
+        journal.get("status") == "failed"
+        and isinstance(receipt, Mapping)
+        and receipt.get("failed_phase") == "validation"
+        and journal.get("validation_stage") != "task_postcondition"
+    )
+    needs_host = retry_validation or "typed_result" not in list(journal.get("completed_phases") or [])
+    recorded = journal.get("plan")
+    if needs_host and isinstance(recorded, Mapping) and recorded.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION:
+        return needs_host, retry_validation, recorded
+    return needs_host, retry_validation, None
+
+
 @single_executor_per_turn_lane(execution_payload)
 def run_loopx_turn_once(
     plan: Mapping[str, Any],
@@ -1583,6 +1604,10 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
 
     with exclusive_file_lock(journal_path):
         journal = _load_journal(journal_path)
+        # Settlement recovery uses its cached host result and must not reserve
+        # another interval. A new host attempt, including failed-result retry,
+        # goes through admission before any host or journal attempt mutation.
+        needs_host, validation_reinvokes_host, recorded_plan = _host_attempt_requirements(journal)
         recovery_decision: dict[str, Any] | None = None
         if journal is not None:
             if goal_admission is not None:
@@ -1592,6 +1617,10 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                     operation="turn_recovery_lifetime",
                 ):
                     pass
+            if recorded_plan is not None:
+                # Session recovery and optional context must use the same
+                # original plan as capture publication.
+                plan, request = recorded_plan, build_loopx_turn_host_request(recorded_plan)
             if settlement_lineage is not None:
                 agent_id = settlement_lineage["agent_id"]
             else:
@@ -1632,22 +1661,6 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 return payload
             request = require_turn_recovery_continuation(assessment)
 
-        # Settlement recovery uses its cached host result and must not reserve
-        # another interval. A new host attempt, including failed-result retry,
-        # goes through admission before any host or journal attempt mutation.
-        receipt = journal.get("receipt") if isinstance(journal, Mapping) else None
-        validation_reinvokes_host = (
-            isinstance(journal, Mapping)
-            and journal.get("status") == "failed"
-            and isinstance(receipt, Mapping)
-            and receipt.get("failed_phase") == "validation"
-            and journal.get("validation_stage") != "task_postcondition"
-        )
-        needs_host = (
-            validation_reinvokes_host
-            or journal is None
-            or "typed_result" not in list(journal.get("completed_phases") or [])
-        )
         if needs_host and goal_admission is not None:
             goal_admission.require_current()
         if needs_host and argv is not None and (reason := host_request_schema_mismatch(
diff --git a/tests/control_plane/test_complete_host_request.py b/tests/control_plane/test_complete_host_request.py
index 463c3e3e2c..b744dab3f2 100644
--- a/tests/control_plane/test_complete_host_request.py
+++ b/tests/control_plane/test_complete_host_request.py
@@ -3,6 +3,7 @@
 
 from copy import deepcopy
 import json
+from pathlib import Path
 
 import pytest
 
@@ -129,6 +130,146 @@ def test_private_plan_lineage_is_not_a_host_wire(complete_host_request):
         complete_host_input(plan)
 
 
+@pytest.mark.parametrize("change", ["capability", "stop", "turn"])
+def test_complete_host_optional_context_requires_original_turn_source(
+    complete_host_request, tmp_path, change,
+):
+    from loopx.control_plane.capability_hooks import (
+        TurnContextHookRegistration, dispatch_turn_context_hooks,
+    )
+    from loopx.control_plane.quota.turn_envelope import compose_turn_decision
+
+    _, original = complete_host_request
+    source = deepcopy({key: value for key, value in original.items()
+                       if key not in {"schema_version", "contract_binding"}})
+    source["observations"] = [1.0, -0.0, 2**60 + 1]
+    contract = compose_turn_decision(source)
+    turn_id = "capability-source"
+    plan = build_loopx_turn_plan(contract, host="generic-cli",
+        execution_mode="isolated-headless", turn_instance_id=turn_id)
+    result = {"readback": {"fixture_readback": "observed"},
+        "context_fields": {"fixture_context": "Advisory context."},
+        "instructions": ["Use the associated observation."],
+        "result_fields": {"fixture_result": {"type": "string", "maxLength": 128}}}
+    dispatch_turn_context_hooks(plan, (TurnContextHookRegistration(
+        "fixture.context", "fixture", ("fixture_readback",), lambda: deepcopy(result),
+    ),))
+    initial = build_loopx_turn_host_request(plan)
+    captured = prepare_turn_decision_capture(plan, journal_path=tmp_path / "capability" / "journal.json",
+        decision=contract, existing_journal=False)
+    bound = build_loopx_turn_host_request(captured)
+    assert bound["detail_ref"]["captured_decision"]["turn_key"] == initial["turn_key"]
+    for request in (initial, bound):
+        assert request["fixture_context"] == result["context_fields"]["fixture_context"]
+        assert request["capability_instructions"] == result["instructions"]
+        assert request["capability_result_fields"] == result["result_fields"]
+
+    if change == "capability":
+        source["selected_todo"]["required_capabilities"] = ["filesystem_read", "network"]
+    elif change == "stop":
+        source["goal_boundary"]["stop_condition"] = "Stop before the fixture external action."
+    else:
+        turn_id = "another-capability-source"
+    candidate = compose_turn_decision(source)
+    changed = build_loopx_turn_plan(candidate, host="generic-cli",
+        execution_mode="isolated-headless", turn_instance_id=turn_id)
+    changed.update({key: deepcopy(plan[key])
+                    for key in ("fixture_readback", "turn_context_contributions")})
+    wire = build_loopx_turn_host_request(changed)
+    assert wire["work_contract"] == candidate
+    assert not {"fixture_context", "capability_instructions", "capability_result_fields"} & wire.keys()
+
+
+def test_recorded_complete_result_recovers_previous_optional_schema(complete_host_request, tmp_path):
+    from loopx.control_plane.capability_hooks import (
+        TurnContextHookRegistration, turn_context_source,
+    )
+    from loopx.control_plane.effect_runtime import effect_runtime_result
+    from loopx.control_plane.turn_driver.executor import (
+        run_loopx_turn_once, validate_loopx_turn_host_result,
+    )
+    from loopx.control_plane.turn_driver.host_candidate import build_result
+    from loopx.control_plane.turn_driver.journal_store import (
+        LOOPX_TURN_JOURNAL_SCHEMA_VERSION, load_loopx_turn_plan_from_journal, turn_journal_path,
+        write_turn_journal_checkpoint,
+    )
+
+    _, contract = complete_host_request
+    plan = build_loopx_turn_plan(contract, host="generic-cli",
+        execution_mode="isolated-headless", turn_instance_id="recorded-capability")
+    goal_id, turn_key = contract["goal_id"], plan["transaction"]["turn_key"]
+    runtime = Path(contract["runtime_root"])
+    journal = turn_journal_path(runtime, goal_id=goal_id, turn_key=turn_key)
+    plan = prepare_turn_decision_capture(plan, journal_path=journal,
+        decision=contract, existing_journal=False)
+    capture = Path(plan["detail_ref"]["captured_decision"]["path"])
+    original = capture.read_bytes()
+    result_fields = {"fixture_result": {"type": "string", "maxLength": 128}}
+    output = {"readback": {"fixture_readback": "observed"},
+        "context_fields": {"fixture_context": "Historical advice."},
+        "instructions": ["Use the original observation."], "result_fields": result_fields}
+    registration = TurnContextHookRegistration(
+        "fixture.context", "fixture", ("fixture_readback",), lambda: output,
+    )
+    # Reproduce the released writer's basis using the historical governing view.
+    contribution = effect_runtime_result("capability_hook.turn_context.validate", {
+        "registration": registration.contract(), "result": output, "source": turn_context_source(plan),
+    })
+    plan.update(contribution.pop("readback"), turn_context_contributions=[contribution])
+    request = build_loopx_turn_host_request(plan)
+    assert "fixture_context" not in request and "capability_result_fields" not in request
+    candidate = build_result({**request, "capability_result_fields": result_fields},
+        {"result_kind": "validated_progress", "classification": "recorded_fixture_progress",
+         "next_action": "Continue the original task.", "delivery_batch_scale": "implementation",
+         "delivery_outcome": "outcome_progress", "work_evidence": ["fixture:artifact"],
+         "validation_evidence": ["fixture:checked"], "fixture_result": "Observed evidence."})
+    assert candidate["fixture_result"] == "Observed evidence."
+    assert not validate_loopx_turn_host_result(plan, candidate)["ok"]
+    for patch in ({"source_decision_hash": "sha256:" + "0" * 64}, {"turn_envelope": {}}):
+        assert not validate_loopx_turn_host_result({**plan, **patch}, candidate, from_journal=True)["ok"]
+    started = {
+        "schema_version": LOOPX_TURN_JOURNAL_SCHEMA_VERSION, "status": "in_progress",
+        "goal_id": goal_id, "turn_key": turn_key, "plan": plan, "host_attempt_count": 1,
+        "completed_phases": [],
+    }
+    write_turn_journal_checkpoint(journal, started)
+    write_turn_journal_checkpoint(journal, {
+        **started, "completed_phases": candidate["completed_phases"], "host_result": candidate,
+    })
+    recovered = load_loopx_turn_plan_from_journal(runtime, goal_id=goal_id, turn_key=turn_key)
+    calls = []
+
+    def forbidden_host(_request):
+        pytest.fail("a recorded typed result must not invoke the Host")
+
+    def validate(_plan, result):
+        assert result["fixture_result"] == candidate["fixture_result"]
+        calls.append("validate")
+        return {"status": "progress", "validator_kind": "fixture",
+                "summary": "Result retained.", "exit_code": 10}
+
+    def writeback(result):
+        assert result["fixture_result"] == candidate["fixture_result"]
+        calls.append("writeback")
+        return {"ok": True, "appended": True}
+
+    def spend():
+        calls.append("spend")
+        return {"ok": True, "appended": True, "slots": 1}
+
+    options = dict(host_runner=forbidden_host, project=tmp_path, runtime_root=runtime,
+        goal_id=goal_id, timeout_seconds=10, execute=True, task_validator=validate,
+        writeback=writeback, spend=spend, scheduler=lambda _: {"completed": True},
+        require_decision_capture=True)
+    outcome = run_loopx_turn_once(recovered, **options)
+    assert outcome["status"] == "committed", outcome.get("reason") or outcome
+    assert calls == ["validate", "writeback", "spend"]
+    assert run_loopx_turn_once(recovered, **options)["replayed"] is True
+    assert calls == ["validate", "writeback", "spend"]
+    assert capture.read_bytes() == original
+    assert json.loads(journal.read_text())["plan"] == plan
+
+
 def test_complete_wire_rejects_downgrade_mixed_carriers_and_capture_drift(complete_host_request, tmp_path):
     original = _captured_request(complete_host_request, tmp_path)
     wire = complete_host_input(original)
diff --git a/tests/control_plane/test_complete_session_binding.py b/tests/control_plane/test_complete_session_binding.py
index bcc5ecb9c6..6add4844cf 100644
--- a/tests/control_plane/test_complete_session_binding.py
+++ b/tests/control_plane/test_complete_session_binding.py
@@ -198,10 +198,13 @@ def scheduler(_spend):
     return plan, common, calls, sources, actions, journal, artifact
 
 
-@pytest.mark.parametrize("historical", [False, True])
+@pytest.mark.parametrize(("historical", "refresh_source"), [(False, False), (False, True), (True, False)])
 def test_failed_session_retry_delivers_original_source_and_settles_once(
-    complete_host_request, tmp_path, historical,
+    complete_host_request, tmp_path, historical, refresh_source,
 ):
+    from loopx.control_plane.capability_hooks import TurnContextHookRegistration, dispatch_turn_context_hooks
+    from loopx.control_plane.quota.turn_envelope import compose_turn_decision
+    from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
     from loopx.control_plane.turn_driver.executor import run_loopx_turn_once, inspect_loopx_turn_journal
     from loopx.control_plane.turn_driver.journal_store import load_loopx_turn_plan_from_journal
 
@@ -209,6 +212,24 @@ def test_failed_session_retry_delivers_original_source_and_settles_once(
         complete_host_request, tmp_path, historical=historical)
     contract = complete_host_request[1]
     before = deepcopy(contract)
+    observed_advice = []
+    if refresh_source:
+        def contribute(candidate, text):
+            dispatch_turn_context_hooks(candidate, (TurnContextHookRegistration(
+                "fixture.context", "fixture", ("fixture_readback",),
+                lambda: {"context_fields": {"fixture_context": text}, "instructions": [],
+                         "result_fields": {}, "readback": {"fixture_readback": text}},
+            ),))
+
+        contribute(plan, "Advice for the original source.")
+        run_host = common["host_runner"]
+
+        def observe_host(request):
+            observed_advice.append(request.get("fixture_context"))
+            return run_host(request)
+
+        common["host_runner"] = observe_host
+
     failed = run_loopx_turn_once(plan, **common)
     assert failed["status"] == "failed" and not artifact.exists()
     assert calls == dict(host=1, writeback=0, spend=0, scheduler=0)
@@ -225,10 +246,23 @@ def test_failed_session_retry_delivers_original_source_and_settles_once(
         retry_failed=True, session_binding_resolver=common["session_binding_resolver"])
     assert inspected["recovery_decision"]["can_continue"] is True
     assert journal.read_bytes() == original_journal
-    recovered = run_loopx_turn_once(saved, retry_failed=True, **common)
+    retry_plan = saved
+    if refresh_source:
+        changed = {key: deepcopy(value) for key, value in contract.items()
+                   if key not in {"schema_version", "contract_binding"}}
+        changed["selected_todo"]["required_capabilities"] = ["filesystem_read", "network"]
+        refreshed = compose_turn_decision(changed)
+        assert refreshed["contract_binding"]["content_hash"] != contract["contract_binding"]["content_hash"]
+        retry_plan = build_loopx_turn_plan(refreshed, host="generic-cli",
+            execution_mode="isolated-headless", turn_instance_id="failed-session-source")
+        assert retry_plan["transaction"]["turn_key"] == saved["transaction"]["turn_key"]
+        contribute(retry_plan, "Advice for the refreshed source.")
+    recovered = run_loopx_turn_once(retry_plan, retry_failed=True, **common)
     assert recovered["status"] == "committed"
     assert actions == ["start_new", "resume"]
     expected = plan["turn_envelope"] if historical else contract
+    if refresh_source:
+        assert observed_advice == ["Advice for the original source."] * 2
     assert sources == [expected, expected]
     assert run_loopx_turn_once(saved, **common)["replayed"] is True
     assert calls == dict(host=2, writeback=1, spend=1, scheduler=1)

From ea032a9d5179a385cc868d19b6bdca186a0fbafa Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 9 Nov 2026 00:42:45 +0800
Subject: [PATCH 2/3] Pin capability context as a canonical digest consumer

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/control_plane_ts/content_digest_single_owner.test.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts
index e9ce2b19c0..af2d2f9675 100644
--- a/tests/control_plane_ts/content_digest_single_owner.test.ts
+++ b/tests/control_plane_ts/content_digest_single_owner.test.ts
@@ -170,6 +170,7 @@ const CANONICAL_CONSUMERS = [
   "control_plane/todos/completion_transaction.ts",
   "control_plane/todos/completion_validation_revision.ts",
   "control_plane/todos/read_transport.ts",
+  "control_plane/turn_driver/capability_context.ts",
   "control_plane/turn_driver/chat_turn_acceptance.ts",
   "control_plane/work_items/operation_agent_handoff.ts",
   "control_plane/work_items/goal_context_source.ts",

From c643d625cd5b20c0e09353f704d2d4af6204aeb6 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 9 Nov 2026 01:42:11 +0800
Subject: [PATCH 3/3] Validate cached Turn results against the recorded plan

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../contract-and-context.md                   |  5 ++-
 .../contract-and-context.zh-CN.md             |  4 +-
 loopx/control_plane/turn_driver/executor.py   |  6 +--
 .../test_complete_host_request.py             | 40 ++++++++++++++++---
 4 files changed, 42 insertions(+), 13 deletions(-)

diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
index a71e12ea01..375aa960ca 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md
@@ -634,8 +634,9 @@ Complete Turn capability advice and its optional Host result fields bind to the
 checked source content hash and Turn key. Changing a selected Todo's required
 capabilities invalidates that advice even when the legacy action projection
 matches. Publishing the same decision under a capture address preserves the
-binding. On retry, Session resolution and Host input both read the journaled
-plan. A journal with an already validated result can still read its original
+binding. On retry, Session resolution, Host input, cached-result task validation
+and settlement use the journaled plan. A journal with an already validated result
+can still read its original
 bounded result schema under the historical binding. A new Host invocation
 requires the complete binding to deliver that advice. Ordinary quota and
 historical v0 retain their existing decoders.
diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
index db98586727..e0fc5b3656 100644
--- a/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
+++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.zh-CN.md
@@ -464,8 +464,8 @@ Task argv；拒绝和原结果恢复保留原会话与一次性效果。
 
 完整 Turn 的 capability 建议及其可选 Host 结果字段绑定到经校验的来源内容哈希和
 Turn key。所选 Todo 的 required capabilities 一旦变化，即使旧 action 投影相同，
-建议也会失效。为同一决策发布 capture 地址时保留绑定。重试时，Session 解析和
-Host 输入都读取日志里的原计划。日志里已校验的结果仍能按
+建议也会失效。为同一决策发布 capture 地址时保留绑定。重试时，Session 解析、
+Host 输入、缓存结果的任务校验和结算都使用日志里的原计划。日志里已校验的结果仍能按
 历史绑定读取原有受限结果 schema。新调用 Host 时，建议须通过完整绑定校验。
 普通 quota 与历史 v0 继续使用各自已有的 decoder。
 
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 9766ec1d58..a45a754f45 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -1489,7 +1489,7 @@ def _host_attempt_requirements(
     )
     needs_host = retry_validation or "typed_result" not in list(journal.get("completed_phases") or [])
     recorded = journal.get("plan")
-    if needs_host and isinstance(recorded, Mapping) and recorded.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION:
+    if isinstance(recorded, Mapping) and recorded.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION:
         return needs_host, retry_validation, recorded
     return needs_host, retry_validation, None
 
@@ -1618,8 +1618,8 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 ):
                     pass
             if recorded_plan is not None:
-                # Session recovery and optional context must use the same
-                # original plan as capture publication.
+                # Recovery uses the recorded source through Host dispatch,
+                # cached-result validation and settlement.
                 plan, request = recorded_plan, build_loopx_turn_host_request(recorded_plan)
             if settlement_lineage is not None:
                 agent_id = settlement_lineage["agent_id"]
diff --git a/tests/control_plane/test_complete_host_request.py b/tests/control_plane/test_complete_host_request.py
index b744dab3f2..b259ecb24c 100644
--- a/tests/control_plane/test_complete_host_request.py
+++ b/tests/control_plane/test_complete_host_request.py
@@ -180,11 +180,13 @@ def test_complete_host_optional_context_requires_original_turn_source(
     assert not {"fixture_context", "capability_instructions", "capability_result_fields"} & wire.keys()
 
 
-def test_recorded_complete_result_recovers_previous_optional_schema(complete_host_request, tmp_path):
+@pytest.mark.parametrize("refresh_source", [False, True], ids=["original", "refreshed-caller"])
+def test_recorded_complete_result_recovers_previous_optional_schema(complete_host_request, tmp_path, refresh_source):
     from loopx.control_plane.capability_hooks import (
         TurnContextHookRegistration, turn_context_source,
     )
     from loopx.control_plane.effect_runtime import effect_runtime_result
+    from loopx.control_plane.quota.turn_envelope import compose_turn_decision
     from loopx.control_plane.turn_driver.executor import (
         run_loopx_turn_once, validate_loopx_turn_host_result,
     )
@@ -195,6 +197,10 @@ def test_recorded_complete_result_recovers_previous_optional_schema(complete_hos
     )
 
     _, contract = complete_host_request
+    source = deepcopy({key: value for key, value in contract.items()
+                       if key not in {"schema_version", "contract_binding"}})
+    source["selected_todo"]["required_capabilities"] = ["filesystem_read", "network"]
+    contract = compose_turn_decision(source)
     plan = build_loopx_turn_plan(contract, host="generic-cli",
         execution_mode="isolated-headless", turn_instance_id="recorded-capability")
     goal_id, turn_key = contract["goal_id"], plan["transaction"]["turn_key"]
@@ -238,13 +244,20 @@ def test_recorded_complete_result_recovers_previous_optional_schema(complete_hos
     })
     recovered = load_loopx_turn_plan_from_journal(runtime, goal_id=goal_id, turn_key=turn_key)
     calls = []
+    checked_sources = []
+    artifact_ready = not refresh_source
 
     def forbidden_host(_request):
         pytest.fail("a recorded typed result must not invoke the Host")
 
-    def validate(_plan, result):
+    def validate(validation_plan, result):
         assert result["fixture_result"] == candidate["fixture_result"]
         calls.append("validate")
+        validation_source = validation_plan["work_contract"]
+        checked_sources.append(validation_source["contract_binding"]["content_hash"])
+        if "network" in validation_source["selected_todo"]["required_capabilities"] and not artifact_ready:
+            return {"status": "failed", "validator_kind": "fixture",
+                    "summary": "The original required artifact is absent."}
         return {"status": "progress", "validator_kind": "fixture",
                 "summary": "Result retained.", "exit_code": 10}
 
@@ -261,11 +274,26 @@ def spend():
         goal_id=goal_id, timeout_seconds=10, execute=True, task_validator=validate,
         writeback=writeback, spend=spend, scheduler=lambda _: {"completed": True},
         require_decision_capture=True)
-    outcome = run_loopx_turn_once(recovered, **options)
+    retry_plan = recovered
+    if refresh_source:
+        first = run_loopx_turn_once(recovered, **options)
+        assert first["status"] == "failed" and calls == ["validate"]
+        source["selected_todo"]["required_capabilities"] = ["filesystem_read"]
+        refreshed = compose_turn_decision(source)
+        assert refreshed["contract_binding"]["content_hash"] != contract["contract_binding"]["content_hash"]
+        retry_plan = build_loopx_turn_plan(refreshed, host="generic-cli",
+            execution_mode="isolated-headless", turn_instance_id="recorded-capability")
+        assert retry_plan["transaction"]["turn_key"] == turn_key
+        failed = run_loopx_turn_once(retry_plan, retry_failed=True, **options)
+        assert failed["status"] == "failed" and calls == ["validate"] * 2
+        artifact_ready = True
+    outcome = run_loopx_turn_once(retry_plan, retry_failed=refresh_source, **options)
     assert outcome["status"] == "committed", outcome.get("reason") or outcome
-    assert calls == ["validate", "writeback", "spend"]
-    assert run_loopx_turn_once(recovered, **options)["replayed"] is True
-    assert calls == ["validate", "writeback", "spend"]
+    expected_calls = ["validate"] * (3 if refresh_source else 1) + ["writeback", "spend"]
+    assert calls == expected_calls
+    assert checked_sources == [contract["contract_binding"]["content_hash"]] * calls.count("validate")
+    assert run_loopx_turn_once(retry_plan, **options)["replayed"] is True
+    assert calls == expected_calls
     assert capture.read_bytes() == original
     assert json.loads(journal.read_text())["plan"] == plan