Preserve restricted agent sandbox compatibility

agents/restricted/tests/test_sandbox.py

@@ -1,6 +1,7 @@
 import asyncio
 import os
 import tempfile
+from concurrent.futures import ProcessPoolExecutor
 
 import pytest
 from hyperforge.codemode import sandbox
@@ -249,6 +250,34 @@ async def callback(_task):
             await asyncio.gather(task, return_exceptions=True)
 
 
+async def test_local_runner_preserves_legacy_python_values():
+    received = []
+
+    async def callback(task):
+        received.append(task.keyword_args["value"])
+        return task.keyword_args["value"]
+
+    with ProcessPoolExecutor(max_workers=1) as pool:
+        runner = sandbox.SandboxRunner.with_pool(pool, callback)
+        await runner.run(
+            WorkerExecutionRequest(
+                code="passthrough(value=value)",
+                question="Q?",
+                local_vars={"value": b"legacy-pickle-value"},
+                global_vars={},
+                function_names={
+                    "self": {
+                        "passthrough": FunctionDefinition(
+                            name="passthrough", description="", parameters={}
+                        )
+                    }
+                },
+            )
+        )
+
+    assert received == [b"legacy-pickle-value"]
+
+
 async def test_sandbox_rejects_concurrent_session(
     sandbox_conn, long_timeout, monkeypatch
 ):
@@ -286,14 +315,15 @@ async def test_sandbox_rejects_concurrent_session(
     assert isinstance(await first_rx.read_message(), SandboxMessage.Done)
 
 
-def test_sandbox_settings_default_to_private_optional_metrics() -> None:
+def test_sandbox_settings_preserve_legacy_metrics_defaults() -> None:
     configured = sandbox.SandboxSettings()
 
-    assert configured.sandbox_metrics_enabled is False
-    assert configured.sandbox_metrics_host == "127.0.0.1"
+    assert configured.sandbox_metrics_enabled is True
+    assert configured.sandbox_metrics_host == "0.0.0.0"
     assert configured.sandbox_socket_mode == "0600"
-    assert configured.sandbox_max_session_runtime_seconds == 60
-    assert configured.sandbox_max_session_memory_bytes == 512 * 1024 * 1024
+    assert configured.sandbox_max_concurrent_sessions is None
+    assert configured.sandbox_max_session_runtime_seconds is None
+    assert configured.sandbox_max_session_memory_bytes is None
 
 
 def test_sandbox_settings_accept_shared_group_socket_mode() -> None:

docs/harness-sdk.md

@@ -334,11 +334,12 @@ The existing exported `codemode` tool remains available for compatibility. It
 discovers registered tools and returns raw JSON-mode outputs. New applications
 that require capability isolation should use `create_codemode_tool()`.
 
-All Code Mode worker transports, including local pool and isolated-process
-execution, now move values over the bounded JSON worker protocol instead of
-pickle. Values crossing the worker boundary must be JSON worker values;
-previously picklable types such as `bytes`, sets, and dataclasses are rejected
-before transport. Convert binary data to base64 strings in tool results.
+Remote and isolated-process Code Mode execution move values over the bounded
+JSON worker protocol. Values crossing those worker boundaries must be JSON
+worker values; types such as `bytes`, sets, and dataclasses are rejected before
+transport. Convert binary data to base64 strings in tool results. The legacy
+local pool used by the restricted agent retains its existing pickle transport
+for backward compatibility.
 
 ### Code Mode Security
 
@@ -361,21 +362,22 @@ development environments.
 
 Sandbox deployment settings use secure defaults:
 
-- `SANDBOX_MAX_CONCURRENT_SESSIONS=4` bounds authenticated worker sessions and
-  concurrent authentication handshakes.
-- `SANDBOX_MAX_SESSION_RUNTIME_SECONDS=60` and
-  `SANDBOX_MAX_SESSION_MEMORY_BYTES=536870912` impose server-owned ceilings even
-  when a client omits limits. The remote client also bounds each connection by
-  the requested runtime (or the session ceiling when no runtime is requested)
-  plus `SANDBOX_TIMEOUT_SLACK_SECONDS=10`, so a hung sandbox cannot stall a
-  caller indefinitely.
+- `SANDBOX_MAX_CONCURRENT_SESSIONS` bounds authenticated worker sessions and
+  concurrent authentication handshakes when configured.
+- `SANDBOX_MAX_SESSION_RUNTIME_SECONDS` and
+  `SANDBOX_MAX_SESSION_MEMORY_BYTES` impose server-owned ceilings when
+  configured. Existing compatibility deployments leave these unset. Scoped
+  Code Mode supplies per-request runtime and memory limits. The remote client
+  bounds connections that have a requested or server runtime plus
+  `SANDBOX_TIMEOUT_SLACK_SECONDS=10`.
 - `SANDBOX_SOCKET_MODE=0600` restricts the socket to its owner.
 - `SANDBOX_SOCKET_GROUP` optionally changes group ownership. Use an explicitly
   provisioned shared group with `SANDBOX_SOCKET_MODE=0660` when the API and
   sandbox run as different non-root users.
-- `SANDBOX_METRICS_ENABLED=false` avoids creating an IP listener. If enabled,
-  `SANDBOX_METRICS_HOST` defaults to `127.0.0.1` and
-  `SANDBOX_METRICS_PORT` defaults to `8091`.
+- `SANDBOX_METRICS_ENABLED=true`, `SANDBOX_METRICS_HOST=0.0.0.0`, and
+  `SANDBOX_METRICS_PORT=8091` preserve the existing metrics endpoint. Set
+  `SANDBOX_METRICS_ENABLED=false` to avoid creating an IP listener, or bind the
+  endpoint to `127.0.0.1` when it does not need to be remotely scraped.
 - Local process IPC and remote length-prefixed JSON messages are bounded;
   oversized run requests, callbacks, responses, and frames are rejected.
 

hyperforge/src/hyperforge/codemode/sandbox.py

@@ -43,14 +43,14 @@ class SandboxSettings(BaseSettings):
     sandbox_socket_mode: str = "0600"
     sandbox_socket_group: str | None = None
     sandbox_metrics_port: int = 8091
-    sandbox_metrics_host: str = "127.0.0.1"
-    sandbox_metrics_enabled: bool = False
+    sandbox_metrics_host: str = "0.0.0.0"
+    sandbox_metrics_enabled: bool = True
     sandbox_token: str | None = None
-    sandbox_max_concurrent_sessions: int = Field(default=4, gt=0)
-    sandbox_max_session_runtime_seconds: float = Field(
-        default=60, gt=0, allow_inf_nan=False
+    sandbox_max_concurrent_sessions: int | None = Field(default=None, gt=0)
+    sandbox_max_session_runtime_seconds: float | None = Field(
+        default=None, gt=0, allow_inf_nan=False
     )
-    sandbox_max_session_memory_bytes: int = Field(default=512 * 1024 * 1024, gt=0)
+    sandbox_max_session_memory_bytes: int | None = Field(default=None, gt=0)
     sandbox_timeout_slack_seconds: float = Field(
         default=10.0, ge=0, allow_inf_nan=False
     )
@@ -154,9 +154,12 @@ async def run(self, request: WorkerExecutionRequest):
                 runtime = message.run.max_runtime_seconds
                 if runtime is None:
                     runtime = settings.sandbox_max_session_runtime_seconds
-                async with asyncio.timeout(
+                timeout = (
                     runtime + settings.sandbox_timeout_slack_seconds
-                ):
+                    if runtime is not None
+                    else None
+                )
+                async with asyncio.timeout(timeout):
                     return await self._run_with_remote_admission(message.run)
             except TimeoutError as exc:
                 raise RuntimeError("Codemode execution timed out") from exc
@@ -174,8 +177,10 @@ async def _run_with_remote_admission(self, request: WorkerExecutionRequest):
     async def _run_in_pool(self, request: WorkerExecutionRequest):
         loop = asyncio.get_running_loop()
         pipe_worker, pipe_restricted = Pipe()
-        worker = PythonAgentWorker(pipe_worker, debug=self.debug)
-        controller_task = asyncio.create_task(self.background_task(pipe_restricted))
+        worker = PythonAgentWorker(pipe_worker, debug=self.debug, json_protocol=False)
+        controller_task = asyncio.create_task(
+            self.background_task(pipe_restricted, json_protocol=False)
+        )
         try:
             await loop.run_in_executor(
                 self.pool,
@@ -324,7 +329,9 @@ def run_in_process(self, request: WorkerExecutionRequest):
         pipe_worker.close()
         return controller_task, process
 
-    async def background_task(self, pipe: Connection) -> bool:
+    async def background_task(
+        self, pipe: Connection, *, json_protocol: bool = True
+    ) -> bool:
         loop = asyncio.get_running_loop()
         context = contextvars.copy_context()
         try:
@@ -334,6 +341,7 @@ async def background_task(self, pipe: Connection) -> bool:
                 pipe,
                 loop,
                 context,
+                json_protocol,
             )
         finally:
             pipe.close()
@@ -383,14 +391,18 @@ def read_write_queue(
         pipe: Connection,
         loop: asyncio.AbstractEventLoop,
         context: contextvars.Context,
+        json_protocol: bool = True,
     ) -> bool:
         try:
             while True:
-                item = decode_protocol_value(
-                    pipe.recv_bytes(MAX_PACKET_BYTES),
-                    "Local sandbox request",
-                    max_bytes=MAX_PACKET_BYTES,
-                )
+                if json_protocol:
+                    item = decode_protocol_value(
+                        pipe.recv_bytes(MAX_PACKET_BYTES),
+                        "Local sandbox request",
+                        max_bytes=MAX_PACKET_BYTES,
+                    )
+                else:
+                    item = pipe.recv()
                 if not isinstance(item, RestrictedPythonTask):
                     raise ValueError("Invalid local sandbox request")
                 if item.function == "_error":
@@ -421,19 +433,22 @@ def copy_result(completed: asyncio.Task[WorkerTypes]) -> None:
                     result = future.result()
                 except Exception as exc:
                     result = WorkerError(error=str(exc))
-                try:
-                    encoded = encode_protocol_value(
-                        result,
-                        "Local sandbox response",
-                        max_bytes=MAX_PACKET_BYTES,
-                    )
-                except ValueError as exc:
-                    encoded = encode_protocol_value(
-                        WorkerError(error=str(exc)),
-                        "Local sandbox response error",
-                        max_bytes=MAX_PACKET_BYTES,
-                    )
-                pipe.send_bytes(encoded)
+                if json_protocol:
+                    try:
+                        encoded = encode_protocol_value(
+                            result,
+                            "Local sandbox response",
+                            max_bytes=MAX_PACKET_BYTES,
+                        )
+                    except ValueError as exc:
+                        encoded = encode_protocol_value(
+                            WorkerError(error=str(exc)),
+                            "Local sandbox response error",
+                            max_bytes=MAX_PACKET_BYTES,
+                        )
+                    pipe.send_bytes(encoded)
+                else:
+                    pipe.send(result)
         except EOFError:
             return False
 
@@ -526,14 +541,18 @@ def __init__(
 
     async def run(self, request: WorkerExecutionRequest):
         memory = settings.sandbox_max_session_memory_bytes
-        if request.max_memory_bytes is not None:
+        if memory is None:
+            memory = request.max_memory_bytes
+        elif request.max_memory_bytes is not None:
             memory = min(memory, request.max_memory_bytes)
         request = request.model_copy(update={"max_memory_bytes": memory})
         self.task, self.process = self.runner.run_in_process(request)
         try:
             await self._start_timeout(self.callback_wait_seconds)
             runtime = settings.sandbox_max_session_runtime_seconds
-            if request.max_runtime_seconds is not None:
+            if runtime is None:
+                runtime = request.max_runtime_seconds
+            elif request.max_runtime_seconds is not None:
                 runtime = min(runtime, request.max_runtime_seconds)
             async with asyncio.timeout(runtime):
                 clean_exit = await self.task
@@ -625,17 +644,19 @@ async def run_sandbox_server(
 
     active_sessions = 0
     admission_lock = asyncio.Lock()
-    handshake_limit = asyncio.Semaphore(settings.sandbox_max_concurrent_sessions)
+    max_sessions = settings.sandbox_max_concurrent_sessions
+    handshake_limit = asyncio.Semaphore(max_sessions) if max_sessions is not None else None
 
     async def handler(rx: asyncio.StreamReader, tx: asyncio.StreamWriter):
         nonlocal active_sessions
         reader = SandboxReader(rx)
         writer = SandboxWriter(tx)
-        if handshake_limit.locked():
+        if handshake_limit is not None and handshake_limit.locked():
             writer.close()
             await writer.wait_closed()
             return
-        await handshake_limit.acquire()
+        if handshake_limit is not None:
+            await handshake_limit.acquire()
         try:
             async with asyncio.timeout(1):
                 msg = await reader.read_message()
@@ -659,10 +680,12 @@ async def handler(rx: asyncio.StreamReader, tx: asyncio.StreamWriter):
             await writer.wait_closed()
             return
         finally:
-            handshake_limit.release()
+            if handshake_limit is not None:
+                handshake_limit.release()
 
         async with admission_lock:
-            if active_sessions >= settings.sandbox_max_concurrent_sessions:
+            active_limit = settings.sandbox_max_concurrent_sessions
+            if active_limit is not None and active_sessions >= active_limit:
                 await writer.write_message(
                     SandboxMessage.Error(error="Sandbox concurrency limit reached")
                 )
@@ -712,6 +735,8 @@ def _consume_task_result(task: asyncio.Future[Any]) -> None:
 
 async def _acquire_remote_session() -> None:
     global _remote_active_sessions
+    if settings.sandbox_max_concurrent_sessions is None:
+        return
     while True:
         with _remote_admission_lock:
             if _remote_active_sessions < settings.sandbox_max_concurrent_sessions:
@@ -722,6 +747,8 @@ async def _acquire_remote_session() -> None:
 
 def _release_remote_session() -> None:
     global _remote_active_sessions
+    if settings.sandbox_max_concurrent_sessions is None:
+        return
     with _remote_admission_lock:
         if _remote_active_sessions <= 0:
             raise RuntimeError("Remote sandbox admission counter underflow")

hyperforge/src/hyperforge/codemode/worker.py

@@ -112,10 +112,13 @@ def guarded_inplace(operator: str, left: Any, right: Any) -> Any:
 class PythonAgentWorker:
     context: Optional[Context] = None
 
-    def __init__(self, pipe: Connection, debug: bool = False):
+    def __init__(
+        self, pipe: Connection, debug: bool = False, json_protocol: bool = True
+    ):
         self.pipe = pipe
         self.functions_agent_id: Dict[str, List[str]] = {}
         self.debug = debug
+        self.json_protocol = json_protocol
 
     def _process_question_context_sync(
         self,
@@ -241,12 +244,18 @@ def valid_functions_call(
         return self._receive()
 
     def _send(self, task: RestrictedPythonTask) -> None:
-        self.pipe.send_bytes(encode_protocol_value(task, "Local sandbox request"))
+        if self.json_protocol:
+            self.pipe.send_bytes(encode_protocol_value(task, "Local sandbox request"))
+        else:
+            self.pipe.send(task)
 
     def _receive(self) -> Any:
-        result = decode_protocol_value(
-            self.pipe.recv_bytes(MAX_PROTOCOL_BYTES), "Local sandbox response"
-        )
+        if self.json_protocol:
+            result = decode_protocol_value(
+                self.pipe.recv_bytes(MAX_PROTOCOL_BYTES), "Local sandbox response"
+            )
+        else:
+            result = self.pipe.recv()
         if isinstance(result, WorkerError):
             raise RuntimeError(result.error)
         return result