From 756b1356d7e7dd8d26138c9a973073d636497941 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 13:36:14 +0000 Subject: [PATCH 01/15] Reach an offboard server over a Unix socket with a `unix://` URL A policy process on the same machine can now serve the offboard protocol on a Unix domain socket, and the rig reaches it as an ordinary remote endpoint. The carrier changes; the protocol does not. `InferenceClient` accepts `unix://[/api/v1/session[/]][?query]`. The socket path runs to the first `/api/v1` segment, so a bare socket path is the default session, and a model id and session params follow the path as they follow a host. The client dials the session with `unix_connect` and reads `/api/v1/models` through an `httpx` transport bound to the same socket. TLS does not apply. `PolicyServer(uds=...)` and `serve --uds` bind that socket path in place of `host:port`, and every vendor server CLI takes the flag. The server removes the socket file an earlier run left, so a restart binds again; anything else at the path stays and the bind fails. Positronic-Robotics/internal#1242 Ticket: Positronic-Robotics/internal#1242 #refs --- docs/inference.md | 2 + positronic/offboard/README.md | 4 + positronic/offboard/client.py | 68 +++++++++--- positronic/offboard/server.py | 43 +++++++- positronic/offboard/tests/conftest.py | 101 +++++++++++++----- .../offboard/tests/test_remote_policy.py | 43 +++++++- positronic/offboard/tests/test_server.py | 57 ++++++++++ 7 files changed, 270 insertions(+), 48 deletions(-) diff --git a/docs/inference.md b/docs/inference.md index 71cc4e9f6..096787aa2 100644 --- a/docs/inference.md +++ b/docs/inference.md @@ -58,6 +58,8 @@ uv run positronic eval run --eval=.sim.positronic.stack_cubes \ Accepted forms: `host`, `host:port`, and `https://host[:port][/api/v1/session[/]]` (`http`, `ws` and `wss` work too), each with an optional query. `https`/`wss` enable TLS. An omitted port is the scheme's own — 443 for TLS and 80 otherwise — so name the port a server listens on (`:8000` for every vendor server's default). Naming no model id serves the checkpoint the server pinned at startup. +**A Unix socket reaches a server on the same machine.** `--uds /run/policy.sock` binds that socket path in place of a host and a port. `--policy.url=unix:///run/policy.sock` dials it, over no network. The socket path runs to the first `/api/v1` segment, so a model id and session params follow it as they follow a host: `unix:///run/policy.sock/api/v1/session/10000?codec.fps=10`. Use this carrier for a policy process that runs beside the harness and has no network interface of its own. + **Credentials stay out of the URL, and out of the command line.** The URL is meant to be safe to paste around, so a token rides a header instead. It stays off the command line too: `save_run_metadata()` writes `sys.argv` beside the run's episodes. Three policy configs build the header: - `.authed_remote` — a bearer token read from `AUTH_TOKEN`, which it raises about when that is unset. Every endpoint [`workflows/nebius/serve.sh`](../workflows/nebius/README.md) creates is gated this way, whether the server checks the token itself or a proxy in front of it does. diff --git a/positronic/offboard/README.md b/positronic/offboard/README.md index ddf7140e1..3c37f4d32 100644 --- a/positronic/offboard/README.md +++ b/positronic/offboard/README.md @@ -74,6 +74,10 @@ Because the whole session configuration fits in the URL, one string is a complet `http(s)`/`ws(s)` URLs — optionally with `/api/v1/session/` — and forwards the query string verbatim. Credentials are the exception and stay a separate `headers` argument, so the URL itself is safe to hand around. +A `unix://` URL reaches a server on the same machine over a Unix socket, which needs no network: the server +binds the path with `--uds`, and `unix:///run/policy.sock[/api/v1/session[/]][?query]` dials it. The +socket path runs to the first `/api/v1` segment; everything after it is the URL path the server reads. + ### WebSocket Flow #### 1. Handshake diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 98369c4f7..15bd9e538 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -1,14 +1,16 @@ import logging +import re import ssl import time import urllib.parse from enum import Enum +from functools import partial from http import HTTPStatus from typing import Any import httpx from websockets.exceptions import ConnectionClosed, InvalidHandshake, InvalidStatus -from websockets.sync.client import connect +from websockets.sync.client import connect, unix_connect from websockets.sync.connection import Connection from . import protocol @@ -117,6 +119,20 @@ def _session_path(path: str, url: str) -> str: return path +def _socket_and_path(split: urllib.parse.SplitResult, url: str) -> tuple[str, str]: + """The socket path a ``unix://`` URL names, and the URL path left over for the server. + + The socket path is absolute and runs to the first ``/api/v1`` segment. A URL that names no such + segment is the bare socket path, which addresses the default session. + """ + if split.netloc or not split.path.startswith('/'): + raise ValueError(f'Socket path must be absolute in {url!r}; write unix:///path/to.sock') + marker = re.search(r'/api/v1(?=/|$)', split.path) + if marker is None: + return split.path, '' + return split.path[: marker.start()], split.path[marker.start() :] + + class _ConnectOutcome(Enum): RETRY = 'retry' SURFACE = 'surface' @@ -156,6 +172,11 @@ class InferenceClient: session — the model id it names and the query it carries as session params — reaches the server exactly as written, so every session opened here serves that model with those params. + ``unix://[/api/v1/session[/]][?query]`` reaches a server on the same + machine over a Unix domain socket, which needs no network. The socket path runs to the first + ``/api/v1`` segment, so ``unix:///run/policy.sock`` is the default session and + ``unix:///run/policy.sock/api/v1/session/10000?fps=10`` names a model and a param. TLS does not apply. + ``headers`` carry auth, whether the server checks it or a proxy in front of it does — credentials stay out of the URL, which is meant to be safe to hand around. @@ -174,22 +195,35 @@ def __init__( infer_timeout: float = DEFAULT_INFER_TIMEOUT, ): split = urllib.parse.urlsplit(url if '://' in url else f'//{url}') - if split.scheme not in ('', 'http', 'ws', 'https', 'wss'): + if split.scheme not in ('', 'http', 'ws', 'https', 'wss', 'unix'): raise ValueError(f'Unsupported scheme {split.scheme!r} in {url!r}') - if not split.hostname: - raise ValueError(f'No host in {url!r}') secure = split.scheme in ('https', 'wss') + if split.scheme == 'unix': + uds, path = _socket_and_path(split, url) + # A socket path is not a host. The server reads the path and the query alone, so the + # handshake asks for them under a host that stands in for the socket. + netloc = 'localhost' + else: + uds = None + if not split.hostname: + raise ValueError(f'No host in {url!r}') + path = split.path + default_port = 443 if secure else 80 + # urlsplit strips the brackets an IPv6 host needs back in a netloc. + host = f'[{split.hostname}]' if ':' in split.hostname else split.hostname + port = default_port if split.port is None else split.port + netloc = host if port == default_port else f'{host}:{port}' ws_scheme = 'wss' if secure else 'ws' http_scheme = 'https' if secure else 'http' - default_port = 443 if secure else 80 - # urlsplit strips the brackets an IPv6 host needs back in a netloc. - host = f'[{split.hostname}]' if ':' in split.hostname else split.hostname - port = default_port if split.port is None else split.port - netloc = host if port == default_port else f'{host}:{port}' # Forwarded verbatim: the server reads each param value as a JSON literal, and only whoever wrote # the URL knows whether `true` means the bool or the string. query = f'?{split.query}' if split.query else '' - self.session_url = f'{ws_scheme}://{netloc}{_session_path(split.path, url)}{query}' + session_path = _session_path(path, url) + self.uds = uds + # The URL the websocket handshake asks for, and the TCP address to dial when there is no socket. + self._ws_uri = f'{ws_scheme}://{netloc}{session_path}{query}' + # What an error names. Over a socket the stand-in host would not say which socket failed. + self.session_url = self._ws_uri if uds is None else f'unix://{uds}{session_path}{query}' self.api_url = f'{http_scheme}://{netloc}/api/v1' self.headers = dict(headers) if headers else None self.open_timeout = open_timeout @@ -207,12 +241,12 @@ def new_session(self) -> InferenceSession: # A proxy between here and the server closes a connection it has read nothing from, often # after 60s — well inside one ``infer_timeout`` inference, which sends nothing until it # answers. The pings keep it open. - ws = connect( - self.session_url, - open_timeout=self.open_timeout, - additional_headers=self.headers, - ping_interval=20.0, + dial = ( + partial(connect, self._ws_uri) + if self.uds is None + else partial(unix_connect, self.uds, uri=self._ws_uri) ) + ws = dial(open_timeout=self.open_timeout, additional_headers=self.headers, ping_interval=20.0) return InferenceSession(ws, infer_timeout=self.infer_timeout) # ``SSLCertVerificationError`` is an ``ssl.SSLError``, but a bad certificate is permanent # misconfiguration, not a cold start — surface it immediately instead of retrying to the deadline. @@ -238,6 +272,8 @@ def new_session(self) -> InferenceSession: def list_models(self) -> list[str]: """List available models from the server.""" - response = httpx.get(f'{self.api_url}/models', headers=self.headers) + transport = None if self.uds is None else httpx.HTTPTransport(uds=self.uds) + with httpx.Client(transport=transport) as client: + response = client.get(f'{self.api_url}/models', headers=self.headers) response.raise_for_status() return response.json()['models'] diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index fbe7d33b7..0f0c97911 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -5,6 +5,7 @@ import json import logging import os +import stat import time from collections import Counter from collections.abc import Callable @@ -202,6 +203,9 @@ class PolicyServer: The default checkpoint is resolved once, at startup, and pinned for every request that names no explicit one — a running server never switches to a newer checkpoint that lands later. A request for /api/v1/session/{model_id} still loads that one on demand. + + ``uds`` binds a Unix socket path instead of ``host:port``, which serves a client on the same machine + over no network. A client reaches it with a ``unix://`` URL. """ def __init__( @@ -212,6 +216,7 @@ def __init__( recording_dir: str | None = None, idle_timeout_min: float | None = None, auth_token: str | None = None, + uds: str | None = None, ): self._pipeline_cfg = pipeline if isinstance(pipeline, cfn.Config) else None self._pipeline = pipeline.instantiate() if isinstance(pipeline, cfn.Config) else pipeline @@ -226,7 +231,11 @@ def __init__( self._manager = PolicyManager(self._source) self.host = host self.port = port - self.metadata: dict[str, Any] = {offboard_keys.HOST: host, offboard_keys.PORT: port} + self.uds = uds + # Where the server listens, as the handshake reports it. A Unix socket has no port. + self.metadata: dict[str, Any] = ( + {offboard_keys.HOST: host, offboard_keys.PORT: port} if uds is None else {offboard_keys.HOST: uds} + ) # Synced once; each session builds its own ``Recorder`` so concurrent streams never mix. self._recording_dir = pos3.sync(recording_dir) if recording_dir else None @@ -407,10 +416,25 @@ async def _idle_watchdog(self, server: uvicorn.Server): server.should_exit = True return + @staticmethod + def clear_stale_socket(path: str) -> None: + """Remove the socket file left by an earlier run, so a restart can bind ``path`` again. + + Anything else at the path stays: the bind then fails rather than deleting a file nobody meant to lose. + """ + try: + mode = os.stat(path).st_mode + except FileNotFoundError: + return + if stat.S_ISSOCK(mode): + os.unlink(path) + def serve(self): async def _run(): await self._startup() - config = uvicorn.Config(self.app, host=self.host, port=self.port, log_level='info') + if self.uds is not None: + self.clear_stale_socket(self.uds) + config = uvicorn.Config(self.app, host=self.host, port=self.port, uds=self.uds, log_level='info') server = uvicorn.Server(config) self._last_activity = time.monotonic() watchdog = None @@ -430,14 +454,24 @@ async def _run(): self._manager.close() -@cfn.config(host='0.0.0.0', port=8000, recording_dir=None, idle_timeout_min=None) -def serve(pipeline: cfn.Config, host: str, port: int, recording_dir: str | None, idle_timeout_min: float | None): +@cfn.config(host='0.0.0.0', port=8000, recording_dir=None, idle_timeout_min=None, uds=None) +def serve( + pipeline: cfn.Config, + host: str, + port: int, + recording_dir: str | None, + idle_timeout_min: float | None, + uds: str | None, +): """The CLI entry point every vendor server exposes: bind ``pipeline``, and the commands are configs of this. Only the socket and the recording taps are flags of their own; everything the served model is — codec, source, checkpoint directory — is reached through the pipeline itself (``--pipeline.source.checkpoints_dir=...``), so each of those values has exactly one name. + ``--uds`` binds that Unix socket path and leaves ``host`` and ``port`` unused, which serves a client + on the same machine over no network. + The bearer token gating the server comes from ``AUTH_TOKEN_ENV`` rather than a flag, which would put a secret in the process arguments; unset serves open. """ @@ -448,4 +482,5 @@ def serve(pipeline: cfn.Config, host: str, port: int, recording_dir: str | None, recording_dir=recording_dir, idle_timeout_min=idle_timeout_min, auth_token=os.environ.get(AUTH_TOKEN_ENV), + uds=uds, ).serve() diff --git a/positronic/offboard/tests/conftest.py b/positronic/offboard/tests/conftest.py index 54dfcbb7a..49ef18ff3 100644 --- a/positronic/offboard/tests/conftest.py +++ b/positronic/offboard/tests/conftest.py @@ -1,5 +1,7 @@ import asyncio +import os import socket +import tempfile import threading import time from collections.abc import Callable, Generator, Mapping @@ -21,41 +23,88 @@ def _find_free_port() -> int: return s.getsockname()[1] +RunningServers = list[tuple[uvicorn.Server, threading.Thread]] + StartServer = Callable[..., tuple[str, int, PolicyServer]] +StartUnixServer = Callable[..., PolicyServer] -@pytest.fixture -def start_server() -> Generator[StartServer, None, None]: - """Factory serving pipelines on daemon threads; every started server is stopped and joined at teardown.""" - running: list[tuple[uvicorn.Server, threading.Thread]] = [] - def start(pipeline, **server_kwargs) -> tuple[str, int, PolicyServer]: - server = PolicyServer(pipeline, host='localhost', port=_find_free_port(), **server_kwargs) - uv_server = uvicorn.Server(uvicorn.Config(server.app, host=server.host, port=server.port, log_level='warning')) - - async def _run(): - await server._startup() - await uv_server.serve() - - thread = threading.Thread(target=asyncio.run, args=(_run(),), daemon=True) - thread.start() - running.append((uv_server, thread)) - - deadline = time.time() + 5.0 - while time.time() < deadline: - try: - with socket.create_connection((server.host, server.port), timeout=0.1): - return server.host, server.port, server - except (ConnectionRefusedError, OSError): - time.sleep(0.05) - raise RuntimeError('Server failed to start') - - yield start +@pytest.fixture +def running_servers() -> Generator[RunningServers, None, None]: + """Every server a test started, stopped and joined at teardown.""" + running: RunningServers = [] + yield running for uv_server, thread in running: uv_server.should_exit = True thread.join(timeout=5.0) +def _serve_in_background(server: PolicyServer, config: uvicorn.Config, running: RunningServers) -> None: + uv_server = uvicorn.Server(config) + + async def _run(): + await server._startup() + await uv_server.serve() + + thread = threading.Thread(target=asyncio.run, args=(_run(),), daemon=True) + thread.start() + running.append((uv_server, thread)) + + +def _wait_until_it_accepts(dial: Callable[[], None]) -> None: + deadline = time.time() + 5.0 + while time.time() < deadline: + try: + dial() + return + except OSError: + time.sleep(0.05) + raise RuntimeError('Server failed to start') + + +@pytest.fixture +def socket_path() -> Generator[str, None, None]: + """A path for a Unix socket, short enough for the 104-byte limit that ``tmp_path`` can pass.""" + with tempfile.TemporaryDirectory(dir='/tmp') as directory: + yield os.path.join(directory, 's.sock') + + +@pytest.fixture +def start_server(running_servers: RunningServers) -> StartServer: + """Factory serving pipelines on daemon threads.""" + + def start(pipeline, **server_kwargs) -> tuple[str, int, PolicyServer]: + server = PolicyServer(pipeline, host='localhost', port=_find_free_port(), **server_kwargs) + config = uvicorn.Config(server.app, host=server.host, port=server.port, log_level='warning') + _serve_in_background(server, config, running_servers) + _wait_until_it_accepts(lambda: socket.create_connection((server.host, server.port), timeout=0.1).close()) + return server.host, server.port, server + + return start + + +def _dial_unix(path: str) -> None: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(0.1) + sock.connect(path) + + +@pytest.fixture +def start_unix_server(running_servers: RunningServers) -> StartUnixServer: + """Factory serving pipelines on a Unix socket, as ``PolicyServer.serve`` binds one.""" + + def start(pipeline, uds: str, **server_kwargs) -> PolicyServer: + server = PolicyServer(pipeline, uds=uds, **server_kwargs) + PolicyServer.clear_stale_socket(uds) + config = uvicorn.Config(server.app, uds=uds, log_level='warning') + _serve_in_background(server, config, running_servers) + _wait_until_it_accepts(lambda: _dial_unix(uds)) + return server + + return start + + @pytest.fixture def open_session() -> Generator[Callable[..., tuple[Session, Executor]], None, None]: """Opens a policy's session against a runtime that serves its functions, as the harness does.""" diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py index 9f2f6201e..6da232cbc 100644 --- a/positronic/offboard/tests/test_remote_policy.py +++ b/positronic/offboard/tests/test_remote_policy.py @@ -117,7 +117,8 @@ def test_new_session_without_headers_passes_none(self): def test_list_models_passes_headers(self): headers = {'Modal-Key': 'k', 'Modal-Secret': 's'} - with patch('positronic.offboard.client.httpx.get') as mock_get: + with patch('positronic.offboard.client.httpx.Client') as mock_client: + mock_get = mock_client.return_value.__enter__.return_value.get mock_get.return_value.json.return_value = {'models': ['m1']} client = InferenceClient('localhost:8000', headers=headers) @@ -127,7 +128,8 @@ def test_list_models_passes_headers(self): assert mock_get.call_args.kwargs['headers'] == headers def test_list_models_without_headers_passes_none(self): - with patch('positronic.offboard.client.httpx.get') as mock_get: + with patch('positronic.offboard.client.httpx.Client') as mock_client: + mock_get = mock_client.return_value.__enter__.return_value.get mock_get.return_value.json.return_value = {'models': []} client = InferenceClient('localhost:8000') client.list_models() @@ -207,6 +209,43 @@ def test_every_session_dials_the_session_url(self): for call in mock_connect.call_args_list: assert call.args[0] == client.session_url == 'ws://localhost:8000/api/v1/session/10000?fps=10' + def test_a_socket_path_alone_is_the_default_session(self): + client = InferenceClient('unix:///run/policy.sock') + assert client.uds == '/run/policy.sock' + assert client.session_url == 'unix:///run/policy.sock/api/v1/session' + assert client.api_url == 'http://localhost/api/v1' + + def test_a_socket_path_ends_at_the_api_segment(self): + client = InferenceClient('unix:///run/policy.sock/api/v1/session/10000?fps=10') + assert client.uds == '/run/policy.sock' + assert client.session_url == 'unix:///run/policy.sock/api/v1/session/10000?fps=10' + + def test_a_query_on_a_bare_socket_path_rides_along(self): + client = InferenceClient('unix:///run/policy.sock?fps=10') + assert client.uds == '/run/policy.sock' + assert client.session_url == 'unix:///run/policy.sock/api/v1/session?fps=10' + + def test_the_api_marker_is_a_whole_segment(self): + """A socket under a directory whose name only starts with the marker is still the whole path.""" + client = InferenceClient('unix:///run/api/v1x/policy.sock') + assert client.uds == '/run/api/v1x/policy.sock' + assert client.session_url == 'unix:///run/api/v1x/policy.sock/api/v1/session' + + def test_a_relative_socket_path_rejected(self): + with pytest.raises(ValueError, match='absolute'): + InferenceClient('unix://policy.sock') + + def test_a_unix_url_dials_the_socket_and_asks_for_the_url_path(self): + with ( + patch('positronic.offboard.client.unix_connect') as mock_connect, + patch('positronic.offboard.client.InferenceSession'), + ): + InferenceClient('unix:///run/policy.sock/api/v1/session/10000?fps=10').new_session() + + mock_connect.assert_called_once() + assert mock_connect.call_args.args[0] == '/run/policy.sock' + assert mock_connect.call_args.kwargs['uri'] == 'ws://localhost/api/v1/session/10000?fps=10' + def _refused(status: HTTPStatus) -> InvalidStatus: return InvalidStatus(Response(status, 'refused', Headers())) diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 2bf3bea0a..21351c573 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -1,4 +1,5 @@ import os +import pathlib import socket import time import urllib.parse @@ -232,6 +233,62 @@ def test_local_stack_declared_in_handshake(start_server, make_mock_policy): session.close() +@pytest.fixture +def unix_stub_server(start_unix_server, socket_path, make_mock_policy) -> tuple[str, MagicMock]: + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + start_unix_server(ChunkedSchedule() | remote | _StubSource(policy), socket_path) + return socket_path, policy + + +def test_a_pipeline_served_over_a_unix_socket(unix_stub_server): + socket_path, policy = unix_stub_server + client = InferenceClient(f'unix://{socket_path}') + + assert client.list_models() == ['stub'] + session = client.new_session() + try: + assert session.metadata['model_name'] == 'stub' + assert session.metadata['local_stack'] == {'name': 'chunked_schedule'} + assert session.metadata[offboard_keys.HOST] == socket_path + assert offboard_keys.PORT not in session.metadata + + obs = {'image': 'test'} + assert session.infer(obs) == [{'action': [1, 2, 3]}] + policy._mock_session.assert_called_with(obs, ANY) + finally: + session.close() + + +def test_a_unix_url_carries_the_model_id_past_the_socket_path(unix_stub_server): + socket_path, _policy = unix_stub_server + + session = InferenceClient(f'unix://{socket_path}/api/v1/session/10000').new_session() + try: + assert session.metadata['checkpoint_id'] == '10000' + finally: + session.close() + + +def test_a_server_binds_over_the_socket_an_earlier_run_left(start_unix_server, socket_path, make_mock_policy): + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as stale: + stale.bind(socket_path) + + start_unix_server(ChunkedSchedule() | remote | _StubSource(policy), socket_path) + + assert InferenceClient(f'unix://{socket_path}').list_models() == ['stub'] + + +def test_a_path_that_is_not_a_socket_is_left_alone(socket_path): + """A wrong ``uds`` fails the bind rather than deleting a file nobody meant to lose.""" + path = pathlib.Path(socket_path) + path.write_text('not a socket') + + PolicyServer.clear_stale_socket(socket_path) + + assert path.read_text() == 'not a socket' + + def test_pipeline_with_no_rig_side_half_refused_at_startup(make_mock_policy): """Nothing left of the marker leaves the rig nothing to run, so the server refuses to serve it.""" stub = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) From 8c57f4a5eb433cc9f06c4812adb68548a3b7a09f Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 13:39:51 +0000 Subject: [PATCH 02/15] Assert the handshake's own keys through `offboard_keys` in the socket tests The two keys the constants already name read through them, beside the `HOST` and `PORT` assertions in the same test, so a rename reaches the test. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/tests/test_server.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 21351c573..7eb866cde 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -248,7 +248,7 @@ def test_a_pipeline_served_over_a_unix_socket(unix_stub_server): session = client.new_session() try: assert session.metadata['model_name'] == 'stub' - assert session.metadata['local_stack'] == {'name': 'chunked_schedule'} + assert session.metadata[offboard_keys.LOCAL_STACK] == {'name': 'chunked_schedule'} assert session.metadata[offboard_keys.HOST] == socket_path assert offboard_keys.PORT not in session.metadata @@ -264,7 +264,7 @@ def test_a_unix_url_carries_the_model_id_past_the_socket_path(unix_stub_server): session = InferenceClient(f'unix://{socket_path}/api/v1/session/10000').new_session() try: - assert session.metadata['checkpoint_id'] == '10000' + assert session.metadata[offboard_keys.CHECKPOINT_ID] == '10000' finally: session.close() From 8ab856e5325faf924df2c1c1cce0268cf8bc77cb Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 13:43:36 +0000 Subject: [PATCH 03/15] State the socket-path rule in one place each The `unix://` grammar has one home per reader: the client's class docstring for a caller, the offboard README for the protocol. The `serve` docstring, the private helper and the inference guide each said it again. Ticket: Positronic-Robotics/internal#1242 #refs --- docs/inference.md | 2 +- positronic/offboard/client.py | 6 +----- positronic/offboard/server.py | 3 +-- 3 files changed, 3 insertions(+), 8 deletions(-) diff --git a/docs/inference.md b/docs/inference.md index 096787aa2..e8ea75737 100644 --- a/docs/inference.md +++ b/docs/inference.md @@ -58,7 +58,7 @@ uv run positronic eval run --eval=.sim.positronic.stack_cubes \ Accepted forms: `host`, `host:port`, and `https://host[:port][/api/v1/session[/]]` (`http`, `ws` and `wss` work too), each with an optional query. `https`/`wss` enable TLS. An omitted port is the scheme's own — 443 for TLS and 80 otherwise — so name the port a server listens on (`:8000` for every vendor server's default). Naming no model id serves the checkpoint the server pinned at startup. -**A Unix socket reaches a server on the same machine.** `--uds /run/policy.sock` binds that socket path in place of a host and a port. `--policy.url=unix:///run/policy.sock` dials it, over no network. The socket path runs to the first `/api/v1` segment, so a model id and session params follow it as they follow a host: `unix:///run/policy.sock/api/v1/session/10000?codec.fps=10`. Use this carrier for a policy process that runs beside the harness and has no network interface of its own. +**A Unix socket reaches a server on the same machine.** `--uds /run/policy.sock` binds that socket path in place of a host and a port. `--policy.url=unix:///run/policy.sock` dials it, over no network. A model id and session params follow the socket path as they follow a host: `unix:///run/policy.sock/api/v1/session/10000?codec.fps=10`. Use this carrier for a policy process that runs beside the harness and has no network interface of its own. **Credentials stay out of the URL, and out of the command line.** The URL is meant to be safe to paste around, so a token rides a header instead. It stays off the command line too: `save_run_metadata()` writes `sys.argv` beside the run's episodes. Three policy configs build the header: diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 15bd9e538..49a0de930 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -120,11 +120,7 @@ def _session_path(path: str, url: str) -> str: def _socket_and_path(split: urllib.parse.SplitResult, url: str) -> tuple[str, str]: - """The socket path a ``unix://`` URL names, and the URL path left over for the server. - - The socket path is absolute and runs to the first ``/api/v1`` segment. A URL that names no such - segment is the bare socket path, which addresses the default session. - """ + """The socket path a ``unix://`` URL names, and the URL path left over for the server.""" if split.netloc or not split.path.startswith('/'): raise ValueError(f'Socket path must be absolute in {url!r}; write unix:///path/to.sock') marker = re.search(r'/api/v1(?=/|$)', split.path) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index 0f0c97911..877051db0 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -469,8 +469,7 @@ def serve( codec, source, checkpoint directory — is reached through the pipeline itself (``--pipeline.source.checkpoints_dir=...``), so each of those values has exactly one name. - ``--uds`` binds that Unix socket path and leaves ``host`` and ``port`` unused, which serves a client - on the same machine over no network. + ``--uds`` binds that Unix socket path and leaves ``host`` and ``port`` unused. The bearer token gating the server comes from ``AUTH_TOKEN_ENV`` rather than a flag, which would put a secret in the process arguments; unset serves open. From 30f6c4bb0d90c96909b31d6c1c5581ec6e7dcbc4 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 13:50:01 +0000 Subject: [PATCH 04/15] Leave a Unix socket a live server listens on `clear_stale_socket` judged a socket stale by its file type alone, so a second server on one path unlinked the first server's address and took its future connections. It now connects to the socket first: a refused connection means nobody listens and the file is stale; an accepted one means the socket has an owner, and the bind fails instead. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/server.py | 13 ++++++++++--- positronic/offboard/tests/test_server.py | 11 +++++++++++ 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index 877051db0..c34e721bf 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -5,6 +5,7 @@ import json import logging import os +import socket import stat import time from collections import Counter @@ -420,14 +421,20 @@ async def _idle_watchdog(self, server: uvicorn.Server): def clear_stale_socket(path: str) -> None: """Remove the socket file left by an earlier run, so a restart can bind ``path`` again. - Anything else at the path stays: the bind then fails rather than deleting a file nobody meant to lose. + A socket that still accepts a connection belongs to a live server and stays, as does anything at + the path that is not a socket. The bind then fails instead of taking an address off its owner. """ try: mode = os.stat(path).st_mode except FileNotFoundError: return - if stat.S_ISSOCK(mode): - os.unlink(path) + if not stat.S_ISSOCK(mode): + return + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe: + try: + probe.connect(path) + except ConnectionRefusedError: + os.unlink(path) def serve(self): async def _run(): diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 7eb866cde..a4c58f7b4 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -289,6 +289,17 @@ def test_a_path_that_is_not_a_socket_is_left_alone(socket_path): assert path.read_text() == 'not a socket' +def test_a_socket_a_live_server_listens_on_is_left_alone(socket_path): + """Unlinking it would take the address off its owner and route new sessions to the wrong server.""" + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as live: + live.bind(socket_path) + live.listen() + + PolicyServer.clear_stale_socket(socket_path) + + assert pathlib.Path(socket_path).is_socket() + + def test_pipeline_with_no_rig_side_half_refused_at_startup(make_mock_policy): """Nothing left of the marker leaves the rig nothing to run, so the server refuses to serve it.""" stub = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) From 2aedb6a2b6d18e9717d5d1b455f0ea2dba0dd7de Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 21:07:34 +0000 Subject: [PATCH 05/15] Refuse a Unix socket path a live server already listens on `clear_stale_socket` left a live socket in place and relied on the bind to fail. It does not: `asyncio.create_unix_server`, which uvicorn calls, unlinks an existing socket file before it binds, so a second server on one path took the first server's future connections and neither side said anything. The check now raises `EADDRINUSE` naming the path, before uvicorn is built. It is `claim_socket_path` for that reason: it takes the path or refuses it, and a name that says only "clear" no longer covers what it does. A refused connection still means nobody listens, so the file is stale and goes. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/server.py | 14 +++++++++----- positronic/offboard/tests/conftest.py | 2 +- positronic/offboard/tests/test_server.py | 21 ++++++++++++++++----- 3 files changed, 26 insertions(+), 11 deletions(-) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index c34e721bf..eeaa0340f 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -1,6 +1,7 @@ """The inference server: serves a policy pipeline (see ``positronic.policy.spec``) over the offboard protocol.""" import asyncio +import errno import hmac import json import logging @@ -418,11 +419,12 @@ async def _idle_watchdog(self, server: uvicorn.Server): return @staticmethod - def clear_stale_socket(path: str) -> None: - """Remove the socket file left by an earlier run, so a restart can bind ``path`` again. + def claim_socket_path(path: str) -> None: + """Take ``path`` for this server: remove the socket an earlier run left, or refuse a live one. - A socket that still accepts a connection belongs to a live server and stays, as does anything at - the path that is not a socket. The bind then fails instead of taking an address off its owner. + The refusal has to happen here, before uvicorn: ``asyncio.create_unix_server`` unlinks an + existing socket file and binds a new one, so a second server would silently take a live + server's future connections. A refused connection means nobody listens and the file is stale. """ try: mode = os.stat(path).st_mode @@ -435,12 +437,14 @@ def clear_stale_socket(path: str) -> None: probe.connect(path) except ConnectionRefusedError: os.unlink(path) + return + raise OSError(errno.EADDRINUSE, f'A server already listens on {path!r}') def serve(self): async def _run(): await self._startup() if self.uds is not None: - self.clear_stale_socket(self.uds) + self.claim_socket_path(self.uds) config = uvicorn.Config(self.app, host=self.host, port=self.port, uds=self.uds, log_level='info') server = uvicorn.Server(config) self._last_activity = time.monotonic() diff --git a/positronic/offboard/tests/conftest.py b/positronic/offboard/tests/conftest.py index 49ef18ff3..285b5e33a 100644 --- a/positronic/offboard/tests/conftest.py +++ b/positronic/offboard/tests/conftest.py @@ -96,7 +96,7 @@ def start_unix_server(running_servers: RunningServers) -> StartUnixServer: def start(pipeline, uds: str, **server_kwargs) -> PolicyServer: server = PolicyServer(pipeline, uds=uds, **server_kwargs) - PolicyServer.clear_stale_socket(uds) + PolicyServer.claim_socket_path(uds) config = uvicorn.Config(server.app, uds=uds, log_level='warning') _serve_in_background(server, config, running_servers) _wait_until_it_accepts(lambda: _dial_unix(uds)) diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index a4c58f7b4..7ada9c727 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -1,3 +1,4 @@ +import errno import os import pathlib import socket @@ -284,20 +285,30 @@ def test_a_path_that_is_not_a_socket_is_left_alone(socket_path): path = pathlib.Path(socket_path) path.write_text('not a socket') - PolicyServer.clear_stale_socket(socket_path) + PolicyServer.claim_socket_path(socket_path) assert path.read_text() == 'not a socket' -def test_a_socket_a_live_server_listens_on_is_left_alone(socket_path): - """Unlinking it would take the address off its owner and route new sessions to the wrong server.""" +@pytest.mark.timeout(30.0) +def test_a_server_refuses_a_socket_a_live_server_listens_on(socket_path, make_mock_policy): + """``asyncio.create_unix_server`` unlinks the file it finds, so only a refusal here keeps the + address with the server that owns it.""" + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + server = PolicyServer(ChunkedSchedule() | remote | _StubSource(policy), uds=socket_path) + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as live: live.bind(socket_path) live.listen() - PolicyServer.clear_stale_socket(socket_path) + with pytest.raises(OSError) as refusal: + server.serve() + assert refusal.value.errno == errno.EADDRINUSE + assert socket_path in str(refusal.value) - assert pathlib.Path(socket_path).is_socket() + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.connect(socket_path) + assert live.accept()[0].close() is None def test_pipeline_with_no_rig_side_half_refused_at_startup(make_mock_policy): From fc0091303b4fc0903319bc4e5b766f7debf40860 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 21:29:37 +0000 Subject: [PATCH 06/15] Claim a Unix socket path by binding it, and serve that descriptor Two defects in the same guard, both from defending a path instead of taking it. The probe told a live server from a stale file but claimed nothing, so two servers starting against an absent or stale path both passed it and the second bind took the first. The bind is the claim now: it happens first, and a path already held fails it. A probe follows only to read what holds the path, and it answers one question -- is this a socket nobody serves, so replacing it costs nobody anything. That probe also blocked with no bound, so a listener with a full backlog held startup open. It waits `LIVE_SOCKET_PROBE_SEC` and reads a wait that runs out as a live server. `serve` hands uvicorn the bound descriptor through `uvicorn.Config(fd=...)` rather than the path. Uvicorn then calls `create_server(sock=...)`, which binds nothing, where `uds=` reaches `create_unix_server(path=...)` and unlinks what it finds. The socket keeps the 0o666 uvicorn gives one it binds itself, and the file stays behind at shutdown: a successor reads it as stale, and an unlink here could take a path that successor has already claimed. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/server.py | 76 ++++++++++++++++++------ positronic/offboard/tests/conftest.py | 14 +++-- positronic/offboard/tests/test_server.py | 36 +++++++++-- 3 files changed, 98 insertions(+), 28 deletions(-) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index eeaa0340f..47bc3ee34 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -35,6 +35,13 @@ AUTH_HEADER = 'Authorization' +# A server whose backlog is full holds a connect open, so the probe for a live socket bounds its wait +# rather than stalling startup on it. A wait that runs out counts as live. +LIVE_SOCKET_PROBE_SEC = 1.0 + +# The mode uvicorn gives a Unix socket it binds itself. +UDS_MODE = 0o666 + def bearer(token: str) -> str: """The ``AUTH_HEADER`` value carrying ``token``.""" @@ -419,43 +426,74 @@ async def _idle_watchdog(self, server: uvicorn.Server): return @staticmethod - def claim_socket_path(path: str) -> None: - """Take ``path`` for this server: remove the socket an earlier run left, or refuse a live one. + def _is_stale_socket(path: str) -> bool: + """Whether ``path`` is a socket no server answers on, so replacing it takes nothing from anybody. - The refusal has to happen here, before uvicorn: ``asyncio.create_unix_server`` unlinks an - existing socket file and binds a new one, so a second server would silently take a live - server's future connections. A refused connection means nobody listens and the file is stale. + A live socket, a probe that runs out of time against a full backlog, and a path that holds + something other than a socket are none of them stale. """ try: - mode = os.stat(path).st_mode + if not stat.S_ISSOCK(os.stat(path).st_mode): + return False except FileNotFoundError: - return - if not stat.S_ISSOCK(mode): - return + return False with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe: + probe.settimeout(LIVE_SOCKET_PROBE_SEC) try: probe.connect(path) except ConnectionRefusedError: + return True + except OSError: + return False + return False + + @staticmethod + def claim_socket_path(path: str) -> socket.socket: + """Bind and listen on ``path``, and return the socket, or refuse a path something already holds. + + The bind is the claim, so two servers starting together cannot both take one path: the loser's + bind fails. Only then is a probe worth making, and only to tell a stale file from a live server. + The caller hands the descriptor to uvicorn, which binds nothing and so never unlinks the path. + """ + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + try: + sock.bind(path) + except OSError as taken: + if taken.errno != errno.EADDRINUSE: + raise + if not PolicyServer._is_stale_socket(path): + raise OSError(errno.EADDRINUSE, f'{path!r} is already in use') from None os.unlink(path) - return - raise OSError(errno.EADDRINUSE, f'A server already listens on {path!r}') + sock.bind(path) + sock.listen() + os.chmod(path, UDS_MODE) + except BaseException: + sock.close() + raise + return sock def serve(self): async def _run(): await self._startup() - if self.uds is not None: - self.claim_socket_path(self.uds) - config = uvicorn.Config(self.app, host=self.host, port=self.port, uds=self.uds, log_level='info') - server = uvicorn.Server(config) - self._last_activity = time.monotonic() - watchdog = None - if self.idle_timeout_min and self.idle_timeout_min > 0: - watchdog = asyncio.create_task(self._idle_watchdog(server)) + sock, watchdog = None, None try: + if self.uds is not None: + sock = self.claim_socket_path(self.uds) + fd = None if sock is None else sock.fileno() + config = uvicorn.Config(self.app, host=self.host, port=self.port, fd=fd, log_level='info') + server = uvicorn.Server(config) + self._last_activity = time.monotonic() + if self.idle_timeout_min and self.idle_timeout_min > 0: + watchdog = asyncio.create_task(self._idle_watchdog(server)) await server.serve() finally: if watchdog is not None: watchdog.cancel() + if sock is not None: + # The file stays: a successor reads it as stale, where an unlink here could take a + # path that successor has already claimed. + sock.close() try: asyncio.run(_run()) diff --git a/positronic/offboard/tests/conftest.py b/positronic/offboard/tests/conftest.py index 285b5e33a..92e13fb3b 100644 --- a/positronic/offboard/tests/conftest.py +++ b/positronic/offboard/tests/conftest.py @@ -91,18 +91,22 @@ def _dial_unix(path: str) -> None: @pytest.fixture -def start_unix_server(running_servers: RunningServers) -> StartUnixServer: - """Factory serving pipelines on a Unix socket, as ``PolicyServer.serve`` binds one.""" +def start_unix_server(running_servers: RunningServers) -> Generator[StartUnixServer, None, None]: + """Factory serving pipelines on a Unix socket, as ``PolicyServer.serve`` claims one.""" + claimed: list[socket.socket] = [] def start(pipeline, uds: str, **server_kwargs) -> PolicyServer: server = PolicyServer(pipeline, uds=uds, **server_kwargs) - PolicyServer.claim_socket_path(uds) - config = uvicorn.Config(server.app, uds=uds, log_level='warning') + sock = PolicyServer.claim_socket_path(uds) + claimed.append(sock) + config = uvicorn.Config(server.app, fd=sock.fileno(), log_level='warning') _serve_in_background(server, config, running_servers) _wait_until_it_accepts(lambda: _dial_unix(uds)) return server - return start + yield start + for sock in claimed: + sock.close() @pytest.fixture diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 7ada9c727..cbff3bb68 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -2,6 +2,7 @@ import os import pathlib import socket +import stat import time import urllib.parse from collections.abc import Callable, Generator @@ -18,7 +19,7 @@ from positronic.offboard import keys as offboard_keys from positronic.offboard.client import InferenceClient, InferenceSession, _ConnectRetries from positronic.offboard.protocol import deserialise -from positronic.offboard.server import AUTH_HEADER, AUTH_TOKEN_ENV, PolicyServer, bearer +from positronic.offboard.server import AUTH_HEADER, AUTH_TOKEN_ENV, UDS_MODE, PolicyServer, bearer from positronic.offboard.server_utils import warmup from positronic.offboard.tests.conftest import round_trip from positronic.policy import Codec, Policy, RemotePolicy, Session @@ -280,16 +281,43 @@ def test_a_server_binds_over_the_socket_an_earlier_run_left(start_unix_server, s assert InferenceClient(f'unix://{socket_path}').list_models() == ['stub'] -def test_a_path_that_is_not_a_socket_is_left_alone(socket_path): - """A wrong ``uds`` fails the bind rather than deleting a file nobody meant to lose.""" +def test_a_path_that_is_not_a_socket_is_refused_and_left_alone(socket_path): + """A wrong ``uds`` is refused rather than emptied of a file nobody meant to lose.""" path = pathlib.Path(socket_path) path.write_text('not a socket') - PolicyServer.claim_socket_path(socket_path) + with pytest.raises(OSError) as refusal: + PolicyServer.claim_socket_path(socket_path) + assert refusal.value.errno == errno.EADDRINUSE assert path.read_text() == 'not a socket' +@pytest.mark.timeout(30.0) +def test_a_second_claim_on_one_path_is_refused_and_the_first_goes_on_serving(socket_path): + """The bind is the claim, so two servers starting on one absent path cannot both pass it.""" + held = PolicyServer.claim_socket_path(socket_path) + try: + with pytest.raises(OSError) as refusal: + PolicyServer.claim_socket_path(socket_path) + assert refusal.value.errno == errno.EADDRINUSE + assert socket_path in str(refusal.value) + + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.connect(socket_path) + assert held.accept()[0].close() is None + finally: + held.close() + + +def test_a_claimed_socket_carries_the_mode_uvicorn_gives_one(socket_path): + sock = PolicyServer.claim_socket_path(socket_path) + try: + assert stat.S_IMODE(os.stat(socket_path).st_mode) == UDS_MODE + finally: + sock.close() + + @pytest.mark.timeout(30.0) def test_a_server_refuses_a_socket_a_live_server_listens_on(socket_path, make_mock_policy): """``asyncio.create_unix_server`` unlinks the file it finds, so only a refusal here keeps the From 58a3ebdb4b05f0b670dfc70553dcd07ded72785a Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 21:33:38 +0000 Subject: [PATCH 07/15] Open two claim comments on their subject Both described the probe by leading with the reason for it. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/server.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index 47bc3ee34..5f3001485 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -35,8 +35,8 @@ AUTH_HEADER = 'Authorization' -# A server whose backlog is full holds a connect open, so the probe for a live socket bounds its wait -# rather than stalling startup on it. A wait that runs out counts as live. +# The probe for a live socket bounds its wait, and reads a wait that runs out as a live server: a +# server whose backlog is full holds a connect open, and an unbounded one would stall startup. LIVE_SOCKET_PROBE_SEC = 1.0 # The mode uvicorn gives a Unix socket it binds itself. @@ -452,8 +452,8 @@ def claim_socket_path(path: str) -> socket.socket: """Bind and listen on ``path``, and return the socket, or refuse a path something already holds. The bind is the claim, so two servers starting together cannot both take one path: the loser's - bind fails. Only then is a probe worth making, and only to tell a stale file from a live server. - The caller hands the descriptor to uvicorn, which binds nothing and so never unlinks the path. + bind fails. A probe follows it only to tell a stale file from a live server. The caller hands the + descriptor to uvicorn, which binds nothing and so never unlinks the path. """ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: From 7ec31e88b1259f30434f3a1a2f2cda27b8df375b Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 21:49:08 +0000 Subject: [PATCH 08/15] Decode the socket path a `unix://` URL names `urlsplit` leaves a path percent-encoded, and the socket path was dialled as it came, so a directory holding a space or a percent named a file that is not there. The socket path is decoded once, after the split. The split still runs over the encoded path, so an escaped separator stays inside a directory name rather than becoming one. The URL path keeps its escapes: the server decodes those, which is what lets a model id carry its own. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/client.py | 11 +++++++--- .../offboard/tests/test_remote_policy.py | 12 +++++++++++ positronic/offboard/tests/test_server.py | 20 +++++++++++++++++++ 3 files changed, 40 insertions(+), 3 deletions(-) diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 49a0de930..f2dcf74ff 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -120,13 +120,18 @@ def _session_path(path: str, url: str) -> str: def _socket_and_path(split: urllib.parse.SplitResult, url: str) -> tuple[str, str]: - """The socket path a ``unix://`` URL names, and the URL path left over for the server.""" + """The socket path a ``unix://`` URL names, decoded, and the URL path left over for the server. + + The split runs over the encoded path, so an escaped separator inside a directory name stays part + of that name. Only the socket path is decoded, and once: it names a file, where the URL path + reaches the server as written, which is what lets a model id carry its own escapes. + """ if split.netloc or not split.path.startswith('/'): raise ValueError(f'Socket path must be absolute in {url!r}; write unix:///path/to.sock') marker = re.search(r'/api/v1(?=/|$)', split.path) if marker is None: - return split.path, '' - return split.path[: marker.start()], split.path[marker.start() :] + return urllib.parse.unquote(split.path), '' + return urllib.parse.unquote(split.path[: marker.start()]), split.path[marker.start() :] class _ConnectOutcome(Enum): diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py index 6da232cbc..8a15bb64c 100644 --- a/positronic/offboard/tests/test_remote_policy.py +++ b/positronic/offboard/tests/test_remote_policy.py @@ -231,6 +231,18 @@ def test_the_api_marker_is_a_whole_segment(self): assert client.uds == '/run/api/v1x/policy.sock' assert client.session_url == 'unix:///run/api/v1x/policy.sock/api/v1/session' + def test_the_socket_path_is_decoded_and_the_session_path_is_not(self): + """The socket path names a file, so its escapes are resolved; the model id reaches the server + as written, which is how an id carrying its own escapes survives.""" + client = InferenceClient('unix:///run/a%20b%25c/policy.sock/api/v1/session/s3%3A//ckpt') + assert client.uds == '/run/a b%c/policy.sock' + assert client.session_url == 'unix:///run/a b%c/policy.sock/api/v1/session/s3%3A//ckpt' + + def test_an_escaped_separator_stays_inside_a_directory_name(self): + """The split runs before the decode, so %2F never becomes a path separator.""" + client = InferenceClient('unix:///run/odd%2Fname/policy.sock') + assert client.uds == '/run/odd/name/policy.sock' + def test_a_relative_socket_path_rejected(self): with pytest.raises(ValueError, match='absolute'): InferenceClient('unix://policy.sock') diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index cbff3bb68..fa20db1e6 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -271,6 +271,26 @@ def test_a_unix_url_carries_the_model_id_past_the_socket_path(unix_stub_server): session.close() +def test_a_socket_path_carrying_url_escapes_is_dialled_as_a_filename(start_unix_server, socket_path, make_mock_policy): + """``urlsplit`` leaves the path encoded, so a directory holding a space or a percent would + otherwise be dialled as a file that does not exist.""" + odd = pathlib.Path(socket_path).parent / 'a b%c' + odd.mkdir() + uds = str(odd / 's.sock') + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + start_unix_server(ChunkedSchedule() | remote | _StubSource(policy), uds) + + client = InferenceClient(f'unix://{urllib.parse.quote(uds)}') + + assert client.uds == uds + assert client.list_models() == ['stub'] + session = client.new_session() + try: + assert session.infer({'obs': 'data'}) == [{'action': [1, 2, 3]}] + finally: + session.close() + + def test_a_server_binds_over_the_socket_an_earlier_run_left(start_unix_server, socket_path, make_mock_policy): policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as stale: From f09910be4905351a22dacb6162995f32a55ec57c Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 22:03:52 +0000 Subject: [PATCH 09/15] Leave a claimed socket's mode to the deployment's umask The claim chmod'd every socket to 0o666, which reads as the mode uvicorn gives one and is not: uvicorn's own bind takes the umask, and it copies an existing file's mode where there is one. So a deployment with a restrictive umask had its socket widened, and an open server behind it became reachable by any local account that can walk the directory. The bind's own mode stands. Two documents said something that is not so. The socket-path docstring claimed an escaped separator stays inside a directory name; the decode resolves every escape, so `%2F` becomes a separator and a socket path cannot hold a name carrying a slash. It says that now, and the test that asserted it is named for it. The README's serve inventory listed four flags and now lists `--uds` as well. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/README.md | 2 +- positronic/offboard/client.py | 8 +++++--- positronic/offboard/server.py | 6 ++---- positronic/offboard/tests/test_remote_policy.py | 4 ++-- positronic/offboard/tests/test_server.py | 14 ++++++++++---- 5 files changed, 20 insertions(+), 14 deletions(-) diff --git a/positronic/offboard/README.md b/positronic/offboard/README.md index 3c37f4d32..ba96c5753 100644 --- a/positronic/offboard/README.md +++ b/positronic/offboard/README.md @@ -227,7 +227,7 @@ PolicyServer(pipeline, host='0.0.0.0', port=8000).serve() `PolicySource` serves one ready in-process policy; vendors instead define a `ModelSource` over a checkpoint directory. Passing a `cfn.Config` that builds the pipeline — as the vendor servers do with their named pipelines — enables [session parameters](#session-parameters); an instantiated pipeline serves exactly as launched. `recording_dir` enables the per-session recording taps described above, and `idle_timeout_min` shuts the server down after that many minutes without activity. ### `server.serve` -The CLI entry point every vendor server exposes. A vendor binds `pipeline` to each of its named pipelines and lists the results as subcommands, so `-server ` launches one. Only `--host`, `--port`, `--recording_dir` and `--idle_timeout_min` are flags of `serve` itself; everything the served model is — codec, source, checkpoint directory — is reached through the pipeline (`--pipeline.source.checkpoints_dir=...`), which is also where a deployment preset binds it. +The CLI entry point every vendor server exposes. A vendor binds `pipeline` to each of its named pipelines and lists the results as subcommands, so `-server ` launches one. Only `--host`, `--port`, `--uds`, `--recording_dir` and `--idle_timeout_min` are flags of `serve` itself (`--uds` binds a Unix socket path in place of `--host`/`--port`); everything the served model is — codec, source, checkpoint directory — is reached through the pipeline (`--pipeline.source.checkpoints_dir=...`), which is also where a deployment preset binds it. ### `client.InferenceClient` A Python client for connecting to an inference server. One URL addresses it, in the same forms diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index f2dcf74ff..93aae5f5c 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -122,9 +122,11 @@ def _session_path(path: str, url: str) -> str: def _socket_and_path(split: urllib.parse.SplitResult, url: str) -> tuple[str, str]: """The socket path a ``unix://`` URL names, decoded, and the URL path left over for the server. - The split runs over the encoded path, so an escaped separator inside a directory name stays part - of that name. Only the socket path is decoded, and once: it names a file, where the URL path - reaches the server as written, which is what lets a model id carry its own escapes. + The split runs over the encoded path, so an escaped ``/api/v1`` cannot be read as the marker. + Decoding follows, and it resolves every escape: ``%2F`` becomes a separator like any other, so a + socket path cannot hold a directory whose own name carries a slash. Only the socket path is + decoded, because it names a file; the URL path reaches the server as written, which is what lets + a model id carry its own escapes. """ if split.netloc or not split.path.startswith('/'): raise ValueError(f'Socket path must be absolute in {url!r}; write unix:///path/to.sock') diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index 5f3001485..25a49b089 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -39,9 +39,6 @@ # server whose backlog is full holds a connect open, and an unbounded one would stall startup. LIVE_SOCKET_PROBE_SEC = 1.0 -# The mode uvicorn gives a Unix socket it binds itself. -UDS_MODE = 0o666 - def bearer(token: str) -> str: """The ``AUTH_HEADER`` value carrying ``token``.""" @@ -466,8 +463,9 @@ def claim_socket_path(path: str) -> socket.socket: raise OSError(errno.EADDRINUSE, f'{path!r} is already in use') from None os.unlink(path) sock.bind(path) + # The mode is the deployment's, through its umask: widening it here would open the socket + # to every local account that can reach the directory. sock.listen() - os.chmod(path, UDS_MODE) except BaseException: sock.close() raise diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py index 8a15bb64c..d33bcd214 100644 --- a/positronic/offboard/tests/test_remote_policy.py +++ b/positronic/offboard/tests/test_remote_policy.py @@ -238,8 +238,8 @@ def test_the_socket_path_is_decoded_and_the_session_path_is_not(self): assert client.uds == '/run/a b%c/policy.sock' assert client.session_url == 'unix:///run/a b%c/policy.sock/api/v1/session/s3%3A//ckpt' - def test_an_escaped_separator_stays_inside_a_directory_name(self): - """The split runs before the decode, so %2F never becomes a path separator.""" + def test_an_escaped_separator_is_a_separator_once_decoded(self): + """The decode resolves every escape, so no socket path can hold a slash inside one name.""" client = InferenceClient('unix:///run/odd%2Fname/policy.sock') assert client.uds == '/run/odd/name/policy.sock' diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index fa20db1e6..37051ad43 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -19,7 +19,7 @@ from positronic.offboard import keys as offboard_keys from positronic.offboard.client import InferenceClient, InferenceSession, _ConnectRetries from positronic.offboard.protocol import deserialise -from positronic.offboard.server import AUTH_HEADER, AUTH_TOKEN_ENV, UDS_MODE, PolicyServer, bearer +from positronic.offboard.server import AUTH_HEADER, AUTH_TOKEN_ENV, PolicyServer, bearer from positronic.offboard.server_utils import warmup from positronic.offboard.tests.conftest import round_trip from positronic.policy import Codec, Policy, RemotePolicy, Session @@ -330,10 +330,16 @@ def test_a_second_claim_on_one_path_is_refused_and_the_first_goes_on_serving(soc held.close() -def test_a_claimed_socket_carries_the_mode_uvicorn_gives_one(socket_path): - sock = PolicyServer.claim_socket_path(socket_path) +def test_a_claimed_socket_keeps_the_mode_the_umask_gives(socket_path): + """A restrictive umask is the deployment's choice, and widening it would open the socket to every + local account that can reach the directory.""" + previous = os.umask(0o077) try: - assert stat.S_IMODE(os.stat(socket_path).st_mode) == UDS_MODE + sock = PolicyServer.claim_socket_path(socket_path) + finally: + os.umask(previous) + try: + assert stat.S_IMODE(os.stat(socket_path).st_mode) & 0o077 == 0 finally: sock.close() From 0a17994fe3e11bceb45eabddcc477b15b0b932c8 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 22:18:33 +0000 Subject: [PATCH 10/15] Wait for a Unix socket a co-located server has not bound yet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `serve` binds only once the model has loaded, so the path is absent for that whole interval. The dial raised `FileNotFoundError` on it at once, which is the state a co-located server is in for the seconds after its unit starts, so the 900-second connect budget bought that deployment nothing. An absent path now rides the deadline, and so does a socket that refuses, which is a server restarting. A refusal from anything else at the path is a wrong path and still surfaces at once, as a permission error always did: the two share `ECONNREFUSED`, so the socket itself is what tells them apart. `_ConnectRetries` holds the whole retry policy now — the 403 budget it already had, plus the deadline and the backoff `new_session` was carrying in locals — so both handlers spend an attempt through one call. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/client.py | 52 ++++++++++++++++++------ positronic/offboard/tests/test_server.py | 29 +++++++++++++ 2 files changed, 69 insertions(+), 12 deletions(-) diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 93aae5f5c..b20ea792b 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -1,6 +1,8 @@ import logging +import os import re import ssl +import stat import time import urllib.parse from enum import Enum @@ -150,8 +152,21 @@ class _ConnectRetries: MAX_FORBIDDEN_ATTEMPTS = 3 - def __init__(self) -> None: + def __init__(self, connect_deadline: float, url: str) -> None: self._forbidden_attempts = 0 + self._deadline = time.monotonic() + connect_deadline + self._backoff = 1.0 + self._url = url + + def wait_or_surface(self, e: Exception) -> None: + """Spend one refused connect against the budget, or let it surface. Call it from the handler.""" + if self.take(e) is _ConnectOutcome.SURFACE: + raise + if time.monotonic() >= self._deadline: + raise TimeoutError(f'{e} (connecting to {self._url})') from e + logger.info('Server not ready (cold start?): %s; retrying in %.0fs', e, self._backoff) + time.sleep(self._backoff) + self._backoff = min(self._backoff * 2, 30.0) def take(self, e: Exception) -> _ConnectOutcome: """Spend a refused connect against the budget.""" @@ -233,11 +248,26 @@ def __init__( self.connect_deadline = connect_deadline self.infer_timeout = infer_timeout + def _socket_may_still_appear(self, e: OSError) -> bool: + """Whether a failed dial is a co-located server that has not bound its socket yet. + + Its ``serve`` binds only once the model has loaded, so the path is absent for that whole + interval, and a socket that refuses is one restarting. A refusal from anything else at the + path is a wrong path, and no waiting clears it. + """ + assert self.uds is not None + if isinstance(e, FileNotFoundError): + return True + if not isinstance(e, ConnectionRefusedError): + return False + try: + return stat.S_ISSOCK(os.stat(self.uds).st_mode) + except FileNotFoundError: + return True + def new_session(self) -> InferenceSession: """Creates a new inference session on the model the URL names.""" - deadline = time.monotonic() + self.connect_deadline - backoff = 1.0 - retries = _ConnectRetries() + retries = _ConnectRetries(self.connect_deadline, self.session_url) while True: ws = None try: @@ -263,15 +293,13 @@ def new_session(self) -> InferenceSession: except (TimeoutError, ssl.SSLError, ConnectionClosed, InvalidHandshake) as e: if ws is not None: ws.close() - if retries.take(e) is _ConnectOutcome.SURFACE: - raise - if time.monotonic() >= deadline: - raise TimeoutError(f'{e} (connecting to {self.session_url})') from e - logger.info('Server not ready (cold start?): %s; retrying in %.0fs', e, backoff) - time.sleep(backoff) - backoff = min(backoff * 2, 30.0) + retries.wait_or_surface(e) except OSError as e: - raise type(e)(f'{e} (connecting to {self.session_url})') from e + if ws is not None: + ws.close() + if self.uds is None or not self._socket_may_still_appear(e): + raise type(e)(f'{e} (connecting to {self.session_url})') from e + retries.wait_or_surface(e) def list_models(self) -> list[str]: """List available models from the server.""" diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 37051ad43..4af51fc26 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -3,6 +3,7 @@ import pathlib import socket import stat +import threading import time import urllib.parse from collections.abc import Callable, Generator @@ -291,6 +292,34 @@ def test_a_socket_path_carrying_url_escapes_is_dialled_as_a_filename(start_unix_ session.close() +@pytest.mark.timeout(60.0) +def test_a_client_waits_for_a_socket_the_server_has_not_bound_yet(start_unix_server, socket_path, make_mock_policy): + """``serve`` binds only once the model has loaded, so a co-located client starting beside its + server finds no socket at all for that interval.""" + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + pipeline = ChunkedSchedule() | remote | _StubSource(policy) + late = threading.Timer(1.5, lambda: start_unix_server(pipeline, socket_path)) + late.start() + + try: + session = InferenceClient(f'unix://{socket_path}', connect_deadline=30.0).new_session() + finally: + late.join() + try: + assert session.metadata['model_name'] == 'stub' + assert session.infer({'obs': 'data'}) == [{'action': [1, 2, 3]}] + finally: + session.close() + + +def test_a_dial_at_a_path_holding_something_that_is_not_a_socket_fails_at_once(socket_path): + """A wrong path refuses like a restarting server does, and no waiting clears it.""" + pathlib.Path(socket_path).write_text('not a socket') + + with pytest.raises(ConnectionRefusedError): + InferenceClient(f'unix://{socket_path}', connect_deadline=30.0).new_session() + + def test_a_server_binds_over_the_socket_an_earlier_run_left(start_unix_server, socket_path, make_mock_policy): policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as stale: From 6b8e1ae0a65137ddd6e3886df5ccac5c245c022d Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 22:34:10 +0000 Subject: [PATCH 11/15] Read the socket path, not the errno, to tell a cold server from a wrong path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit macOS answers a dial at a path holding a regular file with ENOTSOCK, where Linux answers ECONNREFUSED. The check keyed on the exception, so the two platforms classified one state differently, and the test that pinned the Linux errno went red on macOS. The path answers it on both: absent or a socket means a server may still bind there, anything else at the path does not. A permission error stays settled wherever it comes from, since this client cannot read past it to find out. The test asserts what the code promises — a wrong path does not spend the connect deadline — rather than the errno that carries it. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/client.py | 11 ++++++----- positronic/offboard/tests/test_server.py | 8 ++++++-- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index b20ea792b..229407915 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -252,18 +252,19 @@ def _socket_may_still_appear(self, e: OSError) -> bool: """Whether a failed dial is a co-located server that has not bound its socket yet. Its ``serve`` binds only once the model has loaded, so the path is absent for that whole - interval, and a socket that refuses is one restarting. A refusal from anything else at the - path is a wrong path, and no waiting clears it. + interval, and a socket that refuses is one restarting. The path decides that, not the error: + a missing socket and a wrong path report different errno on Linux and on macOS. A refusal + this client has no permission to read past is settled wherever it comes from. """ assert self.uds is not None - if isinstance(e, FileNotFoundError): - return True - if not isinstance(e, ConnectionRefusedError): + if isinstance(e, PermissionError): return False try: return stat.S_ISSOCK(os.stat(self.uds).st_mode) except FileNotFoundError: return True + except OSError: + return False def new_session(self) -> InferenceSession: """Creates a new inference session on the model the URL names.""" diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 4af51fc26..fc76a5963 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -313,12 +313,16 @@ def test_a_client_waits_for_a_socket_the_server_has_not_bound_yet(start_unix_ser def test_a_dial_at_a_path_holding_something_that_is_not_a_socket_fails_at_once(socket_path): - """A wrong path refuses like a restarting server does, and no waiting clears it.""" + """No waiting clears a wrong path. Which errno says so differs by platform, so the deadline it + must not spend is what this asserts.""" pathlib.Path(socket_path).write_text('not a socket') + started = time.monotonic() - with pytest.raises(ConnectionRefusedError): + with pytest.raises(OSError): InferenceClient(f'unix://{socket_path}', connect_deadline=30.0).new_session() + assert time.monotonic() - started < 5.0 + def test_a_server_binds_over_the_socket_an_earlier_run_left(start_unix_server, socket_path, make_mock_policy): policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) From 5312288a4c677e71493b45d3c09542d061ed6c7c Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Thu, 10 Sep 2026 22:36:31 +0000 Subject: [PATCH 12/15] Say three socket-path clauses plainly Two were clefts and one told the reader how to take the fact before it. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/client.py | 10 +++++----- positronic/offboard/tests/test_server.py | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 229407915..98779f906 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -127,8 +127,8 @@ def _socket_and_path(split: urllib.parse.SplitResult, url: str) -> tuple[str, st The split runs over the encoded path, so an escaped ``/api/v1`` cannot be read as the marker. Decoding follows, and it resolves every escape: ``%2F`` becomes a separator like any other, so a socket path cannot hold a directory whose own name carries a slash. Only the socket path is - decoded, because it names a file; the URL path reaches the server as written, which is what lets - a model id carry its own escapes. + decoded, because it names a file; the URL path reaches the server as written, so a model id + carries its own escapes. """ if split.netloc or not split.path.startswith('/'): raise ValueError(f'Socket path must be absolute in {url!r}; write unix:///path/to.sock') @@ -252,9 +252,9 @@ def _socket_may_still_appear(self, e: OSError) -> bool: """Whether a failed dial is a co-located server that has not bound its socket yet. Its ``serve`` binds only once the model has loaded, so the path is absent for that whole - interval, and a socket that refuses is one restarting. The path decides that, not the error: - a missing socket and a wrong path report different errno on Linux and on macOS. A refusal - this client has no permission to read past is settled wherever it comes from. + interval, and a socket that refuses is one restarting. The path answers it, because a missing + socket and a wrong path report different errno on Linux and on macOS. A refusal this client + has no permission to read past is settled wherever it comes from. """ assert self.uds is not None if isinstance(e, PermissionError): diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index fc76a5963..3736bd1a0 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -313,8 +313,8 @@ def test_a_client_waits_for_a_socket_the_server_has_not_bound_yet(start_unix_ser def test_a_dial_at_a_path_holding_something_that_is_not_a_socket_fails_at_once(socket_path): - """No waiting clears a wrong path. Which errno says so differs by platform, so the deadline it - must not spend is what this asserts.""" + """No waiting clears a wrong path. Which errno says so differs by platform, so this asserts the + connect deadline goes unspent.""" pathlib.Path(socket_path).write_text('not a socket') started = time.monotonic() From bcd2b36f07479077806eb93089eb118082af9085 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Fri, 11 Sep 2026 07:00:57 +0000 Subject: [PATCH 13/15] Record a socket path under its own key, and wait only for the two dials that clear MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The handshake metadata put a socket path under `host`, which labels a filesystem path as a network address wherever an episode's metadata is read. It gets `uds`, and a server reports one pair or the other. The client waited out the whole connect deadline on any `OSError` the path answered as a socket, including one this process raised itself — a descriptor limit, a memory limit. Only an absent path and a refusal can mean a co-located server that has not bound yet, so every other dial surfaces at once. Every document that lists the server's flags now names `--uds`: the offboard README, the five vendor READMEs, and the training workflow. The enumeration is complete. Ticket: Positronic-Robotics/internal#1242 #refs --- docs/training-workflow.md | 1 + positronic/offboard/client.py | 10 +++-- positronic/offboard/keys.py | 2 + positronic/offboard/server.py | 4 +- positronic/offboard/tests/test_server.py | 43 +++++++++++++++++++++- positronic/vendors/gr00t/README.md | 1 + positronic/vendors/lerobot/README.md | 1 + positronic/vendors/lerobot_0_3_3/README.md | 1 + positronic/vendors/molmoact2/README.md | 5 ++- positronic/vendors/openpi/README.md | 1 + 10 files changed, 59 insertions(+), 10 deletions(-) diff --git a/docs/training-workflow.md b/docs/training-workflow.md index 62752f7dc..04230a267 100644 --- a/docs/training-workflow.md +++ b/docs/training-workflow.md @@ -219,6 +219,7 @@ cd docker && docker compose run --rm --service-ports openpi-server ee \ | `--pipeline.ee_frame` | OpenPI only: the EE frame the checkpoint speaks, relative to the rig's `default` | `None` | | `--port` | Server port | `8000` (default) | | `--host` | Server host | `0.0.0.0` (default, binds to all interfaces) | +| `--uds` | Unix socket path to bind in place of `--host`/`--port`, for a client on the same machine | `/run/policy.sock` | The subcommand picks the pipeline and `--pipeline.` reaches anywhere inside it, so every value the served model is built from has exactly one name. The same paths are the per-session query params on the client's `--policy.url` (see the [Inference Guide](inference.md)), except `source.*`, which is fixed at launch. diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 98779f906..7d26296db 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -252,12 +252,14 @@ def _socket_may_still_appear(self, e: OSError) -> bool: """Whether a failed dial is a co-located server that has not bound its socket yet. Its ``serve`` binds only once the model has loaded, so the path is absent for that whole - interval, and a socket that refuses is one restarting. The path answers it, because a missing - socket and a wrong path report different errno on Linux and on macOS. A refusal this client - has no permission to read past is settled wherever it comes from. + interval, and a socket that refuses is one restarting. Only an absent path and a refusal say + that; every other dial — a permission the path denies, a descriptor or a memory limit this + process has hit — is settled, and waiting for it spends the whole deadline on an answer that + will not change. A refusal then reads the path, which tells a restarting server from a path + naming something that is not a socket. """ assert self.uds is not None - if isinstance(e, PermissionError): + if not isinstance(e, (FileNotFoundError, ConnectionRefusedError)): return False try: return stat.S_ISSOCK(os.stat(self.uds).st_mode) diff --git a/positronic/offboard/keys.py b/positronic/offboard/keys.py index a9af0a26f..137cd8a99 100644 --- a/positronic/offboard/keys.py +++ b/positronic/offboard/keys.py @@ -4,6 +4,8 @@ # what the rig builds and obeys — the local stack spec, image compression, the positronic version it runs. HOST = 'host' PORT = 'port' +# The socket path a server bound instead of a host and a port. One of the two pairs is present, never both. +UDS = 'uds' CHECKPOINT_ID = 'checkpoint_id' LOCAL_STACK = 'local_stack' COMPRESS_IMAGES = 'compress_images' diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index 25a49b089..d7cec2e1c 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -238,9 +238,9 @@ def __init__( self.host = host self.port = port self.uds = uds - # Where the server listens, as the handshake reports it. A Unix socket has no port. + # Where the server listens, as the handshake reports it. A socket path is not a host, so it has its own key. self.metadata: dict[str, Any] = ( - {offboard_keys.HOST: host, offboard_keys.PORT: port} if uds is None else {offboard_keys.HOST: uds} + {offboard_keys.HOST: host, offboard_keys.PORT: port} if uds is None else {offboard_keys.UDS: uds} ) # Synced once; each session builds its own ``Recorder`` so concurrent streams never mix. self._recording_dir = pos3.sync(recording_dir) if recording_dir else None diff --git a/positronic/offboard/tests/test_server.py b/positronic/offboard/tests/test_server.py index 3736bd1a0..27ce2f44a 100644 --- a/positronic/offboard/tests/test_server.py +++ b/positronic/offboard/tests/test_server.py @@ -8,7 +8,7 @@ import urllib.parse from collections.abc import Callable, Generator from typing import Any -from unittest.mock import ANY, MagicMock +from unittest.mock import ANY, MagicMock, patch import configuronic as cfn import httpx @@ -252,7 +252,8 @@ def test_a_pipeline_served_over_a_unix_socket(unix_stub_server): try: assert session.metadata['model_name'] == 'stub' assert session.metadata[offboard_keys.LOCAL_STACK] == {'name': 'chunked_schedule'} - assert session.metadata[offboard_keys.HOST] == socket_path + assert session.metadata[offboard_keys.UDS] == socket_path + assert offboard_keys.HOST not in session.metadata assert offboard_keys.PORT not in session.metadata obs = {'image': 'test'} @@ -312,6 +313,44 @@ def test_a_client_waits_for_a_socket_the_server_has_not_bound_yet(start_unix_ser session.close() +@pytest.mark.timeout(60.0) +def test_a_client_waits_for_a_server_restarting_over_the_socket_it_left( + start_unix_server, socket_path, make_mock_policy +): + """A bound path whose server has gone refuses the dial, and the successor binds over it. The wait + covers that restart as it covers a first start.""" + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as gone: + gone.bind(socket_path) + policy = make_mock_policy([{'action': [1, 2, 3]}], {'model_name': 'stub'}) + pipeline = ChunkedSchedule() | remote | _StubSource(policy) + late = threading.Timer(1.5, lambda: start_unix_server(pipeline, socket_path)) + late.start() + + try: + session = InferenceClient(f'unix://{socket_path}', connect_deadline=30.0).new_session() + finally: + late.join() + try: + assert session.infer({'obs': 'data'}) == [{'action': [1, 2, 3]}] + finally: + session.close() + + +def test_a_dial_this_process_broke_fails_at_once_over_a_live_socket(unix_stub_server): + """A descriptor limit is this process's own, so no server appearing clears it. The socket is live + and the path says so, which is exactly when reading the path alone would wait out the deadline.""" + socket_path, _policy = unix_stub_server + started = time.monotonic() + + with patch('positronic.offboard.client.unix_connect') as dial: + dial.side_effect = OSError(errno.EMFILE, 'Too many open files') + with pytest.raises(OSError) as refusal: + InferenceClient(f'unix://{socket_path}', connect_deadline=30.0).new_session() + + assert 'Too many open files' in str(refusal.value) + assert time.monotonic() - started < 5.0 + + def test_a_dial_at_a_path_holding_something_that_is_not_a_socket_fails_at_once(socket_path): """No waiting clears a wrong path. Which errno says so differs by platform, so this asserts the connect deadline goes unspent.""" diff --git a/positronic/vendors/gr00t/README.md b/positronic/vendors/gr00t/README.md index 8276a81d7..8729b3e31 100644 --- a/positronic/vendors/gr00t/README.md +++ b/positronic/vendors/gr00t/README.md @@ -128,6 +128,7 @@ cd docker && docker compose run --rm --service-ports groot-server ee_rot6d_joint | `--pipeline.source.checkpoints_dir` | Experiment directory (contains `checkpoint-N` folders) | Required | `~/checkpoints/groot/my_task_v1/` | | `--pipeline.source.checkpoint` | Specific checkpoint ID | Latest | `10000`, `50000` | | `--port` | Server port | `8000` | `8001` | +| `--uds` | Unix socket path to bind in place of `--host`/`--port` | `None` | `/run/policy.sock` | | `--pipeline.source.modality_config` | Override the pipeline's paired modality config | Paired | `ee_rot6d_q` | **Session parameters:** a client can tune the served pipeline per connection via query params on the diff --git a/positronic/vendors/lerobot/README.md b/positronic/vendors/lerobot/README.md index 4fbee6ae0..0bf6ad5a6 100644 --- a/positronic/vendors/lerobot/README.md +++ b/positronic/vendors/lerobot/README.md @@ -109,6 +109,7 @@ cd docker && docker compose run --rm --service-ports lerobot-server ee \ | `--pipeline.source.device` | Torch device the policy runs on | Auto-detected | `cuda`, `mps`, `cpu` | | `--port` | Server port | `8000` | `8001` | | `--host` | Server host | `0.0.0.0` | Binds to all interfaces | +| `--uds` | Unix socket path to bind in place of `--host`/`--port` | `None` | `/run/policy.sock` | | `--recording_dir` | Directory for server-side inference recordings | `None` | `s3://inference/...` | | `--idle_timeout_min` | Shut down after this many idle minutes | `None` | `30` | diff --git a/positronic/vendors/lerobot_0_3_3/README.md b/positronic/vendors/lerobot_0_3_3/README.md index 79dafee7f..91847d9f2 100644 --- a/positronic/vendors/lerobot_0_3_3/README.md +++ b/positronic/vendors/lerobot_0_3_3/README.md @@ -106,6 +106,7 @@ cd docker && docker compose run --rm --service-ports lerobot-0_3_3-server ee \ | `--pipeline.source.model_type` | Names what the factory builds, for the handshake metadata | `act` | `diffusion` | | `--port` | Server port | `8000` | `8001` | | `--host` | Server host | `0.0.0.0` | Binds to all interfaces | +| `--uds` | Unix socket path to bind in place of `--host`/`--port` | `None` | `/run/policy.sock` | | `--recording_dir` | Directory for server-side inference recordings | `None` | `s3://inference/...` | | `--idle_timeout_min` | Shut down after this many idle minutes | `None` | `30` | diff --git a/positronic/vendors/molmoact2/README.md b/positronic/vendors/molmoact2/README.md index 8835be723..640d2a971 100644 --- a/positronic/vendors/molmoact2/README.md +++ b/positronic/vendors/molmoact2/README.md @@ -32,8 +32,9 @@ uv run --python 3.13 --extra molmoact2 python -m positronic.vendors.molmoact2.se The server serves a named policy pipeline — the codec plus the HuggingFace model source. MolmoAct2 ships one pipeline, `droid`, which is the default subcommand. The codec lives server-side, so clients send raw -observations and receive decoded joint commands. `--host`, `--port`, `--recording_dir` and -`--idle_timeout_min` are the server's flags; the model is reached through the pipeline +observations and receive decoded joint commands. `--host`, `--port`, `--uds`, `--recording_dir` and +`--idle_timeout_min` are the server's flags, and `--uds` binds a Unix socket path in place of +`--host`/`--port`; the model is reached through the pipeline (`--pipeline.source.hf_repo`, `.device_map`, `.norm_tag`, `.num_steps`), with defaults in [`server.py`](./server.py). Sanity-check once warm: diff --git a/positronic/vendors/openpi/README.md b/positronic/vendors/openpi/README.md index 32b0c55ce..64c7f3de5 100644 --- a/positronic/vendors/openpi/README.md +++ b/positronic/vendors/openpi/README.md @@ -140,6 +140,7 @@ emits absolute `JointPosition` chunks executed at RoboLab's leaderboard cadence - `--pipeline.source.checkpoint`: (Optional) Specific checkpoint step to load. If omitted, loads the latest checkpoint - `--pipeline.source.config_name`: (Optional) OpenPI config name; overrides the pipeline's pairing (base pipelines use `pi05_positronic_lowmem`) - `--port`: (Optional) Port to serve on (default: 8000) +- `--uds`: (Optional) Unix socket path to bind in place of `--host`/`--port`, for a client on the same machine - `--pipeline.source.openpi_ws_port`: (Optional) Internal port for OpenPI subprocess (default: 8001) - `--recording_dir`: (Optional) Directory for server-side `.rrd` recordings (local or S3) - `--idle_timeout_min`: (Optional) Shut down after this many minutes without activity From afc9af26a0bfabe87b0062cc15359291c38258ed Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Fri, 11 Sep 2026 07:11:06 +0000 Subject: [PATCH 14/15] Say the retry rule in two sentences Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/client.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py index 7d26296db..fd50330c8 100644 --- a/positronic/offboard/client.py +++ b/positronic/offboard/client.py @@ -251,12 +251,10 @@ def __init__( def _socket_may_still_appear(self, e: OSError) -> bool: """Whether a failed dial is a co-located server that has not bound its socket yet. - Its ``serve`` binds only once the model has loaded, so the path is absent for that whole - interval, and a socket that refuses is one restarting. Only an absent path and a refusal say - that; every other dial — a permission the path denies, a descriptor or a memory limit this - process has hit — is settled, and waiting for it spends the whole deadline on an answer that - will not change. A refusal then reads the path, which tells a restarting server from a path - naming something that is not a socket. + Only an absent path and a refusal can mean that; every other ``OSError`` is settled, and + waiting for it spends the whole deadline on an answer that will not change. A refusal then + reads the path, which tells a restarting server from a path naming something that is not a + socket. """ assert self.uds is not None if not isinstance(e, (FileNotFoundError, ConnectionRefusedError)): From 760c35907a6891a538cc9c0cd09e140cf065190b Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Fri, 11 Sep 2026 07:22:55 +0000 Subject: [PATCH 15/15] Keep the probe budget with its user, and the docstring off its caller `LIVE_SOCKET_PROBE_SEC` sat at the top of the file with its one user 400 lines below. It is `PolicyServer`'s now, above the method that spends it. `claim_socket_path`'s docstring described what its caller hands to uvicorn, which nothing keeps true. It states the constraint on the returned socket instead. Ticket: Positronic-Robotics/internal#1242 #refs --- positronic/offboard/server.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/positronic/offboard/server.py b/positronic/offboard/server.py index d7cec2e1c..1f15bbef9 100644 --- a/positronic/offboard/server.py +++ b/positronic/offboard/server.py @@ -35,10 +35,6 @@ AUTH_HEADER = 'Authorization' -# The probe for a live socket bounds its wait, and reads a wait that runs out as a live server: a -# server whose backlog is full holds a connect open, and an unbounded one would stall startup. -LIVE_SOCKET_PROBE_SEC = 1.0 - def bearer(token: str) -> str: """The ``AUTH_HEADER`` value carrying ``token``.""" @@ -422,6 +418,10 @@ async def _idle_watchdog(self, server: uvicorn.Server): server.should_exit = True return + # The probe bounds its wait, and reads a wait that runs out as a live server: a server whose + # backlog is full holds a connect open, and an unbounded one would stall startup. + LIVE_SOCKET_PROBE_SEC = 1.0 + @staticmethod def _is_stale_socket(path: str) -> bool: """Whether ``path`` is a socket no server answers on, so replacing it takes nothing from anybody. @@ -435,7 +435,7 @@ def _is_stale_socket(path: str) -> bool: except FileNotFoundError: return False with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe: - probe.settimeout(LIVE_SOCKET_PROBE_SEC) + probe.settimeout(PolicyServer.LIVE_SOCKET_PROBE_SEC) try: probe.connect(path) except ConnectionRefusedError: @@ -449,8 +449,8 @@ def claim_socket_path(path: str) -> socket.socket: """Bind and listen on ``path``, and return the socket, or refuse a path something already holds. The bind is the claim, so two servers starting together cannot both take one path: the loser's - bind fails. A probe follows it only to tell a stale file from a live server. The caller hands the - descriptor to uvicorn, which binds nothing and so never unlinks the path. + bind fails. A probe follows it only to tell a stale file from a live server. Serve the returned + socket by its descriptor: a server handed the path instead binds again, and unlinks this claim. """ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: