fix(turn): bind capability context to the admitted source

docs/architecture/rfcs/controller-executor-unified-contract-v0/contract-and-context.md

@@ -630,6 +630,17 @@ 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, 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.
+
 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

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

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},
                 },
             )

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 {}))
 
 

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") {

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};
     }

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 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:
+                # 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"]
             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(