mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-14 09:46:03 +00:00
chore(agent-runner): drop tool fix residuals from feature branch
This commit is contained in:
huanghuoguoguo
@@ -46,12 +46,7 @@ class PluginToolLoader(loader.ToolLoader):
|
||||
return tool
|
||||
return None
|
||||
|
||||
async def invoke_tool(self, name: str, parameters: dict, query: pipeline_query.Query | None) -> typing.Any:
|
||||
if query is None:
|
||||
raise ValueError(
|
||||
f'Plugin tool {name} requires a query-based host context. '
|
||||
'Use MCP tools or provide a Host tool implementation that is run-scoped.'
|
||||
)
|
||||
async def invoke_tool(self, name: str, parameters: dict, query: pipeline_query.Query) -> typing.Any:
|
||||
try:
|
||||
return await self.ap.plugin_connector.call_tool(
|
||||
name, parameters, session=query.session, query_id=query.query_id
|
||||
|
||||
@@ -245,8 +245,7 @@ class TestSkillPathHelpers:
|
||||
|
||||
command = wrap_skill_command_with_python_env('python scripts/run.py')
|
||||
|
||||
assert '_LB_SYSTEM_PYTHON="$(command -v python3 || command -v python || true)"' in command
|
||||
assert '"$_LB_SYSTEM_PYTHON" -m venv "$_LB_VENV_DIR"' in command
|
||||
assert 'python -m venv "$_LB_VENV_DIR"' in command
|
||||
assert 'export VIRTUAL_ENV="$_LB_VENV_DIR"' in command
|
||||
assert command.rstrip().endswith('python scripts/run.py')
|
||||
|
||||
@@ -307,130 +306,6 @@ class TestSkillToolLoader:
|
||||
SimpleNamespace(variables={}),
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_activate_rejects_pipeline_hidden_skill(self):
|
||||
from langbot.pkg.provider.tools.loaders.skill_authoring import (
|
||||
ACTIVATE_SKILL_TOOL_NAME,
|
||||
SkillToolLoader,
|
||||
)
|
||||
from langbot.pkg.provider.tools.loaders.skill import (
|
||||
ACTIVATED_SKILLS_KEY,
|
||||
PIPELINE_BOUND_SKILLS_KEY,
|
||||
)
|
||||
|
||||
demo = _make_skill_data(name='demo', package_root='/data/skills/demo', instructions='Demo instructions')
|
||||
hidden = _make_skill_data(
|
||||
name='hidden',
|
||||
package_root='/data/skills/hidden',
|
||||
instructions='Hidden instructions',
|
||||
)
|
||||
ap = _make_ap()
|
||||
ap.skill_mgr = SimpleNamespace(
|
||||
skills={'demo': demo, 'hidden': hidden},
|
||||
)
|
||||
|
||||
loader = SkillToolLoader(ap)
|
||||
query = SimpleNamespace(variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']})
|
||||
|
||||
with pytest.raises(ValueError, match='Available skills: demo'):
|
||||
await loader.invoke_tool(ACTIVATE_SKILL_TOOL_NAME, {'skill_name': 'hidden'}, query)
|
||||
|
||||
result = await loader.invoke_tool(ACTIVATE_SKILL_TOOL_NAME, {'skill_name': 'demo'}, query)
|
||||
|
||||
assert result['activated'] is True
|
||||
assert result['skill_name'] == 'demo'
|
||||
assert set(query.variables[ACTIVATED_SKILLS_KEY].keys()) == {'demo'}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_activate_persists_and_restores_for_next_query_exec(self, tmp_path):
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
from langbot.pkg.agent.runner.persistent_state_store import (
|
||||
get_persistent_state_store,
|
||||
reset_persistent_state_store,
|
||||
)
|
||||
from langbot.pkg.entity.persistence.base import Base
|
||||
from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
|
||||
from langbot.pkg.provider.tools.loaders.skill_authoring import (
|
||||
ACTIVATE_SKILL_TOOL_NAME,
|
||||
SkillToolLoader,
|
||||
)
|
||||
from langbot.pkg.provider.tools.loaders.skill import (
|
||||
ACTIVATED_SKILL_NAMES_STATE_KEY,
|
||||
ACTIVATED_SKILLS_KEY,
|
||||
PIPELINE_BOUND_SKILLS_KEY,
|
||||
restore_activated_skills_from_state,
|
||||
)
|
||||
|
||||
reset_persistent_state_store()
|
||||
engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "state.db"}')
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
try:
|
||||
skill = _make_skill_data(name='demo', package_root=str(tmp_path), instructions='Demo instructions')
|
||||
ap = _make_ap()
|
||||
ap.persistence_mgr.get_db_engine = Mock(return_value=engine)
|
||||
ap.box_service = SimpleNamespace(
|
||||
available=True,
|
||||
default_workspace=str(tmp_path),
|
||||
execute_tool=AsyncMock(return_value={'ok': True}),
|
||||
)
|
||||
ap.skill_mgr = SimpleNamespace(
|
||||
skills={'demo': skill},
|
||||
refresh_skill_from_disk=Mock(),
|
||||
)
|
||||
|
||||
scope_key = 'conversation:plugin:langbot/local-agent/default:binding_001:conv_001'
|
||||
query1 = SimpleNamespace(query_id='q1', variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']})
|
||||
object.__setattr__(
|
||||
query1,
|
||||
'_agent_run_session',
|
||||
{
|
||||
'runner_id': 'plugin:langbot/local-agent/default',
|
||||
'authorization': {
|
||||
'state_policy': {'enable_state': True, 'state_scopes': ['conversation']},
|
||||
'state_context': {
|
||||
'scope_keys': {'conversation': scope_key},
|
||||
'binding_identity': 'binding_001',
|
||||
'conversation_id': 'conv_001',
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
await SkillToolLoader(ap).invoke_tool(ACTIVATE_SKILL_TOOL_NAME, {'skill_name': 'demo'}, query1)
|
||||
|
||||
store = get_persistent_state_store(engine)
|
||||
persisted_names = await store.state_get(scope_key, ACTIVATED_SKILL_NAMES_STATE_KEY)
|
||||
assert persisted_names == ['demo']
|
||||
|
||||
query2 = SimpleNamespace(query_id='q2', variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']})
|
||||
restored = restore_activated_skills_from_state(
|
||||
ap,
|
||||
query2,
|
||||
{'conversation': {ACTIVATED_SKILL_NAMES_STATE_KEY: persisted_names}},
|
||||
)
|
||||
|
||||
assert restored == ['demo']
|
||||
assert set(query2.variables[ACTIVATED_SKILLS_KEY]) == {'demo'}
|
||||
|
||||
result = await NativeToolLoader(ap).invoke_tool(
|
||||
'exec',
|
||||
{
|
||||
'command': 'python /workspace/.skills/demo/scripts/run.py',
|
||||
'workdir': '/workspace/.skills/demo',
|
||||
},
|
||||
query2,
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
ap.box_service.execute_tool.assert_awaited_once()
|
||||
ap.skill_mgr.refresh_skill_from_disk.assert_called_once_with('demo')
|
||||
finally:
|
||||
reset_persistent_state_store()
|
||||
await engine.dispose()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_skill_scans_directory_and_creates_skill(self):
|
||||
from langbot.pkg.provider.tools.loaders.skill_authoring import (
|
||||
@@ -618,35 +493,6 @@ class TestNativeToolLoaderSkillPaths:
|
||||
assert tool_parameters['workdir'] == '/workspace/.skills/demo'
|
||||
ap.skill_mgr.refresh_skill_from_disk.assert_called_once_with('demo')
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exec_requires_skill_activation_even_when_skill_visible(self):
|
||||
from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
|
||||
from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
ap = _make_ap()
|
||||
ap.box_service = SimpleNamespace(
|
||||
available=True,
|
||||
default_workspace=tmpdir,
|
||||
execute_tool=AsyncMock(return_value={'ok': True}),
|
||||
)
|
||||
ap.skill_mgr = SimpleNamespace(skills={'demo': _make_skill_data(name='demo', package_root=tmpdir)})
|
||||
loader = NativeToolLoader(ap)
|
||||
|
||||
query = SimpleNamespace(query_id='q1', variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']})
|
||||
|
||||
with pytest.raises(ValueError, match='must be activated before exec'):
|
||||
await loader.invoke_tool(
|
||||
'exec',
|
||||
{
|
||||
'command': 'python /workspace/.skills/demo/scripts/run.py',
|
||||
'workdir': '/workspace',
|
||||
},
|
||||
query,
|
||||
)
|
||||
|
||||
ap.box_service.execute_tool.assert_not_awaited()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_requires_skill_activation(self):
|
||||
from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
|
||||
|
||||
@@ -3,19 +3,13 @@ from __future__ import annotations
|
||||
import os
|
||||
import tempfile
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
|
||||
|
||||
from langbot.pkg.provider.tools.loader import ToolLoader
|
||||
from langbot.pkg.provider.tools.loaders.native import (
|
||||
_DEFAULT_TOOL_RESULT_MAX_BYTES,
|
||||
_GLOB_MAX_MATCHES,
|
||||
_GREP_MAX_MATCHES,
|
||||
NativeToolLoader,
|
||||
)
|
||||
from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
|
||||
from langbot.pkg.provider.tools.toolmgr import ToolManager
|
||||
|
||||
|
||||
@@ -27,12 +21,6 @@ class StubLoader:
|
||||
async def get_tools(self, *_args, **_kwargs):
|
||||
return self._tools
|
||||
|
||||
async def get_tool(self, name: str):
|
||||
for tool in self._tools:
|
||||
if tool.name == name:
|
||||
return tool
|
||||
return None
|
||||
|
||||
async def has_tool(self, name: str) -> bool:
|
||||
return any(tool.name == name for tool in self._tools)
|
||||
|
||||
@@ -43,14 +31,6 @@ class StubLoader:
|
||||
return None
|
||||
|
||||
|
||||
class DirectLookupLoader(StubLoader):
|
||||
async def get_tools(self, *_args, **_kwargs):
|
||||
raise AssertionError('ToolManager should use the loader get_tool contract')
|
||||
|
||||
async def get_tool(self, name: str):
|
||||
return make_tool(name) if name == 'direct_tool' else None
|
||||
|
||||
|
||||
def make_tool(name: str) -> resource_tool.LLMTool:
|
||||
return resource_tool.LLMTool(
|
||||
name=name,
|
||||
@@ -61,32 +41,6 @@ def make_tool(name: str) -> resource_tool.LLMTool:
|
||||
)
|
||||
|
||||
|
||||
class ListOnlyLoader(ToolLoader):
|
||||
async def get_tools(self, *_args, **_kwargs):
|
||||
return [make_tool('listed_tool')]
|
||||
|
||||
async def has_tool(self, name: str) -> bool:
|
||||
return name == 'listed_tool'
|
||||
|
||||
async def invoke_tool(self, name: str, parameters: dict, query):
|
||||
return parameters
|
||||
|
||||
async def shutdown(self):
|
||||
return None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_loader_get_tool_falls_back_to_public_tool_list():
|
||||
loader = ListOnlyLoader(SimpleNamespace())
|
||||
|
||||
tool = await loader.get_tool('listed_tool')
|
||||
missing_tool = await loader.get_tool('missing_tool')
|
||||
|
||||
assert tool is not None
|
||||
assert tool.name == 'listed_tool'
|
||||
assert missing_tool is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_manager_omits_skill_authoring_tools_by_default():
|
||||
manager = ToolManager(SimpleNamespace())
|
||||
@@ -127,37 +81,6 @@ async def test_tool_manager_routes_native_tool_calls():
|
||||
assert result == {'backend': 'fake'}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_manager_get_tool_by_name_resolves_native_and_skill_tools():
|
||||
manager = ToolManager(SimpleNamespace())
|
||||
manager.native_tool_loader = StubLoader([make_tool('exec')])
|
||||
manager.skill_tool_loader = StubLoader([make_tool('activate')])
|
||||
manager.plugin_tool_loader = StubLoader([make_tool('plugin_tool')])
|
||||
manager.mcp_tool_loader = StubLoader([make_tool('mcp_tool')])
|
||||
|
||||
native_tool = await manager.get_tool_by_name('exec')
|
||||
skill_tool = await manager.get_tool_by_name('activate')
|
||||
|
||||
assert native_tool is not None
|
||||
assert native_tool.name == 'exec'
|
||||
assert skill_tool is not None
|
||||
assert skill_tool.name == 'activate'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_manager_uses_loader_get_tool_contract():
|
||||
manager = ToolManager(SimpleNamespace())
|
||||
manager.native_tool_loader = StubLoader([])
|
||||
manager.skill_tool_loader = StubLoader([])
|
||||
manager.plugin_tool_loader = DirectLookupLoader()
|
||||
manager.mcp_tool_loader = StubLoader([])
|
||||
|
||||
tool = await manager.get_tool_by_name('direct_tool')
|
||||
|
||||
assert tool is not None
|
||||
assert tool.name == 'direct_tool'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_native_tool_loader_hides_tools_when_box_unavailable():
|
||||
loader = NativeToolLoader(SimpleNamespace(box_service=SimpleNamespace(available=False)))
|
||||
@@ -196,7 +119,6 @@ def _make_loader_with_workspace(tmpdir: str) -> tuple[NativeToolLoader, Mock]:
|
||||
def _make_query() -> Mock:
|
||||
q = Mock()
|
||||
q.query_id = 'test-query-1'
|
||||
q.variables = {}
|
||||
return q
|
||||
|
||||
|
||||
@@ -211,9 +133,6 @@ async def test_read_file():
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['content'] == 'hello world'
|
||||
assert result['truncated'] is False
|
||||
assert result['start_line'] == 1
|
||||
assert result['end_line'] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -240,136 +159,6 @@ async def test_read_directory():
|
||||
assert result['ok'] is True
|
||||
assert result['is_directory'] is True
|
||||
assert 'a.txt' in result['content']
|
||||
assert result['total'] == 2
|
||||
assert result['truncated'] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_read_file_supports_line_window():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
loader, _ = _make_loader_with_workspace(tmpdir)
|
||||
content = '\n'.join(f'line-{line_no}' for line_no in range(1, 7))
|
||||
with open(os.path.join(tmpdir, 'large.txt'), 'w') as f:
|
||||
f.write(content)
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'read',
|
||||
{'path': '/workspace/large.txt', 'offset': 2, 'limit': 3},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['content'] == 'line-2\nline-3\nline-4'
|
||||
assert result['truncated'] is True
|
||||
assert result['truncated_by'] == 'lines'
|
||||
assert result['start_line'] == 2
|
||||
assert result['end_line'] == 4
|
||||
assert result['next_offset'] == 5
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_read_file_is_bounded_by_bytes():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
loader, _ = _make_loader_with_workspace(tmpdir)
|
||||
with open(os.path.join(tmpdir, 'wide.txt'), 'w') as f:
|
||||
f.write(('x' * 128) + '\nsecond line')
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'read',
|
||||
{'path': '/workspace/wide.txt', 'max_bytes': 32},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['truncated'] is True
|
||||
assert result['truncated_by'] == 'bytes'
|
||||
assert result['next_offset'] == 1
|
||||
assert result['content'].startswith('[Line 1 exceeds')
|
||||
assert len(result['content']) < 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_skill_read_uses_host_preview_when_package_root_available():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
skill_root = os.path.join(tmpdir, 'skill-demo')
|
||||
os.makedirs(skill_root)
|
||||
with open(os.path.join(skill_root, 'large.txt'), 'w') as f:
|
||||
f.write('first\nsecond\nthird')
|
||||
|
||||
box_service = SimpleNamespace(
|
||||
available=True,
|
||||
default_workspace=tmpdir,
|
||||
read_skill_file=AsyncMock(return_value={'content': 'should not be used'}),
|
||||
)
|
||||
skill_mgr = SimpleNamespace(skills={'demo': {'name': 'demo', 'package_root': skill_root}})
|
||||
loader = NativeToolLoader(SimpleNamespace(box_service=box_service, skill_mgr=skill_mgr, logger=Mock()))
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'read',
|
||||
{'path': '/workspace/.skills/demo/large.txt', 'limit': 1},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['content'] == 'first'
|
||||
assert result['truncated'] is True
|
||||
assert result['next_offset'] == 2
|
||||
box_service.read_skill_file.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_read_truncated_file_returns_host_artifact_ref_for_agent_run():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
engine = object()
|
||||
logger = Mock()
|
||||
box_service = SimpleNamespace(available=True, default_workspace=tmpdir)
|
||||
persistence_mgr = SimpleNamespace(get_db_engine=Mock(return_value=engine))
|
||||
loader = NativeToolLoader(
|
||||
SimpleNamespace(box_service=box_service, persistence_mgr=persistence_mgr, logger=logger)
|
||||
)
|
||||
with open(os.path.join(tmpdir, 'large.txt'), 'w') as f:
|
||||
f.write('first\nsecond\nthird')
|
||||
|
||||
query = _make_query()
|
||||
query.bot_uuid = 'bot-001'
|
||||
query._agent_run_session = {
|
||||
'run_id': 'run-001',
|
||||
'runner_id': 'plugin:test/runner/default',
|
||||
'authorization': {'conversation_id': 'conv-001'},
|
||||
}
|
||||
|
||||
with patch('langbot.pkg.agent.runner.artifact_store.ArtifactStore') as store_cls:
|
||||
store = store_cls.return_value
|
||||
store.register_file_artifact = AsyncMock(return_value='artifact-file-001')
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'read',
|
||||
{'path': '/workspace/large.txt', 'limit': 1},
|
||||
query,
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['content'] == 'first'
|
||||
assert result['preview'] == 'first'
|
||||
assert result['truncated'] is True
|
||||
assert result['artifact_refs'] == [
|
||||
{
|
||||
'artifact_id': 'artifact-file-001',
|
||||
'artifact_type': 'file',
|
||||
'mime_type': 'text/plain',
|
||||
'name': 'large.txt',
|
||||
'size_bytes': os.path.getsize(os.path.join(tmpdir, 'large.txt')),
|
||||
}
|
||||
]
|
||||
store_cls.assert_called_once_with(engine)
|
||||
store.register_file_artifact.assert_awaited_once()
|
||||
call_kwargs = store.register_file_artifact.await_args.kwargs
|
||||
assert call_kwargs['host_path'] == os.path.realpath(os.path.join(tmpdir, 'large.txt'))
|
||||
assert call_kwargs['host_root'] == tmpdir
|
||||
assert call_kwargs['conversation_id'] == 'conv-001'
|
||||
assert call_kwargs['run_id'] == 'run-001'
|
||||
assert call_kwargs['runner_id'] == 'plugin:test/runner/default'
|
||||
assert call_kwargs['metadata']['sandbox_path'] == '/workspace/large.txt'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -459,119 +248,3 @@ async def test_path_escape_blocked():
|
||||
|
||||
with pytest.raises(ValueError, match='escapes'):
|
||||
await loader.invoke_tool('read', {'path': '/workspace/../../etc/passwd'}, _make_query())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_glob_result_is_bounded():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
loader, _ = _make_loader_with_workspace(tmpdir)
|
||||
for index in range(_GLOB_MAX_MATCHES + 5):
|
||||
with open(os.path.join(tmpdir, f'file-{index:03d}.txt'), 'w') as f:
|
||||
f.write(str(index))
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'glob',
|
||||
{'path': '/workspace', 'pattern': '*.txt'},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert len(result['matches']) == _GLOB_MAX_MATCHES
|
||||
assert result['total'] == _GLOB_MAX_MATCHES + 5
|
||||
assert result['truncated'] is True
|
||||
assert result['truncated_by'] == 'matches'
|
||||
assert result['preview'].splitlines() == result['matches']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_grep_result_is_bounded_by_match_count():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
loader, _ = _make_loader_with_workspace(tmpdir)
|
||||
with open(os.path.join(tmpdir, 'hits.txt'), 'w') as f:
|
||||
for index in range(_GREP_MAX_MATCHES + 5):
|
||||
f.write(f'needle {index}\n')
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'grep',
|
||||
{'path': '/workspace', 'pattern': 'needle', 'include': '*.txt'},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert len(result['matches']) == _GREP_MAX_MATCHES
|
||||
assert result['total'] == _GREP_MAX_MATCHES
|
||||
assert result['truncated'] is True
|
||||
assert result['truncated_by'] == 'matches'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_grep_truncates_long_matching_line():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
loader, _ = _make_loader_with_workspace(tmpdir)
|
||||
with open(os.path.join(tmpdir, 'wide.txt'), 'w') as f:
|
||||
f.write('needle ' + ('x' * 600))
|
||||
|
||||
result = await loader.invoke_tool(
|
||||
'grep',
|
||||
{'path': '/workspace', 'pattern': 'needle', 'include': '*.txt'},
|
||||
_make_query(),
|
||||
)
|
||||
|
||||
assert result['ok'] is True
|
||||
assert len(result['matches']) == 1
|
||||
assert result['matches'][0]['content'].endswith('... [truncated]')
|
||||
assert result['truncated'] is True
|
||||
assert result['truncated_by'] == 'line'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exec_result_adds_preview_and_truncated_flag():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
box_service = SimpleNamespace(
|
||||
available=True,
|
||||
default_workspace=tmpdir,
|
||||
execute_tool=AsyncMock(
|
||||
return_value={
|
||||
'ok': True,
|
||||
'stdout': 'stdout text',
|
||||
'stderr': 'stderr text',
|
||||
'stdout_truncated': True,
|
||||
'stderr_truncated': False,
|
||||
}
|
||||
),
|
||||
)
|
||||
loader = NativeToolLoader(SimpleNamespace(box_service=box_service, logger=Mock()))
|
||||
|
||||
result = await loader.invoke_tool('exec', {'command': 'echo ok'}, _make_query())
|
||||
|
||||
assert result['ok'] is True
|
||||
assert result['truncated'] is True
|
||||
assert result['preview'] == 'stdout:\nstdout text\n\nstderr:\nstderr text'
|
||||
box_service.execute_tool.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exec_result_caps_untrusted_large_output():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
box_service = SimpleNamespace(
|
||||
available=True,
|
||||
default_workspace=tmpdir,
|
||||
execute_tool=AsyncMock(
|
||||
return_value={
|
||||
'ok': True,
|
||||
'stdout': 'x' * (_DEFAULT_TOOL_RESULT_MAX_BYTES + 128),
|
||||
'stderr': '',
|
||||
'stdout_truncated': False,
|
||||
'stderr_truncated': False,
|
||||
}
|
||||
),
|
||||
)
|
||||
loader = NativeToolLoader(SimpleNamespace(box_service=box_service, logger=Mock()))
|
||||
|
||||
result = await loader.invoke_tool('exec', {'command': 'echo ok'}, _make_query())
|
||||
|
||||
assert result['ok'] is True
|
||||
assert len(result['stdout'].encode('utf-8')) <= _DEFAULT_TOOL_RESULT_MAX_BYTES
|
||||
assert result['stdout_truncated'] is True
|
||||
assert result['truncated'] is True
|
||||
assert result['preview'] == result['stdout']
|
||||
|
||||
Reference in New Issue
Block a user