diff --git a/docs/reference/protocols/model-behavior-qualification-v0.md b/docs/reference/protocols/model-behavior-qualification-v0.md index f6ae47ef6f..a09e907c5a 100644 --- a/docs/reference/protocols/model-behavior-qualification-v0.md +++ b/docs/reference/protocols/model-behavior-qualification-v0.md @@ -14,8 +14,10 @@ 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 @@ -23,7 +25,20 @@ the pair runner verifies that the candidate's action signature matches and its 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. diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index e3d75877c8..1a0e228b9f 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/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( diff --git a/loopx/control_plane/testing/doubao_model_behavior_actor.py b/loopx/control_plane/testing/doubao_model_behavior_actor.py index d906370ec9..0466cf5ebe 100644 --- a/loopx/control_plane/testing/doubao_model_behavior_actor.py +++ b/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), ) diff --git a/loopx/control_plane/testing/model_behavior_corpus.py b/loopx/control_plane/testing/model_behavior_corpus.py index 44f2186221..d553b81170 100644 --- a/loopx/control_plane/testing/model_behavior_corpus.py +++ b/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( diff --git a/loopx/control_plane/testing/model_behavior_packet_safety.py b/loopx/control_plane/testing/model_behavior_packet_safety.py index 52bea913ce..d3c9318eb2 100644 --- a/loopx/control_plane/testing/model_behavior_packet_safety.py +++ b/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 diff --git a/loopx/control_plane/testing/model_behavior_qualification.py b/loopx/control_plane/testing/model_behavior_qualification.py index 8b662c6496..b19c7a39d9 100644 --- a/loopx/control_plane/testing/model_behavior_qualification.py +++ b/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( diff --git a/tests/control_plane/test_doubao_model_behavior_actor.py b/tests/control_plane/test_doubao_model_behavior_actor.py index 2d79f589c3..86b8f3f472 100644 --- a/tests/control_plane/test_doubao_model_behavior_actor.py +++ b/tests/control_plane/test_doubao_model_behavior_actor.py @@ -457,3 +457,36 @@ def transport(**_: Any) -> Mapping[str, Any]: with pytest.raises(ValueError, match="canonical no-write contract"): actor(request) assert called is False + + +@pytest.mark.parametrize("arm", ["full_packet", "candidate_packet"]) +def test_direct_actor_delivers_complete_source_with_matching_field_instructions(arm: str) -> None: + from loopx.control_plane.quota.turn_envelope import compose_turn_decision + from tests.control_plane.test_model_behavior_qualification import _full_packet + + packet = compose_turn_decision(_full_packet()) + captured = [] + def transport(**kwargs): + captured.append(json.loads(kwargs["body"])) + return {"choices": [{"message": {"content": json.dumps(_decision())}}]} + actor = DoubaoModelBehaviorActor(api_key="fixture-key-not-a-secret", transport=transport) + request = build_model_behavior_actor_request(packet, qualification_id="current-provider-wire", + arm=arm, semantic_contract_required=True, + semantic_contract_fields=("required_reads", "peer_route", "write_scope")) + actor(request) + assert len(captured) == 1 + body = captured[0] + wire = json.loads(body["messages"][1]["content"]) + assert wire["arm"] == arm + assert wire["packet"] == packet + assert wire["packet"]["selected_todo"]["todo_id"] == "todo_fixture001" + assert "tools" not in body + instruction = " ".join(body["messages"][0]["content"].split()) + assert "interaction_contract.agent_channel" in instruction + assert "packet.agent_identity.agent_id" in instruction + assert "packet.selected_todo" in instruction + assert "goal_boundary.write_scope" in instruction + assert "copy packet.required_reads exactly" not in instruction + assert "use packet.action and packet.writeback" not in instruction + assert "action signature" not in instruction + assert "simulator has no external-effect authority" in instruction diff --git a/tests/control_plane/test_model_behavior_corpus.py b/tests/control_plane/test_model_behavior_corpus.py index b9cbdd1ad6..808b09ecc3 100644 --- a/tests/control_plane/test_model_behavior_corpus.py +++ b/tests/control_plane/test_model_behavior_corpus.py @@ -6,7 +6,9 @@ from typing import Any from loopx.control_plane.quota.turn_envelope import ( + COMPLETE_TURN_DECISION_VERSION, build_turn_envelope, + compose_turn_decision, quota_action_signature_document, turn_envelope_action_signature_document, ) @@ -105,7 +107,7 @@ def _actor(request: Mapping[str, Any]) -> dict[str, Any]: packet = request["packet"] signature = ( quota_action_signature_document(packet) - if arm == "full_packet" + if arm == "full_packet" or packet.get("schema_version") == COMPLETE_TURN_DECISION_VERSION else turn_envelope_action_signature_document(packet) ) response_plan = dict(signature.get("response_plan") or {}) @@ -200,6 +202,38 @@ def test_corpus_covers_matrix_retained_counterfactual_and_ablation() -> None: assert result["persistence_boundary"]["raw_packets_persisted"] is False +def test_retained_complete_capture_reaches_both_arms_without_reprojection(monkeypatch) -> None: + packet = compose_turn_decision({**_full_packet(), + "supporting_evidence": {"exact_count": 9007199254740993, "ratio": 1.0}}) + before = json.dumps(packet, sort_keys=True) + + def retired_projection(*args, **kwargs): + raise AssertionError("an original complete source must not become a v0 candidate") + + monkeypatch.setattr("loopx.control_plane.testing.model_behavior_corpus.build_turn_envelope", retired_projection) + corpus = build_model_behavior_corpus(packet, + retained_packets=[{"case_id": "retained-complete", "packet": packet}]) + case = corpus["cases"][0] + assert json.dumps(case["full_packet"], sort_keys=True) == before + assert json.dumps(case["candidate_packet"], sort_keys=True) == before + result = run_model_behavior_corpus(corpus, actor=_actor, repeats=2, seed=7) + assert result["all_cases_passed"] is True + assert result["corpus_gate_passed"] is True + assert result["promotion_eligible"] is False + assert json.dumps(packet, sort_keys=True) == before + + +def test_raw_quota_corpus_retains_its_existing_historical_control() -> None: + raw = _full_packet() + corpus = build_model_behavior_corpus(raw, + retained_packets=[{"case_id": "retained-raw", "packet": raw}]) + case = corpus["cases"][0] + assert "schema_version" not in case["full_packet"] + assert case["candidate_packet"]["schema_version"] == "loopx_turn_envelope_v0" + result = run_model_behavior_corpus(corpus, actor=_actor, repeats=2, seed=7) + assert result["all_cases_passed"] is True + + def test_corpus_reports_hard_and_stochastic_drift_separately() -> None: def drifting_actor(request: Mapping[str, Any]) -> dict[str, Any]: patch = ( diff --git a/tests/control_plane/test_model_behavior_qualification.py b/tests/control_plane/test_model_behavior_qualification.py index 769c667b8e..7df96eff00 100644 --- a/tests/control_plane/test_model_behavior_qualification.py +++ b/tests/control_plane/test_model_behavior_qualification.py @@ -9,7 +9,11 @@ import pytest -from loopx.control_plane.quota.turn_envelope import build_turn_envelope +from loopx.control_plane.quota.turn_envelope import ( + assert_complete_turn_decision, + build_turn_envelope, + compose_turn_decision, +) from loopx.control_plane.quota.live_decision import bind_scheduler_followup_cli_routes from loopx.control_plane.runtime.public_safety import SECRET_LIKE_SURFACE_PATTERN from loopx.control_plane.testing.model_behavior_qualification import ( @@ -925,3 +929,129 @@ def wrong_actor(request: Mapping[str, Any]) -> dict[str, Any]: assert result["semantic_contract_complete"] is False assert result["semantic_contract_drift"] == {} assert result["safety_violations"] == ["semantic_contract_mismatch:write_scope"] + + +@pytest.mark.parametrize("arm", ["full_packet", "candidate_packet"]) +def test_complete_source_reaches_actor_request_without_a_legacy_projection(arm: str) -> None: + full = _full_packet() + full["supporting_evidence"] = {"exact_count": 9007199254740993, "ratio": 1.0} + packet = compose_turn_decision(full) + before = json.dumps(packet, sort_keys=True) + request = build_model_behavior_actor_request(packet, qualification_id="complete-source", arm=arm) + assert request["packet_schema_version"] == "loopx_turn_envelope_v1" + assert json.dumps(request["packet"], sort_keys=True) == before + assert "action_signature" not in request["packet"] + assert request["sandbox"]["tools_enabled"] is False + assert request["sandbox"]["filesystem_writes_allowed"] is False + assert request["sandbox"]["external_writes_allowed"] is False + assert normalize_model_behavior_actor_request(request) == request + semantics = model_behavior_semantic_contract_from_packet(packet, arm=arm) + assert semantics["write_scope"] == ["loopx/**", "tests/**"] + assert semantics["peer_route"]["agent_id"] == "codex-fixture" + + +@pytest.mark.parametrize("captured", [False, True]) +def test_complete_pair_preserves_original_source_and_settlement(captured: bool) -> None: + full = _full_packet() + candidate = compose_turn_decision(full) + source = candidate if captured else full + result = run_model_behavior_qualification_pair(source, candidate, + qualification_id="complete-pair", actor=_actor) + assert result["equivalent"] is True + assert result["safety_violations"] == [] + + +@pytest.mark.parametrize("field,value", [ + ("goal_id", "other-goal"), + ("agent_identity", {"agent_id": "other-agent"}), + ("selected_todo", {"todo_id": "todo_other001", "status": "open"}), + ("recommended_action", "Inspect a different source."), + ("supporting_evidence", {"exact_count": 9007199254740993}), +]) +def test_valid_complete_candidate_from_another_source_is_refused_before_actor(field: str, value: Any) -> None: + full = _full_packet() + candidate = compose_turn_decision({**full, field: value}) + assert_complete_turn_decision(candidate) + calls = [] + with pytest.raises(ValueError, match="does not derive"): + run_model_behavior_qualification_pair(full, candidate, + qualification_id="wrong-complete-source", actor=lambda request: calls.append(request)) + assert calls == [] + + +@pytest.mark.parametrize("mutation", ["version", "binding", "source_hash", "content_hash", "body"]) +def test_malformed_complete_source_never_falls_back_to_legacy(mutation: str) -> None: + packet = compose_turn_decision(_full_packet()) + if mutation == "version": + packet["schema_version"] = "loopx_turn_envelope_v0" + elif mutation == "binding": + del packet["contract_binding"] + elif mutation == "body": + packet["interaction_contract"]["agent_channel"]["delivery_allowed"] = False + else: + key = "source_decision_hash" if mutation == "source_hash" else "content_hash" + packet["contract_binding"][key] = "sha256:" + "0" * 64 + calls = [] + with pytest.raises(ValueError): + run_model_behavior_qualification_pair(_full_packet(), packet, + qualification_id="bad-complete-source", actor=lambda request: calls.append(request)) + assert calls == [] + + +@pytest.mark.parametrize("legacy", [False, True]) +@pytest.mark.parametrize("arm_order", [("full_packet", "candidate_packet"), ("candidate_packet", "full_packet")]) +def test_pair_checks_both_privacy_boundaries_before_either_actor(legacy: bool, arm_order: tuple[str, str]) -> None: + full = {**_full_packet(), "api_key": "synthetic-private-value"} + candidate = build_turn_envelope(full) if legacy else compose_turn_decision(full) + calls = [] + with pytest.raises(ValueError, match="credential-shaped field"): + run_model_behavior_qualification_pair(full, candidate, + qualification_id="pair-private", actor=lambda request: calls.append(request), arm_order=arm_order) + assert calls == [] + + +@pytest.mark.parametrize("operation", ["ack", "host_failure"]) +@pytest.mark.parametrize("private", [False, True]) +def test_complete_candidate_scans_actual_scheduler_source_paths(operation: str, private: bool) -> None: + wire = _scheduler_wire(operation) + if private: + payload = json.loads(zlib.decompress(wire)) + payload["host_facts"]["api_key"] = "synthetic-private-value" + wire = zlib.compress(json.dumps(payload).encode()) + full, _ = _scheduler_transport_packet("full_packet", operation, compressed=wire) + packet = compose_turn_decision(full) + before = json.dumps(packet, sort_keys=True) + if private: + with pytest.raises(ValueError, match="credential-shaped field"): + build_model_behavior_actor_request(packet, qualification_id="complete-wire", arm="candidate_packet") + else: + request = build_model_behavior_actor_request(packet, qualification_id="complete-wire", arm="candidate_packet") + assert json.dumps(request["packet"], sort_keys=True) == before + assert json.dumps(packet, sort_keys=True) == before + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_cli_private_source_is_refused_before_provider_io(tmp_path, monkeypatch, provider) -> None: + from loopx.control_plane.testing.model_behavior_qualification import ModelBehaviorPairValidationError + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, _run_cli, _spend_run_count, _write_fixture, + ) + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path, provider=provider) + rc, source = _run_cli(registry, runtime, "quota", "should-run", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--codex-app", "--context-delivery", "off", + "--turn-instance-id", "qualification-private-source", cwd=project) + assert rc == 0 and source["ok"] is True + candidate = compose_turn_decision(source) + before = json.dumps(candidate, sort_keys=True) + calls = [] + # Real CLI context includes private runtime routing. Complete-source + # support must not authorize sending it to the shadow-model provider. + with pytest.raises(ModelBehaviorPairValidationError, match="local absolute path"): + run_model_behavior_qualification_pair(source, candidate, + qualification_id="private-cli-source", actor=lambda request: calls.append(request)) + assert calls == [] + assert json.dumps(candidate, sort_keys=True) == before + assert _spend_run_count(runtime) == 0 diff --git a/tests/control_plane/test_model_behavior_retained_cases.py b/tests/control_plane/test_model_behavior_retained_cases.py index fcd27f05d0..d23ff34609 100644 --- a/tests/control_plane/test_model_behavior_retained_cases.py +++ b/tests/control_plane/test_model_behavior_retained_cases.py @@ -103,6 +103,27 @@ def test_retained_case_round_trip_is_bounded_private_and_corpus_ready( assert "path" not in first +def test_complete_source_round_trip_preserves_its_original_binding(tmp_path: Path) -> None: + from loopx.control_plane.quota.turn_envelope import compose_turn_decision + + packet = compose_turn_decision({**_full_packet(), + "supporting_evidence": {"count": 9007199254740993, "ratio": 1.0}}) + case = build_model_behavior_retained_case(packet, case_id="complete-shadow-001", + source_kind="owner_approved_public_fixture", recorded_at="2026-07-13T04:00:00+08:00") + runtime = tmp_path / "runtime" + receipt = write_model_behavior_retained_case(case, runtime_root=runtime, goal_id="fixture-goal") + loaded = load_model_behavior_retained_cases(runtime_root=runtime, goal_id="fixture-goal") + corpus = build_model_behavior_corpus(packet, + retained_packets=retained_cases_as_corpus_inputs(loaded)) + assert loaded == [case] + for arm in ("full_packet", "candidate_packet"): + restored = corpus["cases"][0][arm] + assert json.dumps(restored, sort_keys=True) == json.dumps(packet, sort_keys=True) + assert restored["contract_binding"] == packet["contract_binding"] + assert "action_signature" not in restored + assert "packet" not in receipt and "path" not in receipt + + @pytest.mark.parametrize( "patch, message", [