diff --git a/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py b/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py index 3b1a75bcc0..32abb99af7 100644 --- a/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py +++ b/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py @@ -19,6 +19,8 @@ _direct_ark_transport, ) from .model_tool_behavior import ( + QuotaToolContext, + is_context_read_command, DoubaoExecToolClient, argument_value, digest_text, @@ -516,6 +518,7 @@ def qualify( ] steps: list[dict[str, Any]] = [] quota_observed = False + quota_context: QuotaToolContext | None = None callsite_verified = False reentry_completed = False contract: dict[str, Any] | None = None @@ -548,7 +551,9 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: ), ) - if _is_capability_reentry_command(tool_call.command): + if is_context_read_command(tool_call.command): + kind, tool_output = "context_read", None + elif _is_capability_reentry_command(tool_call.command): kind = ( "quota_reentry" if quota_observed and callsite_verified @@ -624,7 +629,8 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: "--turn-instance-id": turn_instance_id, }, ) - contract = _capability_repair_contract(json.loads(tool_output)) + quota_context = QuotaToolContext.from_output(tool_output) + contract = _capability_repair_contract(quota_context.packet) quota_observed = True except (RuntimeError, ValueError, json.JSONDecodeError): return receipt( @@ -642,7 +648,8 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: "--turn-instance-id": turn_instance_id, }, ) - reentry = _capability_reentry_contract(json.loads(tool_output)) + quota_context = QuotaToolContext.from_output(tool_output) + reentry = _capability_reentry_contract(quota_context.packet) contract = {**dict(contract or {}), **reentry} reentry_completed = True return receipt(passed=True, failure_code=None) @@ -651,6 +658,14 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: passed=False, failure_code="quota_reentry_failed", ) + elif kind == "context_read": + try: + if quota_context is None: + raise ValueError("context read before quota") + tool_output = quota_context.read(tool_call.command, + source_root=fixture.source_root, project_root=fixture.project_root) + except (OSError, RuntimeError, ValueError): + return receipt(passed=False, failure_code="context_read_failed") elif kind == "workspace_read": try: tool_output = _execute_workspace_read( diff --git a/loopx/control_plane/testing/model_tool_behavior.py b/loopx/control_plane/testing/model_tool_behavior.py index bfb6e30bb3..f11eff0261 100644 --- a/loopx/control_plane/testing/model_tool_behavior.py +++ b/loopx/control_plane/testing/model_tool_behavior.py @@ -511,6 +511,121 @@ def replace_argument(tokens: list[str], option: str, value: str) -> None: tokens[index + 1] = value +def is_context_read_command(command: str) -> bool: + tokens = loopx_command_tokens(command) + if not tokens or "quota" not in tokens: + return False + index = tokens.index("quota") + return tokens[index:index + 2] == ["quota", "context-read"] + + +def replan_view_command_tokens(command: str) -> list[str] | None: + """Normalize one captured replan reader without broadening shell authority. + + The production reader is rendered through ``python -m loopx.entrypoint`` + so it remains independent of the caller's installed console script. Model + behavior hosts otherwise accept only the bounded ``loopx`` console form. + Keep this exception local to the read-only replan route and return a + console-shaped argv for the existing isolated CLI executor. + """ + + try: + tokens = shlex.split(command) + except ValueError: + return None + if any(token in {";", "&&", "||", "|"} for token in tokens): + return None + normalized: list[str] | None = None + if tokens and Path(tokens[0]).name == "loopx": + normalized = list(tokens) + elif ( + len(tokens) >= 5 + and tokens[1] == "-m" + and tokens[2] in {"loopx.entrypoint", "loopx.cli"} + and Path(tokens[0]).name.startswith("python") + ): + normalized = ["loopx", *tokens[3:]] + if normalized is None or "quota" not in normalized: + return None + quota = normalized.index("quota") + if normalized[quota : quota + 2] != ["quota", "replan-view"]: + return None + return normalized + + +@dataclass(frozen=True) +class QuotaToolContext: + """Grade the current capture separately from the model's actual stdout. + + This private oracle read is not a model read, a retained baseline or a + receipt. Native actors still receive the untouched navigation document; + model-selected reads count against the original tool-call budget. + """ + + packet: dict[str, Any] + output: str + capture_path: Path | None = None + capture_sha256: str | None = None + + @classmethod + def from_output(cls, output: str) -> QuotaToolContext: + from ..turn_driver.decision_capture import ( + read_context_capture_file, read_context_read_reference, + ) + + packet = json.loads(output) + if not isinstance(packet, dict): + raise ValueError("quota result must be an object") + if isinstance(packet.get("interaction_contract"), dict): + return cls(packet, output) + recovery = packet.get("recovery") + if not (packet.get("ok") is True and packet.get("mode") in {"full", "delta"} + and isinstance(recovery, dict)): + return cls(packet, output) # The existing semantic oracle rejects it. + try: + reference = recovery + if "reference_file" in recovery: + _, reference = read_context_read_reference(recovery["reference_file"]) + if "selection" in reference: + raise ValueError("quota recovery must identify the complete current capture") + if not isinstance(reference.get("path"), str) or not isinstance(reference.get("sha256"), str): + raise ValueError("quota recovery requires the original capture and digest") + path, _, current = read_context_capture_file(reference["path"], reference["sha256"]) + except OSError as exc: + raise ValueError("the original current quota capture could not be read") from exc + # Always resolve current recovery, never the delta's saved baseline. + return cls(current, output, path, reference["sha256"]) + + def read(self, command: str, *, source_root: Path, project_root: Path) -> str: + from ..turn_driver.decision_capture import read_context_read_reference + + tokens = loopx_command_tokens(command) + if not is_context_read_command(command) or not tokens or self.capture_path is None: + raise ValueError("context read requires the current quota capture") + # Bind the host's explicit source, as the other behavior hosts do. The + # real CLI below still owns complete argument and selector validation. + # Do not import its full presentation/parser graph into the kernel. + for option in ("--format", "--capture-file", "--capture-sha256", "--reference-file"): + if tokens.count(option) > 1 or any(token.startswith(option + "=") for token in tokens): + raise ValueError("context reader requires one literal source argument") + if (argument_value(tokens, "--format") != "json" + or any(token.partition("=")[0] in {"--codex-context-dir", "--codex-context-root"} for token in tokens)): + raise ValueError("context reader must use the original explicit capture") + path, digest = argument_value(tokens, "--capture-file"), argument_value(tokens, "--capture-sha256") + reference_file = argument_value(tokens, "--reference-file") + if reference_file: + if path or digest: + raise ValueError("context reader cannot mix reference and capture arguments") + reader = Path(reference_file).expanduser() + _, reference = read_context_read_reference(str(reader if reader.is_absolute() else project_root / reader)) + path, digest = reference["path"], reference["sha256"] + source = Path(path).expanduser() if path else None + source = (source if source.is_absolute() else project_root / source) if source else None + if not source or source.resolve() != self.capture_path or digest != self.capture_sha256: + raise ValueError("context reader does not match the current quota capture") + return execute_loopx_cli(command, source_root=source_root, project_root=project_root) + + def execute_loopx_cli( command: str, *, diff --git a/loopx/control_plane/testing/replan_semantic_action_behavior.py b/loopx/control_plane/testing/replan_semantic_action_behavior.py index 7a300f6b86..9a1681c137 100644 --- a/loopx/control_plane/testing/replan_semantic_action_behavior.py +++ b/loopx/control_plane/testing/replan_semantic_action_behavior.py @@ -25,12 +25,15 @@ _direct_ark_transport, ) from .model_tool_behavior import ( + QuotaToolContext, + is_context_read_command, LoopxCliExecutionError, DoubaoExecToolClient, argument_value, digest_text, execute_loopx_cli, loopx_command_tokens, + replan_view_command_tokens, ) from .selected_todo_tool_behavior import bounded_workspace_read_plan from .replan_vision_closeout_behavior import ( @@ -43,7 +46,13 @@ REPLAN_SEMANTIC_ACTION_BEHAVIOR_RECEIPT_SCHEMA_VERSION = ( "replan_semantic_action_behavior_receipt_v0" ) +# Native progressive delivery adds three bounded reads (capture, review, and +# writeback) to the same semantic action path. Keep the original seven-call +# behavior budget for ordinary work and grant a separate allowance only after +# the quota result identifies a native capture. REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS = 7 +REPLAN_SEMANTIC_ACTION_NATIVE_READ_MAX_CALLS = 3 +_NATIVE_CONTEXT_READ_KINDS = frozenset({"context_read", "replan_context_read"}) class _HostToolError(RuntimeError): @@ -92,12 +101,16 @@ class _QualificationState: steps: list[dict[str, Any]] turn_instance_id: str quota_packet: dict[str, Any] | None = None + quota_context: QuotaToolContext | None = None quota_observation: dict[str, Any] | None = None semantic_delta: dict[str, Any] | None = None seen_quota: bool = False seen_clock: bool = False seen_refresh_state_help: bool = False read_only_host_commands_executed: bool = False + delivery_context_read: bool = False + replan_context_read: bool = False + replan_writeback_read: bool = False frontier_context_read: bool = False work_source_read: bool = False created_successor_id: str | None = None @@ -107,8 +120,23 @@ class _QualificationState: @property def tool_call_limit(self) -> int: - return (REQUIRED_VISION_CLOSEOUT_MAX_CALLS if self.fixture.required_vision - else REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS) + if self.fixture.required_vision: + return REQUIRED_VISION_CLOSEOUT_MAX_CALLS + return REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS + ( + REPLAN_SEMANTIC_ACTION_NATIVE_READ_MAX_CALLS + if self.quota_context is not None + and self.quota_context.capture_path is not None + else 0 + ) + + @property + def maximum_tool_call_limit(self) -> int: + if self.fixture.required_vision: + return REQUIRED_VISION_CLOSEOUT_MAX_CALLS + return ( + REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS + + REPLAN_SEMANTIC_ACTION_NATIVE_READ_MAX_CALLS + ) def _digest(value: str) -> str: @@ -619,7 +647,11 @@ def _required_explore_read_command(packet: Mapping[str, Any]) -> str | None: return command -def _quota_behavior_observation(packet: Mapping[str, Any]) -> dict[str, Any]: +def _quota_behavior_observation( + packet: Mapping[str, Any], + *, + replan_view: Mapping[str, Any] | None = None, +) -> dict[str, Any]: _required_explore_read_command(packet) signature = quota_action_signature_document(packet) action = dict(signature.get("action") or {}) @@ -629,15 +661,42 @@ def _quota_behavior_observation(packet: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(obligation, Mapping) or not isinstance(replan_action, Mapping): raise TypeError("quota must project an obligation and replan action packet") context = obligation.get("replan_context") + progressive_view = not isinstance(context, Mapping) + if progressive_view: + context = replan_view if not isinstance(context, Mapping): - raise TypeError("quota must host-project the replan coverage context") + raise TypeError("quota must deliver the replan coverage context") + novelty = obligation.get("replan_novelty_policy") + delivery = ( + str(novelty.get("delivery") or "") + if progressive_view and isinstance(novelty, Mapping) + else str(context.get("delivery") or "") + ) + evidence_source = ( + str(novelty.get("evidence_source") or "") + if progressive_view and isinstance(novelty, Mapping) + else str(context.get("evidence_source") or "") + ) + progressive_identity_valid = bool( + progressive_view + and context.get("ok") is True + and context.get("stage") == "review" + and context.get("goal_id") == _FIXTURE_GOAL_ID + and context.get("agent_id") == _FIXTURE_AGENT_ID + and context.get("obligation_id") == obligation.get("obligation_id") + and str(context.get("source_hash") or "").startswith("sha256:") + ) + inline_delivery_valid = bool( + not progressive_view + and dict(context.get("delivery_receipt") or {}).get("status") == "delivered" + ) if ( packet.get("decision") != "autonomous_replan_required" or replan_action.get("decision") != "replan_required" or replan_action.get("obligation_id") != obligation.get("obligation_id") - or context.get("delivery") != "host_projected" - or context.get("evidence_source") != "compact_run_history" - or dict(context.get("delivery_receipt") or {}).get("status") != "delivered" + or delivery != "host_projected" + or evidence_source != "compact_run_history" + or not (progressive_identity_valid or inline_delivery_valid) ): raise ValueError("quota does not expose the semantic replan contract") if not ( @@ -655,9 +714,18 @@ def _quota_behavior_observation(packet: Mapping[str, Any]) -> dict[str, Any]: "quiet_noop_allowed": False, "external_write_requested": False, "replan_obligation_id": obligation.get("obligation_id"), - "context_delivery": "host_projected", + "context_delivery": ( + "progressive_replan_view" if progressive_view else "host_projected" + ), "required_semantic_outcomes": list( - dict(replan_action.get("uncovered_frontier") or {}).get( + dict( + ( + context.get("uncovered_frontier") + if progressive_view + else replan_action.get("uncovered_frontier") + ) + or {} + ).get( "required_any_of" ) or [] @@ -974,6 +1042,15 @@ def _qualification_receipt( "required_vision_closeout" if state.fixture.required_vision else "semantic_action" ) receipt["tool_call_limit"] = state.tool_call_limit + if not state.fixture.required_vision: + receipt["ordinary_tool_call_limit"] = ( + REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS + ) + receipt["native_context_read_call_limit"] = ( + REPLAN_SEMANTIC_ACTION_NATIVE_READ_MAX_CALLS + if state.tool_call_limit > REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS + else 0 + ) if state.fixture.required_vision: receipt["execution_host"] = "os_isolated_shell" receipt["boundary"]["shell_commands_executed"] = bool(state.steps) @@ -1009,18 +1086,105 @@ def _handle_quota_command(command: str, state: _QualificationState) -> str: fixture=state.fixture, turn_instance_id=state.turn_instance_id, ) - decoded = json.loads(output) - if not isinstance(decoded, dict): - raise ValueError("quota result must be an object") + state.quota_context = QuotaToolContext.from_output(output) + decoded = state.quota_context.packet state.quota_packet = decoded - state.quota_observation = _quota_behavior_observation(decoded) - if state.fixture.required_vision and "required_agent_vision_missing" not in state.quota_observation["trigger_kinds"]: + obligation = decoded.get("autonomous_replan_obligation") + if isinstance(obligation, Mapping) and isinstance( + obligation.get("replan_context"), Mapping + ): + state.quota_observation = _quota_behavior_observation(decoded) + elif not ( + state.quota_context.capture_path is not None + and isinstance(obligation, Mapping) + and isinstance(obligation.get("context_read"), str) + and obligation.get("context_read") + ): + raise TypeError("quota must deliver or reference the replan coverage context") + if ( + state.fixture.required_vision + and state.quota_observation is not None + and "required_agent_vision_missing" + not in state.quota_observation["trigger_kinds"] + ): raise ValueError("required_vision_fixture_trigger_mismatch") state.seen_quota = True state.read_only_host_commands_executed = True return output +def _handle_context_read(command: str, state: _QualificationState) -> str: + if state.quota_context is None: + raise ValueError("context read before quota") + output = state.quota_context.read(command, source_root=state.fixture.source_root, + project_root=state.fixture.project_root) + state.delivery_context_read = True + state.read_only_host_commands_executed = True + return output + + +def _handle_replan_context_read(command: str, state: _QualificationState) -> str: + if state.quota_packet is None or state.quota_context is None: + raise ValueError("replan_context_read_before_quota") + if state.quota_context.capture_path is not None and not state.delivery_context_read: + raise ValueError("replan_context_read_before_delivery_context") + obligation = state.quota_packet.get("autonomous_replan_obligation") + review_command = ( + str(obligation.get("context_read") or "") + if isinstance(obligation, Mapping) + else "" + ) + action = state.quota_packet.get("replan_action_packet") + writeback_command = ( + str(action.get("writeback") or "") if isinstance(action, Mapping) else "" + ) + normalized = replan_view_command_tokens(command) + if normalized is None or command not in {review_command, writeback_command}: + raise ValueError("unexpected_replan_context_read") + is_writeback = command == writeback_command and command != review_command + if is_writeback and not state.replan_context_read: + raise ValueError("replan_writeback_read_before_review") + if is_writeback and state.replan_writeback_read: + raise ValueError("repeated_replan_writeback_read") + if not is_writeback and state.replan_context_read: + raise ValueError("repeated_replan_context_read") + output = execute_loopx_cli( + shlex.join(normalized), + source_root=state.fixture.source_root, + project_root=state.fixture.project_root, + ) + view = json.loads(output) + if not isinstance(view, Mapping): + raise ValueError("replan_context_read_invalid") + if is_writeback: + expected_obligation_id = ( + obligation.get("obligation_id") + if isinstance(obligation, Mapping) + else None + ) + if ( + view.get("stage") != "writeback" + or view.get("obligation_id") != expected_obligation_id + or not isinstance(view.get("writeback_contract"), Mapping) + ): + raise ValueError("replan_writeback_read_invalid") + state.replan_writeback_read = True + else: + state.quota_observation = _quota_behavior_observation( + state.quota_packet, + replan_view=view, + ) + if ( + state.fixture.required_vision + and "required_agent_vision_missing" + not in state.quota_observation["trigger_kinds"] + ): + raise ValueError("required_vision_fixture_trigger_mismatch") + state.replan_context_read = True + state.read_only_host_commands_executed = True + return output + + def _handle_clock_command(output: str, state: _QualificationState) -> str: if state.seen_clock: raise ValueError("repeated_clock") @@ -1078,10 +1242,20 @@ def _require_observed_frontier( ) -> None: if state.quota_packet is None: raise ValueError(f"{prefix}_before_quota") + if state.quota_context is not None and state.quota_context.capture_path is not None: + if not state.delivery_context_read: + raise ValueError(f"{prefix}_before_delivery_context_read") + if state.quota_observation is None: + raise ValueError(f"{prefix}_before_replan_context_read") if not state.frontier_context_read: raise ValueError(f"{prefix}_before_frontier_read") if work_source_required and not state.work_source_read: raise ValueError(f"{prefix}_before_work_source_read") + if ( + state.quota_observation.get("context_delivery") == "progressive_replan_view" + and not state.replan_writeback_read + ): + raise ValueError(f"{prefix}_before_replan_writeback_read") def _handle_successor_command(command: str, state: _QualificationState) -> str: @@ -1118,7 +1292,6 @@ def _handle_successor_command(command: str, state: _QualificationState) -> str: and decoded.get("ok") is True and decoded.get("added") is True and state.created_successor_id - and decoded.get("host_action") == "end_current_heartbeat" ): raise ValueError("successor_create_failed") transition = decoded.get("replan_transition") @@ -1130,6 +1303,17 @@ def _handle_successor_command(command: str, state: _QualificationState) -> str: and transition.get("outcome") == "new_runnable_successor" ): raise ValueError("successor_transition_missing") + immediate_end = ( + transition.get("turn_boundary") == "end_current_heartbeat" + and decoded.get("host_action") == "end_current_heartbeat" + ) + settle_then_end = ( + transition.get("turn_boundary") == "settle_original_turn_before_end" + and isinstance(decoded.get("settlement_identity"), Mapping) + and isinstance(decoded.get("settlement_plan"), Mapping) + ) + if not (immediate_end or settle_then_end): + raise ValueError("successor_create_failed") state.semantic_delta = { "accepted": True, "satisfying_outcomes": ["new_runnable_successor"], @@ -1139,7 +1323,7 @@ def _handle_successor_command(command: str, state: _QualificationState) -> str: fixture=state.fixture, turn_instance_id=f"{state.turn_instance_id}-successor", ) - reentry_packet = json.loads(reentry_output) + reentry_packet = QuotaToolContext.from_output(reentry_output).packet if not isinstance(reentry_packet, dict): raise ValueError("successor_reentry_quota_invalid") state.successor_reentry_observation = _successor_reentry_observation( @@ -1205,7 +1389,7 @@ def _handle_semantic_writeback( fixture=state.fixture, turn_instance_id=f"{state.turn_instance_id}-semantic-reentry", ) - reentry_packet = json.loads(reentry_output) + reentry_packet = QuotaToolContext.from_output(reentry_output).packet if not isinstance(reentry_packet, dict): raise ValueError("semantic_reentry_quota_invalid") state.semantic_reentry_observation = _semantic_reentry_observation( @@ -1223,6 +1407,14 @@ def _dispatch_behavior_command( ) -> tuple[str, str, bool]: if _is_quota_guard(command): return _handle_quota_command(command, state), "quota_should_run", False + if is_context_read_command(command): + return _handle_context_read(command, state), "context_read", False + if replan_view_command_tokens(command) is not None: + return ( + _handle_replan_context_read(command, state), + "replan_context_read", + False, + ) if state.quota_packet is not None and ( required := _required_explore_read_command(state.quota_packet) ): @@ -1270,6 +1462,10 @@ def _behavior_command_kind( if _is_quota_guard(command): return "quota_should_run" + if is_context_read_command(command): + return "context_read" + if replan_view_command_tokens(command) is not None: + return "replan_context_read" if state.quota_packet is not None and ( required := _required_explore_read_command(state.quota_packet) ) and shlex.split(command) == shlex.split(required): @@ -1295,16 +1491,23 @@ def _behavior_command_kind( "repeated_quota_should_run", "repeated_clock", "semantic_action_before_quota", + "semantic_action_before_delivery_context_read", + "semantic_action_before_replan_context_read", "semantic_action_before_frontier_read", "semantic_action_before_work_source_read", + "semantic_action_before_replan_writeback_read", "semantic_action_misses_uncovered_frontier", "refresh_state_help_before_quota", "refresh_state_help_before_frontier_read", "refresh_state_help_before_work_source_read", + "refresh_state_help_before_replan_writeback_read", "repeated_refresh_state_help", "successor_create_before_quota", + "successor_create_before_delivery_context_read", + "successor_create_before_replan_context_read", "successor_create_before_frontier_read", "successor_create_before_work_source_read", + "successor_create_before_replan_writeback_read", "successor_create_obligation_mismatch", "successor_create_composition_binding_mismatch", "repeated_successor_create", @@ -1324,6 +1527,14 @@ def _behavior_command_kind( "semantic_reentry_not_future_monitor_wait", "semantic_reentry_did_not_exit", "manual_evidence_read_is_not_replan", + "replan_context_read_before_quota", + "replan_context_read_before_delivery_context", + "replan_writeback_read_before_review", + "repeated_replan_context_read", + "repeated_replan_writeback_read", + "replan_writeback_read_invalid", + "unexpected_replan_context_read", + "replan_context_read_invalid", "unexpected_command", } ) @@ -1356,6 +1567,19 @@ def _append_tool_response( ) +def _tool_call_budget_exhausted(state: _QualificationState) -> bool: + if state.fixture.required_vision: + return len(state.steps) >= REQUIRED_VISION_CLOSEOUT_MAX_CALLS + ordinary_calls = sum( + step.get("kind") not in _NATIVE_CONTEXT_READ_KINDS + for step in state.steps + ) + return ( + ordinary_calls >= REPLAN_SEMANTIC_ACTION_BEHAVIOR_MAX_CALLS + or len(state.steps) >= state.maximum_tool_call_limit + ) + + def _run_qualification_loop( client: DoubaoExecToolClient, state: _QualificationState, @@ -1363,7 +1587,9 @@ def _run_qualification_loop( qualification_id: str, shell: VisionShellHost | None = None, ) -> dict[str, Any]: - for _ in range(state.tool_call_limit): + for _ in range(state.maximum_tool_call_limit): + if _tool_call_budget_exhausted(state): + break tool_call = client.next_tool_call( state.messages, tool_description=VISION_EXEC_TOOL_DESCRIPTION if state.fixture.required_vision else None, @@ -1379,7 +1605,11 @@ def _run_qualification_loop( kind = "shell" if shell else _behavior_command_kind(tool_call.command, state) try: if shell: - output, exit_code = shell.execute(tool_call.command) + # A captured module reader uses the same supervised CLI bridge, + # without allowing the shell to import the source checkout. + reader = replan_view_command_tokens(tool_call.command) + command = shlex.join(reader) if reader is not None else tool_call.command + output, exit_code = shell.execute(command) observe_shell_evidence(output, state) state.read_only_host_commands_executed = True if exit_code: @@ -1485,7 +1715,14 @@ def qualify( "project. Follow the heartbeat task and use the shell tool. " "The shell working directory is the connected goal project " "root; relative paths in the active-state action resolve there. " - "Treat the quota packet as the control-plane authority. Complete " + "Treat the quota packet as the control-plane authority. If the " + "guard returns native context navigation, execute its recovery " + "reader before acting. If the delivered replan obligation names " + "a context_read command, execute that exact command before " + "choosing a semantic result. Read the work evidence. " + "If the current replan_action_packet supplies a writeback command, " + "execute that exact command to obtain the typed action contract. " + "Otherwise use its inline writeback_contract. Complete " "one bounded in-scope action and persist its result through the " "real LoopX CLI. After quota, `cat` only the exact path named " "in active_state_next_action, then `cat` only the source_ref " @@ -1522,6 +1759,19 @@ def invoke(argv: list[str], cwd: Path, output: str) -> str: if _is_quota_guard(command): return (_execute_loopx(command, fixture=fixture, turn_instance_id=state.turn_instance_id) if state.seen_quota else _handle_quota_command(command, state)) + if is_context_read_command(command): + return _handle_context_read(command, state) + if (reader := replan_view_command_tokens(command)) is not None: + packet = state.quota_packet or {} + obligation = packet.get("autonomous_replan_obligation") + action = packet.get("replan_action_packet") + for expected in ( + obligation.get("context_read") if isinstance(obligation, Mapping) else None, + action.get("writeback") if isinstance(action, Mapping) else None, + ): + if isinstance(expected, str) and replan_view_command_tokens(expected) == reader: + return _handle_replan_context_read(expected, state) + raise ValueError("unexpected_replan_context_read") result = dispatch_vision_closeout(command, state, execute=_execute_loopx) if result is None: raise ValueError("This fixture grants CLI authority for quota, help, vision refresh and settlement only") diff --git a/loopx/control_plane/testing/replan_vision_closeout_behavior.py b/loopx/control_plane/testing/replan_vision_closeout_behavior.py index c25c639537..91273bf7f1 100644 --- a/loopx/control_plane/testing/replan_vision_closeout_behavior.py +++ b/loopx/control_plane/testing/replan_vision_closeout_behavior.py @@ -10,7 +10,7 @@ from collections.abc import Mapping from typing import TYPE_CHECKING, Any, Callable -from .model_tool_behavior import argument_value, loopx_command_tokens +from .model_tool_behavior import QuotaToolContext, argument_value, loopx_command_tokens from .model_behavior_qualification import model_behavior_semantic_contract_from_packet from ..quota.effective_action import EffectiveAction @@ -180,10 +180,10 @@ def dispatch_vision_closeout( if state.vision_closeout is None: raise ValueError("vision_closeout_spend_before_writeback") output = execute(command, fixture=state.fixture, turn_instance_id=state.turn_instance_id) - replay = json.loads(execute(state.fixture.quota_guard_command, fixture=state.fixture, - turn_instance_id=state.turn_instance_id)) - following = json.loads(execute(state.fixture.quota_guard_command, fixture=state.fixture, - turn_instance_id=f"{state.turn_instance_id}-readback")) + replay = QuotaToolContext.from_output(execute(state.fixture.quota_guard_command, fixture=state.fixture, + turn_instance_id=state.turn_instance_id)).packet + following = QuotaToolContext.from_output(execute(state.fixture.quota_guard_command, fixture=state.fixture, + turn_instance_id=f"{state.turn_instance_id}-readback")).packet spends = [row for row in _rows(state) if row.get("classification") == "quota_slot_spent"] if replay.get("effective_action") != EffectiveAction.HEARTBEAT_SETTLED_SKIP.value or len(spends) != 1: raise ValueError("vision_closeout_settlement_readback_failed") diff --git a/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py b/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py index b1db9cec49..ef3b4b9d95 100644 --- a/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py +++ b/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py @@ -19,6 +19,8 @@ _direct_ark_transport, ) from .model_tool_behavior import ( + QuotaToolContext, + is_context_read_command, DoubaoExecToolClient, QUOTA_FIRST_TOOL_INSTRUCTION, digest_text, @@ -378,6 +380,7 @@ def qualify( ] steps: list[dict[str, Any]] = [] quota_observed = False + quota_context: QuotaToolContext | None = None notice_surfaced = False selected_action_executed = False contract: dict[str, Any] | None = None @@ -448,6 +451,8 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: fixture=fixture, quota_observed=quota_observed, ) + if is_context_read_command(tool_call.command): + kind, tool_output = "context_read", None if ( kind == "unexpected_command" and rejected_read_count == 0 @@ -501,14 +506,22 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: "--turn-instance-id": turn_instance_id, }, ) - quota_packet = json.loads(tool_output) - contract = _scoped_gate_quota_contract(quota_packet) + quota_context = QuotaToolContext.from_output(tool_output) + contract = _scoped_gate_quota_contract(quota_context.packet) quota_observed = True except (RuntimeError, ValueError, json.JSONDecodeError): return receipt( passed=False, failure_code="quota_execution_failed", ) + elif kind == "context_read": + try: + if quota_context is None: + raise ValueError("context read before quota") + tool_output = quota_context.read(tool_call.command, + source_root=fixture.source_root, project_root=fixture.project_root) + except (OSError, RuntimeError, ValueError): + return receipt(passed=False, failure_code="context_read_failed") elif kind == "workspace_read": try: tool_output = _execute_workspace_read( diff --git a/loopx/control_plane/testing/selected_todo_tool_behavior.py b/loopx/control_plane/testing/selected_todo_tool_behavior.py index 14e7ba977d..32dede1e7c 100644 --- a/loopx/control_plane/testing/selected_todo_tool_behavior.py +++ b/loopx/control_plane/testing/selected_todo_tool_behavior.py @@ -20,6 +20,8 @@ _direct_ark_transport, ) from .model_tool_behavior import ( + QuotaToolContext, + is_context_read_command, DoubaoExecToolClient, QUOTA_FIRST_TOOL_INSTRUCTION, argument_value, @@ -1093,6 +1095,7 @@ def qualify( ] steps: list[dict[str, Any]] = [] quota_observed = False + quota_context: QuotaToolContext | None = None selection_required = False contract: dict[str, Any] | None = None seen_once: set[str] = set() @@ -1123,6 +1126,8 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: fixture=fixture, quota_observed=quota_observed, ) + if is_context_read_command(tool_call.command): + kind, tool_output = "context_read", None steps.append( { "ordinal": len(steps) + 1, @@ -1175,8 +1180,8 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: "--turn-instance-id": turn_instance_id, }, ) - quota_packet = json.loads(tool_output) - contract = _quota_contract(quota_packet) + quota_context = QuotaToolContext.from_output(tool_output) + contract = _quota_contract(quota_context.packet) quota_observed = True selection_required = bool(contract["selection_required"]) if kind == "quota_action_selection": @@ -1189,6 +1194,14 @@ def receipt(*, passed: bool, failure_code: str | None) -> dict[str, Any]: passed=False, failure_code="quota_execution_failed", ) + elif kind == "context_read": + try: + if quota_context is None: + raise ValueError("context read before quota") + tool_output = quota_context.read(tool_call.command, + source_root=fixture.source_root, project_root=fixture.project_root) + except (OSError, RuntimeError, ValueError): + return receipt(passed=False, failure_code="context_read_failed") elif kind == "candidate_discovery": try: tool_output = execute_loopx_cli( diff --git a/loopx/control_plane/testing/terminal_settlement_tool_behavior.py b/loopx/control_plane/testing/terminal_settlement_tool_behavior.py index bf369e1576..160a925f46 100644 --- a/loopx/control_plane/testing/terminal_settlement_tool_behavior.py +++ b/loopx/control_plane/testing/terminal_settlement_tool_behavior.py @@ -23,6 +23,7 @@ _direct_ark_transport, ) from .model_tool_behavior import ( + QuotaToolContext, DoubaoExecToolClient, QUOTA_FIRST_TOOL_INSTRUCTION, LoopxCliExecutionError, @@ -120,6 +121,7 @@ class _QualificationState: turn_instance_id: str steps: list[dict[str, Any]] = field(default_factory=list) quota_packet: dict[str, Any] | None = None + quota_context: QuotaToolContext | None = None contract: dict[str, Any] | None = None proof_read: bool = False writeback_committed: bool = False @@ -931,10 +933,18 @@ def _execute_terminal_cli( if state.quota_packet is not None: raise _TerminalToolViolation("repeated_quota_should_run") payload = _execute_cli(command, state) - state.quota_packet = payload - state.contract = _quota_contract(payload) + state.quota_context = QuotaToolContext.from_output( + json.dumps(payload, ensure_ascii=False, sort_keys=True)) + state.quota_packet = state.quota_context.packet + state.contract = _quota_contract(state.quota_packet) kind = "quota_should_run" - output = json.dumps(payload, ensure_ascii=False, sort_keys=True) + output = state.quota_context.output + elif command_kind == "quota" and command_action == "context-read": + if state.quota_context is None: + raise _TerminalToolViolation("context_read_before_quota") + output = state.quota_context.read(command, source_root=state.fixture.source_root, + project_root=state.fixture.project_root) + kind = "context_read" elif command_kind == "refresh_state": if state.quota_packet is None: raise _TerminalToolViolation("writeback_before_quota") diff --git a/tests/control_plane/test_capability_monitor_repair_tool_behavior.py b/tests/control_plane/test_capability_monitor_repair_tool_behavior.py index a03a6d031d..9d7110336b 100644 --- a/tests/control_plane/test_capability_monitor_repair_tool_behavior.py +++ b/tests/control_plane/test_capability_monitor_repair_tool_behavior.py @@ -13,6 +13,7 @@ _build_capability_repair_fixture, ) from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, scripted_assistant_response as _scripted_assistant_response, scripted_exec_tool_response as _tool_response, ) @@ -31,10 +32,10 @@ def _projected_reentry_command(request: Mapping[str, Any]) -> str: if not isinstance(message, Mapping) or message.get("role") != "tool": continue try: - packet = json.loads(str(message.get("content") or "")) + packet = QuotaToolContext.from_output(str(message.get("content") or "")).packet except json.JSONDecodeError: continue - reentry = packet.get("runtime_capability_reentry") + reentry = (packet.get("interaction_contract") or {}).get("cli_channel", {}).get("runtime_capability_reentry") if isinstance(reentry, Mapping): return str(reentry["candidates"][0]["command"]) raise AssertionError("quota result did not project runtime capability re-entry") @@ -122,14 +123,13 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: ) assert fixture.task_body in trigger_message - quota_result = json.loads(requests[1]["messages"][-1]["content"]) - quota_keys = list(quota_result) - assert ( - quota_keys.index("heartbeat_receipt") - < quota_keys.index("runtime_capability_reentry") - < quota_keys.index("interaction_contract") - <= 8 - ) + quota_result = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet + # Native output is navigation, not an ordering of duplicated inline keys. + # Preserve receipt metadata and validate the canonical repair owner below; + # capability repair does not itself establish admission or settlement. + assert isinstance(quota_result["heartbeat_receipt"], Mapping) + assert quota_result["interaction_contract"]["cli_channel"][ + "runtime_capability_reentry"]["candidates"] assert quota_result.get("selected_todo") is None assert quota_result["interaction_contract"]["mode"] == ( "capability_bridge_repair" diff --git a/tests/control_plane/test_model_tool_context.py b/tests/control_plane/test_model_tool_context.py new file mode 100644 index 0000000000..c73fefdad8 --- /dev/null +++ b/tests/control_plane/test_model_tool_context.py @@ -0,0 +1,147 @@ +"""Native model stdout and private semantic grading must remain distinct.""" +from __future__ import annotations + +import json +import os +import shlex +from pathlib import Path + +import pytest + +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, + write_context_read_reference, +) + +NATIVE_CAPTURE_TEST = pytest.mark.skipif( + os.name != "posix", reason="Native captured-input transport is POSIX-only; other hosts retain inline delivery") + + +def _captured_output(root: Path, *, mode: str = "full") -> tuple[str, dict]: + root.mkdir(parents=True, exist_ok=True) + packet = {"interaction_contract": {"stop": "Do not publish without acceptance."}, + "selected_todo": {"id": "selected-work", "text": "Read the selected proof."}, + "unchanged": "完整来源 / complete source"} + capture = capture_context_projection(packet, directory=root) + reference = write_context_read_reference(Path(capture["path"]), capture["sha256"]) + output = json.dumps({"ok": True, "schema_version": "context_capture_navigation_v1", + "mode": mode, "reference_file": "not-the-current-source.reader.json", + "recovery": {"reference_file": str(reference)}, + "baseline_reference": {"reference_file": "never-prefetch-this-baseline.reader.json"}}) + return output, packet + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +@NATIVE_CAPTURE_TEST +def test_oracle_uses_current_recovery_without_replacing_model_stdout(tmp_path, mode): + output, packet = _captured_output(tmp_path, mode=mode) + context = QuotaToolContext.from_output(output) + assert context.packet == packet + assert context.output == output + assert "interaction_contract" not in json.loads(context.output) + assert context.capture_path is not None + + +def test_inline_output_retains_the_explicit_opt_out_contract(): + packet = {"ok": True, "interaction_contract": {"must_stop": True}} + output = json.dumps(packet) + context = QuotaToolContext.from_output(output) + assert context.packet == packet and context.output == output + assert context.capture_path is None + + +@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): + output, _ = _captured_output(tmp_path / "current") + context = QuotaToolContext.from_output(output) + command = ["loopx", "--format", "json", "quota", "context-read", + "--capture-file", str(context.capture_path), "--capture-sha256", context.capture_sha256] + if attack == "other_capture": + other_output, _ = _captured_output(tmp_path / "other") + other = QuotaToolContext.from_output(other_output) + command[6] = str(other.capture_path) + command[8] = other.capture_sha256 + elif attack == "other_digest": + command[8] = "0" * 64 + elif attack == "session_lookup": + command = command[:5] + ["--codex-context-dir", str(tmp_path)] + shell = shlex.join(command) + if attack == "shell_chain": + shell += " && loopx quota should-run --goal-id another-goal" + + before = snapshot_disposable_fixture(tmp_path) + with pytest.raises(ValueError): + context.read(shell, source_root=tmp_path, project_root=tmp_path) + assert snapshot_disposable_fixture(tmp_path) == before + + +@NATIVE_CAPTURE_TEST +def test_real_reader_supports_batched_progressive_fields_without_a_guard(tmp_path): + output, packet = _captured_output(tmp_path) + context = QuotaToolContext.from_output(output) + command = shlex.join(["loopx", "--format", "json", "quota", "context-read", + "--capture-file", str(context.capture_path), "--capture-sha256", context.capture_sha256, + "--select", "/selected_todo", "--select", "/interaction_contract"]) + page = json.loads(context.read(command, source_root=Path(__file__).resolve().parents[2], + project_root=tmp_path)) + selected = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} + assert selected == {"/selected_todo": packet["selected_todo"], + "/interaction_contract": packet["interaction_contract"]} + assert page["eof"] + assert not (tmp_path / ".loopx").exists() + + +@NATIVE_CAPTURE_TEST +def test_changed_current_capture_is_not_graded_as_its_previous_decision(tmp_path): + output, _ = _captured_output(tmp_path) + context = QuotaToolContext.from_output(output) + assert context.capture_path is not None + context.capture_path.write_text('{"interaction_contract":{"must_stop":false}}') + with pytest.raises(ValueError, match="digest mismatch"): + QuotaToolContext.from_output(output) + + +@NATIVE_CAPTURE_TEST +def test_actor_receives_native_navigation_then_executes_its_progressive_reader(tmp_path, monkeypatch): + from loopx.control_plane.testing.model_tool_behavior import scripted_exec_tool_response + from loopx.control_plane.testing.selected_todo_tool_behavior import ( + DoubaoSelectedTodoToolBehaviorActor, _build_fixture, + ) + + monkeypatch.delenv("LOOPX_CONTEXT_DELIVERY", raising=False) + fixture = _build_fixture(tmp_path / "oracle") + requests = [] + + def choose_response(**kwargs): + request = json.loads(kwargs["body"]) + requests.append(request) + if len(requests) == 1: + command = fixture.quota_guard_command.replace('"${LOOPX_TURN:?}"', "native-reader-turn") + elif len(requests) == 2: + navigation = json.loads(request["messages"][-1]["content"]) + assert navigation["schema_version"] == "context_capture_navigation_v1" + assert "interaction_contract" not in navigation + command = navigation["recovery"]["command"] + ( + " --select /selected_todo --select /interaction_contract/cli_channel") + else: + page = json.loads(request["messages"][-1]["content"]) + assert page["ok"] and page["eof"] + values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} + assert values["/selected_todo"]["todo_id"] == "todo_portfolio001" + assert values["/interaction_contract/cli_channel"].get("selection_required", False) is False + command = "cat fixture/selected-lane.json" + return scripted_exec_tool_response(f"read-{len(requests)}", command) + + def transport(**kwargs): + try: + return choose_response(**kwargs) + except Exception as exc: + pytest.fail(f"scripted native response: {type(exc).__name__}: {exc}") + + result = DoubaoSelectedTodoToolBehaviorActor(api_key="test-only-placeholder", transport=transport).qualify( + qualification_id="native-progressive-selected-action", fixture_root=tmp_path / "actor") + assert result["qualification_passed"] is True, result + assert result["observed_tool_sequence"] == ["quota_should_run", "context_read", "selected_action"] + assert len(requests) == 3 diff --git a/tests/control_plane/test_quota_shared_json_identity.py b/tests/control_plane/test_quota_shared_json_identity.py index b89f648146..8156acdd00 100644 --- a/tests/control_plane/test_quota_shared_json_identity.py +++ b/tests/control_plane/test_quota_shared_json_identity.py @@ -3,6 +3,7 @@ import json import math +import os from copy import deepcopy import pytest @@ -130,7 +131,11 @@ def emit_with_distinct_sources(payload, args, **kwargs): if native: context = tmp_path / "native-context" context.mkdir(mode=0o700) - args += ["--context-delivery", "on", "--codex-context-dir", str(context)] + args += ["--codex-context-dir", str(context)] # Exercise the supported-host default. + if os.name == "nt": + args += ["--context-delivery", "on"] # Windows retains explicit activation. + else: + args += ["--context-delivery", "off"] code, emitted = call(*args) assert code == 0 and emitted["ok"] and len(observations) == 1, emitted original_bytes = (capture / "decision.json").read_bytes() @@ -138,22 +143,34 @@ def emit_with_distinct_sources(payload, args, **kwargs): before_receipts = fixture._heartbeat_receipt_count(runtime, "source-sharing-turn") assert before_receipts == 1 and fixture._spend_run_count(runtime) == 0 if native: - from pathlib import Path - + from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference recovery = emitted["recovery"] - view = json.loads(Path(recovery["path"]).read_bytes()) + assert emitted["schema_version"] == "context_capture_navigation_v1" + reader = recovery["reference_file"] + offset, bodies = 0, [] + while True: + code, page = call("quota", "context-read", "--reference-file", reader, + "--offset", str(offset), "--page-bytes", "64000", "--max-page-output-bytes", "64000") + assert code == 0 and page["ok"] and page["offset"] == offset, page + bodies.append(page["content"].encode("utf-8")) + offset += len(bodies[-1]) + if page["eof"]: + assert page["next_offset"] is None and offset == page["total_bytes"] + break + assert page["next_offset"] == offset + view = json.loads(b"".join(bodies)) pointers = ["/vision_continuation_audit/sharing_fixture", "/goal_frontier_projection/vision_continuation_audit/sharing_fixture", "/interaction_contract/agent_channel/vision_continuation_audit/sharing_fixture", "/interaction_contract/cli_channel/vision_continuation_audit/sharing_fixture"] selects = [arg for pointer in pointers for arg in ("--select", pointer)] - code, page = call("quota", "context-read", "--capture-file", recovery["path"], - "--capture-sha256", recovery["sha256"], *selects) + code, page = call("quota", "context-read", "--reference-file", reader, *selects) assert code == 0 and page["ok"] and page["eof"], page values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} for pointer in pointers: _assert_scalar(values[pointer]["observation"], True if pointer == pointers[0] else 1) - code, refused = call("quota", "context-read", "--capture-file", recovery["path"], + _, original_reference = read_context_read_reference(reader) + code, refused = call("quota", "context-read", "--capture-file", original_reference["path"], "--capture-sha256", "0" * 64, *selects) assert code == 1 and refused["ok"] is False else: diff --git a/tests/control_plane/test_replan_semantic_action_behavior.py b/tests/control_plane/test_replan_semantic_action_behavior.py index d47997db3f..21db0b4146 100644 --- a/tests/control_plane/test_replan_semantic_action_behavior.py +++ b/tests/control_plane/test_replan_semantic_action_behavior.py @@ -10,10 +10,13 @@ import loopx.control_plane.testing.replan_semantic_action_behavior as replan_behavior from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, ScriptedAssistantAction, - ScriptedDoubaoExecTransport, + ScriptedDoubaoExecTransport as RawScriptedDoubaoExecTransport, ScriptedExecToolAction, + replan_view_command_tokens, ) +from loopx.control_plane.quota.turn_envelope import _turn_envelope_result from loopx.control_plane.testing.replan_semantic_action_behavior import ( DoubaoReplanSemanticActionBehaviorActor, _build_fixture, @@ -21,22 +24,214 @@ ) +def _context_aware_transport( + actions: list[object], + *, + capture_pointers: tuple[str, ...] = (), +) -> RawScriptedDoubaoExecTransport: + """Script the intended work while preserving native progressive reads.""" + + remaining = list(actions) + writeback_command: str | None = None + review_read = False + work_observed = False + writeback_read = False + + def choose(request: Mapping[str, object]) -> object: + nonlocal review_read, work_observed, writeback_command, writeback_read + messages = request.get("messages") + latest = messages[-1] if isinstance(messages, list) and messages else None + raw: object = None + if isinstance(latest, Mapping) and latest.get("role") == "tool": + try: + raw = json.loads(str(latest.get("content") or "")) + except json.JSONDecodeError: + raw = None + if ( + isinstance(raw, Mapping) + and raw.get("schema_version") == "context_capture_navigation_v1" + ): + recovery = raw.get("recovery") + assert isinstance(recovery, Mapping) + command = str(recovery.get("command") or "") + assert command + pointers = ( + "/autonomous_replan_obligation", + "/replan_action_packet", + *capture_pointers, + ) + return ScriptedExecToolAction( + command + "".join(f" --select {pointer}" for pointer in pointers) + ) + if isinstance(raw, Mapping) and isinstance(raw.get("content"), str): + page = json.loads(raw["content"]) + selections = page.get("selections") if isinstance(page, Mapping) else None + if isinstance(selections, list): + values = { + row["pointer"]: row["value"] + for row in selections + if isinstance(row, Mapping) + and isinstance(row.get("pointer"), str) + } + obligation = values.get("/autonomous_replan_obligation") + action_packet = values.get("/replan_action_packet") + if isinstance(action_packet, Mapping) and isinstance( + action_packet.get("writeback"), str + ): + writeback_command = str(action_packet["writeback"]) + if isinstance(obligation, Mapping) and isinstance( + obligation.get("context_read"), str + ): + return ScriptedExecToolAction(str(obligation["context_read"])) + if isinstance(raw, Mapping) and raw.get("stage") == "review": + review_read = True + elif isinstance(raw, Mapping) and raw.get("stage") == "writeback": + writeback_read = True + elif review_read and isinstance(raw, Mapping): + work_observed = True + if not remaining: + raise AssertionError("context-aware scripted transport exhausted its actions") + next_action = remaining[0] + next_name = getattr(next_action, "__name__", "") + next_command = str(getattr(next_action, "command", "")) + read_action = next_name in { + "_inline_explore_context_action", + "_explore_context_action", + "wrong_scope", + } + if ( + writeback_command + and review_read + and work_observed + and not writeback_read + and ( + (callable(next_action) and not read_action) + or "refresh-state --help" in next_command + ) + ): + return ScriptedExecToolAction(writeback_command) + action = remaining.pop(0) + return action(request) if callable(action) else action + + return RawScriptedDoubaoExecTransport([choose] * 12) + + +def _work_sequence(receipt: Mapping[str, object]) -> list[str]: + sequence = receipt.get("observed_tool_sequence") + assert isinstance(sequence, list) + return [ + str(kind) + for kind in sequence + if kind not in {"context_read", "replan_context_read"} + ] + + def _latest_quota_packet(request: Mapping[str, object]) -> dict[str, object]: messages = request.get("messages") if not isinstance(messages, list) or not messages: raise AssertionError("model request must include a real quota result") + replan_view: Mapping[str, object] | None = None + writeback_view: Mapping[str, object] | None = None for message in reversed(messages): if not isinstance(message, Mapping) or message.get("role") != "tool": continue try: - packet = json.loads(str(message.get("content") or "")) + packet = QuotaToolContext.from_output(str(message.get("content") or "")).packet except json.JSONDecodeError: continue + if ( + isinstance(packet, Mapping) + and packet.get("stage") == "review" + and isinstance(packet.get("uncovered_frontier"), Mapping) + ): + replan_view = packet + continue + if ( + isinstance(packet, Mapping) + and packet.get("stage") == "writeback" + and isinstance(packet.get("writeback_contract"), Mapping) + ): + writeback_view = packet + continue + if ( + isinstance(packet, Mapping) + and packet.get("schema_version") == "context_capture_selection_page_v1" + and isinstance(packet.get("content"), str) + ): + selected = json.loads(packet["content"]) + rows = selected.get("selections") if isinstance(selected, Mapping) else None + if isinstance(rows, list): + packet = { + str(row["pointer"]).removeprefix("/"): row["value"] + for row in rows + if isinstance(row, Mapping) + and isinstance(row.get("pointer"), str) + and "value" in row + } if isinstance(packet, dict) and packet.get("replan_action_packet"): + if replan_view is not None: + action = dict(packet["replan_action_packet"]) + action["uncovered_frontier"] = replan_view["uncovered_frontier"] + if writeback_view is not None: + action["writeback_contract"] = writeback_view[ + "writeback_contract" + ] + packet["replan_action_packet"] = action return packet raise AssertionError("semantic action must follow a real quota packet") +def test_progressive_replan_reader_preserves_semantic_observation( + tmp_path: Path, +) -> None: + fixture = _build_fixture(tmp_path / "oracle") + output = _execute_loopx( + fixture.quota_guard_command, + fixture=fixture, + turn_instance_id="progressive-replan-reader", + ) + context = QuotaToolContext.from_output(output) + delivered_obligation = context.packet.get("autonomous_replan_obligation") + progressive = ( + context.packet + if isinstance(delivered_obligation, Mapping) + and isinstance(delivered_obligation.get("context_read"), str) + else _turn_envelope_result( + "replan_cli_view", + payload=context.packet, + ) + ) + obligation = progressive["autonomous_replan_obligation"] + assert isinstance(obligation, Mapping) + command = str(obligation["context_read"]) + assert replan_view_command_tokens(command) is not None + state = replan_behavior._QualificationState( + fixture=fixture, + messages=[], + steps=[], + turn_instance_id="progressive-replan-reader", + quota_packet=progressive, + quota_context=context, + seen_quota=True, + delivery_context_read=True, + ) + + readback = json.loads( + replan_behavior._handle_replan_context_read(command, state) + ) + + assert readback["stage"] == "review" + assert state.replan_context_read is True + assert state.quota_observation is not None + assert state.quota_observation["context_delivery"] == "progressive_replan_view" + assert "new_surface" in state.quota_observation["required_semantic_outcomes"] + with pytest.raises(ValueError, match="unexpected_replan_context_read"): + replan_behavior._handle_replan_context_read( + command.replace("--replan-stage review", "--replan-stage invalid"), + state, + ) + + def _semantic_action( request: Mapping[str, object], *, @@ -137,6 +332,43 @@ def _explore_context_action( return ScriptedExecToolAction(command=read["command"]) +def test_context_delivery_off_keeps_original_ordinary_call_budget( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", "off") + fixture = _build_fixture(tmp_path / "oracle") + transport = _context_aware_transport( + [ + ScriptedExecToolAction(command=fixture.quota_guard_command), + *[ + ScriptedExecToolAction(command="cat replan-frontier.json") + for _ in range(5) + ], + ScriptedExecToolAction(command="cat fixture/permission-config.json"), + _semantic_action, + ] + ) + + receipt = DoubaoReplanSemanticActionBehaviorActor( + api_key="test-only-placeholder", + transport=transport, + ).qualify( + qualification_id="replan-inline-ordinary-budget", + fixture_root=tmp_path / "actor", + ) + + assert receipt["qualification_passed"] is False + assert receipt["failure_code"] == "tool_call_budget_exhausted" + assert receipt["tool_call_count"] == receipt["tool_call_limit"] == 7 + assert receipt["ordinary_tool_call_limit"] == 7 + assert receipt["native_context_read_call_limit"] == 0 + assert "context_read" not in receipt["observed_tool_sequence"] + assert "replan_context_read" not in receipt["observed_tool_sequence"] + assert receipt["semantic_action_accepted"] is False + assert len(transport.requests) == 7 + + def _composition_successor_action( request: Mapping[str, object], @@ -150,23 +382,27 @@ def _composition_successor_action( action = packet["replan_action_packet"] assert isinstance(action, Mapping) bounded = action.get("bounded_frontier") - assert isinstance(bounded, Mapping) - assert bounded["gap_id"] == selected_gap["gap_id"] + if bounded is not None: + assert isinstance(bounded, Mapping) + assert bounded["gap_id"] == selected_gap["gap_id"] writeback = action.get("writeback_contract") assert isinstance(writeback, Mapping) command = str(writeback.get("successor_command") or "") assert command + assert str(selected_gap["experiment_node_ref"]) in command command_tokens = shlex.split(command) assert command_tokens[0] == "loopx" command_tokens[1:1] = ["--format", "json", "--registry", "ignored"] return ScriptedExecToolAction(command=shlex.join(command_tokens)) +@pytest.mark.parametrize("context_mode", ["on", "off"]) def test_real_tool_loop_chooses_and_persists_semantic_replan_action( - tmp_path: Path, + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, context_mode: str, ) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", context_mode) fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command="date -Iseconds"), ScriptedExecToolAction(command=fixture.quota_guard_command), @@ -184,14 +420,26 @@ def test_real_tool_loop_chooses_and_persists_semantic_replan_action( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "clock", "quota_should_run", "workspace_read", "workspace_read", "semantic_replan_writeback", ] - assert receipt["context_delivery"] == "host_projected" + assert ("context_read" in receipt["observed_tool_sequence"]) is (context_mode == "on") + assert receipt["context_delivery"] in { + "host_projected", + "progressive_replan_view", + } + if receipt["context_delivery"] == "progressive_replan_view": + assert "replan_context_read" in receipt["observed_tool_sequence"] + assert receipt["tool_call_limit"] == 10 + assert receipt["native_context_read_call_limit"] == 3 + else: + assert receipt["tool_call_limit"] == 7 + assert receipt["native_context_read_call_limit"] == 0 + assert receipt["ordinary_tool_call_limit"] == 7 assert receipt["semantic_action_accepted"] is True assert "new_surface" in receipt["selected_semantic_outcomes"] assert "new_surface" in receipt["required_semantic_outcomes"] @@ -216,25 +464,34 @@ def test_real_tool_loop_chooses_and_persists_semantic_replan_action( assert "execute the projected typed writeback" in system_prompt assert "execute exactly one loopx refresh-state command" in system_prompt assert "do not create a Todo, chain commands, or combine actions" in system_prompt + assert "If the current replan_action_packet supplies a writeback command" in system_prompt + assert "Otherwise use its inline writeback_contract" in system_prompt + delivered_action = _latest_quota_packet(transport.requests[-1])["replan_action_packet"] + assert ("writeback" in delivered_action) is (context_mode == "on") - quota_packet = json.loads(transport.requests[2]["messages"][-1]["content"]) + quota_packet = QuotaToolContext.from_output(transport.requests[2]["messages"][-1]["content"]).packet assert quota_packet.get("required_reads") in (None, []) assert "successor Todo" not in quota_packet["active_state_next_action"] - assert quota_packet["autonomous_replan_obligation"]["replan_context"][ - "delivery_receipt" - ]["status"] == "delivered" + delivered_obligation = quota_packet["autonomous_replan_obligation"] + if "replan_context" in delivered_obligation: + assert delivered_obligation["replan_context"]["delivery_receipt"][ + "status" + ] == "delivered" + else: + assert replan_view_command_tokens(delivered_obligation["context_read"]) assert quota_packet["replan_action_packet"]["obligation_id"] == receipt[ "replan_obligation_id" ] - assert quota_packet["replan_action_packet"]["writeback_contract"] == {} + assert quota_packet["replan_action_packet"].get("writeback_contract", {}) == {} assert quota_packet["interaction_contract"]["agent_channel"][ "primary_action" ] == "produce one typed outcome from the host-projected replan action packet" - next_cli_actions = quota_packet["interaction_contract"]["cli_channel"][ - "next_cli_actions" - ] - assert any("refresh-state" in action for action in next_cli_actions) - assert not any("successor_command" in action for action in next_cli_actions) + next_cli_actions = quota_packet["interaction_contract"]["cli_channel"].get( + "next_cli_actions", [] + ) + if next_cli_actions: + assert any("refresh-state" in action for action in next_cli_actions) + assert not any("successor_command" in action for action in next_cli_actions) index_path = ( tmp_path @@ -257,7 +514,7 @@ def test_semantic_cli_execution_failure_retains_command_kind( monkeypatch: pytest.MonkeyPatch, ) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -290,7 +547,7 @@ def test_real_tool_loop_allows_bounded_refresh_state_help( tmp_path: Path, ) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -310,7 +567,7 @@ def test_real_tool_loop_allows_bounded_refresh_state_help( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "quota_should_run", "workspace_read", "workspace_read", @@ -324,7 +581,7 @@ def test_real_tool_loop_preserves_projected_targets_through_content_pipelines( tmp_path: Path, ) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction( @@ -345,7 +602,7 @@ def test_real_tool_loop_preserves_projected_targets_through_content_pipelines( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "quota_should_run", "workspace_read", "workspace_read", @@ -358,7 +615,7 @@ def test_refresh_state_help_requires_projected_frontier_reads( tmp_path: Path, ) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="loopx refresh-state --help"), @@ -380,7 +637,7 @@ def test_model_exits_after_replan_confirms_goal_coverage_is_exhausted( tmp_path: Path, ) -> None: fixture = _build_fixture(tmp_path / "oracle", exhausted_frontier=True) - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -398,7 +655,7 @@ def test_model_exits_after_replan_confirms_goal_coverage_is_exhausted( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "quota_should_run", "workspace_read", "semantic_replan_writeback", @@ -472,13 +729,14 @@ def test_real_tool_loop_selects_composition_gap_and_creates_bound_successor( tmp_path / "oracle", composition_frontier=True, ) - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), _inline_explore_context_action, ScriptedExecToolAction(command="cat fixture/permission-config.json"), _composition_successor_action, - ] + ], + capture_pointers=("/interaction_contract", "/bounded_research_frontier"), ) receipt = DoubaoReplanSemanticActionBehaviorActor( api_key="test-only-placeholder", @@ -490,7 +748,7 @@ def test_real_tool_loop_selects_composition_gap_and_creates_bound_successor( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "quota_should_run", "workspace_read", "workspace_read", @@ -507,7 +765,7 @@ def test_real_tool_loop_selects_composition_gap_and_creates_bound_successor( ) assert successor_reentry["composition_status"] == "scheduled" - quota_packet = json.loads(transport.requests[1]["messages"][-1]["content"]) + quota_packet = QuotaToolContext.from_output(transport.requests[1]["messages"][-1]["content"]).packet # The canonical context owner has already delivered the scoped hook body. # Displayed source commands are provenance, not another required tool call. _explore_context_action(transport.requests[1]) @@ -520,9 +778,13 @@ def test_real_tool_loop_selects_composition_gap_and_creates_bound_successor( assert selected_gap["experiment_node_ref"] == ( "experiment-permission-composition" ) - assert quota_packet["replan_action_packet"]["bounded_frontier"][ - "experiment_node_ref" - ] == selected_gap["experiment_node_ref"] + action_packet = quota_packet["replan_action_packet"] + if "bounded_frontier" in action_packet: + assert action_packet["bounded_frontier"]["experiment_node_ref"] == ( + selected_gap["experiment_node_ref"] + ) + else: + assert replan_view_command_tokens(action_packet["writeback"]) @pytest.mark.parametrize("attempt", ["repeat", "wrong_scope"]) @@ -543,7 +805,11 @@ def wrong_scope(request: Mapping[str, object]) -> ScriptedExecToolAction: ) actions.append(wrong_scope) receipt = DoubaoReplanSemanticActionBehaviorActor( - api_key="test-only-placeholder", transport=ScriptedDoubaoExecTransport(actions), + api_key="test-only-placeholder", + transport=_context_aware_transport( + actions, + capture_pointers=("/interaction_contract", "/bounded_research_frontier"), + ), ).qualify( qualification_id=f"composition-read-{attempt}", fixture_root=tmp_path / "actor", composition_frontier=True, @@ -580,7 +846,7 @@ def test_required_explore_read_uses_nested_interaction_contract() -> None: def test_action_outside_observed_frontier_is_rejected(tmp_path: Path) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -619,7 +885,7 @@ def repeated_action(request: Mapping[str, object]) -> ScriptedExecToolAction: "--no-global-sync --suppress-external-sinks" ) - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -644,7 +910,7 @@ def test_manual_evidence_read_is_context_not_replan_completion( tmp_path: Path, ) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction( @@ -670,7 +936,7 @@ def test_manual_evidence_read_is_context_not_replan_completion( def test_semantic_action_must_follow_real_frontier_read(tmp_path: Path) -> None: fixture = _build_fixture(tmp_path / "oracle") - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), _semantic_action, @@ -709,7 +975,7 @@ def create_bound_successor( f"--replan-obligation-id {obligation['obligation_id']}" ) - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction(command=fixture.quota_guard_command), ScriptedExecToolAction(command="cat replan-frontier.json"), @@ -726,7 +992,7 @@ def create_bound_successor( ) assert receipt["qualification_passed"] is True, receipt - assert receipt["observed_tool_sequence"] == [ + assert _work_sequence(receipt) == [ "quota_should_run", "workspace_read", "workspace_read", @@ -789,13 +1055,11 @@ def test_successor_dry_run_shares_quota_agent_scope_for_user_gates( / "ACTIVE_GOAL_STATE.md" ) - quota = json.loads( - _execute_loopx( + quota = QuotaToolContext.from_output(_execute_loopx( fixture.quota_guard_command, fixture=fixture, turn_instance_id="agent-scoped-quota-turn", - ) - ) + )).packet obligation_id = quota["autonomous_replan_obligation"]["obligation_id"] before = state_path.read_text(encoding="utf-8") dry_run = json.loads( @@ -858,13 +1122,11 @@ def test_untyped_successor_is_rejected_before_todo_mutation( / "replan-semantic-action-fixture" / "ACTIVE_GOAL_STATE.md" ) - quota = json.loads( - _execute_loopx( + quota = QuotaToolContext.from_output(_execute_loopx( fixture.quota_guard_command, fixture=fixture, turn_instance_id="untyped-successor-turn", - ) - ) + )).packet obligation_id = quota["autonomous_replan_obligation"]["obligation_id"] before = state_path.read_text(encoding="utf-8") @@ -884,7 +1146,7 @@ def test_untyped_successor_is_rejected_before_todo_mutation( def test_semantic_writeback_before_quota_is_rejected(tmp_path: Path) -> None: - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ ScriptedExecToolAction( command=( @@ -912,7 +1174,7 @@ def test_semantic_writeback_before_quota_is_rejected(tmp_path: Path) -> None: def test_prose_intent_without_typed_action_is_rejected(tmp_path: Path) -> None: - transport = ScriptedDoubaoExecTransport( + transport = _context_aware_transport( [ScriptedAssistantAction(content="I would explore a new surface next.")] ) receipt = DoubaoReplanSemanticActionBehaviorActor( diff --git a/tests/control_plane/test_required_vision_closeout_behavior.py b/tests/control_plane/test_required_vision_closeout_behavior.py index debcf2a5a7..2e1f800bbb 100644 --- a/tests/control_plane/test_required_vision_closeout_behavior.py +++ b/tests/control_plane/test_required_vision_closeout_behavior.py @@ -9,6 +9,7 @@ import pytest from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, EXEC_COMMAND_TOOL, ScriptedAssistantAction, ScriptedDoubaoExecTransport, ScriptedExecToolAction, ) from loopx.control_plane.testing.replan_semantic_action_behavior import ( @@ -21,14 +22,29 @@ def _packet(request: Mapping[str, Any]) -> dict[str, Any]: + packet: dict[str, Any] = {} for message in request["messages"]: if message["role"] == "tool": try: value = json.loads(message["content"]) except json.JSONDecodeError: continue + if not isinstance(value, dict): + continue + if value.get("schema_version") in {"context_capture_selection_page_v1", "context_capture_page_v1"}: + selected = json.loads(value["content"]) + value = {} + for row in selected["selections"]: + if row["pointer"] == "/interaction_contract/cli_channel": + value["interaction_contract"] = {"cli_channel": row["value"]} + else: + value[row["pointer"].removeprefix("/")] = row["value"] if "replan_action_packet" in value: - return value + packet = value + elif value.get("stage") == "writeback" and packet: + packet["interaction_contract"]["cli_channel"].update(value["writeback"]) + if packet: + return packet raise AssertionError("quota was not observed") @@ -62,16 +78,130 @@ def projected_refresh(request: Mapping[str, Any]) -> ScriptedExecToolAction: def projected_spend(request: Mapping[str, Any]) -> ScriptedExecToolAction: - return ScriptedExecToolAction(_packet(request)["interaction_contract"]["cli_channel"]["next_cli_actions"][1]) + cli = _packet(request)["interaction_contract"]["cli_channel"] + for step in cli.get("settlement_plan", {}).get("ordered_steps", []): + if step["kind"] == "quota_spend": + return ScriptedExecToolAction(step["command_template"]) + return ScriptedExecToolAction(cli["next_cli_actions"][1]) + + +def _vision_transport(actions: list[Any]) -> ScriptedDoubaoExecTransport: + remaining = list(actions) + writeback: str | None = None + review_read = False + work_emitted = False + writeback_read = False + + def choose(request: Mapping[str, Any]) -> Any: + nonlocal writeback, review_read, work_emitted, writeback_read + latest = request["messages"][-1] + try: + value = json.loads(latest["content"]) if latest["role"] == "tool" else None + except json.JSONDecodeError: + value = None + if isinstance(value, dict): + if value.get("schema_version") == "context_capture_navigation_v1": + return ScriptedExecToolAction(value["recovery"]["command"] + + " --select /autonomous_replan_obligation --select /replan_action_packet" + " --select /interaction_contract/cli_channel") + if value.get("schema_version") in {"context_capture_selection_page_v1", "context_capture_page_v1"}: + packet = _packet(request) + writeback = packet["replan_action_packet"].get("writeback") + reader = packet["autonomous_replan_obligation"].get("context_read") + if reader: + return ScriptedExecToolAction(reader) + if value.get("stage") == "review": + review_read = True + if value.get("stage") == "writeback": + writeback_read = True + if value.get("error_code") == "shell_nonzero" and writeback and not review_read: + return ScriptedAssistantAction("The projected reader did not return its review.") + if writeback and review_read and work_emitted and not writeback_read: + return ScriptedExecToolAction(writeback) + if not remaining: + return ScriptedAssistantAction("No further scripted action.") + action = remaining.pop(0) + if review_read: + work_emitted = True + return action(request) if callable(action) else action + + return ScriptedDoubaoExecTransport([choose] * 40) def _qualify(tmp_path: Path, actions: list[Any]) -> dict[str, Any]: - transport = ScriptedDoubaoExecTransport(actions) + transport = _vision_transport(actions) return DoubaoReplanSemanticActionBehaviorActor(api_key="test-only-placeholder", transport=transport).qualify( qualification_id="native-vision-closeout", fixture_root=tmp_path / "actor", required_vision=True, ) +def test_projected_module_readers_complete_native_shell_closeout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", "on") + fixture = _build_fixture(tmp_path / "oracle", required_vision=True) + transport = _vision_transport([ + ScriptedExecToolAction(fixture.quota_guard_command), + ScriptedExecToolAction("cat fixture/permission-config.json"), + vision_patch_action, projected_refresh, projected_spend, + ]) + result = DoubaoReplanSemanticActionBehaviorActor( + api_key="test-only-placeholder", transport=transport, + ).qualify(qualification_id="native-module-reader-closeout", + fixture_root=tmp_path / "actor", required_vision=True) + outputs = [json.loads(message["content"]) + for message in transport.requests[-1]["messages"] + if message["role"] == "tool" and message["content"].startswith("{")] + assert [value["stage"] for value in outputs if "stage" in value] == ["review", "writeback"] + packet = _packet(transport.requests[-1]) + reader = packet["autonomous_replan_obligation"]["context_read"] + assert shlex.split(reader)[1:3] == ["-m", "loopx.entrypoint"] + assert result["qualification_passed"] is True, result + assert result["execution_host"] == "os_isolated_shell" + assert result["vision_closeout"]["settled"] is True + assert result["vision_closeout"]["spend_count"] == 1 + assert result["tool_call_limit"] == 40 + + +@pytest.mark.parametrize("change", ["source_hash", "decision_file", "other_module"]) +def test_native_shell_reader_bridge_does_not_expand_authority( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, change: str, +) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", "on") + fixture = _build_fixture(tmp_path / "oracle", required_vision=True) + + def capture(request: Mapping[str, Any]) -> ScriptedExecToolAction: + navigation = json.loads(request["messages"][-1]["content"]) + return ScriptedExecToolAction(navigation["recovery"]["command"] + + " --select /autonomous_replan_obligation --select /replan_action_packet" + " --select /interaction_contract/cli_channel") + + def invalid_reader(request: Mapping[str, Any]) -> ScriptedExecToolAction: + tokens = shlex.split(_packet(request)["autonomous_replan_obligation"]["context_read"]) + if change == "other_module": + tokens[2] = "loopx" + else: + option = "--source-hash" if change == "source_hash" else "--decision-file" + tokens[tokens.index(option) + 1] = "0" * 64 if change == "source_hash" else "foreign.json" + return ScriptedExecToolAction(shlex.join(tokens)) + + def rejected(request: Mapping[str, Any]) -> ScriptedAssistantAction: + error = json.loads(request["messages"][-1]["content"]) + assert error["exit_code"] != 0 + assert error["error_code"] == "shell_nonzero" + return ScriptedAssistantAction("No accepted reader or state mutation.") + + result = DoubaoReplanSemanticActionBehaviorActor( + api_key="test-only-placeholder", transport=ScriptedDoubaoExecTransport([ + ScriptedExecToolAction(fixture.quota_guard_command), capture, invalid_reader, rejected, + ]), + ).qualify(qualification_id="native-reader-boundary", + fixture_root=tmp_path / "actor", required_vision=True) + assert result["qualification_passed"] is False + assert result["semantic_action_accepted"] is False + assert not result.get("vision_closeout") + + @pytest.mark.parametrize("policy", ["as_needed", "repeat_until_closed"]) @pytest.mark.parametrize("compound", [False, True]) def test_native_shell_closes_real_cli_turn_without_command_rituals(tmp_path: Path, policy: str, compound: bool) -> None: @@ -110,7 +240,7 @@ def oversized(request: Mapping[str, Any]) -> ScriptedExecToolAction: return ScriptedExecToolAction(vision_patch_action(request).command.replace( "Reader is default; writing requires an explicit grant.", "x" * 700)) def correct(request: Mapping[str, Any]) -> ScriptedExecToolAction: - response = json.loads(request["messages"][-1]["content"]) + response = QuotaToolContext.from_output(request["messages"][-1]["content"]).packet assert response["exit_code"] != 0 assert "vision_budget_exceeded" in response["output"] return ScriptedExecToolAction(vision_patch_action(request).command.replace("evidence-permission-config", evidence)) @@ -138,7 +268,7 @@ def first_refresh_without_receipt(command: str, **kwargs: Any) -> str: return execute(command, **kwargs) def retry_after_error(request: Mapping[str, Any]) -> ScriptedExecToolAction: - feedback = json.loads(request["messages"][-1]["content"]) + feedback = QuotaToolContext.from_output(request["messages"][-1]["content"]).packet assert feedback["exit_code"] != 0 assert "vision_closeout_durable_writeback_missing" in feedback["output"] return projected_refresh(request) @@ -179,8 +309,12 @@ def refresh(request: Mapping[str, Any]) -> ScriptedExecToolAction: assert not (result.get("vision_closeout") or {}).get("settled") -@pytest.mark.parametrize("extra_reads,passed", [(35, True), (36, False)]) -def test_budget_boundary_still_requires_the_final_spend(tmp_path: Path, extra_reads: int, passed: bool) -> None: +@pytest.mark.parametrize("context_mode", ["on", "off"]) +@pytest.mark.parametrize("passed", [True, False]) +def test_budget_boundary_still_requires_the_final_spend(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, context_mode: str, passed: bool) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", context_mode) + # Native capture/review/writeback consume the original forty-call budget. + extra_reads = (32 if context_mode == "on" else 35) + (0 if passed else 1) fixture = _build_fixture(tmp_path / "oracle", required_vision=True) result = _qualify(tmp_path, [ ScriptedExecToolAction(fixture.quota_guard_command), @@ -193,7 +327,9 @@ def test_budget_boundary_still_requires_the_final_spend(tmp_path: Path, extra_re assert result["vision_closeout"]["settled"] is passed -def test_narrow_actor_retains_its_existing_budget_and_tool(tmp_path: Path) -> None: +@pytest.mark.parametrize("context_mode", ["on", "off"]) +def test_narrow_actor_retains_its_existing_budget_and_tool(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, context_mode: str) -> None: + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", context_mode) fixture = _build_fixture(tmp_path / "oracle") transport = ScriptedDoubaoExecTransport([ ScriptedExecToolAction(fixture.quota_guard_command), @@ -202,7 +338,9 @@ def test_narrow_actor_retains_its_existing_budget_and_tool(tmp_path: Path) -> No result = DoubaoReplanSemanticActionBehaviorActor(api_key="test-only-placeholder", transport=transport).qualify( qualification_id="narrow-unchanged", fixture_root=tmp_path / "actor", ) - assert result["tool_call_limit"] == result["tool_call_count"] == 7 + assert result["ordinary_tool_call_limit"] == result["tool_call_count"] == 7 + assert result["tool_call_limit"] == (10 if context_mode == "on" else 7) + assert len(transport.requests) == 7 assert transport.requests[0]["tools"] == [EXEC_COMMAND_TOOL] @@ -258,8 +396,9 @@ def kill_once(pid: int, sig: int) -> None: def test_actor_cannot_shadow_the_trusted_cli_in_its_writable_project(tmp_path: Path) -> None: fixture = _build_fixture(tmp_path / "oracle", required_vision=True) def check_real_cli(request: Mapping[str, Any]) -> ScriptedAssistantAction: - output = request["messages"][-1]["content"] - assert "loopx --help" in output and "SHADOWED_CLI" not in output + outputs = [message["content"] for message in request["messages"] if message["role"] == "tool"] + assert any("loopx --help" in output for output in outputs) + assert all("SHADOWED_CLI" not in output for output in outputs) return ScriptedAssistantAction("Only checking executor source provenance.") result = _qualify(tmp_path, [ ScriptedExecToolAction(fixture.quota_guard_command), diff --git a/tests/control_plane/test_scoped_gate_successor_tool_behavior.py b/tests/control_plane/test_scoped_gate_successor_tool_behavior.py index 41bf9118c9..aafa7ba825 100644 --- a/tests/control_plane/test_scoped_gate_successor_tool_behavior.py +++ b/tests/control_plane/test_scoped_gate_successor_tool_behavior.py @@ -6,6 +6,7 @@ from typing import Any from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, QUOTA_FIRST_TOOL_INSTRUCTION, scripted_assistant_response as _stop_response, scripted_exec_tool_response as _tool_response, @@ -85,7 +86,7 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: sort_keys=True, ) - quota_result = json.loads(requests[1]["messages"][-1]["content"]) + quota_result = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet assert quota_result["selected_todo"]["todo_id"] == ( SCOPED_GATE_SUCCESSOR_TODO_ID ) diff --git a/tests/control_plane/test_selected_todo_tool_behavior.py b/tests/control_plane/test_selected_todo_tool_behavior.py index 1484e6d480..d97c309baf 100644 --- a/tests/control_plane/test_selected_todo_tool_behavior.py +++ b/tests/control_plane/test_selected_todo_tool_behavior.py @@ -12,6 +12,7 @@ DoubaoActorTransportError, ) from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, DoubaoExecToolClient, QUOTA_FIRST_TOOL_INSTRUCTION, scripted_exec_tool_response as _tool_response, @@ -165,7 +166,7 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: assert "todo_portfolio001" not in first["messages"][1]["content"] assert "use selection_command when required" in first["messages"][1]["content"] - quota_result = json.loads(requests[2]["messages"][-1]["content"]) + quota_result = QuotaToolContext.from_output(requests[2]["messages"][-1]["content"]).packet assert quota_result["selected_todo"]["todo_id"] == "todo_portfolio001" assert "fixture/selected-lane.json" in quota_result["selected_todo"]["text"] @@ -209,11 +210,11 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: "quota_action_selection", "selected_action", ] - first_quota_result = json.loads(requests[1]["messages"][-1]["content"]) + first_quota_result = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet assert first_quota_result["interaction_contract"]["cli_channel"][ "selection_required" ] is True - quota_result = json.loads(requests[2]["messages"][-1]["content"]) + quota_result = QuotaToolContext.from_output(requests[2]["messages"][-1]["content"]).packet assert quota_result["selected_todo"]["todo_id"] == "todo_portfolio001" assert quota_result["selected_todo"]["selection_binding"] == ( "heartbeat_receipt" @@ -248,7 +249,7 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: '"${LOOPX_TURN:?}"', "turn-004" ) elif call_index == 2: - first_quota = json.loads(requests[1]["messages"][-1]["content"]) + first_quota = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet selection = first_quota["interaction_contract"]["cli_channel"][ "selection_command" ] @@ -285,7 +286,7 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: "quota_action_selection", "selected_action", ] - first_quota = json.loads(requests[1]["messages"][-1]["content"]) + first_quota = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet suggested_ids = { item["todo_id"] for item in first_quota["action_portfolio"]["suggested_actions"] @@ -301,9 +302,9 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: "candidate_scope": "current_authoritative_eligible_todos", "suggestions_exhaustive": False, } - discovery = json.loads(requests[2]["messages"][-1]["content"]) + discovery = QuotaToolContext.from_output(requests[2]["messages"][-1]["content"]).packet assert "todo_portfolio001" in json.dumps(discovery, sort_keys=True) - selected_quota = json.loads(requests[3]["messages"][-1]["content"]) + selected_quota = QuotaToolContext.from_output(requests[3]["messages"][-1]["content"]).packet assert selected_quota["selected_todo"]["todo_id"] == "todo_portfolio001" assert selected_quota["selected_todo"]["selection_binding"] == ( "heartbeat_receipt" @@ -349,7 +350,7 @@ def transport(**kwargs: Any) -> Mapping[str, Any]: assert receipt["selected_todo_id"] == SELECTED_TODO_TOOL_FIXTURE_TODO_ID initial_prompt = json.dumps(requests[0]["messages"], ensure_ascii=False) assert SELECTED_TODO_TOOL_FIXTURE_TODO_ID not in initial_prompt - first_quota = json.loads(requests[1]["messages"][-1]["content"]) + first_quota = QuotaToolContext.from_output(requests[1]["messages"][-1]["content"]).packet assert first_quota["action_portfolio"]["primary"]["todo_id"] == ( STALE_DONE_PRIMARY_TODO_ID ) diff --git a/tests/control_plane/test_terminal_settlement_tool_behavior.py b/tests/control_plane/test_terminal_settlement_tool_behavior.py index d7a44164bd..ec2a3ab1ff 100644 --- a/tests/control_plane/test_terminal_settlement_tool_behavior.py +++ b/tests/control_plane/test_terminal_settlement_tool_behavior.py @@ -12,6 +12,7 @@ import pytest from loopx.control_plane.testing.model_tool_behavior import ( + QuotaToolContext, QUOTA_FIRST_TOOL_INSTRUCTION, ScriptedAssistantAction, ScriptedDoubaoExecTransport, @@ -31,7 +32,7 @@ def _latest_quota_packet(request: Mapping[str, Any]) -> dict[str, Any]: if message.get("role") != "tool": continue try: - payload = json.loads(message["content"]) + payload = QuotaToolContext.from_output(message["content"]).packet except (json.JSONDecodeError, TypeError): continue if isinstance(payload, dict) and "interaction_contract" in payload: @@ -46,7 +47,7 @@ def _writeback_action(request: Mapping[str, Any]) -> ScriptedExecToolAction: if message["role"] != "tool": continue try: - result = json.loads(message["content"]) + result = QuotaToolContext.from_output(message["content"]).packet except json.JSONDecodeError: continue if isinstance(result, dict) and result.get("settlement_plan") and result.get("next_cli_actions"): @@ -95,7 +96,7 @@ def _latest_reentry_transition(request: Mapping[str, Any]) -> dict[str, Any]: if message.get("role") != "tool": continue try: - payload = json.loads(message["content"]) + payload = QuotaToolContext.from_output(message["content"]).packet except (json.JSONDecodeError, TypeError): continue transition = ( @@ -209,7 +210,7 @@ def test_real_tool_loop_settles_final_todo_after_spend( TERMINAL_SETTLEMENT_FIXTURE_TODO_ID ) if ordinary_completion: - results = [json.loads(message["content"]) for message in transport.requests[-1]["messages"] + results = [QuotaToolContext.from_output(message["content"]).packet for message in transport.requests[-1]["messages"] if message["role"] == "tool" and message["content"].startswith("{")] completed = next(result for result in results if result.get("status") == "done") assert completed["settlement_progress"]["next_step"] == "durable_writeback" @@ -386,7 +387,7 @@ def authority_files(): def writeback(request, *, omit_scope=False): if correct_input and not omit_scope: - rejected = json.loads(request["messages"][-1]["content"]) + rejected = QuotaToolContext.from_output(request["messages"][-1]["content"]).packet assert rejected["ok"] is False and rejected["appended"] is False assert rejected["error_code"] == "progress_observation_invalid_arguments" assert "--progress-coverage-scope-id" in rejected["error"] @@ -446,7 +447,7 @@ def authority_files() -> dict[str, bytes]: def draft(request, *, corrected=False): if corrected: - rejected = json.loads(request["messages"][-1]["content"]) + rejected = QuotaToolContext.from_output(request["messages"][-1]["content"]).packet assert rejected["ok"] is False and rejected["appended"] is False assert authority_files() == before_rejection vision = {