fix(qualification): preflight source-bound model behavior pairs

docs/reference/protocols/model-behavior-qualification-v0.md

@@ -14,16 +14,31 @@ changing the default quota path.
 
 One qualification case runs the same actor against two public-safe inputs:
 
-1. `full_packet`: the current full `quota should-run` decision;
-2. `candidate_packet`: the candidate `loopx_turn_envelope_v0` projection.
+1. `full_packet`: the full `quota should-run` decision or an explicitly supplied
+   complete `loopx_turn_envelope_v1` source;
+2. `candidate_packet`: a `loopx_turn_envelope_v0` projection or a complete v1
+   source from that same decision.
 
 Both arms share `qualification_id` and `actor_ref`. Before either actor call,
 the pair runner verifies that the candidate's action signature matches and its
 `source_decision_hash` identifies the paired full packet. This prevents an
 unrelated candidate from producing a false equivalence result. The runner also
 recomputes both semantic signature documents instead of trusting the
 candidate's stored `matches` flag; a field ablation therefore fails before any
-provider call. The comparator then checks these hard behavior dimensions:
+provider call. A complete source instead passes the production version,
+content-integrity and source-integrity reader. A raw full packet is composed
+through the existing owner for comparison; a captured v1 full packet retains
+its recorded host and settlement binding without recomposition. The pair must
+match the entire composed source, including unknown fields and exact numeric
+values, as well as its action semantics. Invalid new versions never fall back
+to the historical reader.
+
+Schema and confidentiality checks for **both** arms finish before the first
+actor call, regardless of arm order. A private or malformed second input must
+not consume a first provider call. The validated requests are then used by the
+existing actor/result/receipt path. These checks qualify packet fidelity and
+the no-write boundary; they do not prove model understanding or task outcomes.
+The comparator checks these hard behavior dimensions:
 
 - decision: execute, wait, ask the user, or stop;
 - selected todo;
@@ -106,7 +121,11 @@ matches.
 
 The store is never populated automatically. It contains no model response,
 conversation, credential metadata, or candidate packet; a current candidate is
-rebuilt in memory when the case is loaded into a corpus. Store receipts expose
+rebuilt in memory when the case is loaded into a corpus. Raw quota cases retain
+the existing v0 candidate by default. An explicitly retained complete capture
+stays complete in both arms, preserving its original binding. This support
+does not change the actual-default portfolio, frozen state matrix, scenario
+scoring, live-run permissions or promotion requirements. Store receipts expose
 only case id, digest, created/idempotent status, and count, never the packet or
 local path.
 
@@ -161,6 +180,14 @@ model prompt. The actor disables provider deep thinking for this deterministic
 extraction task and reserves 4096 output tokens so the bounded semantic
 contract is not constrained by the former 1200-token response budget.
 
+The adapter selects field-reading instructions from the validated packet
+schema, independently of the experimental arm label. A complete v1 source uses
+the full-source `interaction_contract`, `agent_identity`, `selected_todo`,
+`goal_boundary` and `scheduler_hint` paths even in `candidate_packet`; a v0
+projection retains its historical paths. The original packet reaches the
+provider unchanged, without an adapter-generated action projection or answer
+key.
+
 The model must derive `selected_todo_id` from the canonical selected-todo field:
 top-level `selected_todo.todo_id` in a full packet or
 `action.selected_todo.todo_id` in a TurnEnvelope, including `null` when absent.

loopx/control_plane/quota/turn_envelope.py

@@ -15,6 +15,7 @@
 )
 
 TURN_ENVELOPE_SCHEMA_VERSION = "loopx_turn_envelope_v0"
+COMPLETE_TURN_DECISION_VERSION = "loopx_turn_envelope_v1"
 TURN_ENVELOPE_BUDGET_BYTES = 8_192
 CONTRACT_CAPSULE_SCHEMA_VERSION = "loopx_contract_capsule_v0"
 ACTION_SIGNATURE_SCHEMA_VERSION = "loopx_action_signature_v0"
@@ -79,7 +80,7 @@ def _turn_envelope_result(
 
     request = {"operation": operation, **params}
     payload = params.get("payload")
-    complete = isinstance(payload, Mapping) and payload.get("schema_version") == "loopx_turn_envelope_v1"
+    complete = isinstance(payload, Mapping) and payload.get("schema_version") == COMPLETE_TURN_DECISION_VERSION
     staged_replan = (
         operation in {"build", "quota_signature", "replan_cli_view", "bind_capture",
                       "capability_automation_source"}
@@ -321,7 +322,7 @@ def _build_turn_projection(
     # A captured complete contract already records the original host binding.
     # Only transport that value when no caller override was supplied; TS checks
     # its integrity and rejects an explicit conflicting binding.
-    if scheduler_execution_context is None and payload.get("schema_version") == "loopx_turn_envelope_v1":
+    if scheduler_execution_context is None and payload.get("schema_version") == COMPLETE_TURN_DECISION_VERSION:
         binding = payload.get("contract_binding")
         scheduler_execution_args = binding.get("scheduler_execution_args") if isinstance(binding, Mapping) else None
     return _turn_envelope_result(

loopx/control_plane/testing/doubao_model_behavior_actor.py

@@ -8,6 +8,7 @@
 from urllib.error import HTTPError, URLError
 from urllib.request import HTTPRedirectHandler, Request, build_opener
 
+from ..quota.turn_envelope import COMPLETE_TURN_DECISION_VERSION
 from .model_behavior_qualification import (
     MODEL_BEHAVIOR_ACTOR_RESULT_SCHEMA_VERSION,
     MODEL_BEHAVIOR_SEMANTIC_CONTRACT_FIELDS,
@@ -237,6 +238,13 @@ def _semantic_contract_field_rules(*, arm: str) -> dict[str, str]:
   null when absent. same_agent_continuation is true only when agent_id and
   selected_todo_claimed_by are equal and continuation_policy is exactly
   same_agent_non_delivery."""
+    if arm == "full_packet":
+        rules["peer_route"] = """always include exactly agent_id,
+  selected_todo_claimed_by, continuation_policy, and same_agent_continuation.
+  Read agent_id from packet.agent_identity.agent_id and the other values from
+  packet.selected_todo, using null when absent. same_agent_continuation is true
+  only when agent_id and selected_todo_claimed_by are equal and
+  continuation_policy is exactly same_agent_non_delivery."""
     return rules
 
 
@@ -291,7 +299,12 @@ def _decision_instruction(
     arm: str,
     semantic_contract_required: bool,
     semantic_contract_fields: Sequence[str] | None = None,
+    packet_schema_version: str | None = None,
 ) -> str:
+    if arm not in ("full_packet", "candidate_packet"):
+        raise ValueError("arm must be full_packet or candidate_packet")
+    # Arm labels describe the experiment, not the packet's field layout.
+    source_arm = "full_packet" if packet_schema_version == COMPLETE_TURN_DECISION_VERSION else arm
     fields = (
         tuple(semantic_contract_fields)
         if semantic_contract_fields is not None
@@ -354,18 +367,18 @@ def _decision_instruction(
 intended_action_kinds does not request an external write. Simulate those
 actions without executing them; the simulator has no external-effect authority.
 """
-    if arm == "candidate_packet":
+    if source_arm == "candidate_packet":
         instruction += """For intended actions, use packet.action and
 packet.writeback. When spend_after_validation=true, end with writeback then
 spend."""
-    elif arm == "full_packet":
+    elif source_arm == "full_packet":
         instruction += """For intended actions, use
 interaction_contract.agent_channel and interaction_contract.cli_channel. When
 spend_after_validation=true, end with writeback then spend."""
     else:
         raise ValueError("arm must be full_packet or candidate_packet")
     if semantic_contract_required:
-        instruction += _semantic_contract_instruction(arm=arm, fields=fields)
+        instruction += _semantic_contract_instruction(arm=source_arm, fields=fields)
     return instruction
 
 
@@ -593,6 +606,7 @@ def __call__(self, request: Mapping[str, Any]) -> Mapping[str, Any]:
                     canonical_request["semantic_contract_required"]
                 ),
                 semantic_contract_fields=semantic_contract_fields,
+                packet_schema_version=str(canonical_request["packet_schema_version"]),
             ),
             provider_input=_provider_input(canonical_request),
         )

loopx/control_plane/testing/model_behavior_corpus.py

@@ -5,7 +5,7 @@
 from collections.abc import Iterable, Mapping
 from typing import Any
 
-from ..quota.turn_envelope import build_turn_envelope
+from ..quota.turn_envelope import COMPLETE_TURN_DECISION_VERSION, build_turn_envelope
 from .model_behavior_qualification import (
     MODEL_BEHAVIOR_HARD_INVARIANT_FIELDS,
     MODEL_BEHAVIOR_SEMANTIC_CONTRACT_FIELDS,
@@ -76,9 +76,12 @@ def _case(
         qualification_id=f"validate-{case_id}",
         arm="full_packet",
     )
+    # Keep the existing raw-quota/v0 control by default. An explicitly retained
+    # complete capture is already its own source; do not reconstruct an old
+    # projection or silently replace its recorded host/settlement binding.
     normalized_candidate = copy.deepcopy(
-        dict(candidate_packet)
-        if candidate_packet is not None
+        dict(candidate_packet) if candidate_packet is not None
+        else normalized_full if normalized_full.get("schema_version") == COMPLETE_TURN_DECISION_VERSION
         else build_turn_envelope(normalized_full)
     )
     build_model_behavior_actor_request(

loopx/control_plane/testing/model_behavior_packet_safety.py

@@ -9,6 +9,8 @@
 from collections.abc import Mapping
 from typing import Any
 
+from ..quota.turn_envelope import COMPLETE_TURN_DECISION_VERSION
+
 _FACTS_FLAG = "--scheduler-host-facts-chunk"
 # Match the existing native follow-up decoder, without invoking its effects.
 _MAX_ENCODED_CHARS = 4_096
@@ -147,7 +149,8 @@ def scheduler_transport_validation_view(packet: Mapping[str, Any], *, arm: str)
     caller must recursively scan it, and must send the original packet onward.
     """
     view: dict[str, Any] = _validation_copy(packet)
-    if arm == "full_packet":
+    source_packet = arm == "full_packet" or packet.get("schema_version") == COMPLETE_TURN_DECISION_VERSION
+    if source_packet:
         scheduler = view.get("scheduler_hint")
     else:
         scheduler = view.get("scheduler")
@@ -156,7 +159,7 @@ def scheduler_transport_validation_view(packet: Mapping[str, Any], *, arm: str)
     app_automation = scheduler.get("app_automation") or scheduler.get("codex_app")
     if not isinstance(app_automation, Mapping):
         return view
-    if arm == "full_packet":
+    if source_packet:
         for kind, command in (("ack", "scheduler-ack-current"), ("failure", "scheduler-fail-current")):
             hint = app_automation.get(kind + "_hint")
             args = hint.get("cli_args") if isinstance(hint, Mapping) else None

loopx/control_plane/testing/model_behavior_qualification.py

@@ -6,9 +6,13 @@
 from hashlib import sha256
 from typing import Any, Protocol
 
+from ..effect_runtime import EffectRuntimeRejected
 from ..quota.turn_envelope import (
+    COMPLETE_TURN_DECISION_VERSION,
     PLANNING_HORIZON_DETAIL_REFS_REF,
     TURN_ENVELOPE_SCHEMA_VERSION,
+    assert_complete_turn_decision,
+    compose_turn_decision,
     quota_action_signature_document,
     turn_envelope_action_signature_document,
 )
@@ -248,9 +252,15 @@ def _reject_private_or_secret_material(value: Any, *, path: str = "packet") -> N
 
 
 def _packet_schema(packet: Mapping[str, Any], *, arm: str) -> str:
+    if arm not in ("full_packet", "candidate_packet"):
+        raise ValueError("arm must be full_packet or candidate_packet")
+    if packet.get("schema_version") == COMPLETE_TURN_DECISION_VERSION or "contract_binding" in packet:
+        assert_complete_turn_decision(packet)
+        return COMPLETE_TURN_DECISION_VERSION
     if arm == "full_packet":
         if (
-            packet.get("mode") != "should-run"
+            "schema_version" in packet
+            or packet.get("mode") != "should-run"
             or not isinstance(packet.get("interaction_contract"), Mapping)
             or not packet.get("goal_id")
         ):
@@ -259,7 +269,7 @@ def _packet_schema(packet: Mapping[str, Any], *, arm: str) -> str:
     if arm == "candidate_packet":
         if packet.get("schema_version") != TURN_ENVELOPE_SCHEMA_VERSION:
             raise ValueError(
-                "candidate_packet must use the current LoopX TurnEnvelope schema"
+                "candidate_packet must use a supported LoopX TurnEnvelope schema"
             )
         return str(TURN_ENVELOPE_SCHEMA_VERSION)
     raise ValueError("arm must be full_packet or candidate_packet")
@@ -270,10 +280,10 @@ def _packet_action_signature(
     *,
     arm: str,
 ) -> dict[str, Any]:
-    _packet_schema(packet, arm=arm)
+    schema = _packet_schema(packet, arm=arm)
     return (
         quota_action_signature_document(packet)
-        if arm == "full_packet"
+        if arm == "full_packet" or schema == COMPLETE_TURN_DECISION_VERSION
         else turn_envelope_action_signature_document(packet)
     )
 
@@ -463,7 +473,7 @@ def model_behavior_semantic_contract_from_packet(
             )
         )
     )
-    if arm == "full_packet":
+    if arm == "full_packet" or packet.get("schema_version") == COMPLETE_TURN_DECISION_VERSION:
         source_horizon = packet.get("planning_horizon")
         source_detail_refs = (
             source_horizon.get("detail_refs")
@@ -663,17 +673,23 @@ def run_model_behavior_qualification_arm(
 ) -> dict[str, Any]:
     """Run one in-memory actor arm and return only a compact decision receipt."""
 
-    required_semantic_fields = _semantic_contract_field_names(
-        required=semantic_contract_required,
-        requested=semantic_contract_fields,
-    )
     request = build_model_behavior_actor_request(
         packet,
         qualification_id=qualification_id,
         arm=arm,
         semantic_contract_required=semantic_contract_required,
-        semantic_contract_fields=required_semantic_fields,
+        semantic_contract_fields=semantic_contract_fields,
     )
+    return _run_model_behavior_qualification_request(request, actor=actor)
+
+
+def _run_model_behavior_qualification_request(
+    request: Mapping[str, Any], *, actor: ModelBehaviorActor,
+) -> dict[str, Any]:
+    packet = request["packet"]
+    arm = str(request["arm"])
+    semantic_contract_required = bool(request["semantic_contract_required"])
+    required_semantic_fields = tuple(request["response_contract"]["semantic_contract_fields"])
     result = normalize_model_behavior_actor_result(
         actor(request),
         semantic_contract_required=semantic_contract_required,
@@ -890,40 +906,59 @@ def run_model_behavior_qualification_pair(
         raise ModelBehaviorPairValidationError(
             "arm_order must contain full_packet and candidate_packet once"
         )
-    action_signature = candidate_packet.get("action_signature")
-    if not isinstance(action_signature, Mapping):
-        raise ModelBehaviorPairValidationError(
-            "candidate_packet is missing its TurnEnvelope action signature"
-        )
-    if action_signature.get("matches") is not True:
-        raise ModelBehaviorPairValidationError(
-            "candidate_packet action signature parity must be verified"
-        )
-    if action_signature.get("source_decision_hash") != _digest(dict(full_packet)):
-        raise ModelBehaviorPairValidationError(
-            "candidate_packet does not derive from the paired full packet"
-        )
-    if quota_action_signature_document(
-        full_packet
-    ) != turn_envelope_action_signature_document(candidate_packet):
-        raise ModelBehaviorPairValidationError(
-            "candidate_packet action semantics drift from the full packet"
-        )
     packets = {
         "full_packet": full_packet,
         "candidate_packet": candidate_packet,
     }
+    try:
+        full_schema = _packet_schema(full_packet, arm="full_packet")
+        candidate_schema = _packet_schema(candidate_packet, arm="candidate_packet")
+        if candidate_schema == COMPLETE_TURN_DECISION_VERSION:
+            # A captured complete source retains its original host binding and
+            # settlement composition. Raw fixtures use the Core's default
+            # complete composition; never infer another binding from a candidate.
+            expected = (dict(full_packet) if full_schema == COMPLETE_TURN_DECISION_VERSION
+                        else compose_turn_decision(full_packet))
+            if _digest(dict(candidate_packet)) != _digest(expected):
+                raise ModelBehaviorPairValidationError(
+                    "candidate_packet does not derive from the paired full packet"
+                )
+        else:
+            # Explicit historical controls keep their original signature and
+            # full-source digest. They are not the default current producer.
+            action_signature = candidate_packet.get("action_signature")
+            if not isinstance(action_signature, Mapping):
+                raise ModelBehaviorPairValidationError(
+                    "candidate_packet is missing its TurnEnvelope action signature"
+                )
+            if action_signature.get("matches") is not True:
+                raise ModelBehaviorPairValidationError(
+                    "candidate_packet action signature parity must be verified"
+                )
+            if action_signature.get("source_decision_hash") != _digest(dict(full_packet)):
+                raise ModelBehaviorPairValidationError(
+                    "candidate_packet does not derive from the paired full packet"
+                )
+        if _packet_action_signature(full_packet, arm="full_packet") != _packet_action_signature(
+            candidate_packet, arm="candidate_packet"
+        ):
+            raise ModelBehaviorPairValidationError(
+                "candidate_packet action semantics drift from the full packet"
+            )
+        # Check both confidentiality boundaries before either provider call.
+        requests = {
+            arm: build_model_behavior_actor_request(
+                packet, qualification_id=qualification_id, arm=arm,
+                semantic_contract_required=semantic_contract_required,
+                semantic_contract_fields=semantic_contract_fields,
+            ) for arm, packet in packets.items()
+        }
+    except (ValueError, EffectRuntimeRejected) as exc:
+        raise ModelBehaviorPairValidationError(str(exc)) from None
     receipts: dict[str, dict[str, Any]] = {}
     for arm in arm_order:
         try:
-            receipts[arm] = run_model_behavior_qualification_arm(
-                packets[arm],
-                qualification_id=qualification_id,
-                arm=arm,
-                actor=actor,
-                semantic_contract_required=semantic_contract_required,
-                semantic_contract_fields=semantic_contract_fields,
-            )
+            receipts[arm] = _run_model_behavior_qualification_request(requests[arm], actor=actor)
         except Exception as exc:
             raise ModelBehaviorArmExecutionError(
                 _arm_failure_receipt(