fix(box): trust Box-reported skill paths when filesystem is not shared

In separated deployments (Docker Compose, k8s sidecar, --standalone-box,
remote runtime.endpoint) the Box runtime owns its own filesystem, so the
skill package_root it reports via list_skills is not resolvable on the
LangBot side. LangBot's reload_skills and build_skill_extra_mounts
validated those paths with os.path.isdir() against its own filesystem,
which silently dropped every skill in such deployments — breaking the
sandbox skill feature for the nsjail/SaaS backend.

Add BoxService.shares_filesystem_with_box, derived from the connector
transport (stdio = shared, WebSocket = separated), with an explicit
override seam for tests/embedders. Gate both isdir() guards on it: keep
local validation in shared-fs stdio mode, trust Box-reported paths
otherwise. The Box runtime only reports skills found on its own
filesystem, so those paths are valid there by construction.

Adds topology-derivation tests (real connector, no mocks) and
skill-retention tests for both shared and separated filesystems.
This commit is contained in:
RockChinQ
2026-06-07 12:46:52 -04:00
parent 47fe9bde03
commit 882c9ae8f5
5 changed files with 205 additions and 16 deletions

View File

@@ -190,6 +190,66 @@ async def test_box_service_without_explicit_client_initializes_internal_connecto
connector.initialize.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
@pytest.mark.asyncio
async def test_box_service_get_sessions_delegates_to_client():
client = Mock()
@@ -1342,11 +1402,16 @@ class TestBuildSkillExtraMounts:
the backend never sees a bad mount.
"""
def _make_service(self, logger, skills):
def _make_service(self, logger, skills, *, shares_filesystem=True):
app = make_app(logger)
app.skill_mgr = SimpleNamespace(skills=skills)
client = Mock(spec=BoxRuntimeClient)
return BoxService(app, client=client)
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()
@@ -1373,6 +1438,30 @@ class TestBuildSkillExtraMounts:
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 = {
@@ -1383,6 +1472,14 @@ class TestBuildSkillExtraMounts:
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)

View File

@@ -62,15 +62,17 @@ class TestSkillManagerCache:
@pytest.mark.asyncio
async def test_reload_skills_drops_box_skills_with_missing_package_root(self):
"""When Box reports a skill whose package_root is gone from the
LangBot-visible filesystem, the cache must drop it instead of
keeping a stale entry that would later produce a bad mount."""
"""When LangBot shares a filesystem with Box (local stdio mode) and Box
reports a skill whose package_root is gone from that shared filesystem,
the cache must drop it instead of keeping a stale entry that would later
produce a bad mount."""
from langbot.pkg.skill.manager import SkillManager
with tempfile.TemporaryDirectory() as live_dir:
ghost_dir = os.path.join(live_dir, '_does_not_exist')
box_service = SimpleNamespace(
available=True,
shares_filesystem_with_box=True,
list_skills=AsyncMock(
return_value=[
_make_skill_data(name='alive', package_root=live_dir),
@@ -90,6 +92,37 @@ class TestSkillManagerCache:
warning_messages = [str(call.args[0]) for call in ap.logger.warning.call_args_list]
assert any('ghost' in msg and 'package_root missing' in msg for msg in warning_messages)
@pytest.mark.asyncio
async def test_reload_skills_trusts_box_paths_when_filesystem_not_shared(self):
"""In separated deployments (Docker Compose, k8s sidecar,
--standalone-box, remote endpoint) the package_root reported by Box
lives on the Box runtime's filesystem and is not resolvable on the
LangBot side. The cache must keep every Box-reported skill rather than
dropping them all via a local isdir() check."""
from langbot.pkg.skill.manager import SkillManager
box_service = SimpleNamespace(
available=True,
shares_filesystem_with_box=False,
list_skills=AsyncMock(
return_value=[
_make_skill_data(name='alpha', package_root='/box/skills/alpha'),
_make_skill_data(name='beta', package_root='/box/skills/beta'),
]
),
)
ap = _make_ap()
ap.box_service = box_service
mgr = SkillManager(ap)
await mgr.reload_skills()
assert sorted(mgr.skills) == ['alpha', 'beta']
# No skill dropped → no "package_root missing" warning.
warning_messages = [str(call.args[0]) for call in ap.logger.warning.call_args_list]
assert not any('package_root missing' in msg for msg in warning_messages)
class TestSkillActivationHelper:
"""Skill activation is now Tool-Call based.