From c76503629007d48208d72afd94a559fb4db31edc Mon Sep 17 00:00:00 2001 From: Lihua <1017343802@qq.com> Date: Wed, 4 Nov 2026 20:05:49 -0800 Subject: [PATCH] test(quota): follow captured context in recovery fixtures Read the native immutable capture before scripted model decisions and reuse the current CLI policy oracle for bound-wait assertions. Preserve full-byte, digest, freshness and original receipt checks without increasing actor tool budgets. Signed-off-by: Lihua <1017343802@qq.com> --- tests/control_plane/quota_context_fixture.py | 103 ++++++++++++++++++ ...actual_default_model_behavior_portfolio.py | 39 ++----- .../control_plane/test_model_tool_context.py | 39 +++++++ tests/test_quota_bound_wait_recovery.py | 14 ++- 4 files changed, 164 insertions(+), 31 deletions(-) create mode 100644 tests/control_plane/quota_context_fixture.py diff --git a/tests/control_plane/quota_context_fixture.py b/tests/control_plane/quota_context_fixture.py new file mode 100644 index 0000000000..f95d4293cd --- /dev/null +++ b/tests/control_plane/quota_context_fixture.py @@ -0,0 +1,103 @@ +"""Consume the current CLI capture through its reader, without new admission.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from hashlib import sha256 +import json +from typing import Any + +from loopx.control_plane.testing.model_tool_behavior import ( + ScriptedAssistantAction, + ScriptedDoubaoExecTransport, + ScriptedExecToolAction, + is_context_read_command, +) + + +def _object(value: Any) -> dict[str, Any] | None: + try: + payload = json.loads(value) + except (ValueError, TypeError): + return None + return payload if isinstance(payload, dict) else None + + +def _reader(command: Any) -> str: + if not isinstance(command, str) or not is_context_read_command(command): + raise ValueError("the current capture must supply its own context reader") + return command + + +def _retained_packet(pages: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """EOF alone is insufficient: retain contiguous UTF-8 bytes and their digest.""" + if not pages: + raise ValueError("the current quota capture was not read") + digest, total = pages[-1].get("sha256"), pages[-1].get("total_bytes") + if not isinstance(digest, str) or type(total) is not int or total <= 0: + raise ValueError("the current capture identity is incomplete") + retained = bytearray() + for page in sorted(pages, key=lambda item: item.get("offset", -1)): + offset, content = page.get("offset"), page.get("content") + if (page.get("ok") is not True or page.get("schema_version") != "context_capture_page_v1" + or page.get("sha256") != digest or page.get("total_bytes") != total + or type(offset) is not int or not 0 <= offset <= len(retained) + or not isinstance(content, str)): + raise ValueError("the current capture pages do not share one contiguous source") + data = content.encode("utf-8") + end = offset + len(data) + if (not data or end > total or page.get("eof") is not (end == total) + or page.get("next_offset") != (None if end == total else end)): + raise ValueError("the current capture page byte range is invalid") + overlap = min(len(retained) - offset, len(data)) + if retained[offset:offset + overlap] != data[:overlap]: + raise ValueError("repeated capture bytes changed") + retained.extend(data[overlap:]) + if len(retained) != total or sha256(retained).hexdigest() != digest: + raise ValueError("the complete current capture was not retained") + payload = _object(retained) + if payload is None: + raise ValueError("the complete capture is not a JSON object") + return payload + + +def latest_quota_payload(request: Mapping[str, Any]) -> dict[str, Any]: + packet, pages, navigation = None, [], False + for message in request["messages"]: + if message.get("role") != "tool": + continue + value = _object(message.get("content")) + if value is None: + continue + if value.get("schema_version") == "context_capture_navigation_v1": + packet, pages, navigation = None, [], True + elif value.get("schema_version") == "context_capture_page_v1": + pages.append(value) + elif isinstance(value.get("interaction_contract"), dict): + packet, pages, navigation = value, [], False + if pages: + packet = _retained_packet(pages) + if packet is None or "interaction_contract" not in packet: + suffix = "current capture was not read" if navigation else "quota result was not observed" + raise AssertionError(suffix) + return packet + + +def context_reading_transport(actions: Sequence[Any]) -> ScriptedDoubaoExecTransport: + """Script native reads as model tool calls within the actor's unchanged budget.""" + remaining = list(actions) + + def choose(request: Mapping[str, Any]) -> Any: + message = request["messages"][-1] + value = _object(message.get("content")) if message.get("role") == "tool" else None + if value is not None: + if value.get("schema_version") == "context_capture_navigation_v1": + return ScriptedExecToolAction(_reader(value["recovery"]["command"])) + if value.get("schema_version") == "context_capture_page_v1" and value.get("eof") is not True: + return ScriptedExecToolAction(_reader(value.get("next_command"))) + if not remaining: + return ScriptedAssistantAction("No further scripted action.") + action = remaining.pop(0) + return action(request) if callable(action) else action + + return ScriptedDoubaoExecTransport([choose] * 40) diff --git a/tests/control_plane/test_actual_default_model_behavior_portfolio.py b/tests/control_plane/test_actual_default_model_behavior_portfolio.py index 16a03ba944..94d89b0425 100644 --- a/tests/control_plane/test_actual_default_model_behavior_portfolio.py +++ b/tests/control_plane/test_actual_default_model_behavior_portfolio.py @@ -11,6 +11,10 @@ import pytest +from control_plane.quota_context_fixture import ( + context_reading_transport, + latest_quota_payload, +) from loopx.control_plane.quota.cli_projection import ( compact_quota_should_run_cli_payload, ) @@ -42,7 +46,6 @@ ) from loopx.control_plane.testing.model_tool_behavior import ( ScriptedAssistantAction, - ScriptedDoubaoExecTransport, ScriptedExecToolAction, ) from loopx.control_plane.testing.quota_fixtures import ( @@ -494,16 +497,7 @@ def run_actual_default_model_behavior_portfolio( def _latest_tool_payload(request: Mapping[str, Any]) -> dict[str, Any]: - for message in reversed(request["messages"]): - if message.get("role") != "tool": - continue - try: - payload = json.loads(message["content"]) - except (json.JSONDecodeError, TypeError): - continue - if isinstance(payload, dict) and "interaction_contract" in payload: - return payload - raise AssertionError("scripted model did not receive a quota tool result") + return latest_quota_payload(request) def _selected_read_action(request: Mapping[str, Any]) -> ScriptedExecToolAction: @@ -523,16 +517,6 @@ def _selected_read_action(request: Mapping[str, Any]) -> ScriptedExecToolAction: return ScriptedExecToolAction(f"cat {shlex.quote(match.group(0))}") -def _replan_frontier_read_action( - request: Mapping[str, Any], -) -> ScriptedExecToolAction: - payload = _latest_tool_payload(request) - action = payload["replan_action_packet"] - assert "fresh_vision_path_outcome" in action["uncovered_frontier"]["required_any_of"] - assert "replan-frontier.json" in payload["active_state_next_action"] - return ScriptedExecToolAction("cat replan-frontier.json") - - def _capability_callsite_action( request: Mapping[str, Any], ) -> ScriptedExecToolAction: @@ -617,7 +601,7 @@ def run_root(actor_kind: str, run_id: str) -> Path: def selected_todo_actor(run_id: str) -> Mapping[str, Any]: fixture_root = run_root("selected", run_id) fixture = _build_selected_fixture(fixture_root / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = context_reading_transport( [ ScriptedExecToolAction(fixture.quota_guard_command), _selected_read_action, @@ -633,14 +617,13 @@ def selected_todo_actor(run_id: str) -> Mapping[str, Any]: def replan_semantic_action_actor(run_id: str) -> Mapping[str, Any]: from tests.control_plane.test_required_vision_closeout_behavior import ( - vision_patch_action, projected_refresh, projected_spend, + vision_patch_action, projected_refresh, projected_spend, _vision_transport, ) fixture_root = run_root("replan", run_id) fixture = _build_replan_fixture(fixture_root / "oracle", required_vision=True) - transport = ScriptedDoubaoExecTransport( + transport = _vision_transport( [ ScriptedExecToolAction(fixture.quota_guard_command), - _replan_frontier_read_action, ScriptedExecToolAction("cat fixture/permission-config.json"), vision_patch_action, projected_refresh, projected_spend, ] @@ -657,7 +640,7 @@ def replan_semantic_action_actor(run_id: str) -> Mapping[str, Any]: def scoped_gate_successor_actor(run_id: str) -> Mapping[str, Any]: fixture_root = run_root("scoped-gate", run_id) fixture = _build_scoped_gate_fixture(fixture_root / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = context_reading_transport( [ ScriptedExecToolAction(fixture.quota_guard_command), _selected_read_action, @@ -675,7 +658,7 @@ def scoped_gate_successor_actor(run_id: str) -> Mapping[str, Any]: def capability_monitor_repair_actor(run_id: str) -> Mapping[str, Any]: fixture_root = run_root("capability-repair", run_id) fixture = _build_capability_repair_fixture(fixture_root / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = context_reading_transport( [ ScriptedExecToolAction(fixture.quota_guard_command), _capability_callsite_action, @@ -693,7 +676,7 @@ def capability_monitor_repair_actor(run_id: str) -> Mapping[str, Any]: def terminal_settlement_actor(run_id: str) -> Mapping[str, Any]: fixture_root = run_root("terminal-settlement", run_id) fixture = _build_terminal_settlement_fixture(fixture_root / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = context_reading_transport( [ ScriptedExecToolAction(fixture.quota_guard_command), ScriptedExecToolAction("cat fixture/settlement-proof.json"), diff --git a/tests/control_plane/test_model_tool_context.py b/tests/control_plane/test_model_tool_context.py index c73fefdad8..744df20385 100644 --- a/tests/control_plane/test_model_tool_context.py +++ b/tests/control_plane/test_model_tool_context.py @@ -1,6 +1,7 @@ """Native model stdout and private semantic grading must remain distinct.""" from __future__ import annotations +import argparse import json import os import shlex @@ -8,6 +9,8 @@ import pytest +from control_plane.quota_context_fixture import latest_quota_payload +from loopx.cli_commands.quota_capture import read_context_capture from loopx.control_plane.testing.model_tool_behavior import QuotaToolContext, snapshot_disposable_fixture from loopx.control_plane.turn_driver.decision_capture import ( capture_context_projection, @@ -51,6 +54,42 @@ def test_inline_output_retains_the_explicit_opt_out_contract(): assert context.capture_path is None +@pytest.mark.parametrize("attack", ["missing_middle", "changed_bytes", "unread_new_capture"]) +def test_scripted_consumer_cannot_replace_current_retained_context_with_eof(tmp_path, attack): + _, packet = _captured_output(tmp_path / "original") + packet["padding"] = "保留完整来源🙂" * 200 + (tmp_path / "paged").mkdir() + capture = capture_context_projection(packet, directory=tmp_path / "paged") + pages, offset = [], 0 + while True: + page = read_context_capture(argparse.Namespace( + capture_file=capture["path"], capture_sha256=capture["sha256"], + offset=offset, page_bytes=256, + )) + assert page["ok"], page + pages.append(page) + if page["eof"]: + break + offset = page["next_offset"] + assert len(pages) > 2 + messages = [{"role": "tool", "content": json.dumps(page)} for page in pages] + assert latest_quota_payload({"messages": messages}) == packet + if attack == "missing_middle": + messages.pop(len(messages) // 2) + elif attack == "changed_bytes": + first = dict(pages[0]) + assert "Do not" in first["content"] + first["content"] = first["content"].replace("Do not", "Please", 1) + messages[0]["content"] = json.dumps(first) + else: + messages.append({"role": "tool", "content": json.dumps({ + "ok": True, "schema_version": "context_capture_navigation_v1", + "recovery": {"command": "a new current reader has not been read"}, + })}) + with pytest.raises((ValueError, AssertionError)): + latest_quota_payload({"messages": messages}) + + @pytest.mark.parametrize("attack", ["other_capture", "other_digest", "session_lookup", "shell_chain"]) @NATIVE_CAPTURE_TEST def test_model_read_cannot_escape_the_current_capture(tmp_path, attack): diff --git a/tests/test_quota_bound_wait_recovery.py b/tests/test_quota_bound_wait_recovery.py index 298708b653..4eec955bb7 100644 --- a/tests/test_quota_bound_wait_recovery.py +++ b/tests/test_quota_bound_wait_recovery.py @@ -13,6 +13,7 @@ initialize_canonical_authority, read_fixture_todos, isolate_sqlite_runtime, ) +from control_plane.codex_checkpoint_fixture import read_cli_policy_payload from loopx.cli import main as cli_main from control_plane.test_quota_settlement_cli import ( AGENT_ID, @@ -62,7 +63,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: *args, ] ) - return rc, json.loads(capsys.readouterr().out) + return rc, read_cli_policy_payload(capsys.readouterr().out, returncode=rc) state = _configure_completion_validation_todo(project) _configure_selectable_alternative(project) @@ -197,6 +198,12 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert command_args[command_args.index("--registry") + 1] == str(registry) assert command_args[command_args.index("--runtime-root") + 1] == str(runtime) assert command_args[command_args.index("--todo-id") + 1] == TODO_ID + def work_identity(): + fields = ("status", "claimed_by", "resume_when", "resume_monitor_generation", "successor_todo_ids") + return {todo["todo_id"]: {field: todo.get(field) for field in fields} + for todo in read_fixture_todos(state)} + + original_todos = work_identity() rc, rejected = cli( "quota", "should-run", @@ -214,8 +221,9 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert rejected["error_code"] == "heartbeat_receipt_identity_conflict" assert rejected["heartbeat_receipt"]["status"] == "replayed" assert rejected["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID - assert rejected["effective_action"] != "heartbeat_receipt_write_failed" - assert "without" in rejected["recommended_action"] + assert rejected["should_run"] is False + assert rejected["heartbeat_receipt"]["event_id"] == replay["heartbeat_receipt"]["event_id"] + assert work_identity() == original_todos # Execute the advertised recovery action, supplying independently verified # evidence and the actual delivery workspace, not a parallel test-only command. refresh_args = tuple(