[codex] Follow captured quota context in recovery fixtures

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)

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

tests/control_plane/test_model_tool_context.py

@@ -1,13 +1,16 @@
 """Native model stdout and private semantic grading must remain distinct."""
 from __future__ import annotations
 
+import argparse
 import json
 import os
 import shlex
 from pathlib import Path
 
 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):

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(