mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-08-22 10:17:13 +00:00
DongXiaoming
e934f08adf
* fix(wecombot): align media upload protocol * fix(wecombot): deliver outbox media in reply and fix tool call recording - Integrate _send_media into reply_message and reply_message_chunk so sandbox outbox images/voices/files are uploaded and sent instead of being silently dropped. - Add missing import base64 that caused _send_media to fail with a NameError swallowed by its except clause. - Change yiri2target to return component dicts (text/image/voice/file) so callers can distinguish text from media. - Fix _get_message_for_tool_context using result.first()/row[0] which returned a raw string instead of the ORM object, causing "'str' object has no attribute 'pipeline_id'" in tool call recording. Use result.scalars().first() per SQLAlchemy 2.0 convention. * fix(pipeline): collect outbox attachments on final chunk with empty content When the last streaming chunk has is_final=True but empty content (e.g. the LLM sends all text in earlier chunks), the 'if result.content' branch is skipped entirely, so _append_outbound_attachments never runs and sandbox outbox images are silently dropped. Add an elif branch for _is_final_assistant_message that creates an empty MessageChain and still collects outbox attachments, so images are delivered even when the final chunk carries no text. * fix(box): bypass stdout truncation when reading outbox via exec _read_outbox_via_exec used execute_tool which returns _serialize_result where stdout is truncated to output_limit_chars (4000). A 7KB JPEG encodes to ~9400 base64 chars, so the JSON payload was truncated and json.loads failed silently, returning an empty list. Call client.execute directly to get the raw BoxExecutionResult with untruncated stdout, so base64 file data is preserved. * fix(tests): adapt box and wrapper tests for client.execute and strict is_final check - wrapper.py: restrict outbox collection on empty-content chunks to actual MessageChunk instances with is_final=True, not generic Mock objects that happen to have role='assistant' - test_box_service.py: update _read_outbox_via_exec tests to mock client.execute (returning BoxExecutionResult) instead of execute_tool, matching the implementation change * chore(wecombot): remove temporary upload log * test(box): preserve direct outbox read and cleanup coverage --------- Co-authored-by: fdc310 <2213070223@qq.com> Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>
2511 lines
96 KiB
Python
2511 lines
96 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import datetime as dt
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
|
|
|
|
from langbot_plugin.box.backend import BaseSandboxBackend
|
|
from langbot_plugin.box.client import BoxRuntimeClient, ActionRPCBoxClient
|
|
from langbot_plugin.box.errors import (
|
|
BoxBackendUnavailableError,
|
|
BoxError,
|
|
BoxSessionConflictError,
|
|
BoxSessionNotFoundError,
|
|
BoxValidationError,
|
|
)
|
|
from langbot_plugin.box.models import (
|
|
BUILTIN_PROFILES,
|
|
BoxExecutionResult,
|
|
BoxExecutionStatus,
|
|
BoxHostMountMode,
|
|
BoxManagedProcessSpec,
|
|
BoxNetworkMode,
|
|
BoxSessionInfo,
|
|
BoxSpec,
|
|
)
|
|
from langbot_plugin.box.runtime import BoxRuntime
|
|
from langbot_plugin.box.security import (
|
|
BOX_CONTROL_TOKEN_HEADER,
|
|
BOX_INSTANCE_HEADER,
|
|
BOX_PLACEMENT_GENERATION_HEADER,
|
|
BOX_WORKSPACE_HEADER,
|
|
)
|
|
from langbot_plugin.entities.io.context import ActionContext
|
|
from langbot.pkg.api.http.context import ExecutionContext
|
|
from langbot.pkg.box.service import BoxService
|
|
|
|
_UTC = dt.timezone.utc
|
|
_CONTEXT = ExecutionContext(
|
|
instance_uuid='instance-a',
|
|
workspace_uuid='workspace-a',
|
|
placement_generation=1,
|
|
)
|
|
_ACTION_CONTEXT = ActionContext(
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
)
|
|
|
|
|
|
class _InProcessBoxRuntimeClient(BoxRuntimeClient):
|
|
"""Test-only client that wraps a BoxRuntime in-process (no HTTP)."""
|
|
|
|
def __init__(self, logger, runtime=None):
|
|
self._runtime = runtime or BoxRuntime(logger=logger)
|
|
|
|
async def initialize(self):
|
|
await self._runtime.initialize()
|
|
|
|
async def execute(self, spec, *, action_context=None):
|
|
return await self._runtime.execute(spec)
|
|
|
|
async def shutdown(self):
|
|
await self._runtime.shutdown()
|
|
|
|
async def get_status(self, *, action_context=None):
|
|
return await self._runtime.get_status()
|
|
|
|
async def get_sessions(self, *, action_context=None):
|
|
return self._runtime.get_sessions()
|
|
|
|
async def get_backend_info(self):
|
|
return await self._runtime.get_backend_info()
|
|
|
|
async def delete_session(self, session_id, *, action_context=None):
|
|
await self._runtime.delete_session(session_id)
|
|
|
|
async def create_session(self, spec, *, action_context=None):
|
|
return await self._runtime.create_session(spec)
|
|
|
|
async def start_managed_process(
|
|
self,
|
|
session_id: str,
|
|
spec: BoxManagedProcessSpec,
|
|
*,
|
|
action_context=None,
|
|
):
|
|
return await self._runtime.start_managed_process(session_id, spec)
|
|
|
|
async def get_managed_process(
|
|
self,
|
|
session_id: str,
|
|
process_id: str = 'default',
|
|
*,
|
|
action_context=None,
|
|
):
|
|
return self._runtime.get_managed_process(session_id, process_id)
|
|
|
|
async def stop_managed_process(
|
|
self,
|
|
session_id: str,
|
|
process_id: str = 'default',
|
|
*,
|
|
action_context=None,
|
|
):
|
|
await self._runtime.stop_managed_process(session_id, process_id)
|
|
|
|
async def get_session(self, session_id: str, *, action_context=None):
|
|
return self._runtime.get_session(session_id)
|
|
|
|
async def init(self, config: dict) -> None:
|
|
self._runtime.init(config)
|
|
|
|
async def verify_shared_workspace(self, marker_name: str) -> dict:
|
|
return self._runtime.verify_shared_workspace(marker_name)
|
|
|
|
|
|
class FakeBackend(BaseSandboxBackend):
|
|
def __init__(self, logger: Mock, available: bool = True):
|
|
super().__init__(logger)
|
|
self.name = 'fake'
|
|
self.available = available
|
|
self.start_calls: list[str] = []
|
|
self.start_specs: list[BoxSpec] = []
|
|
self.exec_calls: list[tuple[str, str]] = []
|
|
self.stop_calls: list[str] = []
|
|
|
|
async def is_available(self) -> bool:
|
|
return self.available
|
|
|
|
async def start_session(self, spec: BoxSpec) -> BoxSessionInfo:
|
|
self.start_calls.append(spec.session_id)
|
|
self.start_specs.append(spec)
|
|
now = dt.datetime.now(_UTC)
|
|
return BoxSessionInfo(
|
|
session_id=spec.session_id,
|
|
backend_name=self.name,
|
|
backend_session_id=f'backend-{spec.session_id}',
|
|
image=spec.image,
|
|
network=spec.network,
|
|
host_path=spec.host_path,
|
|
host_path_mode=spec.host_path_mode,
|
|
mount_path=spec.mount_path,
|
|
cpus=spec.cpus,
|
|
memory_mb=spec.memory_mb,
|
|
pids_limit=spec.pids_limit,
|
|
read_only_rootfs=spec.read_only_rootfs,
|
|
created_at=now,
|
|
last_used_at=now,
|
|
)
|
|
|
|
async def exec(self, session: BoxSessionInfo, spec: BoxSpec) -> BoxExecutionResult:
|
|
self.exec_calls.append((session.session_id, spec.cmd))
|
|
return BoxExecutionResult(
|
|
session_id=session.session_id,
|
|
backend_name=self.name,
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout=f'executed: {spec.cmd}',
|
|
stderr='',
|
|
duration_ms=12,
|
|
)
|
|
|
|
async def stop_session(self, session: BoxSessionInfo):
|
|
self.stop_calls.append(session.session_id)
|
|
|
|
|
|
def make_query(query_id: int = 42) -> pipeline_query.Query:
|
|
return pipeline_query.Query.model_construct(
|
|
query_id=query_id,
|
|
query_uuid=f'query-{query_id}',
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
bot_uuid='bot-a',
|
|
pipeline_uuid='pipeline-a',
|
|
launcher_type='person',
|
|
launcher_id='test_user',
|
|
sender_id='test_user',
|
|
variables={
|
|
'launcher_type': 'person',
|
|
'launcher_id': 'test_user',
|
|
'sender_id': 'test_user',
|
|
'query_id': str(query_id),
|
|
},
|
|
)
|
|
|
|
|
|
def make_app(
|
|
logger: Mock,
|
|
allowed_mount_roots: list[str] | None = None,
|
|
profile: str = 'default',
|
|
host_root: str = '',
|
|
workspace_quota_mb: int | None = None,
|
|
enabled: bool = True,
|
|
force_box_session_id_template: str = '',
|
|
):
|
|
box_config = {
|
|
'enabled': enabled,
|
|
'backend': 'local',
|
|
'runtime': {'endpoint': ''},
|
|
'local': {
|
|
'profile': profile,
|
|
'host_root': host_root,
|
|
'allowed_mount_roots': allowed_mount_roots or [],
|
|
'default_workspace': '',
|
|
},
|
|
'e2b': {'api_key': '', 'api_url': '', 'template': ''},
|
|
}
|
|
if workspace_quota_mb is not None:
|
|
box_config['local']['workspace_quota_mb'] = workspace_quota_mb
|
|
|
|
workspace_service = SimpleNamespace(
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
get_execution_binding=AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
)
|
|
),
|
|
)
|
|
return SimpleNamespace(
|
|
logger=logger,
|
|
workspace_service=workspace_service,
|
|
instance_config=SimpleNamespace(
|
|
data={
|
|
'box': box_config,
|
|
'system': {'limitation': {'force_box_session_id_template': force_box_session_id_template}},
|
|
}
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_without_explicit_client_initializes_internal_connector(monkeypatch: pytest.MonkeyPatch):
|
|
connector = Mock()
|
|
connector.client = Mock()
|
|
connector.initialize = AsyncMock()
|
|
|
|
monkeypatch.setattr('langbot.pkg.box.service.BoxRuntimeConnector', Mock(return_value=connector))
|
|
|
|
service = BoxService(make_app(Mock()))
|
|
await service.initialize()
|
|
|
|
assert service.client is connector.client
|
|
connector.initialize.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_initialize_validation_failure_closes_connector_and_cancels_reconnect(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
logger = Mock()
|
|
app = make_app(logger)
|
|
app.deployment = SimpleNamespace(multi_workspace_enabled=True)
|
|
app.instance_config.data['box'].update(
|
|
{
|
|
'backend': 'nsjail',
|
|
'admission': {'required': True, 'workspace_quota_mb': 32},
|
|
}
|
|
)
|
|
connector = Mock()
|
|
connector.client = Mock(spec=BoxRuntimeClient)
|
|
connector.initialize = AsyncMock()
|
|
connector.aclose = AsyncMock()
|
|
connector.runtime_disconnect_callback = Mock()
|
|
monkeypatch.setattr('langbot.pkg.box.service.BoxRuntimeConnector', Mock(return_value=connector))
|
|
service = BoxService(app)
|
|
service._ensure_default_workspace = Mock()
|
|
readiness_error = BoxValidationError('Cloud Box nsjail isolation readiness failed')
|
|
service._verify_cloud_runtime = AsyncMock(side_effect=readiness_error)
|
|
reconnect_task = asyncio.create_task(asyncio.Event().wait())
|
|
service._reconnect_task = reconnect_task
|
|
service._reconnecting = True
|
|
|
|
with pytest.raises(BoxValidationError) as exc_info:
|
|
await service.initialize()
|
|
|
|
assert exc_info.value is readiness_error
|
|
assert service.available is False
|
|
assert service._closing is True
|
|
assert service._reconnecting is False
|
|
assert service._reconnect_task is None
|
|
assert reconnect_task.cancelled()
|
|
assert connector.runtime_disconnect_callback is None
|
|
connector.aclose.assert_awaited_once()
|
|
|
|
|
|
class TestSharesFilesystemWithBox:
|
|
"""``shares_filesystem_with_box`` must reflect the real LangBot<->Box
|
|
filesystem topology, which is derived from the connector transport:
|
|
|
|
- stdio (local child process) → shared filesystem → True
|
|
- WebSocket (Docker / sidecar / --standalone-box / remote) → separated → False
|
|
|
|
This drives whether LangBot validates Box-reported skill paths locally.
|
|
Getting it wrong silently drops every skill in separated deployments.
|
|
"""
|
|
|
|
def test_true_for_stdio_connector(self, monkeypatch: pytest.MonkeyPatch):
|
|
# Non-Docker Unix, no endpoint, not standalone → stdio transport.
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.get_platform', lambda: 'linux')
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.standalone_box', False)
|
|
|
|
service = BoxService(make_app(Mock()))
|
|
|
|
assert service._runtime_connector is not None
|
|
assert service._runtime_connector.uses_websocket() is False
|
|
assert service.shares_filesystem_with_box is True
|
|
|
|
def test_false_for_websocket_connector_via_endpoint(self, monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.get_platform', lambda: 'linux')
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.standalone_box', False)
|
|
app = make_app(Mock())
|
|
app.instance_config.data['box']['runtime']['endpoint'] = 'ws://pod-x-box:5410'
|
|
|
|
service = BoxService(app)
|
|
|
|
assert service._runtime_connector is not None
|
|
assert service._runtime_connector.uses_websocket() is True
|
|
assert service.shares_filesystem_with_box is False
|
|
|
|
def test_false_for_websocket_connector_in_docker(self, monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.get_platform', lambda: 'docker')
|
|
monkeypatch.setattr('langbot.pkg.utils.platform.standalone_box', False)
|
|
|
|
service = BoxService(make_app(Mock()))
|
|
|
|
assert service.shares_filesystem_with_box is False
|
|
|
|
def test_false_when_client_injected_without_connector(self):
|
|
# Injected client (no connector) → unknown topology → conservative False
|
|
# so LangBot never wrongly drops Box-reported skills.
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
|
|
assert service._runtime_connector is None
|
|
assert service.shares_filesystem_with_box is False
|
|
|
|
def test_explicit_override_wins(self):
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
|
|
service._shares_filesystem_with_box_override = True
|
|
assert service.shares_filesystem_with_box is True
|
|
|
|
service._shares_filesystem_with_box_override = False
|
|
assert service.shares_filesystem_with_box is False
|
|
|
|
|
|
def test_separated_box_runtime_does_not_create_default_workspace_in_langbot(tmp_path):
|
|
logger = Mock()
|
|
runtime = BoxRuntime(logger=logger, backends=[FakeBackend(logger)], session_ttl_sec=300)
|
|
host_root = tmp_path / 'box'
|
|
service = BoxService(make_app(logger, host_root=str(host_root)), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
service._shares_filesystem_with_box_override = False
|
|
|
|
service._ensure_default_workspace()
|
|
|
|
assert not (host_root / 'default').exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_initialize_fails_when_core_and_runtime_volumes_are_separated(tmp_path):
|
|
logger = Mock()
|
|
core_root = tmp_path / 'core-box'
|
|
runtime_root = tmp_path / 'runtime-box'
|
|
(core_root / 'default').mkdir(parents=True)
|
|
runtime = BoxRuntime(logger=logger, backends=[FakeBackend(logger)], session_ttl_sec=300)
|
|
runtime.init(
|
|
{
|
|
'local': {
|
|
'host_root': str(runtime_root),
|
|
'default_workspace': 'default',
|
|
'allowed_mount_roots': [str(runtime_root)],
|
|
}
|
|
}
|
|
)
|
|
app = make_app(logger, host_root=str(core_root))
|
|
app.deployment = SimpleNamespace(multi_workspace_enabled=True)
|
|
app.instance_config.data['box'].update(
|
|
{
|
|
'backend': 'nsjail',
|
|
'admission': {'required': True, 'workspace_quota_mb': 32},
|
|
}
|
|
)
|
|
service = BoxService(
|
|
app,
|
|
client=_InProcessBoxRuntimeClient(logger, runtime),
|
|
)
|
|
|
|
with pytest.raises(BoxValidationError, match='shared durable Workspace volume'):
|
|
await service.initialize()
|
|
|
|
assert service.available is False
|
|
assert list((core_root / 'default').glob('.langbot-box-volume-probe-*')) == []
|
|
await runtime.shutdown()
|
|
|
|
|
|
def test_separated_box_runtime_allows_box_owned_missing_host_path(tmp_path):
|
|
logger = Mock()
|
|
runtime = BoxRuntime(logger=logger, backends=[FakeBackend(logger)], session_ttl_sec=300)
|
|
host_root = tmp_path / 'box'
|
|
service = BoxService(make_app(logger, host_root=str(host_root)), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
service._shares_filesystem_with_box_override = False
|
|
|
|
spec = service.build_spec({'cmd': 'echo hi', 'session_id': 'missing-host-path'})
|
|
|
|
assert spec.host_path == str(host_root / 'default')
|
|
assert not (host_root / 'default').exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_get_sessions_delegates_to_client():
|
|
client = Mock()
|
|
client.get_sessions = AsyncMock(return_value=[{'session_id': 'test-session'}])
|
|
|
|
service = BoxService(make_app(Mock()), client=client)
|
|
service._available = True
|
|
|
|
sessions = await service.get_sessions(_CONTEXT)
|
|
|
|
assert sessions == [{'session_id': 'test-session'}]
|
|
client.get_sessions.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_relay_connection_is_binding_checked_and_scoped():
|
|
app = make_app(Mock())
|
|
client = Mock()
|
|
client.get_managed_process_websocket_url = Mock(
|
|
return_value='ws://box/v1/sessions/physical/managed-process/server-a/ws'
|
|
)
|
|
connector = Mock()
|
|
connector.ws_relay_base_url = 'http://box:5410'
|
|
connector.get_relay_headers = Mock(
|
|
return_value={
|
|
BOX_CONTROL_TOKEN_HEADER: 'secret',
|
|
BOX_INSTANCE_HEADER: _CONTEXT.instance_uuid,
|
|
BOX_WORKSPACE_HEADER: _CONTEXT.workspace_uuid,
|
|
BOX_PLACEMENT_GENERATION_HEADER: '1',
|
|
}
|
|
)
|
|
service = BoxService(app, client=client)
|
|
service._runtime_connector = connector
|
|
|
|
url, headers = await service.get_managed_process_websocket_connection(
|
|
_CONTEXT,
|
|
'mcp-shared',
|
|
'server-a',
|
|
)
|
|
|
|
assert url == 'ws://box/v1/sessions/physical/managed-process/server-a/ws'
|
|
assert headers[BOX_WORKSPACE_HEADER] == _CONTEXT.workspace_uuid
|
|
assert headers[BOX_PLACEMENT_GENERATION_HEADER] == '1'
|
|
app.workspace_service.get_execution_binding.assert_awaited_once_with(
|
|
_CONTEXT.workspace_uuid,
|
|
expected_generation=_CONTEXT.placement_generation,
|
|
)
|
|
action_context = connector.get_relay_headers.call_args.args[0]
|
|
assert action_context == _ACTION_CONTEXT
|
|
client.get_managed_process_websocket_url.assert_called_once_with(
|
|
'mcp-shared',
|
|
'http://box:5410',
|
|
'server-a',
|
|
action_context=_ACTION_CONTEXT,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_relay_connection_rejects_stale_binding():
|
|
app = make_app(Mock())
|
|
app.workspace_service.get_execution_binding.return_value = SimpleNamespace(
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=2,
|
|
)
|
|
client = Mock()
|
|
service = BoxService(app, client=client)
|
|
service._runtime_connector = Mock()
|
|
|
|
with pytest.raises(BoxValidationError, match='stale Workspace placement'):
|
|
await service.get_managed_process_websocket_connection(
|
|
_CONTEXT,
|
|
'mcp-shared',
|
|
'server-a',
|
|
)
|
|
|
|
service._runtime_connector.get_relay_headers.assert_not_called()
|
|
|
|
|
|
def test_box_service_dispose_delegates_to_internal_connector(monkeypatch: pytest.MonkeyPatch):
|
|
connector = Mock()
|
|
connector.client = Mock()
|
|
|
|
monkeypatch.setattr('langbot.pkg.box.service.BoxRuntimeConnector', Mock(return_value=connector))
|
|
|
|
service = BoxService(make_app(Mock()))
|
|
service.dispose()
|
|
|
|
connector.dispose.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_dispose_schedules_shutdown_on_event_loop(monkeypatch: pytest.MonkeyPatch):
|
|
connector = Mock()
|
|
connector.client = Mock()
|
|
connector.dispose = Mock()
|
|
|
|
monkeypatch.setattr('langbot.pkg.box.service.BoxRuntimeConnector', Mock(return_value=connector))
|
|
|
|
app = make_app(Mock())
|
|
loop = asyncio.get_running_loop()
|
|
app.event_loop = loop
|
|
|
|
service = BoxService(app)
|
|
service.shutdown = AsyncMock()
|
|
|
|
service.dispose()
|
|
await asyncio.sleep(0)
|
|
|
|
connector.dispose.assert_not_called()
|
|
service.shutdown.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_shutdown_reaps_connector_when_runtime_rpc_is_offline(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
connector = Mock()
|
|
connector.client = Mock()
|
|
connector.client.shutdown = AsyncMock(side_effect=RuntimeError('offline'))
|
|
connector.aclose = AsyncMock()
|
|
|
|
monkeypatch.setattr('langbot.pkg.box.service.BoxRuntimeConnector', Mock(return_value=connector))
|
|
|
|
service = BoxService(make_app(Mock()))
|
|
await service.shutdown()
|
|
|
|
connector.client.shutdown.assert_awaited_once()
|
|
connector.aclose.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_reconnect_restores_workspace_and_runs_cleanup(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
app = make_app(Mock())
|
|
app.skill_mgr = SimpleNamespace(reload_skills=AsyncMock())
|
|
service = BoxService(app, client=Mock(spec=BoxRuntimeClient))
|
|
connector = Mock()
|
|
connector.reconnect = AsyncMock()
|
|
service._ensure_default_workspace = Mock()
|
|
service._purge_attachment_dirs = AsyncMock()
|
|
monkeypatch.setattr('langbot.pkg.box.service.asyncio.sleep', AsyncMock())
|
|
|
|
await service._reconnect_loop(connector)
|
|
|
|
connector.reconnect.assert_awaited_once()
|
|
service._ensure_default_workspace.assert_called_once()
|
|
service._purge_attachment_dirs.assert_awaited_once()
|
|
app.skill_mgr.reload_skills.assert_awaited_once()
|
|
assert service.available is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_box_service_reconnect_does_not_reload_unscoped_skills(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
app = make_app(Mock())
|
|
app.skill_mgr = SimpleNamespace(reload_skills=AsyncMock())
|
|
service = BoxService(app, client=Mock(spec=BoxRuntimeClient))
|
|
service._cloud_managed = True
|
|
connector = Mock()
|
|
connector.reconnect = AsyncMock()
|
|
service._ensure_default_workspace = Mock()
|
|
service._verify_cloud_runtime = AsyncMock()
|
|
monkeypatch.setattr('langbot.pkg.box.service.asyncio.sleep', AsyncMock())
|
|
|
|
await service._reconnect_loop(connector)
|
|
|
|
connector.reconnect.assert_awaited_once()
|
|
service._verify_cloud_runtime.assert_awaited_once()
|
|
app.skill_mgr.reload_skills.assert_not_awaited()
|
|
assert service.available is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_runtime_reuses_request_session():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
first = BoxSpec.model_validate({'cmd': 'echo first', 'session_id': 'req-1'})
|
|
second = BoxSpec.model_validate({'cmd': 'echo second', 'session_id': 'req-1'})
|
|
|
|
await runtime.execute(first)
|
|
await runtime.execute(second)
|
|
|
|
assert backend.start_calls == ['req-1']
|
|
assert backend.exec_calls == [('req-1', 'echo first'), ('req-1', 'echo second')]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_defaults_session_id_from_query():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'pwd'}, make_query(7))
|
|
|
|
assert result['session_id'] == 'person_test_user'
|
|
assert result['ok'] is True
|
|
assert backend.start_calls == ['person_test_user']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_session_id_uses_query_attributes_without_variables():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
query = pipeline_query.Query.model_construct(
|
|
query_id=7,
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
launcher_type='group',
|
|
launcher_id='room-1',
|
|
)
|
|
result = await service.execute_tool({'command': 'pwd'}, query)
|
|
|
|
assert result['session_id'] == 'group_room-1'
|
|
assert result['ok'] is True
|
|
assert backend.start_calls == ['group_room-1']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_session_id_falls_back_to_query_id_for_synthetic_queries():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
query = pipeline_query.Query.model_construct(
|
|
query_id=7,
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
)
|
|
result = await service.execute_tool({'command': 'pwd'}, query)
|
|
|
|
assert result['session_id'] == 'query_7'
|
|
assert result['ok'] is True
|
|
assert backend.start_calls == ['query_7']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_forced_global_scope_overrides_pipeline_template():
|
|
"""SaaS guard: a non-empty ``force_box_session_id_template`` pins every
|
|
query to one shared sandbox regardless of the pipeline's own scope."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(
|
|
make_app(logger, force_box_session_id_template='{global}'),
|
|
client=_InProcessBoxRuntimeClient(logger, runtime),
|
|
)
|
|
await service.initialize()
|
|
|
|
# Two distinct callers that would otherwise get separate sandboxes.
|
|
q1 = pipeline_query.Query.model_construct(
|
|
query_id=1,
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
launcher_type='group',
|
|
launcher_id='room-1',
|
|
)
|
|
q2 = pipeline_query.Query.model_construct(
|
|
query_id=2,
|
|
instance_uuid=_CONTEXT.instance_uuid,
|
|
workspace_uuid=_CONTEXT.workspace_uuid,
|
|
placement_generation=_CONTEXT.placement_generation,
|
|
launcher_type='person',
|
|
launcher_id='alice',
|
|
)
|
|
|
|
r1 = await service.execute_tool({'command': 'pwd'}, q1)
|
|
r2 = await service.execute_tool({'command': 'pwd'}, q2)
|
|
|
|
assert r1['session_id'] == 'global'
|
|
assert r2['session_id'] == 'global'
|
|
# Only one sandbox was ever started — the shared global one.
|
|
assert backend.start_calls == ['global']
|
|
|
|
|
|
def test_box_service_forced_template_ignores_pipeline_config():
|
|
"""The forced template wins even when the pipeline explicitly sets a
|
|
per-user scope — proving the override is not bypassable via pipeline config."""
|
|
logger = Mock()
|
|
service = BoxService(
|
|
make_app(logger, force_box_session_id_template='{global}'),
|
|
client=Mock(spec=BoxRuntimeClient),
|
|
)
|
|
query = pipeline_query.Query.model_construct(
|
|
query_id=7,
|
|
launcher_type='person',
|
|
launcher_id='test_user',
|
|
sender_id='test_user',
|
|
pipeline_config={
|
|
'ai': {'local-agent': {'box-session-id-template': '{launcher_type}_{launcher_id}_{sender_id}'}}
|
|
},
|
|
)
|
|
|
|
assert service.resolve_box_session_id(query) == 'global'
|
|
|
|
|
|
def test_box_service_empty_forced_template_respects_pipeline_config():
|
|
"""An empty/whitespace forced template is a no-op: the pipeline's own
|
|
scope template is honoured (default non-SaaS behaviour)."""
|
|
logger = Mock()
|
|
service = BoxService(
|
|
make_app(logger, force_box_session_id_template=' '),
|
|
client=Mock(spec=BoxRuntimeClient),
|
|
)
|
|
query = pipeline_query.Query.model_construct(
|
|
query_id=7,
|
|
launcher_type='group',
|
|
launcher_id='room-1',
|
|
pipeline_config={'ai': {'local-agent': {'box-session-id-template': '{launcher_type}_{launcher_id}'}}},
|
|
)
|
|
|
|
assert service.resolve_box_session_id(query) == 'group_room-1'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_fails_closed_when_backend_unavailable():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger, available=False)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
with pytest.raises(BoxBackendUnavailableError):
|
|
await service.execute_tool({'command': 'echo hello'}, make_query(9))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_allows_host_mount_under_configured_root(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'mounted-workspace'
|
|
host_dir.mkdir()
|
|
service = BoxService(make_app(logger, [str(tmp_path)]), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_spec_payload(
|
|
{
|
|
'cmd': 'pwd',
|
|
'host_path': str(host_dir),
|
|
'host_path_mode': BoxHostMountMode.READ_WRITE.value,
|
|
'session_id': '11',
|
|
},
|
|
make_query(11),
|
|
)
|
|
|
|
assert result['ok'] is True
|
|
assert backend.start_calls == ['11']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_uses_default_workspace_when_host_path_omitted(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'default-workspace'
|
|
host_dir.mkdir()
|
|
app = make_app(logger, [str(tmp_path)])
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(host_dir)
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'pwd'}, make_query(15))
|
|
|
|
assert result['ok'] is True
|
|
assert backend.start_calls == ['person_test_user']
|
|
assert backend.exec_calls == [('person_test_user', 'pwd')]
|
|
assert backend.start_specs[0].host_path == service._tenant_workspace(_CONTEXT)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_creates_default_workspace_on_initialize(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
allowed_root = tmp_path / 'allowed-root'
|
|
allowed_root.mkdir()
|
|
default_workspace = allowed_root / 'default-workspace'
|
|
app = make_app(logger, [str(allowed_root)])
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(default_workspace)
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
service._shares_filesystem_with_box_override = True
|
|
|
|
await service.initialize()
|
|
|
|
assert default_workspace.is_dir()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_derives_workspace_and_allowed_root_from_host_root(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
shared_root = tmp_path / 'shared-box-root'
|
|
app = make_app(logger, host_root=str(shared_root))
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
service._shares_filesystem_with_box_override = True
|
|
|
|
await service.initialize()
|
|
|
|
assert service.host_root == os.path.realpath(shared_root)
|
|
assert service.default_workspace == os.path.realpath(shared_root / 'default')
|
|
assert service.allowed_mount_roots == [os.path.realpath(shared_root)]
|
|
assert (shared_root / 'default').is_dir()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_rejects_host_mount_outside_allowed_roots(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
allowed_root = tmp_path / 'allowed'
|
|
disallowed_root = tmp_path / 'disallowed'
|
|
allowed_root.mkdir()
|
|
disallowed_root.mkdir()
|
|
service = BoxService(make_app(logger, [str(allowed_root)]), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
with pytest.raises(BoxValidationError):
|
|
await service.execute_spec_payload(
|
|
{
|
|
'cmd': 'pwd',
|
|
'host_path': str(disallowed_root),
|
|
'session_id': '12',
|
|
},
|
|
make_query(12),
|
|
)
|
|
|
|
|
|
class TestGetSystemGuidance:
|
|
"""``get_system_guidance`` must ALWAYS advertise the per-query outbox path
|
|
when given a ``query_id`` — even with no inbound attachment — so files the
|
|
agent generates (QR codes, charts, rendered docs) are actually delivered.
|
|
|
|
The wrapper collects the outbox on every turn regardless of inbound files;
|
|
before this, the agent was only told the outbox path inside the
|
|
inbound-attachment note, so pure-generation turns produced files that were
|
|
silently dropped.
|
|
"""
|
|
|
|
def _service(self, logger=None):
|
|
logger = logger or Mock()
|
|
runtime = BoxRuntime(logger=logger, backends=[FakeBackend(logger)], session_ttl_sec=300)
|
|
return BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
def test_guidance_includes_outbox_when_query_id_given(self):
|
|
service = self._service()
|
|
guidance = service.get_system_guidance(42)
|
|
assert f'{service.OUTBOX_MOUNT_DIR}/42' in guidance
|
|
assert 'delivered to the user automatically' in guidance
|
|
|
|
def test_guidance_omits_outbox_without_query_id(self):
|
|
service = self._service()
|
|
guidance = service.get_system_guidance()
|
|
assert service.OUTBOX_MOUNT_DIR not in guidance
|
|
# core exec guidance is still present
|
|
assert 'exec tool' in guidance
|
|
|
|
def test_guidance_outbox_independent_of_inbound_attachments(self):
|
|
# A bare query_id (the pure-generation case) still gets the outbox note.
|
|
service = self._service()
|
|
assert f'{service.OUTBOX_MOUNT_DIR}/0' in service.get_system_guidance(0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_runtime_rejects_host_mount_conflict_in_same_session(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
first_host_dir = tmp_path / 'first'
|
|
second_host_dir = tmp_path / 'second'
|
|
first_host_dir.mkdir()
|
|
second_host_dir.mkdir()
|
|
|
|
first = BoxSpec.model_validate(
|
|
{
|
|
'cmd': 'echo first',
|
|
'session_id': 'req-mount',
|
|
'host_path': os.path.realpath(first_host_dir),
|
|
}
|
|
)
|
|
second = BoxSpec.model_validate(
|
|
{
|
|
'cmd': 'echo second',
|
|
'session_id': 'req-mount',
|
|
'host_path': os.path.realpath(second_host_dir),
|
|
}
|
|
)
|
|
|
|
await runtime.execute(first)
|
|
|
|
with pytest.raises(BoxSessionConflictError):
|
|
await runtime.execute(second)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_runtime_rejects_resource_limit_conflict_in_same_session():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
first = BoxSpec.model_validate({'cmd': 'echo first', 'session_id': 'req-resource', 'cpus': 1.0})
|
|
second = BoxSpec.model_validate({'cmd': 'echo second', 'session_id': 'req-resource', 'cpus': 2.0})
|
|
|
|
await runtime.execute(first)
|
|
|
|
with pytest.raises(BoxSessionConflictError):
|
|
await runtime.execute(second)
|
|
|
|
|
|
# ── Truncation tests ──────────────────────────────────────────────────
|
|
|
|
|
|
class FakeBackendWithOutput(FakeBackend):
|
|
"""FakeBackend that returns configurable stdout/stderr."""
|
|
|
|
def __init__(self, logger: Mock, stdout: str = '', stderr: str = ''):
|
|
super().__init__(logger)
|
|
self._stdout = stdout
|
|
self._stderr = stderr
|
|
|
|
async def exec(self, session: BoxSessionInfo, spec: BoxSpec) -> BoxExecutionResult:
|
|
self.exec_calls.append((session.session_id, spec.cmd))
|
|
return BoxExecutionResult(
|
|
session_id=session.session_id,
|
|
backend_name=self.name,
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout=self._stdout,
|
|
stderr=self._stderr,
|
|
duration_ms=5,
|
|
)
|
|
|
|
|
|
class FakeBackendWritingFiles(FakeBackend):
|
|
"""Fake backend that writes files into the mounted host workspace during exec."""
|
|
|
|
def __init__(self, logger: Mock, files_to_write: list[tuple[str, int]]):
|
|
super().__init__(logger)
|
|
self._files_to_write = files_to_write
|
|
|
|
async def exec(self, session: BoxSessionInfo, spec: BoxSpec) -> BoxExecutionResult:
|
|
self.exec_calls.append((session.session_id, spec.cmd))
|
|
if session.host_path:
|
|
for relative_path, size in self._files_to_write:
|
|
host_path = os.path.join(session.host_path, relative_path)
|
|
os.makedirs(os.path.dirname(host_path), exist_ok=True)
|
|
with open(host_path, 'wb') as f:
|
|
f.write(b'x' * size)
|
|
return BoxExecutionResult(
|
|
session_id=session.session_id,
|
|
backend_name=self.name,
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout='wrote files',
|
|
stderr='',
|
|
duration_ms=5,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncate_short_output_unchanged():
|
|
logger = Mock()
|
|
backend = FakeBackendWithOutput(logger, stdout='hello world')
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime), output_limit_chars=100)
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'echo hello'}, make_query(20))
|
|
|
|
assert result['stdout'] == 'hello world'
|
|
assert result['stdout_truncated'] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncate_preserves_head_and_tail():
|
|
logger = Mock()
|
|
# Build output: "AAAA...BBB..." where each section is identifiable
|
|
head_marker = 'HEAD_START|'
|
|
tail_marker = '|TAIL_END'
|
|
filler = 'x' * 500
|
|
big_output = f'{head_marker}{filler}{tail_marker}'
|
|
|
|
backend = FakeBackendWithOutput(logger, stdout=big_output)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
limit = 100
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime), output_limit_chars=limit)
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'cat big'}, make_query(21))
|
|
|
|
assert result['stdout_truncated'] is True
|
|
stdout = result['stdout']
|
|
# Head part should contain the head marker
|
|
assert stdout.startswith(head_marker)
|
|
# Tail part should contain the tail marker
|
|
assert stdout.endswith(tail_marker)
|
|
# Should contain the truncation notice
|
|
assert 'characters truncated' in stdout
|
|
assert len(stdout) <= limit
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncate_at_exact_limit_not_truncated():
|
|
logger = Mock()
|
|
exact_output = 'a' * 200
|
|
backend = FakeBackendWithOutput(logger, stdout=exact_output)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime), output_limit_chars=200)
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'echo a'}, make_query(22))
|
|
|
|
assert result['stdout'] == exact_output
|
|
assert result['stdout_truncated'] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncate_stderr_independently():
|
|
logger = Mock()
|
|
backend = FakeBackendWithOutput(logger, stdout='short', stderr='E' * 300)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime), output_limit_chars=100)
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'fail'}, make_query(23))
|
|
|
|
assert result['stdout_truncated'] is False
|
|
assert result['stderr_truncated'] is True
|
|
assert 'characters truncated' in result['stderr']
|
|
assert len(result['stderr']) <= 100
|
|
|
|
|
|
# ── Profile tests ─────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_default_provides_defaults():
|
|
"""When tool call omits network/image, profile defaults are used."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'echo hi'}, make_query(30))
|
|
|
|
assert result['ok'] is True
|
|
spec = backend.start_specs[0]
|
|
profile = BUILTIN_PROFILES['default']
|
|
assert spec.network == BoxNetworkMode.OFF
|
|
assert spec.image == profile.image
|
|
assert spec.timeout_sec == profile.timeout_sec
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_unlocked_field_can_be_overridden():
|
|
"""Spec payload can override unlocked profile fields."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_spec_payload(
|
|
{'cmd': 'echo hi', 'timeout_sec': 60, 'network': 'on', 'session_id': '31'},
|
|
make_query(31),
|
|
)
|
|
|
|
assert result['ok'] is True
|
|
spec = backend.start_specs[0]
|
|
assert spec.timeout_sec == 60
|
|
assert spec.network == BoxNetworkMode.ON
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_locked_field_cannot_be_overridden():
|
|
"""offline_readonly profile locks network and host_path_mode."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(
|
|
make_app(logger, profile='offline_readonly'), client=_InProcessBoxRuntimeClient(logger, runtime)
|
|
)
|
|
await service.initialize()
|
|
|
|
result = await service.execute_spec_payload(
|
|
{'cmd': 'echo hi', 'network': 'on', 'host_path_mode': 'rw', 'session_id': '32'},
|
|
make_query(32),
|
|
)
|
|
|
|
assert result['ok'] is True
|
|
spec = backend.start_specs[0]
|
|
assert spec.network == BoxNetworkMode.OFF
|
|
assert spec.host_path_mode == BoxHostMountMode.READ_ONLY
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_timeout_clamped_to_max():
|
|
"""timeout_sec exceeding max_timeout_sec is clamped."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
result = await service.execute_tool({'command': 'echo hi', 'timeout_sec': 999}, make_query(33))
|
|
|
|
assert result['ok'] is True
|
|
spec = backend.start_specs[0]
|
|
# default profile max_timeout_sec = 120
|
|
assert spec.timeout_sec == 120
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('timeout_value', ['999', 999.0])
|
|
async def test_profile_timeout_clamped_for_coercible_inputs(timeout_value):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
await service.execute_tool({'command': 'echo hi', 'timeout_sec': timeout_value}, make_query(34))
|
|
|
|
spec = backend.start_specs[0]
|
|
assert spec.timeout_sec == 120
|
|
|
|
|
|
def test_unknown_profile_raises_error():
|
|
"""Config referencing a non-existent profile name raises immediately."""
|
|
logger = Mock()
|
|
runtime = BoxRuntime(logger=logger, backends=[FakeBackend(logger)], session_ttl_sec=300)
|
|
with pytest.raises(BoxValidationError, match='unknown box profile'):
|
|
BoxService(make_app(logger, profile='nonexistent'), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
|
|
def test_builtin_profiles_are_consistent():
|
|
"""Basic sanity check on all built-in profiles."""
|
|
assert 'default' in BUILTIN_PROFILES
|
|
assert 'offline_readonly' in BUILTIN_PROFILES
|
|
assert 'network_basic' in BUILTIN_PROFILES
|
|
assert 'network_extended' in BUILTIN_PROFILES
|
|
|
|
offline = BUILTIN_PROFILES['offline_readonly']
|
|
assert offline.network == BoxNetworkMode.OFF
|
|
assert offline.host_path_mode == BoxHostMountMode.READ_ONLY
|
|
assert 'network' in offline.locked
|
|
assert 'host_path_mode' in offline.locked
|
|
assert 'read_only_rootfs' in offline.locked
|
|
assert offline.max_timeout_sec <= BUILTIN_PROFILES['default'].max_timeout_sec
|
|
|
|
basic = BUILTIN_PROFILES['network_basic']
|
|
assert basic.network == BoxNetworkMode.ON
|
|
assert basic.read_only_rootfs is True
|
|
|
|
extended = BUILTIN_PROFILES['network_extended']
|
|
assert extended.network == BoxNetworkMode.ON
|
|
assert extended.read_only_rootfs is False
|
|
assert extended.cpus > BUILTIN_PROFILES['default'].cpus
|
|
assert extended.memory_mb > BUILTIN_PROFILES['default'].memory_mb
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_default_applies_resource_limits():
|
|
"""Default profile resource limits are applied to BoxSpec."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
await service.execute_tool({'command': 'echo hi'}, make_query(40))
|
|
|
|
spec = backend.start_specs[0]
|
|
profile = BUILTIN_PROFILES['default']
|
|
assert spec.cpus == profile.cpus
|
|
assert spec.memory_mb == profile.memory_mb
|
|
assert spec.pids_limit == profile.pids_limit
|
|
assert spec.read_only_rootfs == profile.read_only_rootfs
|
|
assert spec.workspace_quota_mb == profile.workspace_quota_mb
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_applies_workspace_quota_from_config(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'default-workspace'
|
|
host_dir.mkdir()
|
|
app = make_app(logger, [str(tmp_path)], workspace_quota_mb=32)
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(host_dir)
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
await service.initialize()
|
|
await service.execute_tool({'command': 'echo hi'}, make_query(43))
|
|
|
|
assert backend.start_specs[0].workspace_quota_mb == 32
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_rejects_execution_when_workspace_already_exceeds_quota(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'quota-workspace'
|
|
host_dir.mkdir()
|
|
app = make_app(logger, [str(tmp_path)], workspace_quota_mb=1)
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(host_dir)
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
tenant_host_dir = service._tenant_workspace(_CONTEXT)
|
|
assert tenant_host_dir is not None
|
|
os.makedirs(tenant_host_dir, exist_ok=True)
|
|
with open(os.path.join(tenant_host_dir, 'already-too-large.bin'), 'wb') as handle:
|
|
handle.write(b'x' * (2 * 1024 * 1024))
|
|
|
|
await service.initialize()
|
|
|
|
with pytest.raises(BoxValidationError, match='workspace quota exceeded before execution'):
|
|
await service.execute_tool({'command': 'echo hi'}, make_query(44))
|
|
|
|
assert backend.start_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_rejects_and_cleans_up_when_execution_exceeds_workspace_quota(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackendWritingFiles(logger, files_to_write=[('output.bin', 2 * 1024 * 1024)])
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'quota-workspace-post'
|
|
host_dir.mkdir()
|
|
app = make_app(logger, [str(tmp_path)], workspace_quota_mb=1)
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(host_dir)
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
await service.initialize()
|
|
|
|
with pytest.raises(BoxValidationError, match='workspace quota exceeded after execution'):
|
|
await service.execute_tool({'command': 'generate-output'}, make_query(45))
|
|
|
|
assert backend.start_calls == ['person_test_user']
|
|
assert backend.stop_calls == ['person_test_user']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_box_service_rejects_workspace_inode_bomb_before_execution(tmp_path):
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
host_dir = tmp_path / 'quota-workspace-entries'
|
|
host_dir.mkdir()
|
|
app = make_app(logger, [str(tmp_path)], workspace_quota_mb=1)
|
|
app.instance_config.data['box']['local']['default_workspace'] = str(host_dir)
|
|
app.instance_config.data['box']['limits'] = {'max_workspace_entries': 2}
|
|
service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
|
|
tenant_host_dir = service._tenant_workspace(_CONTEXT)
|
|
assert tenant_host_dir is not None
|
|
os.makedirs(tenant_host_dir, exist_ok=True)
|
|
for index in range(3):
|
|
pathlib.Path(tenant_host_dir, f'tiny-{index}').write_bytes(b'x')
|
|
|
|
await service.initialize()
|
|
|
|
with pytest.raises(BoxValidationError, match='workspace entry limit exceeded before execution'):
|
|
await service.execute_tool({'command': 'echo hi'}, make_query(46))
|
|
|
|
assert backend.start_calls == []
|
|
|
|
|
|
def test_box_service_workspace_entry_limit_is_hard_clamped():
|
|
app = make_app(Mock())
|
|
app.instance_config.data['box']['limits'] = {'max_workspace_entries': 10_000_000}
|
|
service = BoxService(app, client=Mock(spec=BoxRuntimeClient))
|
|
assert service._max_workspace_entries() == 1_000_000
|
|
|
|
app.instance_config.data['box']['limits']['max_workspace_entries'] = 0
|
|
assert service._max_workspace_entries() == 1
|
|
|
|
app.instance_config.data['box']['limits']['max_workspace_entries'] = 'invalid'
|
|
assert service._max_workspace_entries() == 100_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_offline_readonly_locks_read_only_rootfs():
|
|
"""offline_readonly locks read_only_rootfs so it cannot be overridden."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(
|
|
make_app(logger, profile='offline_readonly'), client=_InProcessBoxRuntimeClient(logger, runtime)
|
|
)
|
|
await service.initialize()
|
|
|
|
await service.execute_spec_payload(
|
|
{'cmd': 'echo hi', 'read_only_rootfs': False, 'session_id': '41'}, make_query(41)
|
|
)
|
|
|
|
spec = backend.start_specs[0]
|
|
assert spec.read_only_rootfs is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_profile_network_extended_has_relaxed_limits():
|
|
"""network_extended profile provides higher resource limits."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(
|
|
make_app(logger, profile='network_extended'), client=_InProcessBoxRuntimeClient(logger, runtime)
|
|
)
|
|
await service.initialize()
|
|
|
|
await service.execute_tool({'command': 'echo hi'}, make_query(42))
|
|
|
|
spec = backend.start_specs[0]
|
|
assert spec.network == BoxNetworkMode.ON
|
|
assert spec.cpus == 2.0
|
|
assert spec.memory_mb == 1024
|
|
assert spec.read_only_rootfs is False
|
|
|
|
|
|
def test_box_spec_validates_resource_limits():
|
|
"""BoxSpec rejects invalid resource limit values."""
|
|
with pytest.raises(Exception):
|
|
BoxSpec.model_validate({'cmd': 'echo', 'session_id': 's1', 'cpus': 0})
|
|
with pytest.raises(Exception):
|
|
BoxSpec.model_validate({'cmd': 'echo', 'session_id': 's1', 'memory_mb': 10})
|
|
with pytest.raises(Exception):
|
|
BoxSpec.model_validate({'cmd': 'echo', 'session_id': 's1', 'pids_limit': 0})
|
|
with pytest.raises(Exception):
|
|
BoxSpec.model_validate({'cmd': 'echo', 'session_id': 's1', 'workspace_quota_mb': -1})
|
|
|
|
|
|
# ── Observability tests ───────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_get_status_reports_backend_and_sessions():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
status = await runtime.get_status()
|
|
assert status['backend']['name'] == 'fake'
|
|
assert status['backend']['available'] is True
|
|
assert status['active_sessions'] == 0
|
|
|
|
await runtime.execute(BoxSpec.model_validate({'cmd': 'echo', 'session_id': 'obs-1'}))
|
|
status = await runtime.get_status()
|
|
assert status['active_sessions'] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_get_sessions_returns_session_info():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
await runtime.execute(BoxSpec.model_validate({'cmd': 'echo', 'session_id': 'obs-2'}))
|
|
sessions = runtime.get_sessions()
|
|
assert len(sessions) == 1
|
|
assert sessions[0]['session_id'] == 'obs-2'
|
|
assert sessions[0]['backend_name'] == 'fake'
|
|
assert 'created_at' in sessions[0]
|
|
assert 'last_used_at' in sessions[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_get_backend_info_when_no_backend():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger, available=False)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
info = await runtime.get_backend_info()
|
|
assert info['name'] is None
|
|
assert info['available'] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_records_errors_on_failure():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger, available=False)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
with pytest.raises(Exception):
|
|
await service.execute_tool({'command': 'echo hello'}, make_query(50))
|
|
|
|
errors = service.get_recent_errors(_CONTEXT)
|
|
assert len(errors) == 1
|
|
assert errors[0]['type'] == 'BoxBackendUnavailableError'
|
|
assert errors[0]['query_id'] == '50'
|
|
assert 'timestamp' in errors[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_error_ring_buffer_capped():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger, available=False)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
for i in range(60):
|
|
with pytest.raises(Exception):
|
|
await service.execute_tool({'command': 'fail'}, make_query(100 + i))
|
|
|
|
errors = service.get_recent_errors(_CONTEXT)
|
|
assert len(errors) == 50
|
|
# Oldest should have been evicted, newest kept
|
|
assert errors[0]['query_id'] == '110'
|
|
assert errors[-1]['query_id'] == '159'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_get_status_aggregates_runtime_and_profile():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
|
|
await service.initialize()
|
|
|
|
status = await service.get_status(_CONTEXT)
|
|
assert status['profile'] == 'default'
|
|
assert status['backend']['name'] == 'fake'
|
|
assert status['backend']['available'] is True
|
|
assert status['active_sessions'] == 0
|
|
assert status['recent_error_count'] == 0
|
|
|
|
|
|
# ── In-process RPC client/server tests ─────────────────────────────────
|
|
|
|
|
|
class _QueueConnection:
|
|
"""In-process Connection backed by asyncio Queues — no real IO."""
|
|
|
|
def __init__(self, rx: asyncio.Queue[str], tx: asyncio.Queue[str]):
|
|
self._rx = rx
|
|
self._tx = tx
|
|
|
|
async def send(self, message: str) -> None:
|
|
await self._tx.put(message)
|
|
|
|
async def receive(self) -> str:
|
|
return await self._rx.get()
|
|
|
|
async def close(self) -> None:
|
|
pass
|
|
|
|
|
|
def _make_queue_connection_pair():
|
|
"""Return (client_conn, server_conn) linked by queues."""
|
|
c2s: asyncio.Queue[str] = asyncio.Queue()
|
|
s2c: asyncio.Queue[str] = asyncio.Queue()
|
|
client_conn = _QueueConnection(rx=s2c, tx=c2s)
|
|
server_conn = _QueueConnection(rx=c2s, tx=s2c)
|
|
return client_conn, server_conn
|
|
|
|
|
|
async def _make_rpc_pair(runtime: BoxRuntime):
|
|
"""Create an in-process (ActionRPCBoxClient, server_task, client_task) connected via queues."""
|
|
from langbot_plugin.box.server import BoxServerHandler
|
|
from langbot_plugin.runtime.io.handler import Handler
|
|
|
|
client_conn, server_conn = _make_queue_connection_pair()
|
|
|
|
server_handler = BoxServerHandler(
|
|
server_conn,
|
|
runtime,
|
|
host_control_authenticated=True,
|
|
trusted_instance_uuid=_CONTEXT.instance_uuid,
|
|
)
|
|
server_task = asyncio.create_task(server_handler.run())
|
|
|
|
client_handler = Handler.__new__(Handler)
|
|
Handler.__init__(client_handler, client_conn)
|
|
client_task = asyncio.create_task(client_handler.run())
|
|
|
|
client = ActionRPCBoxClient(logger=Mock())
|
|
client.set_handler(client_handler)
|
|
|
|
return client, server_task, client_task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_execute():
|
|
"""ActionRPCBoxClient correctly calls server and parses result."""
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec = BoxSpec.model_validate({'cmd': 'echo remote', 'session_id': 'r-1'})
|
|
result = await client.execute(spec, action_context=_ACTION_CONTEXT)
|
|
|
|
assert result.session_id == 'r-1'
|
|
assert result.status == BoxExecutionStatus.COMPLETED
|
|
assert result.exit_code == 0
|
|
assert result.stdout == 'executed: echo remote'
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_get_sessions():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec = BoxSpec.model_validate({'cmd': 'echo hi', 'session_id': 'r-2'})
|
|
await client.execute(spec, action_context=_ACTION_CONTEXT)
|
|
|
|
sessions = await client.get_sessions(action_context=_ACTION_CONTEXT)
|
|
assert len(sessions) == 1
|
|
assert sessions[0]['session_id'] == 'r-2'
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_generation_advance_retires_old_session_and_rejects_old_context():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
second_context = _ACTION_CONTEXT.model_copy(update={'placement_generation': 2})
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec = BoxSpec.model_validate({'cmd': 'echo generation', 'session_id': 'shared'})
|
|
await client.execute(spec, action_context=_ACTION_CONTEXT)
|
|
await client.execute(spec, action_context=second_context)
|
|
|
|
assert len(backend.start_calls) == 2
|
|
assert backend.start_calls[0] != backend.start_calls[1]
|
|
assert backend.stop_calls == [backend.start_calls[0]]
|
|
assert [session['session_id'] for session in await client.get_sessions(action_context=second_context)] == [
|
|
'shared'
|
|
]
|
|
with pytest.raises(BoxError, match='Stale Box placement generation'):
|
|
await client.execute(spec, action_context=_ACTION_CONTEXT)
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_get_status():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
status = await client.get_status(action_context=_ACTION_CONTEXT)
|
|
|
|
assert 'backend' in status
|
|
assert 'active_sessions' in status
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_get_backend_info():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
info = await client.get_backend_info()
|
|
|
|
assert info['name'] == 'fake'
|
|
assert info['available'] is True
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
# ── RPC-based delete/create/conflict tests ────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_delete_session():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec = BoxSpec.model_validate({'cmd': 'echo hi', 'session_id': 'r-del-1'})
|
|
await client.execute(spec, action_context=_ACTION_CONTEXT)
|
|
|
|
await client.delete_session('r-del-1', action_context=_ACTION_CONTEXT)
|
|
|
|
sessions = await client.get_sessions(action_context=_ACTION_CONTEXT)
|
|
assert len(sessions) == 0
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_delete_session_raises_not_found():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
with pytest.raises(BoxSessionNotFoundError):
|
|
await client.delete_session('nonexistent', action_context=_ACTION_CONTEXT)
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_create_session():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec = BoxSpec.model_validate({'cmd': 'placeholder', 'session_id': 'r-create-1'})
|
|
info = await client.create_session(spec, action_context=_ACTION_CONTEXT)
|
|
assert info['session_id'] == 'r-create-1'
|
|
assert info['backend_name'] == 'fake'
|
|
|
|
sessions = await client.get_sessions(action_context=_ACTION_CONTEXT)
|
|
assert len(sessions) == 1
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_client_exec_raises_conflict_error():
|
|
logger = Mock()
|
|
backend = FakeBackend(logger)
|
|
runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
|
|
await runtime.initialize()
|
|
|
|
client, server_task, client_task = await _make_rpc_pair(runtime)
|
|
try:
|
|
spec1 = BoxSpec.model_validate({'cmd': 'echo first', 'session_id': 'r-conflict-1', 'network': 'off'})
|
|
await client.execute(spec1, action_context=_ACTION_CONTEXT)
|
|
|
|
spec2 = BoxSpec.model_validate({'cmd': 'echo second', 'session_id': 'r-conflict-1', 'network': 'on'})
|
|
with pytest.raises(BoxSessionConflictError):
|
|
await client.execute(spec2, action_context=_ACTION_CONTEXT)
|
|
finally:
|
|
server_task.cancel()
|
|
client_task.cancel()
|
|
await runtime.shutdown()
|
|
|
|
|
|
# ── BoxHostMountMode.NONE tests ─────────────────────────────────────
|
|
|
|
|
|
class TestBoxHostMountModeNone:
|
|
def test_none_mode_is_valid_enum(self):
|
|
assert BoxHostMountMode.NONE.value == 'none'
|
|
|
|
def test_spec_with_none_mode_skips_workdir_check(self):
|
|
"""When host_path_mode is NONE, workdir validation is skipped."""
|
|
spec = BoxSpec(
|
|
session_id='test',
|
|
cmd='echo hi',
|
|
host_path='/home/user/data',
|
|
host_path_mode=BoxHostMountMode.NONE,
|
|
workdir='/opt/custom', # Not under /workspace, should be allowed
|
|
)
|
|
assert spec.host_path_mode == BoxHostMountMode.NONE
|
|
assert spec.workdir == '/opt/custom'
|
|
|
|
def test_spec_with_rw_mode_requires_workspace_workdir(self):
|
|
"""When host_path_mode is RW, workdir must be under mount_path."""
|
|
with pytest.raises(Exception):
|
|
BoxSpec(
|
|
session_id='test',
|
|
cmd='echo hi',
|
|
host_path='/home/user/data',
|
|
host_path_mode=BoxHostMountMode.READ_WRITE,
|
|
workdir='/opt/custom',
|
|
)
|
|
|
|
def test_spec_with_ro_mode_requires_workspace_workdir(self):
|
|
"""When host_path_mode is RO, workdir must be under mount_path."""
|
|
with pytest.raises(Exception):
|
|
BoxSpec(
|
|
session_id='test',
|
|
cmd='echo hi',
|
|
host_path='/home/user/data',
|
|
host_path_mode=BoxHostMountMode.READ_ONLY,
|
|
workdir='/opt/custom',
|
|
)
|
|
|
|
def test_spec_with_custom_mount_path_allows_matching_workdir(self):
|
|
spec = BoxSpec(
|
|
session_id='test',
|
|
cmd='echo hi',
|
|
host_path='/home/user/data',
|
|
host_path_mode=BoxHostMountMode.READ_WRITE,
|
|
mount_path='/project',
|
|
workdir='/project/src',
|
|
)
|
|
assert spec.mount_path == '/project'
|
|
assert spec.workdir == '/project/src'
|
|
|
|
def test_spec_with_custom_mount_path_rejects_outside_workdir(self):
|
|
with pytest.raises(Exception):
|
|
BoxSpec(
|
|
session_id='test',
|
|
cmd='echo hi',
|
|
host_path='/home/user/data',
|
|
host_path_mode=BoxHostMountMode.READ_WRITE,
|
|
mount_path='/project',
|
|
workdir='/workspace',
|
|
)
|
|
|
|
|
|
class TestBoxDisabledByConfig:
|
|
"""``box.enabled = false`` must keep the BoxService usable as a status
|
|
surface but skip every connection attempt and report unavailable."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_skips_connector_when_disabled(self):
|
|
logger = Mock()
|
|
app = make_app(logger, enabled=False)
|
|
client = Mock(spec=BoxRuntimeClient)
|
|
client.initialize = AsyncMock()
|
|
service = BoxService(app, client=client)
|
|
|
|
await service.initialize()
|
|
|
|
# The client must not be touched; we did not even open a connection.
|
|
client.initialize.assert_not_awaited()
|
|
assert service.enabled is False
|
|
assert service.available is False
|
|
# The reason is captured so the dashboard / UI can show it.
|
|
assert 'disabled' in service._connector_error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_reports_disabled(self):
|
|
logger = Mock()
|
|
service = BoxService(make_app(logger, enabled=False), client=Mock(spec=BoxRuntimeClient))
|
|
await service.initialize()
|
|
|
|
status = await service.get_status(_CONTEXT)
|
|
|
|
assert status['available'] is False
|
|
assert status['enabled'] is False
|
|
assert 'disabled' in status['connector_error'].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_distinguishes_enabled_but_unavailable(self):
|
|
logger = Mock()
|
|
client = Mock(spec=BoxRuntimeClient)
|
|
client.initialize = AsyncMock(side_effect=RuntimeError('docker daemon not running'))
|
|
service = BoxService(make_app(logger, enabled=True), client=client)
|
|
|
|
await service.initialize()
|
|
|
|
status = await service.get_status(_CONTEXT)
|
|
assert status['available'] is False
|
|
assert status['enabled'] is True
|
|
assert 'docker daemon' in status['connector_error']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_downgrades_available_when_backend_dead(self):
|
|
"""The connector can be healthy while the runtime reports no usable
|
|
backend (operator selected nsjail but binary missing, Docker daemon
|
|
crashed after handshake, ...). The top-level ``available`` must
|
|
reflect the combined state so the dashboard / useBoxStatus hook /
|
|
skill_service gate stay consistent with the native-tool gate."""
|
|
logger = Mock()
|
|
client = Mock(spec=BoxRuntimeClient)
|
|
client.initialize = AsyncMock()
|
|
client.get_status = AsyncMock(
|
|
return_value={
|
|
'backend': {'name': 'nsjail', 'available': False},
|
|
'active_sessions': 0,
|
|
}
|
|
)
|
|
service = BoxService(make_app(logger, enabled=True), client=client)
|
|
await service.initialize()
|
|
|
|
status = await service.get_status(_CONTEXT)
|
|
assert status['available'] is False
|
|
assert status['enabled'] is True
|
|
# The detailed backend object is preserved for the dialog
|
|
assert status['backend'] == {'name': 'nsjail', 'available': False}
|
|
assert 'nsjail' in status['connector_error']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_status_keeps_available_true_when_backend_ok(self):
|
|
logger = Mock()
|
|
client = Mock(spec=BoxRuntimeClient)
|
|
client.initialize = AsyncMock()
|
|
client.get_status = AsyncMock(
|
|
return_value={
|
|
'backend': {'name': 'docker', 'available': True},
|
|
'active_sessions': 2,
|
|
}
|
|
)
|
|
service = BoxService(make_app(logger, enabled=True), client=client)
|
|
await service.initialize()
|
|
|
|
status = await service.get_status(_CONTEXT)
|
|
assert status['available'] is True
|
|
assert status['backend'] == {'name': 'docker', 'available': True}
|
|
# No spurious connector_error overlay when everything is healthy
|
|
assert 'connector_error' not in status or not status['connector_error']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_callback_is_no_op_when_disabled(self):
|
|
logger = Mock()
|
|
service = BoxService(make_app(logger, enabled=False), client=Mock(spec=BoxRuntimeClient))
|
|
|
|
# Should be safe to fire; must not flip reconnect state on a disabled
|
|
# service. If it tried to schedule a reconnect, the test would hang.
|
|
await service._on_runtime_disconnect(connector=Mock())
|
|
|
|
assert service._reconnecting is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_callback_does_not_schedule_on_closing_event_loop(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
closed_loop = Mock()
|
|
closed_loop.is_closed.return_value = True
|
|
monkeypatch.setattr('langbot.pkg.box.service.asyncio.get_running_loop', Mock(return_value=closed_loop))
|
|
|
|
await service._on_runtime_disconnect(connector=Mock())
|
|
|
|
closed_loop.create_task.assert_not_called()
|
|
assert service._reconnect_task is None
|
|
assert service._reconnecting is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_callback_closes_reconnect_coroutine_when_task_creation_races_with_loop_close(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
loop = Mock()
|
|
loop.is_closed.return_value = False
|
|
loop.create_task.side_effect = RuntimeError('event loop is closed')
|
|
monkeypatch.setattr('langbot.pkg.box.service.asyncio.get_running_loop', Mock(return_value=loop))
|
|
|
|
async def reconnect():
|
|
await asyncio.Event().wait()
|
|
|
|
reconnect_coroutine = reconnect()
|
|
service._reconnect_loop = Mock(return_value=reconnect_coroutine)
|
|
|
|
await service._on_runtime_disconnect(connector=Mock())
|
|
|
|
loop.create_task.assert_called_once_with(reconnect_coroutine)
|
|
assert reconnect_coroutine.cr_frame is None
|
|
assert service._reconnect_task is None
|
|
assert service._reconnecting is False
|
|
|
|
|
|
def test_disconnect_callback_does_not_schedule_without_running_event_loop():
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
callback = service._on_runtime_disconnect(connector=Mock())
|
|
|
|
with pytest.raises(StopIteration):
|
|
callback.send(None)
|
|
|
|
assert service._reconnect_task is None
|
|
assert service._reconnecting is False
|
|
|
|
|
|
class TestBuildSkillExtraMounts:
|
|
"""Robustness of skill mount construction against a stale skill cache.
|
|
|
|
The three sandbox backends behave inconsistently when a skill's
|
|
package_root no longer exists on disk (nsjail aborts the whole sandbox
|
|
start, Docker silently auto-creates a root-owned empty directory, E2B
|
|
silently skips). Mount construction must filter these out up front so
|
|
the backend never sees a bad mount.
|
|
"""
|
|
|
|
def _make_service(self, logger, skills, *, shares_filesystem=True):
|
|
app = make_app(logger)
|
|
app.skill_mgr = SimpleNamespace(skills=skills, get_skills=Mock(return_value=skills))
|
|
client = Mock(spec=BoxRuntimeClient)
|
|
service = BoxService(app, client=client)
|
|
# Tests construct BoxService with an injected client (no connector), so
|
|
# set the topology explicitly. Most cases exercise the shared-fs (local
|
|
# stdio) path where local package_root validation applies.
|
|
service._shares_filesystem_with_box_override = shares_filesystem
|
|
return service
|
|
|
|
def test_skips_skill_with_missing_package_root(self):
|
|
logger = Mock()
|
|
with tempfile.TemporaryDirectory() as live_dir:
|
|
skills = {
|
|
'alive': {'name': 'alive', 'package_root': live_dir},
|
|
'ghost': {'name': 'ghost', 'package_root': '/nonexistent/path/should/never/exist'},
|
|
}
|
|
service = self._make_service(logger, skills)
|
|
query = make_query()
|
|
|
|
mounts = service.build_skill_extra_mounts(query)
|
|
|
|
assert mounts == [
|
|
{
|
|
'host_path': live_dir,
|
|
'mount_path': '/workspace/.skills/alive',
|
|
'mode': 'rw',
|
|
}
|
|
]
|
|
# Warning logged so operators can see what was dropped
|
|
assert any(
|
|
'ghost' in str(call.args[0]) and 'package_root missing' in str(call.args[0])
|
|
for call in logger.warning.call_args_list
|
|
)
|
|
|
|
def test_trusts_box_paths_when_filesystem_not_shared(self):
|
|
"""In separated deployments (Docker Compose, k8s sidecar,
|
|
--standalone-box, remote endpoint) the Box runtime owns its own
|
|
filesystem. package_root values it reports are NOT resolvable on the
|
|
LangBot side, so LangBot must trust them rather than dropping every
|
|
skill via a local isdir() check."""
|
|
logger = Mock()
|
|
skills = {
|
|
'a': {'name': 'a', 'package_root': '/box/skills/a'},
|
|
'b': {'name': 'b', 'package_root': '/box/skills/b'},
|
|
}
|
|
service = self._make_service(logger, skills, shares_filesystem=False)
|
|
|
|
mounts = service.build_skill_extra_mounts(make_query())
|
|
|
|
assert mounts == [
|
|
{'host_path': '/box/skills/a', 'mount_path': '/workspace/.skills/a', 'mode': 'rw'},
|
|
{'host_path': '/box/skills/b', 'mount_path': '/workspace/.skills/b', 'mode': 'rw'},
|
|
]
|
|
# No skill is dropped, so no "missing" warning should be logged.
|
|
assert not any('package_root missing' in str(call.args[0]) for call in logger.warning.call_args_list)
|
|
|
|
def test_skips_skill_with_empty_package_root(self):
|
|
logger = Mock()
|
|
skills = {
|
|
'no_root': {'name': 'no_root', 'package_root': ''},
|
|
'whitespace': {'name': 'whitespace', 'package_root': ' '},
|
|
}
|
|
service = self._make_service(logger, skills)
|
|
|
|
assert service.build_skill_extra_mounts(make_query()) == []
|
|
|
|
def test_empty_package_root_skipped_even_when_not_shared(self):
|
|
"""An empty package_root is always invalid regardless of topology."""
|
|
logger = Mock()
|
|
skills = {'no_root': {'name': 'no_root', 'package_root': ''}}
|
|
service = self._make_service(logger, skills, shares_filesystem=False)
|
|
|
|
assert service.build_skill_extra_mounts(make_query()) == []
|
|
|
|
def test_returns_empty_when_no_skill_manager(self):
|
|
logger = Mock()
|
|
app = make_app(logger)
|
|
# no skill_mgr attribute
|
|
service = BoxService(app, client=Mock(spec=BoxRuntimeClient))
|
|
|
|
assert service.build_skill_extra_mounts(make_query()) == []
|
|
|
|
|
|
# ── Attachment passthrough (inbound / outbound) ─────────────────────────────
|
|
|
|
|
|
class TestAttachmentHelpers:
|
|
def test_sanitize_attachment_name_strips_traversal(self):
|
|
assert BoxService._sanitize_attachment_name('../../etc/passwd', 'fb') == 'passwd'
|
|
assert BoxService._sanitize_attachment_name('/a/b/c.png', 'fb') == 'c.png'
|
|
assert BoxService._sanitize_attachment_name('a b c.txt', 'fb') == 'a_b_c.txt'
|
|
assert BoxService._sanitize_attachment_name('', 'fallback.bin') == 'fallback.bin'
|
|
assert BoxService._sanitize_attachment_name('...', 'fb.bin') == 'fb.bin'
|
|
# weird unicode / shell chars dropped, but keeps a usable name
|
|
out = BoxService._sanitize_attachment_name('rm -rf $(x).png', 'fb')
|
|
assert '/' not in out and '$' not in out and out.endswith('.png')
|
|
|
|
def test_classify_outbound_entries_by_extension(self):
|
|
entries = [
|
|
{'name': 'chart.png', 'b64': 'AAA'},
|
|
{'name': 'clip.mp3', 'b64': 'BBB'},
|
|
{'name': 'report.pdf', 'b64': 'CCC'},
|
|
{'name': 'sub/dir/photo.JPG', 'b64': 'DDD'},
|
|
{'name': 'noext', 'b64': 'EEE'},
|
|
{'name': 'skip', 'b64': ''}, # dropped (no payload)
|
|
]
|
|
out = BoxService._classify_outbound_entries(entries)
|
|
by_name = {a['name']: a for a in out}
|
|
assert by_name['chart.png']['type'] == 'Image'
|
|
assert by_name['chart.png']['base64'].startswith('data:image/png;base64,')
|
|
assert by_name['clip.mp3']['type'] == 'Voice'
|
|
assert by_name['clip.mp3']['base64'].startswith('data:audio/mp3;base64,')
|
|
assert by_name['report.pdf']['type'] == 'File'
|
|
assert by_name['report.pdf']['base64'] == 'CCC' # raw b64, no data: prefix
|
|
# nested path collapses to basename, case-insensitive ext
|
|
assert by_name['photo.JPG']['type'] == 'Image'
|
|
assert by_name['noext']['type'] == 'File'
|
|
assert 'skip' not in by_name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_component_to_bytes_from_data_uri(self):
|
|
import base64
|
|
|
|
raw = b'hello-bytes'
|
|
data_uri = 'data:text/plain;base64,' + base64.b64encode(raw).decode()
|
|
component = SimpleNamespace(base64=data_uri, url=None, path=None)
|
|
result = await BoxService._component_to_bytes(component)
|
|
assert result is not None
|
|
data, mime = result
|
|
assert data == raw
|
|
assert mime == 'text/plain'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_component_to_bytes_returns_none_when_empty(self):
|
|
component = SimpleNamespace(base64=None, url=None, path=None)
|
|
assert await BoxService._component_to_bytes(component) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_component_to_bytes_rejects_oversized_base64(self, monkeypatch):
|
|
monkeypatch.setattr(BoxService, '_ATTACHMENT_MAX_BYTES', 4)
|
|
component = SimpleNamespace(
|
|
base64='data:application/octet-stream;base64,' + ('A' * 12),
|
|
url=None,
|
|
path=None,
|
|
)
|
|
|
|
assert await BoxService._component_to_bytes(component) is None
|
|
|
|
|
|
class TestInboundOutboundRoundTrip:
|
|
def _service(self) -> BoxService:
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
service._available = True
|
|
return service
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_materialize_inbound_writes_and_describes(self):
|
|
import base64
|
|
|
|
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
|
|
|
service = self._service()
|
|
|
|
img_bytes = b'\x89PNG\r\n\x1a\n fake png'
|
|
img_b64 = 'data:image/png;base64,' + base64.b64encode(img_bytes).decode()
|
|
|
|
query = make_query()
|
|
query.message_chain = platform_message.MessageChain(
|
|
[
|
|
platform_message.Plain(text='please resize this'),
|
|
platform_message.Image(base64=img_b64),
|
|
]
|
|
)
|
|
|
|
# Mock the sandbox write path: echo back the written paths.
|
|
async def fake_execute_tool(parameters, q):
|
|
assert '/workspace/inbox/' in parameters['command']
|
|
return {
|
|
'ok': True,
|
|
'stdout': '["/workspace/inbox/query-42/image_1.png"]',
|
|
'stderr': '',
|
|
}
|
|
|
|
service.execute_tool = AsyncMock(side_effect=fake_execute_tool)
|
|
|
|
descriptors = await service.materialize_inbound_attachments(query)
|
|
assert len(descriptors) == 1
|
|
d = descriptors[0]
|
|
assert d['type'] == 'Image'
|
|
assert d['path'] == '/workspace/inbox/query-42/image_1.png'
|
|
assert d['size'] == len(img_bytes)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_materialize_inbound_noop_without_attachments(self):
|
|
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
|
|
|
service = self._service()
|
|
query = make_query()
|
|
query.message_chain = platform_message.MessageChain([platform_message.Plain(text='just text')])
|
|
service.execute_tool = AsyncMock()
|
|
assert await service.materialize_inbound_attachments(query) == []
|
|
service.execute_tool.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_outbound_reads_and_clears(self):
|
|
service = self._service()
|
|
query = make_query()
|
|
|
|
calls = []
|
|
|
|
async def fake_client_execute(spec):
|
|
cmd = spec.cmd
|
|
calls.append(cmd)
|
|
if 'os.scandir' in cmd:
|
|
return BoxExecutionResult(
|
|
session_id='s',
|
|
backend_name='test',
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout='[{"name": "out.png", "b64": "QUJD"}]',
|
|
duration_ms=10,
|
|
)
|
|
# the rm -rf cleanup call
|
|
return BoxExecutionResult(
|
|
session_id='s',
|
|
backend_name='test',
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout='',
|
|
duration_ms=10,
|
|
)
|
|
|
|
service.client.execute = AsyncMock(side_effect=fake_client_execute)
|
|
service.execute_tool = AsyncMock(return_value={'ok': True, 'stdout': '', 'stderr': ''})
|
|
|
|
attachments = await service.collect_outbound_attachments(query)
|
|
assert len(attachments) == 1
|
|
assert attachments[0]['type'] == 'Image'
|
|
assert attachments[0]['name'] == 'out.png'
|
|
# cleanup (rm -rf) must have been issued after a successful collection
|
|
service.execute_tool.assert_awaited_once()
|
|
assert 'rm -rf' in service.execute_tool.await_args.args[0]['command']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_outbound_empty_still_clears(self):
|
|
# An empty collection MUST still clear the per-query outbox, so a later
|
|
# turn reusing the same query_id (the counter resets across restarts)
|
|
# cannot inherit stale files left from a prior run.
|
|
service = self._service()
|
|
query = make_query()
|
|
|
|
calls = []
|
|
|
|
async def fake_client_execute(spec):
|
|
cmd = spec.cmd
|
|
calls.append(cmd)
|
|
if 'os.scandir' in cmd:
|
|
return BoxExecutionResult(
|
|
session_id='s',
|
|
backend_name='test',
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout='[]',
|
|
duration_ms=10,
|
|
)
|
|
return BoxExecutionResult(
|
|
session_id='s',
|
|
backend_name='test',
|
|
status=BoxExecutionStatus.COMPLETED,
|
|
exit_code=0,
|
|
stdout='',
|
|
duration_ms=10,
|
|
)
|
|
|
|
service.client.execute = AsyncMock(side_effect=fake_client_execute)
|
|
service.execute_tool = AsyncMock(return_value={'ok': True, 'stdout': '', 'stderr': ''})
|
|
assert await service.collect_outbound_attachments(query) == []
|
|
# cleanup (rm -rf) is issued unconditionally now
|
|
service.execute_tool.assert_awaited_once()
|
|
assert 'rm -rf' in service.execute_tool.await_args.args[0]['command']
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passthrough_noop_when_unavailable(self):
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
service._available = False
|
|
query = make_query()
|
|
assert await service.materialize_inbound_attachments(query) == []
|
|
assert await service.collect_outbound_attachments(query) == []
|
|
|
|
|
|
class TestAttachmentHostPath:
|
|
"""Direct host-filesystem transfer path (bind-mounted workspace).
|
|
|
|
When ``default_workspace`` is a real local dir, inbound/outbound bypass the
|
|
exec channel entirely (no ARG_MAX / stdout-truncation limits) and read/write
|
|
the bind-mounted host dir directly.
|
|
"""
|
|
|
|
def _service_with_workspace(self, tmp_path):
|
|
default_workspace = str(tmp_path / 'box' / 'default')
|
|
os.makedirs(default_workspace, exist_ok=True)
|
|
app = make_app(Mock(), allowed_mount_roots=[str(tmp_path)], host_root=str(tmp_path / 'box'))
|
|
service = BoxService(app, client=Mock(spec=BoxRuntimeClient))
|
|
service._available = True
|
|
# Force the default_workspace to our tmp dir so _host_query_dir resolves.
|
|
service.default_workspace = default_workspace
|
|
tenant_workspace = service._tenant_workspace(_CONTEXT)
|
|
assert tenant_workspace is not None
|
|
os.makedirs(tenant_workspace, exist_ok=True)
|
|
return service, tenant_workspace
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inbound_writes_to_host_no_exec(self, tmp_path):
|
|
import base64
|
|
|
|
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
|
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
# Big payload that would blow ARG_MAX on the exec path:
|
|
big = b'\x89PNG\r\n\x1a\n' + b'x' * (300 * 1024)
|
|
b64 = 'data:image/png;base64,' + base64.b64encode(big).decode()
|
|
query = make_query()
|
|
query.message_chain = platform_message.MessageChain([platform_message.Image(base64=b64)])
|
|
# execute_tool must NOT be called on the host path.
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used on host path'))
|
|
|
|
descriptors = await service.materialize_inbound_attachments(query)
|
|
assert len(descriptors) == 1
|
|
d = descriptors[0]
|
|
assert d['type'] == 'Image'
|
|
assert d['size'] == len(big)
|
|
# File actually landed on the host workspace.
|
|
host_file = os.path.join(ws, 'inbox', str(query.query_uuid), d['name'])
|
|
assert os.path.isfile(host_file)
|
|
assert pathlib.Path(host_file).read_bytes() == big
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inbound_host_clears_stale_query_dir(self, tmp_path):
|
|
import base64
|
|
|
|
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
|
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
# Seed a stale file under the same query_id (simulates webchat id reuse).
|
|
stale_dir = os.path.join(ws, 'inbox', 'query-42')
|
|
os.makedirs(stale_dir, exist_ok=True)
|
|
pathlib.Path(stale_dir, 'image_1.png').write_bytes(b'STALE-OLD-IMAGE')
|
|
|
|
new = b'\x89PNG\r\n\x1a\n NEW'
|
|
b64 = 'data:image/png;base64,' + base64.b64encode(new).decode()
|
|
query = make_query(query_id=42)
|
|
query.message_chain = platform_message.MessageChain([platform_message.Image(base64=b64)])
|
|
service.execute_tool = AsyncMock()
|
|
descriptors = await service.materialize_inbound_attachments(query)
|
|
# The new write recreated the dir; the stale file is gone, new bytes present.
|
|
host_file = os.path.join(stale_dir, descriptors[0]['name'])
|
|
host_bytes = pathlib.Path(host_file).read_bytes()
|
|
assert host_bytes == new
|
|
# No leftover content from the stale image.
|
|
assert b'STALE-OLD-IMAGE' not in host_bytes
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inbound_host_replaces_query_symlink_without_touching_other_workspace(self, tmp_path):
|
|
import base64
|
|
|
|
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
|
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
other_workspace = tmp_path / 'other-workspace'
|
|
other_workspace.mkdir()
|
|
protected = other_workspace / 'protected.txt'
|
|
protected.write_bytes(b'workspace-b-secret')
|
|
inbox = os.path.join(ws, 'inbox')
|
|
os.makedirs(inbox, exist_ok=True)
|
|
os.symlink(other_workspace, os.path.join(inbox, 'query-42'))
|
|
|
|
query = make_query()
|
|
payload = b'workspace-a-input'
|
|
query.message_chain = platform_message.MessageChain(
|
|
[platform_message.File(name='input.bin', base64=base64.b64encode(payload).decode())]
|
|
)
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used on host path'))
|
|
|
|
descriptors = await service.materialize_inbound_attachments(query)
|
|
|
|
assert descriptors[0]['path'] == '/workspace/inbox/query-42/input.bin'
|
|
assert protected.read_bytes() == b'workspace-b-secret'
|
|
assert not os.path.islink(os.path.join(inbox, 'query-42'))
|
|
assert pathlib.Path(inbox, 'query-42', 'input.bin').read_bytes() == payload
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbound_reads_host_and_clears(self, tmp_path):
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
query = make_query()
|
|
outbox = os.path.join(ws, 'outbox', str(query.query_uuid))
|
|
os.makedirs(outbox, exist_ok=True)
|
|
# A large file that would be truncated on the exec/stdout path:
|
|
big_png = b'\x89PNG\r\n\x1a\n' + b'y' * (400 * 1024)
|
|
pathlib.Path(outbox, 'result.png').write_bytes(big_png)
|
|
pathlib.Path(outbox, 'notes.txt').write_bytes(b'hello')
|
|
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used on host path'))
|
|
attachments = await service.collect_outbound_attachments(query)
|
|
by_name = {a['name']: a for a in attachments}
|
|
assert by_name['result.png']['type'] == 'Image'
|
|
assert by_name['notes.txt']['type'] == 'File'
|
|
# Full image survived (no truncation).
|
|
import base64
|
|
|
|
raw = base64.b64decode(by_name['result.png']['base64'].split(',', 1)[-1])
|
|
assert raw == big_png
|
|
# Outbox cleared after collection.
|
|
assert os.listdir(outbox) == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbound_host_never_follows_query_or_file_symlinks(self, tmp_path):
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
query = make_query()
|
|
other_workspace = tmp_path / 'other-workspace'
|
|
other_workspace.mkdir()
|
|
secret = other_workspace / 'secret.txt'
|
|
secret.write_bytes(b'workspace-b-secret')
|
|
outbox_root = os.path.join(ws, 'outbox')
|
|
os.makedirs(outbox_root, exist_ok=True)
|
|
|
|
# A hostile query-directory replacement is rejected rather than read.
|
|
query_dir = os.path.join(outbox_root, str(query.query_uuid))
|
|
os.symlink(other_workspace, query_dir)
|
|
with pytest.raises(BoxValidationError, match='symbolic link'):
|
|
await service.collect_outbound_attachments(query)
|
|
assert secret.read_bytes() == b'workspace-b-secret'
|
|
|
|
os.unlink(query_dir)
|
|
os.makedirs(query_dir)
|
|
os.symlink(secret, os.path.join(query_dir, 'leak.txt'))
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used on host path'))
|
|
assert await service.collect_outbound_attachments(query) == []
|
|
assert secret.read_bytes() == b'workspace-b-secret'
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbound_host_fails_closed_on_inode_bomb(self, tmp_path):
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
query = make_query()
|
|
outbox = os.path.join(ws, 'outbox', str(query.query_uuid))
|
|
os.makedirs(outbox, exist_ok=True)
|
|
harmless_target = tmp_path / 'harmless-target'
|
|
harmless_target.write_bytes(b'x')
|
|
# Symlinks do not count toward the 20 returned files, so this proves
|
|
# traversal itself has a bounded entry budget.
|
|
for index in range(513):
|
|
os.symlink(harmless_target, os.path.join(outbox, f'entry-{index}'))
|
|
|
|
with pytest.raises(BoxValidationError, match='symbolic link'):
|
|
await service.collect_outbound_attachments(query)
|
|
assert harmless_target.read_bytes() == b'x'
|
|
|
|
def test_host_attachment_directories_use_query_uuid_not_process_local_id(self, tmp_path):
|
|
service, _ws = self._service_with_workspace(tmp_path)
|
|
first = make_query(query_id=7)
|
|
second = make_query(query_id=7)
|
|
object.__setattr__(first, 'query_uuid', 'replica-a-query')
|
|
object.__setattr__(second, 'query_uuid', 'replica-b-query')
|
|
|
|
first_path = service._host_query_dir(service.OUTBOX_SUBDIR, first)
|
|
second_path = service._host_query_dir(service.OUTBOX_SUBDIR, second)
|
|
|
|
assert first_path is not None and first_path.endswith('/outbox/replica-a-query')
|
|
assert second_path is not None and second_path.endswith('/outbox/replica-b-query')
|
|
assert first_path != second_path
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbound_empty_clears_stale_host_dir(self, tmp_path):
|
|
# Reusing a query_id (counter resets on restart) must not re-send files
|
|
# a previous run left in the outbox: an empty collection still clears it.
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
query = make_query()
|
|
outbox = os.path.join(ws, 'outbox', str(query.query_uuid))
|
|
os.makedirs(outbox, exist_ok=True)
|
|
# Stale file from a prior turn; the agent produced nothing this turn —
|
|
# but _read_outbox_host would still pick it up, so collection must drop
|
|
# it and then wipe the dir. Simulate "nothing produced this turn" by
|
|
# treating any present file as stale and asserting it is not re-sent
|
|
# across a second, genuinely-empty collection.
|
|
pathlib.Path(outbox, 'stale.png').write_bytes(b'\x89PNG\r\n\x1a\n old')
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used on host path'))
|
|
|
|
# First collection drains + clears the dir.
|
|
first = await service.collect_outbound_attachments(query)
|
|
assert {a['name'] for a in first} == {'stale.png'}
|
|
assert os.listdir(outbox) == []
|
|
|
|
# Second collection (no new files) returns nothing and leaves a clean dir.
|
|
second = await service.collect_outbound_attachments(query)
|
|
assert second == []
|
|
assert os.listdir(outbox) == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_purge_attachment_dirs_wipes_host_owned_leftovers_on_init(self, tmp_path):
|
|
# Leftover inbox/outbox dirs from a previous process (same reset
|
|
# query_id counter) must be removed at startup. Host-owned files are
|
|
# cleared without any sandbox exec.
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
for sub in ('inbox', 'outbox'):
|
|
d = os.path.join(ws, sub, '0')
|
|
os.makedirs(d, exist_ok=True)
|
|
pathlib.Path(d, 'leftover.bin').write_bytes(b'from a previous process')
|
|
service.execute_tool = AsyncMock(side_effect=AssertionError('exec must not be used for host-owned files'))
|
|
|
|
await service._purge_attachment_dirs()
|
|
|
|
assert not os.path.exists(os.path.join(ws, 'inbox'))
|
|
assert not os.path.exists(os.path.join(ws, 'outbox'))
|
|
# The workspace root itself survives.
|
|
assert os.path.isdir(ws)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_purge_attachment_dirs_never_uses_unscoped_exec_for_root_owned(self, tmp_path, monkeypatch):
|
|
# Startup has no trusted Workspace context. If host deletion cannot
|
|
# remove root-owned output, cleanup must fail closed instead of issuing
|
|
# an unscoped Box exec that could cross a tenant boundary.
|
|
service, ws = self._service_with_workspace(tmp_path)
|
|
outbox = os.path.join(ws, 'outbox')
|
|
os.makedirs(os.path.join(outbox, '0'), exist_ok=True)
|
|
|
|
# Simulate a host delete that cannot remove the root-owned outbox.
|
|
from langbot.pkg.box import secure_fs
|
|
|
|
real_purge = secure_fs.purge_subdirectory
|
|
|
|
def fake_purge(root, subdir):
|
|
if os.path.abspath(os.path.join(root, subdir)) == os.path.abspath(outbox):
|
|
raise PermissionError('root-owned')
|
|
return real_purge(root, subdir)
|
|
|
|
monkeypatch.setattr(secure_fs, 'purge_subdirectory', fake_purge)
|
|
|
|
service.build_spec = Mock()
|
|
service.client.execute = AsyncMock()
|
|
|
|
await service._purge_attachment_dirs()
|
|
|
|
assert os.path.isdir(outbox)
|
|
service.build_spec.assert_not_called()
|
|
service.client.execute.assert_not_awaited()
|
|
assert any(
|
|
'no trusted Workspace context' in str(call.args[0]) for call in service.ap.logger.warning.call_args_list
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_purge_attachment_dirs_noop_without_workspace(self):
|
|
# No bind-mounted workspace (E2B / remote): purge is a safe no-op.
|
|
service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
|
|
service.default_workspace = None
|
|
# Must not raise.
|
|
await service._purge_attachment_dirs()
|