refactor(skill): compose generic box mounts in core

tests/unit_tests/box/test_box_service.py

@@ -41,6 +41,7 @@ from langbot_plugin.box.security import (
 from langbot_plugin.entities.io.context import ActionContext
 from langbot.pkg.api.http.context import ExecutionContext
 from langbot.pkg.box.service import BoxService
+from langbot.pkg.provider.tools.loaders import skill as skill_loader
 
 _UTC = dt.timezone.utc
 _CONTEXT = ExecutionContext(
@@ -301,7 +302,7 @@ class TestSharesFilesystemWithBox:
     - 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.
+    This drives whether LangBot can safely perform local workspace operations.
     Getting it wrong silently drops every skill in separated deployments.
     """
 
@@ -338,7 +339,7 @@ class TestSharesFilesystemWithBox:
 
     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.
+        # so LangBot does not assume a shared local filesystem.
         service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient))
 
         assert service._runtime_connector is None
@@ -552,7 +553,6 @@ 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()
@@ -565,16 +565,14 @@ async def test_box_service_reconnect_restores_workspace_and_runs_cleanup(
     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(
+async def test_cloud_box_service_reconnect_restores_runtime_only(
     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()
@@ -587,7 +585,6 @@ async def test_cloud_box_service_reconnect_does_not_reload_unscoped_skills(
 
     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
 
 
@@ -1941,7 +1938,7 @@ def test_disconnect_callback_does_not_schedule_without_running_event_loop():
     assert service._reconnecting is False
 
 
-class TestBuildSkillExtraMounts:
+class TestBuildSkillExecutionMounts:
     """Robustness of skill mount construction against a stale skill cache.
 
     The three sandbox backends behave inconsistently when a skill's
@@ -1951,16 +1948,10 @@ class TestBuildSkillExtraMounts:
     the backend never sees a bad mount.
     """
 
-    def _make_service(self, logger, skills, *, shares_filesystem=True):
+    def _make_app(self, logger, skills):
         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
+        return app
 
     def test_skips_skill_with_missing_package_root(self):
         logger = Mock()
@@ -1969,10 +1960,10 @@ class TestBuildSkillExtraMounts:
                 'alive': {'name': 'alive', 'package_root': live_dir},
                 'ghost': {'name': 'ghost', 'package_root': '/nonexistent/path/should/never/exist'},
             }
-            service = self._make_service(logger, skills)
+            app = self._make_app(logger, skills)
             query = make_query()
 
-            mounts = service.build_skill_extra_mounts(query)
+            mounts = skill_loader.build_execution_mounts(app, query)
 
             assert mounts == [
                 {
@@ -1987,27 +1978,19 @@ 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."""
+    def test_rejects_missing_core_paths_when_filesystem_not_shared(self):
+        """Core owns package paths even when Box is a separate process."""
         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)
+        app = self._make_app(logger, skills)
 
-        mounts = service.build_skill_extra_mounts(make_query())
+        mounts = skill_loader.build_execution_mounts(app, make_query())
 
-        assert mounts == [
-            {'host_path': '/box/skills/a', 'mount_path': '/workspace/.skills/a', 'mode': 'ro'},
-            {'host_path': '/box/skills/b', 'mount_path': '/workspace/.skills/b', 'mode': 'ro'},
-        ]
-        # 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)
+        assert mounts == []
+        assert len(logger.warning.call_args_list) == 2
 
     def test_skips_skill_with_empty_package_root(self):
         logger = Mock()
@@ -2015,25 +1998,23 @@ class TestBuildSkillExtraMounts:
             'no_root': {'name': 'no_root', 'package_root': ''},
             'whitespace': {'name': 'whitespace', 'package_root': '   '},
         }
-        service = self._make_service(logger, skills)
+        app = self._make_app(logger, skills)
 
-        assert service.build_skill_extra_mounts(make_query()) == []
+        assert skill_loader.build_execution_mounts(app, 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)
+        app = self._make_app(logger, skills)
 
-        assert service.build_skill_extra_mounts(make_query()) == []
+        assert skill_loader.build_execution_mounts(app, 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()) == []
+        assert skill_loader.build_execution_mounts(app, make_query()) == []
 
 
 # ── Attachment passthrough (inbound / outbound) ─────────────────────────────

tests/unit_tests/box/test_workspace.py

@@ -13,8 +13,8 @@ from langbot.pkg.box.workspace import (
     classify_python_workspace,
     infer_workspace_host_path,
     rewrite_mounted_path,
-    wrap_python_command_with_env,
 )
+from langbot.pkg.utils.python_workspace import wrap_python_command_with_env
 
 
 _CONTEXT = ExecutionContext(

tests/unit_tests/provider/test_skill_tools.py

@@ -614,7 +614,7 @@ class TestNativeToolLoaderSkillPaths:
             ap.skill_repository.read_skill_file.assert_awaited_once_with(_CONTEXT, 'demo', 'SKILL.md')
 
     @pytest.mark.asyncio
-    async def test_external_runtime_rejects_skill_host_fallback_without_protocol_capability(self):
+    async def test_core_owned_skill_path_does_not_depend_on_runtime_topology(self):
         from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
         from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY
 
@@ -634,18 +634,20 @@ class TestNativeToolLoaderSkillPaths:
                 variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']},
             )
 
-            with pytest.raises(ValueError, match='owned by the Box Runtime'):
-                await loader.invoke_tool(
-                    'grep',
-                    {
-                        'path': '/workspace/.skills/demo',
-                        'pattern': 'core-host-secret',
-                    },
-                    query,
-                )
+            result = await loader.invoke_tool(
+                'grep',
+                {
+                    'path': '/workspace/.skills/demo',
+                    'pattern': 'core-host-secret',
+                },
+                query,
+            )
+
+            assert result['ok'] is True
+            assert result['total'] == 1
 
     @pytest.mark.asyncio
-    async def test_exec_in_activated_skill_mount_rewrites_command_and_refreshes(self):
+    async def test_exec_in_activated_skill_mount_rewrites_command_without_mutating_skill(self):
         from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
         from langbot.pkg.provider.tools.loaders.skill import register_activated_skill
 
@@ -656,11 +658,15 @@ class TestNativeToolLoaderSkillPaths:
                 default_workspace=tmpdir,
                 execute_tool=AsyncMock(return_value={'ok': True}),
             )
-            ap.skill_mgr = SimpleNamespace(refresh_skill_from_disk=Mock())
+            skill_data = _make_skill_data(name='demo', package_root=tmpdir)
+            ap.skill_mgr = _make_skill_manager(
+                {'demo': skill_data},
+                refresh_skill_from_disk=Mock(),
+            )
             loader = NativeToolLoader(ap)
 
             query = _make_query(query_id='q1', launcher_type='person', launcher_id='123')
-            register_activated_skill(query, _make_skill_data(name='demo', package_root=tmpdir))
+            register_activated_skill(query, skill_data)
 
             result = await loader.invoke_tool(
                 'exec',
@@ -675,8 +681,8 @@ class TestNativeToolLoaderSkillPaths:
             tool_parameters = ap.box_service.execute_tool.await_args.args[0]
             assert tool_parameters['command'] == 'python /workspace/.skills/demo/scripts/run.py'
             assert tool_parameters['workdir'] == '/workspace/.skills/demo'
-            assert ap.box_service.execute_tool.await_args.kwargs['skill_name'] == 'demo'
-            ap.skill_mgr.refresh_skill_from_disk.assert_called_once_with(_CONTEXT, 'demo')
+            assert 'skill_name' not in ap.box_service.execute_tool.await_args.kwargs
+            ap.skill_mgr.refresh_skill_from_disk.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_external_runtime_python_skill_uses_trusted_metadata_and_writable_env(self):
@@ -689,17 +695,18 @@ class TestNativeToolLoaderSkillPaths:
             shares_filesystem_with_box=False,
             execute_tool=AsyncMock(return_value={'ok': True}),
         )
-        ap.skill_mgr = SimpleNamespace(refresh_skill_from_disk=Mock())
+        skill_data = _make_skill_data(
+            name='demo',
+            package_root='/box-runtime/skills/tenants/workspace/demo',
+            python_project=True,
+        )
+        ap.skill_mgr = _make_skill_manager(
+            {'demo': skill_data},
+            refresh_skill_from_disk=Mock(),
+        )
         loader = NativeToolLoader(ap)
         query = _make_query(query_id='q-external', launcher_type='person', launcher_id='123')
-        register_activated_skill(
-            query,
-            _make_skill_data(
-                name='demo',
-                package_root='/box-runtime/skills/tenants/workspace/demo',
-                python_project=True,
-            ),
-        )
+        register_activated_skill(query, skill_data)
 
         result = await loader.invoke_tool(
             'exec',
@@ -716,7 +723,7 @@ class TestNativeToolLoaderSkillPaths:
         assert '_LB_VENV_DIR="/workspace/.skill-envs/demo/.venv"' in wrapped
         assert 'root = "/workspace/.skills/demo"' in wrapped
         assert '/box-runtime/skills/tenants/workspace/demo' not in wrapped
-        assert ap.box_service.execute_tool.await_args.kwargs['skill_name'] == 'demo'
+        assert 'skill_name' not in ap.box_service.execute_tool.await_args.kwargs
 
     @pytest.mark.asyncio
     async def test_write_requires_skill_activation(self):

tests/unit_tests/test_skill_repository.py

@@ -28,11 +28,11 @@ def _repository(tmp_path) -> SkillRepository:
         ),
         instance_config=SimpleNamespace(
             data={
+                'skills': {'root': str(tmp_path / 'skill-store')},
                 'box': {
                     'enabled': False,
                     'local': {
                         'host_root': str(tmp_path / 'box'),
-                        'skills_root': 'skills',
                     },
                 }
             }
@@ -41,6 +41,32 @@ def _repository(tmp_path) -> SkillRepository:
     return SkillRepository(app)
 
 
+def test_repository_prefers_standalone_skill_root(tmp_path):
+    repository = _repository(tmp_path)
+
+    assert repository._store.root == str((tmp_path / 'skill-store').resolve())
+
+
+def test_repository_keeps_old_box_root_only_for_online_upgrade(tmp_path):
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(get_execution_binding=_binding),
+        instance_config=SimpleNamespace(
+            data={
+                'box': {
+                    'local': {
+                        'host_root': str(tmp_path / 'box'),
+                        'skills_root': 'legacy-skills',
+                    }
+                }
+            }
+        ),
+    )
+
+    repository = SkillRepository(app)
+
+    assert repository._store.root == str((tmp_path / 'box' / 'legacy-skills').resolve())
+
+
 @pytest.mark.asyncio
 async def test_repository_crud_and_reads_do_not_require_box(tmp_path):
     repository = _repository(tmp_path)