refactor(box): derive paths from shared host root

docker/docker-compose.yaml

@@ -13,7 +13,7 @@ services:
       # Uncomment the one that matches your container runtime:
       # - /var/run/podman/podman.sock:/var/run/podman/podman.sock   # Podman
       - /var/run/docker.sock:/var/run/docker.sock                   # Docker
-      - ./data/box-workspaces:/workspaces
+      - ./data/box:/workspaces
     ports:
       - 5410:5410
     restart: on-failure
@@ -41,7 +41,7 @@ services:
     container_name: langbot
     volumes:
       - ./data:/app/data
-      - ./data/box-workspaces:/workspaces
+      - ./data/box:/workspaces
     restart: on-failure
     environment:
       - TZ=Asia/Shanghai

src/langbot/pkg/box/service.py

@@ -50,6 +50,7 @@ class BoxService:
             client = self._runtime_connector.client
         self.client = client
         self.output_limit_chars = output_limit_chars
+        self.shared_host_root = self._load_shared_host_root()
         self.allowed_host_mount_roots = self._load_allowed_host_mount_roots()
         self.default_host_workspace = self._load_default_host_workspace()
         self.profile = self._load_profile()
@@ -73,13 +74,17 @@ class BoxService:
     def available(self) -> bool:
         return self._available
 
-    async def execute_sandbox_tool(self, parameters: dict, query: 'pipeline_query.Query') -> dict:
+    async def execute_spec_payload(
+        self,
+        spec_payload: dict,
+        query: 'pipeline_query.Query',
+        *,
+        skip_host_mount_validation: bool = False,
+    ) -> dict:
         if not self._available:
             raise BoxError('Box runtime is not available. Install and start Podman or Docker to use sandbox features.')
-        spec_payload = dict(parameters)
-        spec_payload.setdefault('session_id', str(query.query_id))
         try:
-            spec = self.build_spec(spec_payload)
+            spec = self.build_spec(spec_payload, skip_host_mount_validation=skip_host_mount_validation)
         except BoxError as exc:
             self._record_error(exc, query)
             raise
@@ -100,6 +105,11 @@ class BoxService:
         )
         return self._serialize_result(result)
 
+    async def execute_sandbox_tool(self, parameters: dict, query: 'pipeline_query.Query') -> dict:
+        spec_payload = dict(parameters)
+        spec_payload.setdefault('session_id', str(query.query_id))
+        return await self.execute_spec_payload(spec_payload, query)
+
     async def shutdown(self):
         await self.client.shutdown()
 
@@ -250,14 +260,30 @@ class BoxService:
                 continue
             normalized_roots.append(os.path.realpath(os.path.abspath(root_value)))
 
+        if not normalized_roots and self.shared_host_root is not None:
+            normalized_roots.append(self.shared_host_root)
+
         return normalized_roots
 
+    def _load_shared_host_root(self) -> str | None:
+        shared_host_root = str(_get_box_config(self.ap).get('shared_host_root', '')).strip()
+        if not shared_host_root:
+            return None
+        return os.path.realpath(os.path.abspath(shared_host_root))
+
     def _load_default_host_workspace(self) -> str | None:
         default_host_workspace = str(_get_box_config(self.ap).get('default_host_workspace', '')).strip()
         if not default_host_workspace:
-            return None
+            if self.shared_host_root is None:
+                return None
+            default_host_workspace = os.path.join(self.shared_host_root, 'default')
         return os.path.realpath(os.path.abspath(default_host_workspace))
 
+    def get_managed_skills_root(self) -> str | None:
+        if self.shared_host_root is None:
+            return None
+        return os.path.join(self.shared_host_root, 'skills')
+
     def _ensure_default_host_workspace(self):
         if self.default_host_workspace is None:
             return

src/langbot/templates/config.yaml

@@ -90,9 +90,9 @@ monitoring:
 box:
     profile: 'default'
     runtime_url: ''  # Leave empty to use defaults: http://127.0.0.1:5410 locally, http://langbot_box_runtime:5410 in Docker
-    default_host_workspace: './data/box-workspaces/default'  # For Docker deployment, use '/workspaces/default'
-    allowed_host_mount_roots:  # For Docker deployment, use '/workspaces' instead
-        - './data/box-workspaces'
+    shared_host_root: './data/box'  # For Docker deployment, use '/workspaces'
+    default_host_workspace: ''  # Defaults to '<shared_host_root>/default'
+    allowed_host_mount_roots:  # Defaults to ['<shared_host_root>'] when left empty
         - '/tmp'
 space:
     # Space service URL for OAuth and API

tests/unit_tests/box/test_box_service.py

@@ -128,13 +128,19 @@ def make_query(query_id: int = 42) -> pipeline_query.Query:
     return pipeline_query.Query.model_construct(query_id=query_id)
 
 
-def make_app(logger: Mock, allowed_host_mount_roots: list[str] | None = None, profile: str = 'default'):
+def make_app(
+    logger: Mock,
+    allowed_host_mount_roots: list[str] | None = None,
+    profile: str = 'default',
+    shared_host_root: str = '',
+):
     return SimpleNamespace(
         logger=logger,
         instance_config=SimpleNamespace(
             data={
                 'box': {
                     'profile': profile,
+                    'shared_host_root': shared_host_root,
                     'allowed_host_mount_roots': allowed_host_mount_roots or [],
                     'default_host_workspace': '',
                 }
@@ -309,6 +315,23 @@ async def test_box_service_creates_default_host_workspace_on_initialize(tmp_path
     assert default_host_workspace.is_dir()
 
 
+@pytest.mark.asyncio
+async def test_box_service_derives_workspace_and_allowed_root_from_shared_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, shared_host_root=str(shared_root))
+    service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime))
+
+    await service.initialize()
+
+    assert service.shared_host_root == os.path.realpath(shared_root)
+    assert service.default_host_workspace == os.path.realpath(shared_root / 'default')
+    assert service.allowed_host_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()