feat(box): unify native agent tools around exec/read/write/edit

tests/unit_tests/provider/test_tool_manager_native.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import os
+import tempfile
 from types import SimpleNamespace
 from unittest.mock import Mock
 
@@ -42,41 +44,191 @@ def make_tool(name: str) -> resource_tool.LLMTool:
 @pytest.mark.asyncio
 async def test_tool_manager_lists_native_tools_first():
     manager = ToolManager(SimpleNamespace())
-    manager.native_tool_loader = StubLoader([make_tool('sandbox_exec')])
+    manager.native_tool_loader = StubLoader([make_tool('exec')])
     manager.plugin_tool_loader = StubLoader([make_tool('plugin_tool')])
     manager.mcp_tool_loader = StubLoader([make_tool('mcp_tool')])
 
     tools = await manager.get_all_tools()
 
-    assert [tool.name for tool in tools] == ['sandbox_exec', 'plugin_tool', 'mcp_tool']
+    assert [tool.name for tool in tools] == ['exec', 'plugin_tool', 'mcp_tool']
 
 
 @pytest.mark.asyncio
 async def test_tool_manager_routes_native_tool_calls():
     app = SimpleNamespace()
     manager = ToolManager(app)
-    manager.native_tool_loader = StubLoader([make_tool('sandbox_exec')], invoke_result={'backend': 'fake'})
+    manager.native_tool_loader = StubLoader([make_tool('exec')], invoke_result={'backend': 'fake'})
     manager.plugin_tool_loader = StubLoader([make_tool('plugin_tool')])
     manager.mcp_tool_loader = StubLoader([make_tool('mcp_tool')])
 
-    result = await manager.execute_func_call('sandbox_exec', {'cmd': 'pwd'}, query=Mock())
+    result = await manager.execute_func_call('exec', {'command': 'pwd'}, query=Mock())
 
     assert result == {'backend': 'fake'}
 
 
 @pytest.mark.asyncio
-async def test_native_tool_loader_hides_sandbox_exec_when_box_unavailable():
+async def test_native_tool_loader_hides_tools_when_box_unavailable():
     loader = NativeToolLoader(SimpleNamespace(box_service=SimpleNamespace(available=False)))
 
     assert await loader.get_tools() == []
-    assert await loader.has_tool('sandbox_exec') is False
+    assert await loader.has_tool('exec') is False
+    assert await loader.has_tool('read') is False
+    assert await loader.has_tool('write') is False
+    assert await loader.has_tool('edit') is False
 
 
 @pytest.mark.asyncio
-async def test_native_tool_loader_exposes_sandbox_exec_when_box_available():
+async def test_native_tool_loader_exposes_all_tools_when_box_available():
     loader = NativeToolLoader(SimpleNamespace(box_service=SimpleNamespace(available=True)))
 
     tools = await loader.get_tools()
 
-    assert [tool.name for tool in tools] == ['sandbox_exec']
-    assert await loader.has_tool('sandbox_exec') is True
+    assert [tool.name for tool in tools] == ['exec', 'read', 'write', 'edit']
+    assert await loader.has_tool('exec') is True
+    assert await loader.has_tool('read') is True
+    assert await loader.has_tool('write') is True
+    assert await loader.has_tool('edit') is True
+
+
+# ── read/write/edit file tool tests ─────────────────────────────
+
+
+def _make_loader_with_workspace(tmpdir: str) -> tuple[NativeToolLoader, Mock]:
+    logger = Mock()
+    box_service = SimpleNamespace(available=True, default_host_workspace=tmpdir)
+    ap = SimpleNamespace(box_service=box_service, logger=logger)
+    return NativeToolLoader(ap), logger
+
+
+def _make_query() -> Mock:
+    q = Mock()
+    q.query_id = 'test-query-1'
+    return q
+
+
+@pytest.mark.asyncio
+async def test_read_file():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+        with open(os.path.join(tmpdir, 'hello.txt'), 'w') as f:
+            f.write('hello world')
+
+        result = await loader.invoke_tool('read', {'path': '/workspace/hello.txt'}, _make_query())
+
+        assert result['ok'] is True
+        assert result['content'] == 'hello world'
+
+
+@pytest.mark.asyncio
+async def test_read_nonexistent_file():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+
+        result = await loader.invoke_tool('read', {'path': '/workspace/no_such.txt'}, _make_query())
+
+        assert result['ok'] is False
+        assert 'not found' in result['error'].lower()
+
+
+@pytest.mark.asyncio
+async def test_read_directory():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+        os.makedirs(os.path.join(tmpdir, 'subdir'))
+        with open(os.path.join(tmpdir, 'a.txt'), 'w') as f:
+            f.write('a')
+
+        result = await loader.invoke_tool('read', {'path': '/workspace'}, _make_query())
+
+        assert result['ok'] is True
+        assert result['is_directory'] is True
+        assert 'a.txt' in result['content']
+
+
+@pytest.mark.asyncio
+async def test_write_creates_file():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+
+        result = await loader.invoke_tool(
+            'write', {'path': '/workspace/new.txt', 'content': 'new content'}, _make_query()
+        )
+
+        assert result['ok'] is True
+        with open(os.path.join(tmpdir, 'new.txt')) as f:
+            assert f.read() == 'new content'
+
+
+@pytest.mark.asyncio
+async def test_write_creates_subdirectories():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+
+        result = await loader.invoke_tool(
+            'write', {'path': '/workspace/sub/deep/file.txt', 'content': 'nested'}, _make_query()
+        )
+
+        assert result['ok'] is True
+        with open(os.path.join(tmpdir, 'sub', 'deep', 'file.txt')) as f:
+            assert f.read() == 'nested'
+
+
+@pytest.mark.asyncio
+async def test_edit_replaces_unique_string():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+        with open(os.path.join(tmpdir, 'code.py'), 'w') as f:
+            f.write('def foo():\n    return 1\n')
+
+        result = await loader.invoke_tool(
+            'edit',
+            {'path': '/workspace/code.py', 'old_string': 'return 1', 'new_string': 'return 42'},
+            _make_query(),
+        )
+
+        assert result['ok'] is True
+        with open(os.path.join(tmpdir, 'code.py')) as f:
+            assert f.read() == 'def foo():\n    return 42\n'
+
+
+@pytest.mark.asyncio
+async def test_edit_rejects_ambiguous_match():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+        with open(os.path.join(tmpdir, 'dup.txt'), 'w') as f:
+            f.write('aaa\naaa\n')
+
+        result = await loader.invoke_tool(
+            'edit',
+            {'path': '/workspace/dup.txt', 'old_string': 'aaa', 'new_string': 'bbb'},
+            _make_query(),
+        )
+
+        assert result['ok'] is False
+        assert '2' in result['error']
+
+
+@pytest.mark.asyncio
+async def test_edit_rejects_missing_string():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+        with open(os.path.join(tmpdir, 'x.txt'), 'w') as f:
+            f.write('hello')
+
+        result = await loader.invoke_tool(
+            'edit',
+            {'path': '/workspace/x.txt', 'old_string': 'nope', 'new_string': 'yes'},
+            _make_query(),
+        )
+
+        assert result['ok'] is False
+        assert 'not found' in result['error'].lower()
+
+
+@pytest.mark.asyncio
+async def test_path_escape_blocked():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        loader, _ = _make_loader_with_workspace(tmpdir)
+
+        with pytest.raises(ValueError, match='escapes'):
+            await loader.invoke_tool('read', {'path': '/workspace/../../etc/passwd'}, _make_query())