diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_tool_context.py b/loopx/capabilities/benchmark_toolkit/trajectory_tool_context.py index 5bbe4c6e06..0cf8c0647e 100644 --- a/loopx/capabilities/benchmark_toolkit/trajectory_tool_context.py +++ b/loopx/capabilities/benchmark_toolkit/trajectory_tool_context.py @@ -35,19 +35,26 @@ def _argv(command: str, depth: int = 0) -> list[str] | None: def _literal_exec_argv(script: object) -> list[str] | None: """Recognize one complete literal exec wrapper, never arbitrary JavaScript. - Codex's custom exec tool commonly wraps one exec_command in const/await/text. + Accept the stored-result and directly returned forms of one exec_command. Accept only JSON-valued object properties and one return; calls, spreads, interpolation, multiple commands and computed properties remain unknown. """ if not isinstance(script, str): return None - match = re.fullmatch( - r"\s*const\s+([A-Za-z_]\w*)\s*=\s*await\s+tools\.exec_command\(\s*\{(.*)\}\s*\)\s*;" - r"\s*text\(\s*\1(?:\.output)?\s*\)\s*;?\s*", script, re.S, + invocation = r"await\s+tools\.exec_command\(\s*\{(?P.*)\}\s*\)" + wrappers = ( + r"const\s+(?P[A-Za-z_]\w*)\s*=\s*" + invocation + + r"\s*;\s*text\(\s*(?P=result)(?:\.output)?\s*\)", + r"text\(\s*" + invocation + r"\s*\)", + r"text\(\s*\(\s*" + invocation + r"\s*\)\.output\s*\)", ) - if match is None: + for wrapper in wrappers: + match = re.fullmatch(wrapper + r"\s*;?", script.strip(), re.S) + if match is not None: + break + else: return None - rest, values = match[2].strip(), {} + rest, values = match["options"].strip(), {} decoder = json.JSONDecoder() while rest: key = re.match(r'(?:"([A-Za-z_]\w*)"|([A-Za-z_]\w*))\s*:\s*', rest) diff --git a/tests/capabilities/test_trajectory_input_capture.py b/tests/capabilities/test_trajectory_input_capture.py index 3753ae217f..a2e4c59817 100644 --- a/tests/capabilities/test_trajectory_input_capture.py +++ b/tests/capabilities/test_trajectory_input_capture.py @@ -107,11 +107,15 @@ def test_invalid_optional_observation_cannot_erase_native_counters_or_escape_roo assert usage["projection"]["loopx_input"]["status"] == "unavailable" -def test_tool_inventory_is_opt_in_separate_and_retains_native_counter_identity(tmp_path): +@pytest.mark.parametrize('wrapper', [ + 'const r = await tools.exec_command({cmd: "loopx quota context-read"}); text(r.output);', + 'text(await tools.exec_command({cmd: "loopx quota context-read"}))', + 'text((await tools.exec_command({cmd: "loopx quota context-read"})).output)', +]) +def test_tool_inventory_is_opt_in_separate_and_retains_native_counter_identity(tmp_path, wrapper): _, rows = native_trace(tmp_path) tool = [{'type': 'response_item', 'payload': {'type': 'custom_tool_call', - 'name': 'exec', 'call_id': 'tool', 'input': - 'const r = await tools.exec_command({cmd: "loopx quota context-read"}); text(r.output);'}}, + 'name': 'exec', 'call_id': 'tool', 'input': wrapper}}, {'type': 'response_item', 'payload': {'type': 'custom_tool_call_output', 'call_id': 'tool', 'output': '状态已保存'}}] path = tmp_path / 'usage.jsonl' @@ -128,10 +132,17 @@ def test_tool_inventory_is_opt_in_separate_and_retains_native_counter_identity(t assert path.read_bytes() == raw -def test_real_cli_registers_recorded_component_without_model_execution(tmp_path): +@pytest.mark.parametrize("direct_call", [False, True]) +def test_real_cli_registers_recorded_component_without_model_execution(tmp_path, direct_call): from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import inspect_pack - source, _ = native_trace(tmp_path) + source, rows = native_trace(tmp_path) + if direct_call: + call = {"type": "custom_tool_call", "call_id": "tool", "name": "exec", + "input": 'text(await tools.exec_command({cmd: "loopx status"}))'} + reply = {"type": "custom_tool_call_output", "call_id": "tool", "output": "saved"} + items = [{"type": "response_item", "payload": item} for item in (call, reply)] + (tmp_path / "usage.jsonl").write_text("\n".join(map(json.dumps, rows[:2] + items + rows[2:]))) evidence = "Synthetic fixed negative result." (tmp_path / "evidence.txt").write_text(evidence) data = {"schema_version": "trajectory_assessment_input_v1", @@ -155,6 +166,11 @@ def test_real_cli_registers_recorded_component_without_model_execution(tmp_path) registered = json.loads(result.stdout) view = inspect_pack(runtime_root=runtime, owner_goal_id="research", pack_id=registered["pack_id"]) assert view["solver_usage"]["projection"]["loopx_input"]["input_fraction"] == 0.02 - assert view["solver_usage"]["projection"]["tool_context"]["loopx_tool_output_fraction"] is None + inventory = view["solver_usage"]["projection"]["tool_context"] + if direct_call: + assert inventory["operations"] == [{"operation": "status", "calls": 1, "bytes": 5}] + assert inventory["loopx_tool_output_fraction"] == 1.0 + else: + assert inventory["loopx_tool_output_fraction"] is None assert view["job"] is None and view["result"] is None assert "hello world" not in json.dumps(view) diff --git a/tests/capabilities/test_trajectory_tool_context.py b/tests/capabilities/test_trajectory_tool_context.py index 9cbf4b1bcb..a7a1cec8d1 100644 --- a/tests/capabilities/test_trajectory_tool_context.py +++ b/tests/capabilities/test_trajectory_tool_context.py @@ -30,6 +30,25 @@ def test_literal_shell_and_exec_wrapper_are_parsed_without_execution(tmp_path): assert not marker.exists() +@pytest.mark.parametrize('script', [ + 'text(await tools.exec_command({cmd: "loopx --format json quota context-read"}))', + 'text((await tools.exec_command({\n cmd: "loopx --format json quota context-read", tty: false})).output);', +]) +def test_direct_result_wrappers_preserve_one_literal_invocation(script): + assert _literal_exec_argv(script) == ['loopx', '--format', 'json', 'quota', 'context-read'] + + +@pytest.mark.parametrize('script', [ + 'text(await tools.exec_command({cmd: "loopx status"})); text(await tools.exec_command({cmd: "pytest"}));', + 'text((await tools.exec_command({cmd: "loopx status"})).output + extra)', + 'text(await tools.exec_command({cmd: "loopx status && pytest"}))', + 'text((await tools.exec_command({...opts, cmd: "loopx status"})).output)', + 'text(await tools.exec_command({cmd: `loopx ${command}`}))', +]) +def test_direct_wrapper_rejects_mixed_or_dynamic_work(script): + assert _literal_exec_argv(script) is None + + def test_projection_uses_native_response_items_not_mirrors_or_model_prose(): def item(payload): return {'type': 'response_item', 'payload': payload} @@ -43,13 +62,13 @@ def item(payload): item({'type': 'message', 'role': 'assistant', 'content': [{'text': 'loopx status'}]}), item({'type': 'custom_tool_call', 'call_id': 'js', 'name': 'exec', 'input': 'text(await tools.exec_command({cmd: "loopx status"}));'}), - item({'type': 'custom_tool_call_output', 'call_id': 'js', 'output': 'unassigned'})] + item({'type': 'custom_tool_call_output', 'call_id': 'js', 'output': 'saved'})] result = project_tool_context(rows) assert result[0]['argv'] == ['loopx', 'status'] assert result[1]['bytes'] == 15 assert result[1]['sha256'] == hashlib.sha256(output.encode()).hexdigest() assert result[2] == {'line': 4, 'kind': 'message', 'bytes': 6, 'unmeasured_parts': 1} - assert result[3]['argv'] is None + assert result[3]['argv'] == ['loopx', 'status'] assert output not in json.dumps(result, ensure_ascii=False) assert len(result) == 5