Recognize direct exec results in the trajectory tool inventory

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<options>.*)\}\s*\)"
+    wrappers = (
+        r"const\s+(?P<result>[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)

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)

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