refactor(box): clean up sandbox subsystem code quality and efficiency

- Fix O(n²) stderr trimming in runtime.py with running length tracker
  - Remove dead code: RESERVED_CONTAINER_PATHS, _subprocess_wait_task,
    unused config_hash computation, unused imports
  - Deduplicate connection callback in BoxRuntimeConnector, parse URL once
  - Use enum comparison instead of stringly-typed spec.network.value check
  - Replace manual _result_to_dict/_session_to_dict with model_dump()
  - Cache NativeToolLoader tool definition and sandbox system guidance
  - Extract _is_path_under() helper to eliminate duplicated path checks
  - Import SANDBOX_EXEC_TOOL_NAME from native.py instead of redefining
  - Add JSON startswith guard in logging_utils to skip futile json.loads
  - Fix ruff lint errors (F401 unused imports, F841 unused variables)

src/langbot/pkg/box/backend.py

@@ -4,17 +4,14 @@ import abc
 import asyncio
 import dataclasses
 import datetime as dt
-import hashlib
-import json
 import logging
 import re
 import shlex
 import shutil
-import typing
 import uuid
 
 from .errors import BoxError
-from .models import DEFAULT_BOX_MOUNT_PATH, BoxExecutionResult, BoxExecutionStatus, BoxHostMountMode, BoxSessionInfo, BoxSpec
+from .models import DEFAULT_BOX_MOUNT_PATH, BoxExecutionResult, BoxExecutionStatus, BoxHostMountMode, BoxNetworkMode, BoxSessionInfo, BoxSpec
 from .security import validate_sandbox_security
 
 # Hard cap on raw subprocess output to prevent unbounded memory usage.
@@ -102,20 +99,7 @@ class CLISandboxBackend(BaseSandboxBackend):
             f'langbot.box.instance_id={self.instance_id}',
         ]
 
-        # Config hash label for identifying configuration drift
-        config_hash = hashlib.sha256(json.dumps({
-            'image': spec.image,
-            'network': spec.network.value,
-            'host_path': spec.host_path,
-            'host_path_mode': spec.host_path_mode.value,
-            'cpus': spec.cpus,
-            'memory_mb': spec.memory_mb,
-            'pids_limit': spec.pids_limit,
-            'read_only_rootfs': spec.read_only_rootfs,
-        }, sort_keys=True).encode()).hexdigest()[:16]
-        args.extend(['--label', f'langbot.box.config_hash={config_hash}'])
-
-        if spec.network.value == 'off':
+        if spec.network == BoxNetworkMode.OFF:
             args.extend(['--network', 'none'])
 
         # Resource limits
@@ -353,7 +337,7 @@ class CLISandboxBackend(BaseSandboxBackend):
 
     @staticmethod
     async def _read_stream(
-        stream: typing.Optional[asyncio.StreamReader],
+        stream: asyncio.StreamReader | None,
         limit: int = _MAX_RAW_OUTPUT_BYTES,
     ) -> tuple[bytes, int]:
         if stream is None:

src/langbot/pkg/box/connector.py

@@ -4,6 +4,7 @@ import asyncio
 import os
 import sys
 from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 from langbot_plugin.entities.io.actions.enums import CommonAction
 from langbot_plugin.runtime.io.handler import Handler
@@ -32,7 +33,11 @@ class BoxRuntimeConnector:
         self._handler_task: asyncio.Task | None = None
         self._ctrl_task: asyncio.Task | None = None
         self._subprocess: asyncio.subprocess.Process | None = None
-        self._subprocess_wait_task: asyncio.Task | None = None
+
+        # Parse the relay URL once for reuse
+        parsed = urlparse(self.ws_relay_base_url)
+        self._relay_host = parsed.hostname or '127.0.0.1'
+        self._relay_port = parsed.port or 5410
 
     async def initialize(self) -> None:
         if self.manages_local_runtime:
@@ -40,6 +45,26 @@ class BoxRuntimeConnector:
         else:
             await self._connect_remote_ws()
 
+    def _make_connection_callback(
+        self, transport_name: str, connected: asyncio.Event, connect_error: list[Exception],
+    ):
+        async def new_connection_callback(connection: Connection) -> None:
+            handler = Handler(connection)
+            self._handler = handler
+            self.client.set_handler(handler)
+            self._handler_task = asyncio.create_task(handler.run())
+            try:
+                await handler.call_action(CommonAction.PING, {})
+                self.ap.logger.info(f'Connected to Box runtime via {transport_name}.')
+                connected.set()
+                await self._handler_task
+            except Exception as exc:
+                if not connected.is_set():
+                    connect_error.append(exc)
+                    connected.set()
+
+        return new_connection_callback
+
     async def _start_local_stdio(self) -> None:
         """Launch box server as subprocess and connect via stdio."""
         from langbot_plugin.runtime.io.controllers.stdio.client import StdioClientController
@@ -50,29 +75,15 @@ class BoxRuntimeConnector:
         connected = asyncio.Event()
         connect_error: list[Exception] = []
 
-        async def new_connection_callback(connection: Connection) -> None:
-            handler = Handler.__new__(Handler)
-            Handler.__init__(handler, connection)
-            self._handler = handler
-            self.client.set_handler(handler)
-            self._handler_task = asyncio.create_task(handler.run())
-            try:
-                await handler.call_action(CommonAction.PING, {})
-                self.ap.logger.info('Connected to Box runtime via stdio.')
-                connected.set()
-                await self._handler_task
-            except Exception as exc:
-                if not connected.is_set():
-                    connect_error.append(exc)
-                    connected.set()
-
         ctrl = StdioClientController(
             command=python_path,
-            args=['-m', 'langbot.pkg.box.server', '--port', str(self._get_ws_relay_port())],
+            args=['-m', 'langbot.pkg.box.server', '--port', str(self._relay_port)],
             env=env,
         )
         self._subprocess = None  # StdioClientController manages the subprocess
-        self._ctrl_task = asyncio.create_task(ctrl.run(new_connection_callback))
+        self._ctrl_task = asyncio.create_task(
+            ctrl.run(self._make_connection_callback('stdio', connected, connect_error))
+        )
 
         # Wait for connection or failure
         try:
@@ -90,33 +101,19 @@ class BoxRuntimeConnector:
         """Connect to a remote box server via WebSocket."""
         from langbot_plugin.runtime.io.controllers.ws.client import WebSocketClientController
 
-        ws_url = self._get_rpc_ws_url()
+        ws_url = f'ws://{self._relay_host}:{self._relay_port + 1}'
 
         connected = asyncio.Event()
         connect_error: list[Exception] = []
 
-        async def new_connection_callback(connection: Connection) -> None:
-            handler = Handler.__new__(Handler)
-            Handler.__init__(handler, connection)
-            self._handler = handler
-            self.client.set_handler(handler)
-            self._handler_task = asyncio.create_task(handler.run())
-            try:
-                await handler.call_action(CommonAction.PING, {})
-                self.ap.logger.info('Connected to Box runtime via WebSocket.')
-                connected.set()
-                await self._handler_task
-            except Exception as exc:
-                if not connected.is_set():
-                    connect_error.append(exc)
-                    connected.set()
-
         async def on_connect_failed(ctrl, exc):
             connect_error.append(exc or BoxRuntimeUnavailableError('ws connection failed'))
             connected.set()
 
         ctrl = WebSocketClientController(ws_url=ws_url, make_connection_failed_callback=on_connect_failed)
-        self._ctrl_task = asyncio.create_task(ctrl.run(new_connection_callback))
+        self._ctrl_task = asyncio.create_task(
+            ctrl.run(self._make_connection_callback('WebSocket', connected, connect_error))
+        )
 
         try:
             await asyncio.wait_for(connected.wait(), timeout=30.0)
@@ -139,29 +136,8 @@ class BoxRuntimeConnector:
             self.ap.logger.info('Terminating managed box runtime process...')
             self._subprocess.terminate()
 
-        if self._subprocess_wait_task is not None:
-            self._subprocess_wait_task.cancel()
-            self._subprocess_wait_task = None
-
     def _load_configured_runtime_url(self) -> str:
         return str(get_box_config(self.ap).get('runtime_url', '')).strip()
 
     def _should_manage_local_runtime(self) -> bool:
         return not self.configured_runtime_url and platform.get_platform() != 'docker'
-
-    def _get_ws_relay_port(self) -> int:
-        """Extract the port for ws relay from ws_relay_base_url."""
-        from urllib.parse import urlparse
-        parsed = urlparse(self.ws_relay_base_url)
-        return parsed.port or 5410
-
-    def _get_rpc_ws_url(self) -> str:
-        """Derive the action RPC ws URL from the configured runtime URL.
-
-        The RPC endpoint is on port+1 relative to the ws relay port.
-        """
-        from urllib.parse import urlparse
-        parsed = urlparse(self.ws_relay_base_url)
-        host = parsed.hostname or '127.0.0.1'
-        port = (parsed.port or 5410) + 1
-        return f'ws://{host}:{port}'

src/langbot/pkg/box/runtime.py

@@ -37,6 +37,7 @@ class _ManagedProcess:
     started_at: dt.datetime
     attach_lock: asyncio.Lock
     stderr_chunks: collections.deque[str]
+    stderr_total_len: int = 0
     exit_code: int | None = None
     exited_at: dt.datetime | None = None
 
@@ -306,10 +307,10 @@ class BoxRuntime:
                 if not text:
                     continue
                 managed_process.stderr_chunks.append(text)
-                preview = '\n'.join(managed_process.stderr_chunks)
-                while len(preview) > _MANAGED_PROCESS_STDERR_PREVIEW_LIMIT and managed_process.stderr_chunks:
-                    managed_process.stderr_chunks.popleft()
-                    preview = '\n'.join(managed_process.stderr_chunks)
+                managed_process.stderr_total_len += len(text) + 1  # +1 for '\n' separator
+                while managed_process.stderr_total_len > _MANAGED_PROCESS_STDERR_PREVIEW_LIMIT and managed_process.stderr_chunks:
+                    removed = managed_process.stderr_chunks.popleft()
+                    managed_process.stderr_total_len -= len(removed) + 1
                 self.logger.info(f'LangBot Box managed process stderr: session_id={session_id} {text}')
         except Exception as exc:
             self.logger.warning(f'Failed to drain managed process stderr for {session_id}: {exc}')
@@ -378,18 +379,4 @@ class BoxRuntime:
 
     @staticmethod
     def _session_to_dict(info: BoxSessionInfo) -> dict:
-        return {
-            'session_id': info.session_id,
-            'backend_name': info.backend_name,
-            'backend_session_id': info.backend_session_id,
-            'image': info.image,
-            'network': info.network.value,
-            'host_path': info.host_path,
-            'host_path_mode': info.host_path_mode.value,
-            'cpus': info.cpus,
-            'memory_mb': info.memory_mb,
-            'pids_limit': info.pids_limit,
-            'read_only_rootfs': info.read_only_rootfs,
-            'created_at': info.created_at.isoformat(),
-            'last_used_at': info.last_used_at.isoformat(),
-        }
+        return info.model_dump(mode='json')

src/langbot/pkg/box/security.py

@@ -20,13 +20,6 @@ BLOCKED_HOST_PATHS = frozenset({
     '/var/run/podman',
 })
 
-RESERVED_CONTAINER_PATHS = frozenset({
-    '/workspace',
-    '/tmp',
-    '/var/tmp',
-    '/run',
-})
-
 
 def validate_sandbox_security(spec: BoxSpec) -> None:
     """Validate that a BoxSpec does not request dangerous container config.

src/langbot/pkg/box/server.py

@@ -26,7 +26,6 @@ from langbot_plugin.runtime.io.handler import Handler
 
 from .actions import LangBotToBoxAction
 from .errors import (
-    BoxError,
     BoxManagedProcessConflictError,
     BoxManagedProcessNotFoundError,
     BoxSessionNotFoundError,
@@ -38,15 +37,7 @@ logger = logging.getLogger('langbot.box.server')
 
 
 def _result_to_dict(result: BoxExecutionResult) -> dict:
-    return {
-        'session_id': result.session_id,
-        'backend_name': result.backend_name,
-        'status': result.status.value,
-        'exit_code': result.exit_code,
-        'stdout': result.stdout,
-        'stderr': result.stderr,
-        'duration_ms': result.duration_ms,
-    }
+    return result.model_dump(mode='json')
 
 
 class BoxServerHandler(Handler):

src/langbot/pkg/box/service.py

@@ -26,6 +26,11 @@ _INT_ADAPTER = pydantic.TypeAdapter(int)
 _UTC = _dt.timezone.utc
 _MAX_RECENT_ERRORS = 50
 
+
+def _is_path_under(path: str, root: str) -> bool:
+    """Check whether *path* equals *root* or is a child of *root*."""
+    return path == root or path.startswith(f'{root}{os.sep}')
+
 if TYPE_CHECKING:
     from ..core import app as core_app
     import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -274,7 +279,7 @@ class BoxService:
             )
 
         for allowed_root in self.allowed_host_mount_roots:
-            if self.default_host_workspace == allowed_root or self.default_host_workspace.startswith(f'{allowed_root}{os.sep}'):
+            if _is_path_under(self.default_host_workspace, allowed_root):
                 os.makedirs(self.default_host_workspace, exist_ok=True)
                 return
 
@@ -293,7 +298,7 @@ class BoxService:
             raise BoxValidationError('host_path mounting is disabled because no allowed_host_mount_roots are configured')
 
         for allowed_root in self.allowed_host_mount_roots:
-            if host_path == allowed_root or host_path.startswith(f'{allowed_root}{os.sep}'):
+            if _is_path_under(host_path, allowed_root):
                 return
 
         allowed_roots = ', '.join(self.allowed_host_mount_roots)