diff --git a/.basedpyright/baseline.json b/.basedpyright/baseline.json index 220d04b59..8e1741cd1 100644 --- a/.basedpyright/baseline.json +++ b/.basedpyright/baseline.json @@ -5628,24 +5628,6 @@ } } ], - "./positronic/simulator/env_server/client.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 22, - "endColumn": 33, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 26, - "endColumn": 50, - "lineCount": 1 - } - } - ], "./positronic/simulator/env_server/proxy.py": [ { "code": "reportOptionalMemberAccess", @@ -5656,24 +5638,6 @@ } } ], - "./positronic/simulator/env_server/server.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 40, - "endColumn": 60, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 28, - "endColumn": 42, - "lineCount": 1 - } - } - ], "./positronic/simulator/env_server/tests/mujoco_env.py": [ { "code": "reportArgumentType", @@ -6364,16 +6328,6 @@ } } ], - "./positronic/simulator/robolab/launcher.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 24, - "endColumn": 42, - "lineCount": 1 - } - } - ], "./positronic/simulator/robolab/make_fixture.py": [ { "code": "reportMissingImports", diff --git a/.dockerignore b/.dockerignore index 851c4e6eb..401a0701b 100644 --- a/.dockerignore +++ b/.dockerignore @@ -15,6 +15,7 @@ build # Virtual environments .venv +.venv-* venv env ENV diff --git a/docker/CONTEXTS.md b/docker/CONTEXTS.md index 84e2d59ba..50a6b1982 100644 --- a/docker/CONTEXTS.md +++ b/docker/CONTEXTS.md @@ -15,9 +15,13 @@ | `positro/openpi` | OpenPI training and inference | | `positro/dreamzero` | DreamZero inference (1+ GPU, H100 80GB recommended) | | `positro/robolab` | RoboLab (Isaac Lab) eval — runs `positronic eval run`, which spawns the Isaac sim subprocess in-container; needs an RTX-class GPU | +| `positro/galaxea` | G0.5-DROID inference, internal non-commercial evaluation only; isolated Galaxea and Positronic Python environments | Build and push all: `make push` +Galaxea is opt-in: `make build-galaxea`. Its evaluation-only image is excluded from +aggregate builds and pushes; see [the vendor README](../positronic/vendors/galaxea/README.md). + ## References - Service definitions and compose commands: `docker-compose.yml` diff --git a/docker/Dockerfile.galaxea b/docker/Dockerfile.galaxea new file mode 100644 index 000000000..8421b7a9b --- /dev/null +++ b/docker/Dockerfile.galaxea @@ -0,0 +1,45 @@ +# Internal, non-commercial evaluation only. See positronic/vendors/galaxea/NOTICE and LICENSE-G0.5. +FROM ubuntu:22.04 + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/ + +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential cmake git ca-certificates ffmpeg \ + libgl1 libglib2.0-0 libsm6 libxext6 libturbojpeg \ + && rm -rf /var/lib/apt/lists/* + +# Keep interpreter symlinks outside the host-mounted cache directories. +ENV UV_PYTHON_INSTALL_DIR=/opt/uv/python \ + UV_LINK_MODE=copy \ + UV_HTTP_TIMEOUT=600 \ + NVIDIA_VISIBLE_DEVICES=all \ + NVIDIA_DRIVER_CAPABILITIES=compute,utility + +ARG GALAXEA_SHA=89f2322b4ad016e192437adc1a2c253b05bab246 +RUN git clone https://github.com/OpenGalaxea/GalaxeaVLA.git /galaxea \ + && git -C /galaxea checkout ${GALAXEA_SHA} + +WORKDIR /galaxea +COPY positronic/vendors/galaxea/requirements-inference.txt /galaxea/requirements-inference.txt +# Export versions without installing Galaxea's training and simulation dependencies. +RUN --mount=type=cache,target=/root/.cache/uv \ + uv export --frozen --no-dev --no-emit-project --no-hashes --format requirements-txt \ + --output-file /galaxea/constraints.txt \ + && uv venv --python 3.10 /galaxea/.venv \ + && uv pip install --python /galaxea/.venv/bin/python \ + --index https://pypi.org/simple --default-index https://download.pytorch.org/whl/cu128 \ + --index-strategy unsafe-best-match \ + --constraint /galaxea/constraints.txt --requirement /galaxea/requirements-inference.txt + +WORKDIR /positronic +COPY . /positronic +ENV UV_PROJECT_ENVIRONMENT=/opt/positronic-venv \ + PYTHONPATH=/positronic +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --frozen --no-dev --python 3.13 + +LABEL org.opencontainers.image.description="G0.5-DROID for internal, non-commercial evaluation only" \ + org.opencontainers.image.licenses="Apache-2.0 AND LicenseRef-G0.5-Community-1.0" + +EXPOSE 8000 +ENTRYPOINT ["/opt/positronic-venv/bin/python", "-m", "positronic.vendors.galaxea.server"] diff --git a/docker/Makefile b/docker/Makefile index d8840cab7..5f1b764b9 100644 --- a/docker/Makefile +++ b/docker/Makefile @@ -1,6 +1,6 @@ .PHONY: all build tag push clean prune help build-training tag-training push-training build-openpi tag-openpi push-openpi build-groot tag-groot push-groot build-dreamzero-base push-dreamzero-base build-dreamzero tag-dreamzero push-dreamzero build-robolab tag-robolab push-robolab nebius-login push-robolab-cr +.PHONY: build-galaxea tag-galaxea push-galaxea -# Image configuration IMAGE_NAME_TRAINING := positro/positronic IMAGE_NAME_OPENPI := positro/openpi IMAGE_NAME_GROOT := positro/gr00t @@ -8,14 +8,13 @@ IMAGE_NAME_DREAMZERO := positro/dreamzero IMAGE_NAME_DREAMZERO_BASE := positro/dreamzero-base IMAGE_NAME_OPENPI_BASE := positro/openpi-base IMAGE_NAME_ROBOLAB := positro/robolab +IMAGE_NAME_GALAXEA := positro/galaxea -# Extract version from pyproject.toml (first literal version entry) VERSION := $(shell sed -n 's/^version = "\([^"]*\)"/\1/p' ../pyproject.toml | head -n 1) -# Get git commit SHA (short) GIT_SHA := $(shell git rev-parse --short HEAD 2>/dev/null || echo "unknown") -# Branch name sanitized for Docker tags (replace / with -) +# Docker tags cannot contain slashes. BRANCH := $(shell git rev-parse --abbrev-ref HEAD 2>/dev/null | tr '/' '-') # IMAGE_TAG: branch name locally, overridable via `make push IMAGE_TAG=foo` @@ -23,16 +22,14 @@ IMAGE_TAG ?= $(BRANCH) REGISTRY_URL ?= docker.io -# Nebius Container Registry — opt-in path for an in-region pull from serverless -# jobs (the `push-*` targets above stay on Docker Hub and are unchanged). Override -# NEBIUS_REGISTRY for a different project/registry; set NEBIUS_PROFILE to use a +# Nebius publishing is opt-in for in-region pulls; default push targets use Docker Hub. +# Override NEBIUS_REGISTRY for a different project/registry; set NEBIUS_PROFILE to use a # non-default `nebius` CLI profile (e.g. a service-account profile for headless # CI). Auth goes through the nebius credential helper — no registry secret here. NEBIUS_REGISTRY ?= cr.eu-north1.nebius.cloud/e00a0ahqzcp9x0xczz NEBIUS_REGISTRY_HOST := $(firstword $(subst /, ,$(NEBIUS_REGISTRY))) NEBIUS_PROFILE_FLAG := $(if $(NEBIUS_PROFILE),--profile $(NEBIUS_PROFILE),) -# Local build tags LOCAL_TAG_TRAINING := $(IMAGE_NAME_TRAINING):local # Base image tags (not parameterized — these are heavy foundational images) @@ -52,6 +49,7 @@ help: @echo "" @echo "Targets:" @echo " make build-training Build the training image" + @echo " make build-galaxea Build the internal non-commercial evaluation image" @echo " make build Build training, openpi and groot images" @echo "" @echo " make tag-training Tag the training image (depends on build-training)" @@ -99,6 +97,17 @@ build-robolab: @echo "Building $(IMAGE_NAME_ROBOLAB)..." docker build --platform linux/amd64 -f Dockerfile.robolab -t $(IMAGE_NAME_ROBOLAB):local .. +build-galaxea: + docker build --platform linux/amd64 -f Dockerfile.galaxea -t $(IMAGE_NAME_GALAXEA):local .. + +tag-galaxea: build-galaxea + docker tag $(IMAGE_NAME_GALAXEA):local $(IMAGE_NAME_GALAXEA):$(IMAGE_TAG) + docker tag $(IMAGE_NAME_GALAXEA):local $(IMAGE_NAME_GALAXEA):$(GIT_SHA) + +push-galaxea: tag-galaxea + docker push $(IMAGE_NAME_GALAXEA):$(IMAGE_TAG) + docker push $(IMAGE_NAME_GALAXEA):$(GIT_SHA) + build: build-training build-openpi build-groot build-dreamzero build-robolab tag-training: build-training @@ -252,6 +261,7 @@ endif all: push +# Optional image tags can be absent; removal errors must not stop attempts on the other tags. clean: @echo "Removing all local images..." -docker rmi $(LOCAL_TAG_TRAINING) @@ -270,7 +280,10 @@ clean: -docker rmi $(IMAGE_NAME_ROBOLAB):local -docker rmi $(IMAGE_NAME_ROBOLAB):$(IMAGE_TAG) -docker rmi $(IMAGE_NAME_ROBOLAB):$(GIT_SHA) - @echo "All local images removed." + -docker rmi $(IMAGE_NAME_GALAXEA):local + -docker rmi $(IMAGE_NAME_GALAXEA):$(IMAGE_TAG) + -docker rmi $(IMAGE_NAME_GALAXEA):$(GIT_SHA) + @echo "Image cleanup attempts complete. Check the output for removal errors." prune: @echo "Pruning dangling and unused Docker images..." diff --git a/docker/README.md b/docker/README.md index d74e05903..c27a0b1be 100644 --- a/docker/README.md +++ b/docker/README.md @@ -44,3 +44,12 @@ docker/build.sh If you customize `docker-compose.yml` volumes, **do not bind-mount** your host `~/.local/share/uv` into `/root/.local/share/uv` for `positro/gr00t` images. GR00T's `/.venv/bin/python` can be a symlink into the image's own uv-managed CPython under `/root/.local/share/uv/python/...`, and the bind mount can hide that target and cause `/.venv/bin/python` to fail with `ENOENT`. + +## Galaxea: internal evaluation access + +The `galaxea-server` service publishes port 8000 on the Docker host's localhost. +Use Docker Engine 28 or newer with default bridge networking. Remote DROID clients +connect through SSH forwarding; RoboLab uses `galaxea-server:8000` on the same +Compose network and Docker daemon. Start the server with `--service-ports --use-aliases`. +See the [Galaxea setup](../positronic/vendors/galaxea/README.md#start-the-server) +for the commands and internal, non-commercial usage conditions. diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index a02fdb4ff..cf069636e 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -3,7 +3,6 @@ services: image: positro/positronic:${IMAGE_TAG:-latest} pull_policy: always - # Enable interactive terminal stdin_open: true tty: true @@ -16,7 +15,6 @@ services: count: all capabilities: [compute, graphics, utility] - # Volume mounts volumes: - ${CACHE_ROOT:-${HOME}}/.cache:/root/.cache - ${CACHE_ROOT:-${HOME}}/.aws:/root/.aws:ro @@ -35,7 +33,6 @@ services: positronic: &openpi-common image: positro/openpi:${IMAGE_TAG:-latest} - # Enable interactive terminal stdin_open: true tty: true @@ -48,7 +45,6 @@ services: count: all capabilities: [compute, graphics, utility] - # Volume mounts volumes: - ${CACHE_ROOT:-${HOME}}/.cache:/root/.cache - ${CACHE_ROOT:-${HOME}}/.aws:/root/.aws:ro @@ -164,7 +160,6 @@ services: positronic-groot: &groot-common image: positro/gr00t:${IMAGE_TAG:-latest} - # Enable interactive terminal stdin_open: true tty: true @@ -177,7 +172,6 @@ services: count: all capabilities: [compute, graphics, utility] - # Volume mounts volumes: - ${CACHE_ROOT:-${HOME}}/.cache:/root/.cache - ${CACHE_ROOT:-${HOME}}/.aws:/root/.aws:ro @@ -200,7 +194,7 @@ services: <<: *groot-common container_name: groot-train shm_size: 8g - ipc: host # Enable shared memory access for training + ipc: host entrypoint: ["uv", "run", "--python", "3.13", "python", "-m", "positronic.vendors.gr00t.train"] groot-server: &groot-server-common @@ -213,11 +207,9 @@ services: positronic-dreamzero: &dreamzero-common image: positro/dreamzero:${IMAGE_TAG:-latest} - # Enable interactive terminal stdin_open: true tty: true - # GPU support deploy: resources: reservations: @@ -226,7 +218,6 @@ services: count: all capabilities: [compute, graphics, utility] - # Volume mounts volumes: - ${CACHE_ROOT:-${HOME}}/.cache:/root/.cache - ${CACHE_ROOT:-${HOME}}/.aws:/root/.aws:ro @@ -257,6 +248,25 @@ services: ipc: host entrypoint: ["uv", "run", "--python", "3.13", "python", "-m", "positronic.vendors.dreamzero.train"] + # Galaxea model use is restricted to internal, non-commercial evaluation; see vendors/galaxea/NOTICE. + galaxea-server: + image: positro/galaxea:${IMAGE_TAG:-latest} + init: true + shm_size: 8g + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: all + capabilities: [compute, utility] + volumes: + - ${CACHE_ROOT:-${HOME}}/.cache:/root/.cache + - ${CACHE_ROOT:-${HOME}}/.aws:/root/.aws:ro + - ${CACHE_ROOT:-${HOME}}/.cache/galaxea/checkpoints:/galaxea/checkpoints + ports: + - "127.0.0.1:8000:8000" + robolab-eval: image: positro/robolab:${IMAGE_TAG:-latest} diff --git a/positronic/simulator/env_server/client.py b/positronic/simulator/env_server/client.py index f89963e4f..12825c728 100644 --- a/positronic/simulator/env_server/client.py +++ b/positronic/simulator/env_server/client.py @@ -1,8 +1,4 @@ -"""Synchronous client for the env server: the lockstep ``tasks``/``reset``/``step``/``close`` round-trips. - -Positronic-free (``websockets`` + the wire codec). ``RemoteEnvControlSystem`` wraps this as a pimm -control system; tests use it directly to compare a socket rollout against an in-process one. -""" +"""Synchronous env-server client with no dependencies on Positronic.""" import logging import time @@ -11,34 +7,38 @@ from websockets.exceptions import ConnectionClosed from websockets.sync.client import connect -from .protocol import decode, encode +from . import protocol logger = logging.getLogger(__name__) -# How long ``close`` waits to be acknowledged. Teardown often runs while the peer is on its way out, and a -# simulator wedged in its own destructor holds the socket open without ever answering — an unbounded wait there -# hangs the run in place of ending it. +# Bound cleanup time when the server keeps the socket open without answering. _CLOSE_ACK_TIMEOUT = 5.0 class EnvConnection: - """One websocket to an ``EnvServer``, opened with retry. Every command blocks on the round-trip. - - There is no handshake: the first ``reset`` constructs the env server-side and returns - ``{'obs', 'meta', 'control_dt'}``. + """Connect to an ``EnvServer`` with retry; each command blocks for its response. - The connect deadline must cover a first boot on a fresh machine: a heavy simulator can spend many minutes - bringing its runtime up — compiling shaders, loading assets — before it binds the port. + Requests need no application handshake; ``reset`` returns the initial scene frame. + The connect deadline must cover simulator startup, which can take many minutes on a fresh machine. """ - def __init__(self, host: str, port: int, *, open_timeout: float = 10.0, connect_deadline: float = 1800.0): + def __init__( + self, + host: str, + port: int, + *, + open_timeout: float = 10.0, + connect_deadline: float = 1800.0, + ping_timeout: float = 600.0, + ): uri = f'ws://{host}:{port}/' deadline = time.monotonic() + connect_deadline backoff = 0.5 while True: try: # Camera + full-state observations routinely exceed websockets' 1 MiB default frame size. - self._ws = connect(uri, open_timeout=open_timeout, max_size=None) + # Native scene loading can block the server's heartbeat replies for minutes. + self._ws = connect(uri, open_timeout=open_timeout, max_size=None, ping_timeout=ping_timeout) break except (TimeoutError, OSError) as e: if time.monotonic() >= deadline: @@ -47,30 +47,28 @@ def __init__(self, host: str, port: int, *, open_timeout: float = 10.0, connect_ backoff = min(backoff * 2, 5.0) def tasks(self, spec: Any) -> list[dict[str, Any]]: - return self._request({'cmd': 'tasks', 'spec': spec})['tasks'] + return self._request({protocol.CMD: protocol.Command.TASKS.value, protocol.SPEC: spec})[protocol.TASKS] def reset(self, token: Any) -> dict[str, Any]: - return self._request({'cmd': 'reset', 'token': token}) + return self._request({protocol.CMD: protocol.Command.RESET.value, protocol.TOKEN: token}) def step(self, action: dict[str, Any]) -> dict[str, Any]: - return self._request({'cmd': 'step', 'action': action}) + return self._request({protocol.CMD: protocol.Command.STEP.value, protocol.ACTION: action}) def _request(self, msg: dict[str, Any]) -> dict[str, Any]: - self._ws.send(encode(msg)) - result = decode(self._ws.recv()) - if 'error' in result: - raise RuntimeError(f'env server: {result["error"]}') + self._ws.send(protocol.encode(msg)) + result = protocol.decode(self._ws.recv()) + if protocol.ERROR in result: + raise RuntimeError(f'env server: {result[protocol.ERROR]}') return result def close(self) -> None: try: - self._ws.send(encode({'cmd': 'close'})) + self._ws.send(protocol.encode({protocol.CMD: protocol.Command.CLOSE.value})) self._ws.recv(timeout=_CLOSE_ACK_TIMEOUT) except ConnectionClosed: - pass # a peer already gone has released whatever the acknowledgement would have reported + pass # A closed peer cannot acknowledge the close request. except TimeoutError: - # Abandoning it is still better than hanging the run here, but a server that took the request and - # never answered is wedged rather than finished, and its resources are nobody's to reclaim now. logger.error('Env server did not acknowledge close within %.1fs; abandoning it', _CLOSE_ACK_TIMEOUT) finally: self._ws.close() diff --git a/positronic/simulator/env_server/protocol.py b/positronic/simulator/env_server/protocol.py index f24362d37..53ca48079 100644 --- a/positronic/simulator/env_server/protocol.py +++ b/positronic/simulator/env_server/protocol.py @@ -1,19 +1,32 @@ -"""Wire codec for the remote env-server boundary: msgpack with a numpy envelope. +"""Env-server wire names and msgpack encoding for numpy arrays and plain data. -This module is **positronic-free** — it imports only ``msgpack`` and ``numpy`` — so it can be -imported (or copied) into a benchmark's isolated interpreter alongside the dumb server without -dragging in pimm or the rest of positronic. Only raw numpy arrays and plain-data dicts cross the -wire; every canonical<->raw mapping lives client-side in the ``EnvAdapter``. - -Arrays travel as raw bytes plus their ``dtype.str`` and shape, so a numpy-2 server round-trips a -numpy-1 client unchanged. +This module must work in an isolated interpreter without Positronic installed. +Arrays use raw bytes, ``dtype.str``, and shape for compatibility between numpy versions. """ import functools +from collections.abc import Callable +from enum import Enum +from typing import Any, cast import msgpack import numpy as np +CMD = 'cmd' +OK = 'ok' +TASKS = 'tasks' +SPEC = 'spec' +TOKEN = 'token' +ACTION = 'action' +ERROR = 'error' + + +class Command(Enum): + TASKS = 'tasks' + RESET = 'reset' + STEP = 'step' + CLOSE = 'close' + def _pack(obj): if isinstance(obj, np.ndarray): @@ -27,13 +40,13 @@ def _pack(obj): def _unpack(obj): if b'__ndarray__' in obj: - # ``bytearray`` (not the raw msgpack ``bytes``) backs a writable array, so the socket path - # matches the in-process path for envs/adapters that mutate a decoded buffer in place. + # A bytearray keeps the decoded array writable. return np.ndarray(buffer=bytearray(obj[b'data']), dtype=np.dtype(obj[b'dtype']), shape=obj[b'shape']) if b'__npgeneric__' in obj: return np.dtype(obj[b'dtype']).type(obj[b'data']) return obj -encode = functools.partial(msgpack.packb, default=_pack) +# msgpack's default autoreset=True makes packb return bytes. +encode = cast(Callable[[Any], bytes], functools.partial(msgpack.packb, default=_pack)) decode = functools.partial(msgpack.unpackb, object_hook=_unpack) diff --git a/positronic/simulator/env_server/server.py b/positronic/simulator/env_server/server.py index 2aa4256ca..102070e27 100644 --- a/positronic/simulator/env_server/server.py +++ b/positronic/simulator/env_server/server.py @@ -1,73 +1,58 @@ -"""The dumb remote env-server: a benchmark env behind ``tasks``/``reset``/``step``/``close`` over websockets. +"""Synchronous env server for one client with one outstanding request. -Positronic-free by contract — it depends only on ``websockets``, plus the wire codec and the -``EnvProtocol`` an env implements. A benchmark runs this in its own isolated interpreter (where -positronic can't be installed); the native fixture runs it in-process against ``MujocoSim``. The -server is lockstep request-response — one client, one outstanding request — so the World's virtual -clock advances unchanged while a step round-trips. - -There is no build phase: ``reset`` constructs (or reuses a cached) env from its opaque token and -re-randomizes it; ``control_dt`` rides every observation (``reset`` and each ``step``), so a benchmark -may even vary its control period per step. Heavy construction is the env's own concern (cache it). +This module must work in an isolated interpreter without Positronic installed. Protocol (msgpack frames, see ``protocol``): client ``{'cmd': 'tasks', 'spec': ...}`` -> server ``{'tasks': [{...}, ...]}`` client ``{'cmd': 'reset', 'token': ...}`` -> server ``{'obs', 'meta', 'robot_meta', 'control_dt'}`` client ``{'cmd': 'step', 'action': {...}}`` -> server ``{'obs', 'done', 'control_dt'}`` client ``{'cmd': 'close'}`` -> server ``{'ok': True}`` -Any command whose handling raises returns ``{'error': str}`` instead, which the client re-raises. +Command handling failures return ``{'error': str}`` without closing the session; the client re-raises them. + +``control_dt`` is the control period in seconds and can vary per step. +``meta`` identifies the scene; ``robot_meta`` identifies the robot model. +Either metadata dict can be empty when the client supplies that information. """ -import logging from abc import ABC, abstractmethod from typing import Any from websockets.sync.server import ServerConnection, serve -# This module + ``protocol`` copy into a benchmark's isolated interpreter as a self-contained unit -# (importing ``positronic.*`` would run the package's installed-version lookup and fail there). -# Relative when they land as a package, top-level when copied in flat. +# Isolated interpreters can import these modules as a package or as flat files. try: - from .protocol import decode, encode + from . import protocol except ImportError: - from protocol import decode, encode - -logger = logging.getLogger(__name__) + import protocol class EnvProtocol(ABC): - """A benchmark env behind the four methods the server exposes; the wire contract, positronic-free. + """An environment exchanging raw arrays and plain data with no Positronic dependencies. - ``tasks``, ``reset`` and ``step`` exchange raw plain data (numpy arrays + scalars) — the canonical<->raw - mapping is the client's ``EnvAdapter``, never the server's. Heavy, per-task construction is the - env's own concern (cache it, keyed by the token's structural part); the protocol has no build phase. + Canonical observation and command conversion belongs in the client's ``EnvAdapter``. + Implementations own environment construction and caching across resets. """ @abstractmethod def tasks(self, spec: dict[str, Any]) -> list[dict[str, Any]]: - """The benchmark's task records for ``spec``: the selection arguments an eval binds, keyed by the eval - config's own parameter names. + """Task records selected by ``spec``, each with a ``name`` field. - An absent key does not narrow that axis; an unknown value raises. Every record carries ``name``, the id - of one task. + An absent key leaves that axis unrestricted; an unknown value raises. """ @abstractmethod def reset(self, token: Any) -> dict[str, Any]: - """Construct (cached) + re-randomize from a token; returns ``obs``, ``meta``, ``robot_meta``, ``control_dt``. + """Construct or reuse the environment and re-randomize it from an opaque token. - ``control_dt`` is the env's control period: the client paces one ``step`` per ``control_dt`` and advances - the World's virtual clock by it each step. ``meta`` is the scene identity the policy reads its instruction - from (the language goal, scene ids); ``robot_meta`` is the robot model identity (URDF / joint names / - control frame) recorded into the episode. Either is ``{}`` when the client owns that side — a static - instruction, or an embodiment that ships its own model. + Return ``obs``, scene ``meta``, ``robot_meta``, and ``control_dt`` in seconds. + Either metadata dict may be empty when the client supplies it. """ @abstractmethod def step(self, action: dict[str, Any]) -> dict[str, Any]: - """Apply a raw action for one control period; returns ``{'obs', 'done', 'control_dt'}``. + """Apply a raw action for one control period; return ``obs``, ``done``, and ``control_dt``. - ``control_dt`` is re-reported every step (the wait until the next one), so it can vary. + ``control_dt`` is the wait until the next step and may vary each step. """ @abstractmethod @@ -76,12 +61,9 @@ def close(self) -> None: class EnvServer: - """Serves one ``EnvProtocol`` over a synchronous websocket — one client per server, lockstep. + """Serve one client on the calling thread; ``shutdown`` releases the owned environment. - The single client is accepted and served on the thread that calls ``serve_forever`` (the subprocess main - thread), not a per-connection websocket thread: a render backend can be thread-affine — macOS GLFW must - initialize on the main thread — and a sim is single-threaded anyway. The env lives for the server's - lifetime; ``shutdown`` releases it. + ``serve_forever`` ends when that client's session ends or shutdown is requested. """ def __init__(self, env: EnvProtocol, host: str, port: int): @@ -93,40 +75,31 @@ def __init__(self, env: EnvProtocol, host: str, port: int): self._shutdown = False def _handle(self, connection: ServerConnection) -> None: - # Reaching here means a client completed the websocket handshake: it is the one client this server - # serves, so ``serve_forever`` exits once this returns. A failure handling a command (a rejected action, - # a sim blow-up, an unknown command) crosses back as an error frame the client re-raises, rather than - # killing the connection. self._served = True for raw in connection: - msg = decode(raw) + msg = protocol.decode(raw) try: - match msg['cmd']: - case 'close': - connection.send(encode({'ok': True})) + match protocol.Command(msg[protocol.CMD]): + case protocol.Command.CLOSE: + connection.send(protocol.encode({protocol.OK: True})) return - case 'tasks': - result = {'tasks': self._env.tasks(msg['spec'])} - case 'reset': - result = self._env.reset(msg['token']) - case 'step': - result = self._env.step(msg['action']) - case other: - raise ValueError(f'Unknown command: {other!r}') + case protocol.Command.TASKS: + result = {protocol.TASKS: self._env.tasks(msg[protocol.SPEC])} + case protocol.Command.RESET: + result = self._env.reset(msg[protocol.TOKEN]) + case protocol.Command.STEP: + result = self._env.step(msg[protocol.ACTION]) except Exception as e: - result = {'error': f'{type(e).__name__}: {e}'} - connection.send(encode(result)) + result = {protocol.ERROR: f'{type(e).__name__}: {e}'} + connection.send(protocol.encode(result)) def serve_forever(self) -> None: - # A reset token can be a large opaque blob (e.g. an exact-replay scene), so don't cap frame size. - # ``serve`` would spawn a thread per connection; instead the accept loop runs here and calls the - # handshake + ``_handle`` inline, so the env runs on this thread (the subprocess main thread — macOS GLFW - # must init there). The loop skips bare TCP probes (which never handshake) and exits once the one client - # has been served. + # Reset tokens can exceed the default frame size. + # Accept and handle connections inline: macOS GLFW requires environment calls on the main thread. + # Bare TCP probes do not count as the one websocket session. with serve(self._handle, self._host, self._port, max_size=None) as server: self._server = server - # Time out ``accept`` so the loop periodically observes ``shutdown`` even with no client connecting — - # closing the listening socket from another thread does not reliably wake a blocking ``accept``. + # Closing the socket from another thread does not reliably wake a blocking accept. server.socket.settimeout(0.5) while not self._served and not self._shutdown: try: @@ -139,7 +112,6 @@ def serve_forever(self) -> None: server.handler(sock, addr) def shutdown(self) -> None: - # Stop accepting (the flag breaks the timed accept loop) and release the env. self._shutdown = True if self._server is not None: self._server.shutdown() diff --git a/positronic/simulator/env_server/tests/test_remote_env.py b/positronic/simulator/env_server/tests/test_remote_env.py index 6c8c07f17..bd29820c8 100644 --- a/positronic/simulator/env_server/tests/test_remote_env.py +++ b/positronic/simulator/env_server/tests/test_remote_env.py @@ -7,6 +7,9 @@ import numpy as np import pos3 import pytest +from websockets.exceptions import ConnectionClosedError +from websockets.frames import OP_PING +from websockets.sync.client import connect as websocket_connect from websockets.sync.server import serve as websocket_serve import pimm @@ -25,6 +28,7 @@ from positronic.policy.codec import ActionTimestamp from positronic.policy.layers import ChunkedSchedule from positronic.policy.tests.test_harness import StubPolicy +from positronic.simulator.env_server import protocol from positronic.simulator.env_server.adapter import EnvAdapter, _in_env_control_frame, _wire_command from positronic.simulator.env_server.client import _CLOSE_ACK_TIMEOUT, EnvConnection from positronic.simulator.env_server.launcher import free_port @@ -82,8 +86,7 @@ def _assert_obs_equal(a: dict, b: dict) -> None: @pytest.mark.timeout(60.0) def test_transport_is_transparent(env_server): - """The wire round-trips faithfully: the same seed + action sequence through the env in-process and - behind the socket produce identical raw observations. This is the parity oracle for the protocol.""" + """The same seed and actions must yield identical raw observations in-process and over the socket.""" host, port = env_server seed = 7 @@ -109,7 +112,7 @@ def test_transport_is_transparent(env_server): @pytest.mark.timeout(60.0) def test_the_env_answers_its_own_task_list(env_server): - """The env answers its own records, and a spec it does not know raises through to the client.""" + """Unknown task specifications must reach the environment and return its errors.""" host, port = env_server conn = EnvConnection(host, port) assert conn.tasks({}) == [{'name': SCENE_NAME}] @@ -121,11 +124,7 @@ def test_the_env_answers_its_own_task_list(env_server): @contextmanager def _mute_server(): - """A peer that accepts the connection and then answers nothing, holding the socket open. - - What a simulator looks like once it is wedged in its own teardown: the process is past serving but the - socket outlives it, so nothing ever closes the connection from that end. - """ + """Hold the socket open without answering requests, simulating a peer stuck in teardown.""" host, port = 'localhost', free_port() release = threading.Event() @@ -146,7 +145,6 @@ def handler(connection): @pytest.mark.timeout(60.0) def test_close_gives_up_on_a_peer_that_never_answers(): - """Teardown ends the run: an unanswered goodbye is as good as a closed socket, and is not waited out.""" with _mute_server() as (host, port): conn = EnvConnection(host, port) started = time.monotonic() @@ -154,11 +152,76 @@ def test_close_gives_up_on_a_peer_that_never_answers(): assert time.monotonic() - started < _CLOSE_ACK_TIMEOUT + 10.0 +@pytest.fixture +def server_without_heartbeat(monkeypatch): + monkeypatch.setattr( + 'positronic.simulator.env_server.client.connect', + partial(websocket_connect, ping_interval=0.01, ping_timeout=0.02), + ) + ignored_pings = [] + release = threading.Event() + + def handler(connection): + receive_frame = connection.protocol.recv_frame + + def ignore_ping(frame): + if frame.opcode == OP_PING: + ignored_pings.append(frame) + else: + receive_frame(frame) + + monkeypatch.setattr(connection.protocol, 'recv_frame', ignore_ping) + for raw in connection: + if protocol.Command(protocol.decode(raw)[protocol.CMD]) is protocol.Command.CLOSE: + connection.send(protocol.encode({protocol.OK: True})) + return + if release.wait(timeout=0.2): + return + connection.send(protocol.encode({'obs': {'ready': True}})) + + host, port = 'localhost', free_port() + with websocket_serve(handler, host, port) as server: + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield host, port, ignored_pings + finally: + release.set() + server.shutdown() + thread.join(timeout=2.0) + + +@pytest.mark.timeout(10.0) +def test_scene_reset_survives_delayed_heartbeat_replies(server_without_heartbeat): + host, port, ignored_pings = server_without_heartbeat + conn = EnvConnection(host, port) + try: + assert conn.reset({}) == {'obs': {'ready': True}} + assert ignored_pings + finally: + conn.close() + + +@pytest.mark.timeout(10.0) +@pytest.mark.parametrize('command', [EnvConnection.tasks, EnvConnection.reset, EnvConnection.step]) +def test_unanswered_heartbeat_closes_pending_requests(server_without_heartbeat, command): + host, port, ignored_pings = server_without_heartbeat + conn = EnvConnection(host, port, ping_timeout=0.02) + try: + with pytest.raises(ConnectionClosedError, match='keepalive ping timeout'): + command(conn, {}) + assert ignored_pings + with pytest.raises(ConnectionClosedError): + conn.step({}) + finally: + conn.close() + + _HOLD = {'command': {'type': 'hold'}, 'grip': 0.0} def _settle(env, action: dict, steps: int) -> np.ndarray: - """Apply ``action`` once, then idle ``steps`` ticks while the position actuators settle; return the final eef.""" + """Apply the action once, hold for ``steps`` ticks, and return the settled end-effector position.""" env.step(action) out = {'obs': None} for _ in range(steps): @@ -167,7 +230,7 @@ def _settle(env, action: dict, steps: int) -> np.ndarray: def test_a_pinned_control_mode_rides_the_wire(): - """Honoring a mode is the env's, so the adapter delivers it rather than deciding for every env.""" + """Control modes must pass through for the environment to interpret.""" mode = roboarm_command.Impedance(kq=(40.0,) * 7, kqd=(4.0,) * 7, kx=(750.0,) * 6, kxd=(37.0,) * 6) wired = _wire_command(roboarm_command.JointPosition(np.zeros(7), mode=mode)) assert wired['mode'] == roboarm_command.to_wire(mode) @@ -175,11 +238,7 @@ def test_a_pinned_control_mode_rides_the_wire(): class TestEnvControlFrame: - """An env measuring somewhere other than the embodiment's ``default`` gets commands re-expressed for it. - - ``RobolabAdapter`` is the live case, and this file covers the adapter wire contract, so its frame - round-trip runs here. - """ + """Commands must target the frame the environment measures, even when it differs from the default.""" frame = geom.Transform3D(np.array([0.0, 0.0, 0.1]), geom.Rotation.from_euler([0.0, 0.0, np.pi / 2])) rotmat = geom.Rotation.Representation.ROTATION_MATRIX @@ -196,7 +255,6 @@ def test_a_delta_already_in_the_env_frame_wires_bare(self): np.testing.assert_allclose(wired['delta'], delta.as_vector(self.rotmat), atol=1e-12) def test_a_command_re_expressed_for_the_env_keeps_its_mode(self): - """The frame a command is measured in has nothing to do with the law that drives to it.""" mode = roboarm_command.Impedance(kq=(40.0,) * 7, kqd=(4.0,) * 7, kx=(750.0,) * 6, kxd=(37.0,) * 6) pose = geom.Transform3D(np.array([0.4, 0.1, 0.3]), geom.Rotation.identity) moved = _in_env_control_frame(roboarm_command.CartesianPosition(pose, mode=mode), self.frame) @@ -205,7 +263,7 @@ def test_a_command_re_expressed_for_the_env_keeps_its_mode(self): assert delta.mode == mode def test_a_delta_outside_the_env_frame_is_refused(self): - """The env anchors on its own measured pose, so a delta meant for another frame has no wire form.""" + """A delta needs the measured pose in its own frame, which the wire does not supply.""" delta = geom.Transform3D(np.array([0.0, 0.0, 0.04]), geom.Rotation.identity) with pytest.raises(ValueError, match='control frame'): _wire_command(_in_env_control_frame(roboarm_command.CartesianDelta(delta), self.frame)) @@ -227,9 +285,10 @@ def test_robolab_reports_and_drives_the_same_frame(self): @pytest.mark.timeout(60.0) def test_cartesian_delta_matches_absolute_target(): - """A CartesianDelta settles to the same eef as the absolute CartesianPosition for the composed target: it - confirms the world-frame compose and that the delta fires once — a delta re-applied every tick would overshoot. - Comparing the two paths cancels the actuators' shared steady-state offset, so the match is exact.""" + """A one-shot delta must settle at the composed absolute target without accumulating on idle ticks. + + Comparing both paths cancels their shared actuator steady-state offset. + """ rotmat = geom.Rotation.Representation.ROTATION_MATRIX seed, settle = 11, 300 lift = np.array([0.0, 0.0, 0.04]) @@ -250,18 +309,15 @@ def test_cartesian_delta_matches_absolute_target(): delta_env.close() assert ee_delta[2] > ee0[2] + 0.01, 'the delta did not lift the arm' - np.testing.assert_allclose(ee_delta, ee_abs, atol=1e-4) # same composed target -> same settled eef - np.testing.assert_allclose(ee_idle, ee_delta, atol=1e-3) # one-shot: idle ticks add no motion + np.testing.assert_allclose(ee_delta, ee_abs, atol=1e-4) + np.testing.assert_allclose(ee_idle, ee_delta, atol=1e-3) -# The one task ``_CountdownEnv`` serves. _COUNTDOWN = 'countdown' class _CountdownEnv(EnvProtocol): - """A degenerate env exercising the proxy's terminal and free-run paths without the real ``stack_cubes`` - wrapper. Obs encodes the step count (``reset`` is step 0, each ``step`` increments) so a reader can - tell whether the proxy stepped; ``done`` fires after ``done_after`` steps (``None`` → never).""" + """Observe step counts starting at zero on reset; ``done_after=None`` never terminates.""" def __init__(self, done_after: int | None = None, control_dt: float = 0.1): self._done_after = done_after @@ -277,7 +333,7 @@ def tasks(self, spec): def reset(self, token): self._steps = 0 - meta = {'task': _COUNTDOWN} # scene meta the env reports only at reset; ``step`` omits it + meta = {'task': _COUNTDOWN} return { 'obs': {'q': np.full(7, self._steps, dtype=np.float64)}, 'meta': meta, @@ -316,8 +372,7 @@ def terminal(self, result): @pytest.mark.timeout(60.0) def test_the_proxy_connects_on_a_tasks_call_before_any_reset(): - """``tasks`` runs before the first trial, so it starts the server; the reset that follows shares its - connection.""" + """Task listing must start the server and leave the connection usable for reset.""" with serve_env(_CountdownEnv()) as (host, port), pimm.World(virtual_time=True) as world: proxy = RemoteEnvControlSystem(_CountdownAdapter(), nullcontext((host, port))) obs_rx = world.pair(proxy.observations['value']) @@ -331,7 +386,7 @@ def test_the_proxy_connects_on_a_tasks_call_before_any_reset(): @pytest.mark.timeout(60.0) def test_a_selection_naming_no_task_is_refused(): - """A sweep of no trials ends at once and reads as a run that succeeded, so an empty listing raises.""" + """Reject empty task selections so zero-trial runs cannot appear successful.""" with serve_env(_CountdownEnv()) as (host, port): proxy = RemoteEnvControlSystem(_CountdownAdapter(), nullcontext((host, port))) with pytest.raises(ValueError, match='no task'): @@ -340,7 +395,7 @@ def test_a_selection_naming_no_task_is_refused(): @pytest.mark.timeout(60.0) def test_a_refused_listing_stops_the_server_it_started(): - """The scheduler enters ``run``, whose teardown stops the server, only after the listing.""" + """Listing can fail before the scheduler starts, so cleanup cannot depend on its teardown.""" with serve_env(_CountdownEnv()) as address: stopped = False @@ -360,9 +415,7 @@ def serve(): @pytest.mark.timeout(60.0) def test_proxy_publishes_the_reset_frame_then_free_runs(): - """``reset`` publishes the env's frame (step 0) and clears ``done``, then the proxy free-runs — it steps - the env every active tick (physics progresses through the inference window). The step-count obs makes it - observable: the reset publishes step 0, then it advances each tick with no command needed.""" + """Reset must publish step zero and clear termination; active ticks advance physics without commands.""" with serve_env(_CountdownEnv()) as (host, port), pimm.World(virtual_time=True) as world: proxy = RemoteEnvControlSystem(_CountdownAdapter(), nullcontext((host, port))) obs_rx = world.pair(proxy.observations['value']) @@ -381,25 +434,21 @@ def test_proxy_publishes_the_reset_frame_then_free_runs(): @pytest.mark.timeout(60.0) def test_proxy_caches_reset_meta_as_live_instruction_source(): - """The env reports scene meta only at ``reset`` (``step`` omits it); the proxy caches it so a ``Task`` - reads its language live off ``proxy.meta`` — the callable-instruction path LIBERO relies on — and the - cached value holds across the steps that follow.""" + """Live instruction callbacks must retain reset metadata across steps that omit it.""" with serve_env(_CountdownEnv()) as (host, port), pimm.World(virtual_time=True) as world: proxy = RemoteEnvControlSystem(_CountdownAdapter(), nullcontext((host, port))) task = Task(instruction_source=lambda: proxy.meta['task'], timeout_sec=1.0) scheduler = world.start([proxy]) proxy.reset({eval_keys.SEED: 0}) - assert task.instruction == 'countdown' # resolved live off the cached reset meta - drive_scheduler(scheduler, steps=4) # the env steps, each ``step`` omitting meta ... - assert task.instruction == 'countdown' # ... yet the reset-scoped cache holds + assert task.instruction == 'countdown' + drive_scheduler(scheduler, steps=4) + assert task.instruction == 'countdown' @pytest.mark.timeout(60.0) def test_remote_eval_runs_to_timeout_without_done(env_server, tmp_path): - """The real ``stack_cubes`` wrapper, end to end: no terminal, so the trial runs to the task timeout - (``eval.terminated`` False, ``eval.success`` absent) and records the canonical signals under the shared - camera key.""" + """A timed-out trial must record canonical signals without reporting termination or success.""" host, port = env_server with pos3.mirror(): ev = remote_stack_cubes_eval(host, port, camera_dict=CAMERAS) @@ -430,15 +479,13 @@ def test_remote_eval_runs_to_timeout_without_done(env_server, tmp_path): 'eval_cfg', [libero_cfg.spatial, robolab_cfg.benchmark, native_cfg.stack_cubes], ids=['libero', 'robolab', 'mujoco'] ) def test_every_sim_eval_publishes_the_shared_camera_keys(eval_cfg): - """A codec names the camera it wants, so a sim spelling its cameras its own way can be scored only by a codec - written for it. Every sim publishes the shared pair, whatever the benchmark calls those cameras.""" + """Shared camera names let the same policy codec serve different simulators.""" observations = eval_cfg.instantiate().embodiment.observations assert {keys.EXTERIOR_IMAGE, keys.WRIST_IMAGE} <= set(observations) class _JointposChunks(Policy): - """Chunks exactly as long as the intended open-loop cadence; ``target_grip`` encodes - ``chunk * 100 + step`` so the recorded wire signals show which actions executed, and when.""" + """Encode ``chunk * 100 + step`` in grip values to identify executed actions in recordings.""" def __init__(self, command: roboarm_command.CommandType, chunk_len: int): self.command = command @@ -463,10 +510,10 @@ def __call__(self, obs, time_ns): @pytest.mark.timeout(60.0) def test_full_chunk_executes_between_replans(env_server, tmp_path): - """The recording proves the contract the DROID jointpos codec makes with RoboLab's client: every action - of every chunk lands on the wire — including the final one, which ``ActionTimestamp``'s validity - sentinel gives a full period before ``ChunkedSchedule`` re-infers — and replans arrive exactly - ``chunk_len`` control periods apart.""" + """Every chunk action must execute, with a full control period for the final action. + + ``ActionTimestamp``'s validity sentinel must keep replans ``chunk_len`` control periods apart. + """ host, port = env_server probe = make_mujoco_env([]) control_dt = probe.reset(0)['control_dt'] @@ -496,14 +543,19 @@ def test_full_chunk_executes_between_replans(env_server, tmp_path): @pytest.mark.timeout(60.0) -def test_server_failure_crosses_as_error_frame(env_server): - """A command the env rejects comes back as an error the client re-raises — the connection survives - rather than dying on the server-side exception, and the next command still works.""" +@pytest.mark.parametrize( + 'message', + [ + {protocol.CMD: 'bogus'}, + {protocol.CMD: protocol.Command.STEP.value, protocol.ACTION: {'command': {'type': 'bogus'}, 'grip': 0.0}}, + ], +) +def test_server_failure_crosses_as_error_frame(env_server, message): + """Rejected commands must reach the client as errors while leaving the connection usable.""" host, port = env_server conn = EnvConnection(host, port) conn.reset(7) with pytest.raises(RuntimeError, match='bogus'): - conn.step({'command': {'type': 'bogus'}, 'grip': 0.0}) - # The socket is still usable after a delivered failure. + conn._request(message) assert 'obs' in conn.step({'command': {'type': 'joint_pos', 'q': np.zeros(7)}, 'grip': 0.0}) conn.close() diff --git a/positronic/vendors/galaxea/LICENSE-G0.5 b/positronic/vendors/galaxea/LICENSE-G0.5 new file mode 100644 index 000000000..27e017e02 --- /dev/null +++ b/positronic/vendors/galaxea/LICENSE-G0.5 @@ -0,0 +1,107 @@ +G0.5 COMMUNITY LICENSE AGREEMENT (NON-COMMERCIAL + LIMITED PATENT LICENSE) +Release Date: [2026-06-16] + +IMPORTANT: THIS IS A LICENSE AGREEMENT, NOT A SALE. BY DOWNLOADING, ACCESSING, USING, REPRODUCING, MODIFYING, OR DISTRIBUTING ANY PORTION OF THE G0.5 MATERIALS, YOU WILL BE DEEMED TO HAVE ACCEPTED AND AGREED TO BE BOUND BY THIS AGREEMENT, WHICH IS EFFECTIVE IMMEDIATELY. YOU ARE GRANTED THE LICENSE IN CONSIDERATION OF YOUR ACCEPTANCE OF THESE TERMS AND CONDITIONS, AND GALAXEA GRANTS YOU SUCH LICENSE IN CONSIDERATION OF BENEFITS WE RECEIVE FROM MAKING THE G0.5 MATERIALS AVAILABLE UNDER THESE TERMS AND CONDITIONS. + +1. DEFINITIONS +1.1 “Licensor”, “Galaxea”, “We” or “Us” means Xinghaitu (Beijing) AI Technology Co., Ltd., and its affiliates, with an address at Building 14, Compound No. 3, Jinghai 5th Road, Beijing Economic-Technological Development Area (Tongzhou), Beijing, China. +1.2 “You”, “Licensee” or “Recipient” means you, or your employer or any other person or entity (if you are entering into this Agreement on such person or entity’s behalf), of the age required under applicable laws, rules or regulations to provide legal consent and that has legal authority to bind your employer or such other person or entity if you are entering in this Agreement on their behalf. +1.2A “Affiliate” means any entity that directly or indirectly controls, is controlled by, or is under common control with You, where “control” means ownership of more than 50% of the voting power or the power to direct management. +1.2B “Corporate Group” means You and Your Affiliates collectively. +1.3 “Agreement” means this G0.5 Community License Agreement, including any exhibits referenced herein. +1.4 “G0.5” means the vision-language-action model released by Licensor under this Agreement and all technical solutions and knowledge related to this model, including model code, weights, configurations, training/inference scripts, model parameters, documentation, and other accompanying materials. +1.5 “Materials” means, collectively, G0.5 and the Documentation made available by Licensor under this Agreement. +1.6 “Documentation” means technical documentation, specifications, manuals, model cards, README files, and other documentation for the Materials. +1.7 “Output” means the data, actions, predictions, embeddings, logs, or other results generated by running the Materials. +1.8 “Derivative Works” means modifications, adaptations, or derivative works of the Materials, including any modified versions of model code, weights, or configurations. +1.9 “Permitted Purpose” means and is strictly limited to: + (a) Academic Research (non-profit research conducted by a non-commercial academic institution), + (b) Personal Use (non-commercial use by an individual for private purposes), + (c) Education (teaching, coursework, and classroom use), and + (d) Evaluation (testing, benchmarking, or internal technical evaluation that is not Commercial Use), including Internal Evaluation PoC (as defined below) even if performed by a for-profit entity, provided it meets all Internal Evaluation PoC conditions. +For the avoidance of doubt, Evaluation does NOT include production deployment, providing services to third parties, or distribution as part of a commercial product. +1.9A “Internal Evaluation PoC” means a limited, internal proof-of-concept or evaluation performed solely to assess technical feasibility, where: +(i) the Materials are run only on Authorized Hardware using an Authorized Image; +(ii) access is restricted to personnel and contractors of Your Corporate Group who are bound by confidentiality obligations at least as protective as those applicable to Your own confidential information; +(iii) no access to the Materials, Model, or Output is provided to any party outside Your Corporate Group (including customers, partners, or service recipients); +(iv) it is not used in production; +(v) it is not distributed, embedded, shipped, or deployed with any commercial product or service; and +(vi) it does not generate revenue from, and is not used to provide any service to, any party outside Your Corporate Group. +1.10 “Commercial Use” means any use, directly or indirectly, for commercial advantage or monetary compensation, or in connection with any for-profit activity, including without limitation: + (a) selling, licensing, leasing, or otherwise transferring the Materials or any Derivative Works; + (b) manufacturing, selling, offering to sell, distributing, or providing any product or service that incorporates, utilizes, is enabled by, or is derived from the Materials (including Output) or the underlying technology; + (c) providing hosted or managed services (e.g., SaaS, API access, cloud inference/training, consulting, integration, or support services) based on the Materials, whether or not for a fee; + (d) use by a for-profit entity for internal business operations, development, manufacturing, or production deployment, other than Internal Evaluation PoC that fully complies with Section 1.9A; or + (e) using the Materials to train, fine-tune, or improve any model for a commercial purpose. +(Commercial Use is intentionally broad. If in doubt, contact us for a commercial license.) +1.11 “IPR” (Intellectual Property Rights) means all worldwide intellectual property rights related to the Materials, including copyrights, patents and patent applications, trade secrets, and trademarks. +1.12 “Patent Claims” means any patent claim(s) owned or controlled by Licensor that would be infringed by making, using, selling, offering for sale, importing, or otherwise exploiting the Materials. +1.13 “Notice” means copyright, acknowledgement and trademark notices, modification notices, and all notices that refer to this Agreement and/or warranty disclaimers included with the Materials. + +2. LICENSE GRANT (NON-COMMERCIAL) +2.1 Copyright License (Permitted Purpose Only). +Subject to the terms and conditions of this Agreement, Licensor grants You a worldwide, non-exclusive, non-transferable, non-sublicensable, royalty-free license to: + (a) download, reproduce, and use the Materials; + (b) modify the Materials and create Derivative Works; and + (c) distribute the Materials or Derivative Works, +strictly and only for the Permitted Purpose. +2.2 Distribution Conditions. +If You copy, distribute, or make available to any third party any Materials or Derivative Works, You must: + (a) provide recipients a copy of this Agreement (or a link to it) and clearly state the Materials are licensed under this Agreement; + (b) retain all Notices in the Materials and include a NOTICE file as described in Section 2.2(d); + (c) cause any modified files to carry prominent notices stating that You changed the files; and + (d) include a text file named “NOTICE” with the following (or substantially similar) notice: “G0.5 is licensed under the G0.5 Community License Agreement (Non-Commercial + Limited Patent License), not sold, Copyright © 2026 Galaxea. All rights reserved by Galaxea. ‘Galaxea’ and related marks are trademarks of Galaxea or its affiliates.” +2.3 Attribution. +You must not state or imply that Licensor endorses You or Your use of the Materials. You may use the name “G0.5” solely to describe that Your work is built using the Materials, provided You comply with this Agreement and applicable trademark law. + +3. PATENT LICENSE (LIMITED) + NO COMMERCIAL PATENT RIGHTS +3.1 Limited Patent License for Permitted Purpose. +Subject to the terms and conditions of this Agreement, Licensor grants You a limited, non-exclusive, non-transferable, non-sublicensable, royalty-free patent license under Licensor’s Patent Claims solely to make, use, and modify the Materials as necessary to exercise the rights granted in Section 2 for the Permitted Purpose only. +3.2 No Patent License for Commercial Use. +NO PATENT LICENSE IS GRANTED FOR ANY COMMERCIAL USE. Any Commercial Use that would infringe Licensor’s Patent Claims requires a separate written commercial license agreement with Licensor. +3.3 Patent Retaliation. +If You (or any entity controlling, controlled by, or under common control with You) initiate or voluntarily participate in any claim or proceeding alleging that the Materials (or any portion thereof) infringe any patent or any intellectual property claim or proceeding, then all licenses granted to You under this Agreement terminate automatically as of the date such claim or proceeding is filed. + +4. RESTRICTIONS +4.1 No Commercial Use. +You must not use the Materials or Derivative Works for any Commercial Use. +4.2 No Circumvention. +You must not remove or alter Notices or attempt to circumvent license restrictions through technical measures, contractual terms, or otherwise. +4.3 Compliance with Law. +Your use must comply with applicable laws and regulations, including trade compliance laws. +4.4 Internal Evaluation PoC Boundary. +If You rely on the Internal Evaluation PoC allowance, You must ensure that no third party (including Your customers, partners, or affiliates acting as service recipients) is provided access to the Materials, Model, or Output as a service or demonstration environment, and that the Materials are not used to support any external-facing function. Any such external exposure constitutes Commercial Use and is prohibited without a commercial license. + +5. THIRD-PARTY COMPONENTS +The Materials may include third-party software or data under separate license terms. Those third-party license terms apply to the extent required, and nothing in this Agreement limits Your rights under those third-party licenses. You are responsible for complying with all third-party license obligations. + +6. AUTHORIZED HARDWARE / PREINSTALLATION TERMS +6.1 Use on Authorized Hardware. Licensor permits You to run, evaluate, develop, and (for Permitted Purpose) train or fine-tune the Model on Authorized Hardware using an Authorized Image, subject to this Agreement. +6.2 No Implied Commercial Rights. Preinstallation, shipment, or availability of the Materials on Authorized Hardware does not grant any Commercial Use rights or any patent license for Commercial Use. +6.3 No Extraction for External or Commercial Deployment. You may not extract, export, deploy, or run the Materials, Model, or weights from Authorized Hardware / Authorized Image into any non-authorized environment for any external-facing use or any Commercial Use without a separate commercial license. +6.4 Internal Evaluation PoC on Authorized Hardware. For clarity, Internal Evaluation PoC (Section 1.9A) is permitted for Your Corporate Group only when conducted on Authorized Hardware using an Authorized Image and remains strictly internal (no external access). +6.5 No External Demos or Customer Access. You may not provide any customer, partner, or other third party access to a demo environment, hosted endpoint, API, or system that exposes the Materials, Model, or Output, even if no fee is charged, unless You have a separate commercial license. +6.6 Trademarks. This Agreement does not grant permission to use Licensor’s trademarks, logos, or brand names except to make truthful statements that You are using the Materials as provided by Licensor, without implying endorsement, certification, or partnership. + +7. OWNERSHIP +7.1 Licensor Ownership. +Licensor retains all right, title, and interest in and to the Materials, including all IPR therein, except for the limited licenses expressly granted in this Agreement. +7.2 Your Derivative Works. +As between You and Licensor, You own Your modifications and Derivative Works, subject to Licensor’s ownership of the underlying Materials and subject to Your ongoing compliance with this Agreement. + +8. DISCLAIMER OF WARRANTY AND LIMITATION OF LIABILITY +8.1 No Support. +Licensor is not obligated to provide support, updates, or maintenance. +8.2 Disclaimer. +THE MATERIALS ARE PROVIDED “AS IS” WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING ANY WARRANTIES OF TITLE, MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, OR NON-INFRINGEMENT. +8.3 Limitation of Liability. +TO THE MAXIMUM EXTENT PERMITTED BY LAW, IN NO EVENT SHALL LICENSOR OR ITS AFFILIATES BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, CONSEQUENTIAL, EXEMPLARY, OR PUNITIVE DAMAGES, OR LOST PROFITS, ARISING FROM OR RELATED TO THIS AGREEMENT OR THE MATERIALS. + +9. TERMINATION +9.1 Termination for Breach. +This Agreement is effective until terminated. Licensor may terminate this Agreement immediately upon Your breach. Upon termination, You must promptly cease all use of the Materials and delete any copies in Your possession or control. +9.2 Survival. +Sections 3, 5, 7, 8, 9.2, and 10 survive termination. + +10. GOVERNING LAW AND JURISDICTION +This Agreement is governed by the laws of the Hong Kong Special Administrative Region of the People’s Republic of China, without regard to conflict of law principles. Exclusive jurisdiction and venue shall be in a court of competent jurisdiction in Hong Kong SAR. diff --git a/positronic/vendors/galaxea/NOTICE b/positronic/vendors/galaxea/NOTICE new file mode 100644 index 000000000..07af57b0e --- /dev/null +++ b/positronic/vendors/galaxea/NOTICE @@ -0,0 +1,14 @@ +This integration is intended solely for internal, non-commercial evaluation. +It does not grant permission for production use, customer demonstrations, hosted +services, or redistribution of Galaxea materials in a commercial product. + +G0.5 is licensed under the G0.5 Community License Agreement (Non-Commercial + +Limited Patent License), not sold, Copyright © 2026 Galaxea. All rights reserved +by Galaxea. ‘Galaxea’ and related marks are trademarks of Galaxea or its affiliates. + +The upstream terms are reproduced in LICENSE-G0.5. Model code and weights remain +subject to those terms and their third-party licenses. Internal company evaluation +requires Authorized Hardware and an Authorized Image under that agreement; +confirm eligibility with Galaxea before running the model. + +The Positronic adapter does not imply endorsement or certification by Galaxea. diff --git a/positronic/vendors/galaxea/README.md b/positronic/vendors/galaxea/README.md new file mode 100644 index 000000000..faa7968a3 --- /dev/null +++ b/positronic/vendors/galaxea/README.md @@ -0,0 +1,184 @@ +# Galaxea G0.5-DROID in Positronic + +**For internal, non-commercial evaluation only.** This integration is not intended +for production, customer demonstrations, or hosted services. Galaxea's model code, +weights, and associated materials are governed by [LICENSE-G0.5](LICENSE-G0.5), +including its Authorized Hardware and Authorized Image conditions for company +evaluation. Confirm that your environment qualifies with Galaxea before running +the model. See [NOTICE](NOTICE). The adapter does not grant additional model rights. + +## Scope + +Pretrained DROID/Franka inference through Positronic's standard remote policy API. +The backend calls Galaxea's `PolicyInferencer` and returns the **entire predicted +chunk in one response**. It does not use the upstream step cache or its +`action_steps` setting. The checkpoint determines the prediction length; the +adapter neither truncates nor repeats steps. GPU and real-robot evaluation are +required to establish performance on a particular rig. + +The server-side [codec](codecs.py) handles all embodiment conversion: + +| Direction | Conversion | +| --- | --- | +| Observation | Exterior + wrist RGB from HWC to CHW, zero dummy wrist view, 7 joint angles, instruction, 15 Hz | +| Gripper state | Positronic `0=open, 1=closed` → Galaxea `1=open, 0=closed`, using `1 - grip` | +| Arm action | Absolute joint targets in radians → `JointPosition` with DROID impedance gains | +| Gripper action | Galaxea `1=open, 0=closed` → Positronic `0=open, 1=closed`, using `clip(1 - predicted_grip, 0, 1)` | +| Missing gripper prediction | Emit no new gripper command; the driver retains its last target | +| Missing arm / malformed prediction | Fail the request | + +Galaxea's DROID dataset loader flips both recorded gripper state and action with +`1 - x` before training. Its evaluation client applies the same conversion when +sending observations and receiving predictions. The model therefore represents +openness, while Positronic and DROID's robot interface represent closure. +See upstream [`DroidLerobotDataset._slice_meta_feature`](https://github.com/OpenGalaxea/GalaxeaVLA/blob/89f2322b4ad016e192437adc1a2c253b05bab246/src/g05/data/droid/droid_lerobot_dataset.py#L319). + +Galaxea's processor performs image resizing, state normalization, and action +denormalization. There is no gripper conversion on the robot client. Every step +receives its 15 Hz timestamp and the standard end-of-chunk timestamp. Positronic's +`ChunkedSchedule` executes the complete chunk before asking for a new prediction. +The backend has no per-episode action cache; closing or cancelling a session +cannot carry cached actions into another episode. + +## Docker setup + +The dedicated `positro/galaxea` image contains Galaxea's Python 3.10 environment +and a separate Positronic Python 3.13 environment. Galaxea is pinned to +`89f2322b4ad016e192437adc1a2c253b05bab246`. Its [inference dependencies](requirements-inference.txt) +are constrained to the versions in that revision's lockfile. PyTorch supplies CUDA +12.8 runtime libraries; the image uses an Ubuntu base and omits Galaxea's training +and simulation packages. Positronic installs its frozen lockfile without extras. +Model weights are mounted separately. +Inference requires a CUDA GPU; Galaxea's DROID guide estimates about 12 GB of free +GPU memory. RoboLab also needs GPU memory and graphics support. + +Build from the Positronic repository root: + +```bash +make -C docker build-galaxea +``` + +Galaxea's image targets are opt-in; aggregate image publishing does not include +this evaluation-only vendor. + +### Checkpoint access + +Download directly from [OpenGalaxea/G05](https://huggingface.co/OpenGalaxea/G05) +using your own Hugging Face account. Hugging Face grants gated model access +[to individual users](https://huggingface.co/docs/hub/models-gated), so each +user must accept Galaxea's conditions and obtain access for their account. +That access remains subject to the model license, including the hardware and +image conditions above. + +On the machine holding the Docker bind-mounted cache, log in and check the account: + +```bash +uvx hf auth login +uvx hf auth whoami +``` + +If downloading returns `403` with a request to enable public gated repositories, +edit the active token in [Hugging Face settings](https://huggingface.co/settings/tokens) +and enable **Read access to contents of all public gated repos you can access**. +The account must also have access to G05; the token permission alone does not grant it. + +Download the pinned DROID checkpoint and shared resources (about 12 GB): + +```bash +uvx hf download OpenGalaxea/G05 \ + --revision e312be81e90c56a55bcb26b57429bd39a335b449 \ + --local-dir "$HOME/.cache/galaxea/checkpoints" \ + --include 'g05-droid/*' \ + --include 'action_tokenizer.pt' \ + --include 'qwen3_5_2b_base_processor/*' \ + --include 'licenses/*' +``` + +### Start the server + +Keep `.hydra/config.yaml`, `dataset_stats.json`, and `checkpoints/model_state_dict.pt` +inside `checkpoints/g05-droid/`. Start the server on the GPU host: + +```bash +IMAGE_TAG=local docker compose -f docker/docker-compose.yml \ + run --rm --service-ports --use-aliases galaxea-server --port=8000 +``` + +The published API binds to `127.0.0.1:8000` on the Docker host. Use Docker Engine +28 or newer: older releases can expose localhost ports to the local network +([Docker reference](https://docs.docker.com/engine/network/port-publishing/)). +Keep Docker's default bridge networking without external direct routing. Access +to the host and its Compose network must remain within the authorized corporate group. + +With a remote Docker context, set `CACHE_ROOT` to the cache owner's home directory +on that host. At startup the server loads `g05-droid` by launching +[backend.py](backend.py) on the container's private `127.0.0.1:9000` endpoint. +The HTTP API, including `/api/v1/models`, becomes available after the model is ready. +Unloading the policy stops the child process. Each request runs fresh inference and returns +the full chunk. The first request can include model compilation latency; set +`--pipeline.source.infer_timeout=300` if needed. + +The `droid` pipeline places the vendor codec after the remote boundary, so existing +clients use `.remote` without Galaxea dependencies. For a source installation, +create Galaxea's `.venv` with its locked dependencies. Link or copy the complete +downloaded checkpoint bundle to `/checkpoints`. The upstream config +resolves the shared action tokenizer and processor files from this location. Start +the server in the Positronic environment with an explicit localhost bind: + +```bash +uv run --locked python -m positronic.vendors.galaxea.server \ + --host=127.0.0.1 --port=8000 \ + --pipeline.source.galaxea_root=/path/to/GalaxeaVLA \ + --pipeline.source.checkpoint_path=/path/to/GalaxeaVLA/checkpoints/g05-droid/checkpoints/model_state_dict.pt +``` + +For DROID clients on another machine, forward the API through SSH to the GPU host +(the Docker daemon's host when using a remote context). Run this on the client +machine and keep it open while evaluating: + +```bash +ssh -N -L 127.0.0.1:8000:127.0.0.1:8000 user@host +``` + +Use the existing DROID evaluation configuration in another terminal. The same +localhost URL works for clients running directly on the GPU host: + +```bash +uv run --locked positronic eval run --eval=.real.droid.pick_place \ + --policy=.remote --policy.url=localhost:8000 \ + --output_dir=/path/to/evaluation-recordings +``` + +For RoboLab, the Docker host also needs RTX graphics support. Run both services +through the same Docker context/daemon and Compose project (the same `-f` path +and, if set, `--project-name`). The server's `--use-aliases` flag registers +`galaxea-server` on their shared private bridge network. RoboLab connects directly +to that service name; its container's localhost is separate from the host. +Use a unique output directory for each run: + +```bash +IMAGE_TAG=latest docker compose -f docker/docker-compose.yml run --rm robolab-eval \ + --eval=.sim.robolab.banana_in_bowl --eval.trial_count=1 \ + --policy=.remote --policy.url=galaxea-server:8000 \ + --output_dir=s3://inference/tmp/galaxea-robolab// +``` + +The private backend protocol is `galaxea-full-chunk-v1`; connecting this adapter to +Galaxea's stock step-serving endpoint raises a protocol error. Transport timeouts +close the connection, and an in-flight result cancelled by the caller is discarded. +Reconnect after a transport failure. Fine-tuning and dataset conversion are outside +this vendor's scope. + +## Validation + +```bash +uv run --locked pytest positronic/vendors/galaxea/tests positronic/tests/test_vendor_boundary.py +``` + +CPU tests cover full-chunk conversion, the live WebSocket adapter, action timing, +gripper conventions, omitted predictions, cancellation, and subprocess ownership. +They do not establish GPU compatibility or robot performance. + +Upstream references: [DROID deployment](https://github.com/OpenGalaxea/GalaxeaVLA/tree/89f2322b4ad016e192437adc1a2c253b05bab246/experiments/droid), +[model weights](https://huggingface.co/OpenGalaxea/G05), +[inferencer](https://github.com/OpenGalaxea/GalaxeaVLA/blob/89f2322b4ad016e192437adc1a2c253b05bab246/src/g05/models/g05/inferencer.py). diff --git a/positronic/vendors/galaxea/__init__.py b/positronic/vendors/galaxea/__init__.py new file mode 100644 index 000000000..e581012d7 --- /dev/null +++ b/positronic/vendors/galaxea/__init__.py @@ -0,0 +1 @@ +"""Galaxea G0.5 integration for internal, non-commercial evaluation only. See README.md and NOTICE.""" diff --git a/positronic/vendors/galaxea/backend.py b/positronic/vendors/galaxea/backend.py new file mode 100644 index 000000000..6065a4805 --- /dev/null +++ b/positronic/vendors/galaxea/backend.py @@ -0,0 +1,91 @@ +"""Full-chunk G0.5 backend for internal, non-commercial evaluation only; see README.md and NOTICE. + +Run as galaxea.backend with Galaxea's Python 3.10 environment; see server.py for module paths. +""" + +import argparse +import logging +import threading +from pathlib import Path + +import torch +from g05.models.g05.inferencer import PolicyInferencer +from g05.utils.checkpoint.ckpt_utils import find_run_dir, load_config_from_run_dir +from g05.utils.eval.eval_utils import filter_embodiment +from g05.utils.websocket import packb, unpackb +from scripts.serve_policy import build_obs_dict, setup +from websockets.sync.server import serve + +from . import protocol + +logger = logging.getLogger(__name__) + +_ABSENT_KEYS = '_absent_keys' +_COT_TEXT = '_cot_text' + + +class ChunkBackend: + """Serialize access to the shared model; every request recomputes a complete trajectory.""" + + def __init__(self, inferencer: PolicyInferencer, processor): + self._inferencer = inferencer + self._processor = processor + self._lock = threading.Lock() + + def infer(self, obs: dict) -> dict: + with self._lock: + prediction = self._inferencer.infer([build_obs_dict(obs, self._processor)])[0] + absent = prediction.pop(_ABSENT_KEYS, set()) + prediction.pop(_COT_TEXT, None) + actions = {} + for name, value in prediction.items(): + if name in absent: + continue + if not isinstance(value, torch.Tensor) or value.ndim != 3 or value.shape[0] != 1: + raise ValueError(f'Expected {name} as a (1, T, D) tensor') + actions[name] = value[0].float().cpu().numpy() + return protocol.chunk_response(actions) + + def handle(self, connection): + connection.send(packb({protocol.PROTOCOL: protocol.FULL_CHUNK_V1})) + for message in connection: + try: + response = self.infer(unpackb(message)) + except Exception as exc: + logger.exception('G0.5 inference failed') + response = {protocol.ERROR: str(exc)} + connection.send(packb(response)) + + +_EVAL_EMBODIMENT = 'eval_embodiment' +_DISCRETE_ACTION = 'model.model_arch.discrete_action' +_CONTINUOUS_ACTION = 'model.model_arch.continuous_action' + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--checkpoint', type=Path, required=True) + parser.add_argument('--host', default='127.0.0.1') + parser.add_argument('--port', type=int, default=9000) + parser.add_argument('--device', default='cuda') + args = parser.parse_args() + checkpoint = args.checkpoint.absolute() + if not checkpoint.is_file(): + raise FileNotFoundError(checkpoint) + overrides = [ + f'{_EVAL_EMBODIMENT}={protocol.DROID_FRANKA}', + f'{_DISCRETE_ACTION}=true', + f'{_CONTINUOUS_ACTION}=false', + ] + cfg = load_config_from_run_dir(find_run_dir(str(checkpoint)), str(checkpoint), overrides) + filter_embodiment(cfg, protocol.DROID_FRANKA) + model, processor = setup(cfg, device=args.device) + backend = ChunkBackend(PolicyInferencer(model, processor, device=args.device), processor) + logger.info('G0.5 backend: internal, non-commercial evaluation only') + with serve(backend.handle, args.host, args.port, compression=None, max_size=None) as server: + server.serve_forever() + + +if __name__ == '__main__': + logging.basicConfig(level=logging.INFO) + main() diff --git a/positronic/vendors/galaxea/codecs.py b/positronic/vendors/galaxea/codecs.py new file mode 100644 index 000000000..3eab5bdb8 --- /dev/null +++ b/positronic/vendors/galaxea/codecs.py @@ -0,0 +1,82 @@ +"""Server-side DROID observation and action conversion for non-commercial G0.5 evaluation.""" + +import configuronic as cfn +import numpy as np + +from positronic import keys +from positronic.cfg import codecs +from positronic.drivers.roboarm import command +from positronic.policy.codec import ActionTimestamp, Codec +from positronic.vendors.galaxea import protocol + + +class DroidCodec(Codec): + """Map canonical 0=open, 1=closed grip to Galaxea's inverted convention in both directions. + + An absent gripper prediction emits no gripper command, preserving the driver's last target. + Arm predictions are required. The upstream processor owns resizing and normalization. + """ + + def __init__( + self, + exterior_camera: str = keys.EXTERIOR_IMAGE, + wrist_camera: str = keys.WRIST_IMAGE, + joint_key: str = keys.JOINTS, + grip_key: str = keys.GRIP, + task_key: str = keys.TASK, + fps: float = 15.0, + ): + if not np.isfinite(fps) or fps <= 0: + raise ValueError('fps must be finite and positive') + self._cameras = {protocol.EXTERIOR_IMAGE: exterior_camera, protocol.WRIST_IMAGE: wrist_camera} + self._joint_key = joint_key + self._grip_key = grip_key + self._task_key = task_key + self.fps = fps + + @staticmethod + def _vector(value, size: int, name: str) -> np.ndarray: + array = np.asarray(value, dtype=np.float32) + if size == 1 and array.ndim == 0: + array = array.reshape(1) + if array.shape != (size,) or not np.isfinite(array).all(): + raise ValueError(f'{name} must be a finite ({size},) vector, got {array.shape}') + return array + + @staticmethod + def _image(value, name: str) -> np.ndarray: + image = np.asarray(value) + if image.ndim != 3 or image.shape[-1] != 3 or image.dtype != np.uint8 or min(image.shape) == 0: + raise ValueError(f'{name} must be a nonempty uint8 HWC RGB image, got {image.shape}, {image.dtype}') + return np.ascontiguousarray(image.transpose(2, 0, 1)) + + def encode(self, data: dict) -> dict: + grip = self._vector(data[self._grip_key], 1, self._grip_key) + if np.any((grip < 0) | (grip > 1)): + raise ValueError('Observed grip must be in [0, 1]') + return { + protocol.IMAGES: { + **{name: self._image(data[key], key) for name, key in self._cameras.items()}, + protocol.DUMMY_WRIST_RIGHT: np.zeros((3, 224, 224), dtype=np.uint8), + }, + protocol.STATE: { + protocol.RIGHT_ARM: self._vector(data[self._joint_key], 7, self._joint_key), + protocol.RIGHT_GRIPPER: 1.0 - grip, + }, + protocol.TASK: data[self._task_key], + protocol.FREQUENCY: self.fps, + protocol.EMBODIMENT_TYPE: protocol.DROID_FRANKA, + } + + def _decode_single(self, data: dict) -> dict: + joints = self._vector(data[protocol.RIGHT_ARM], 7, protocol.RIGHT_ARM) + result = {keys.ROBOT_COMMAND: command.JointPosition(positions=joints)} + if protocol.RIGHT_GRIPPER in data: + grip = self._vector(data[protocol.RIGHT_GRIPPER], 1, protocol.RIGHT_GRIPPER) + result[keys.TARGET_GRIP] = float(np.clip(1.0 - grip[0], 0.0, 1.0)) + return result + + +@cfn.config(codec=cfn.Config(DroidCodec)) +def droid(codec: DroidCodec): + return ActionTimestamp(fps=codec.fps) | codecs.droid_execution(action=codec) diff --git a/positronic/vendors/galaxea/protocol.py b/positronic/vendors/galaxea/protocol.py new file mode 100644 index 000000000..cf6421f09 --- /dev/null +++ b/positronic/vendors/galaxea/protocol.py @@ -0,0 +1,35 @@ +"""Full-chunk transport contract for non-commercial G0.5 evaluation. + +Imported directly by the isolated Python 3.10 backend; no Positronic imports belong here. +Observations use Galaxea's msgpack/numpy encoding. Responses contain plain lists and scalars. +""" + +import numpy as np + +PROTOCOL = 'protocol' +FULL_CHUNK_V1 = 'galaxea-full-chunk-v1' +MODEL_ID = 'g05-droid' +ACTIONS = 'actions' +ERROR = 'error' +IMAGES = 'images' +STATE = 'state' +TASK = 'task' +FREQUENCY = 'frequency' +EMBODIMENT_TYPE = 'embodiment_type' +DROID_FRANKA = 'Droid_Franka' +EXTERIOR_IMAGE = 'exterior_image' +WRIST_IMAGE = 'wrist_image' +DUMMY_WRIST_RIGHT = 'dummy_wrist_right' +RIGHT_ARM = 'right_arm' +RIGHT_GRIPPER = 'right_gripper' + + +def chunk_response(actions: dict[str, np.ndarray]) -> dict: + """Serialize every predicted step. Each part is an unbatched (time, dimensions) array.""" + arm = actions[RIGHT_ARM] + if arm.ndim != 2 or arm.shape[0] == 0 or arm.shape[1] != 7: + raise ValueError(f'Expected a nonempty (T, 7) arm chunk, got {arm.shape}') + for name, values in actions.items(): + if values.ndim != 2 or values.shape[0] != arm.shape[0] or not np.isfinite(values).all(): + raise ValueError(f'Invalid chunk for {name}: expected {arm.shape[0]} finite steps, got {values.shape}') + return {ACTIONS: [{name: values[i].tolist() for name, values in actions.items()} for i in range(len(arm))]} diff --git a/positronic/vendors/galaxea/requirements-inference.txt b/positronic/vendors/galaxea/requirements-inference.txt new file mode 100644 index 000000000..6de889d89 --- /dev/null +++ b/positronic/vendors/galaxea/requirements-inference.txt @@ -0,0 +1,18 @@ +# Internal, non-commercial evaluation only; see NOTICE and LICENSE-G0.5. +# Versions are constrained by the pinned Galaxea checkout's exported uv.lock. +accelerate +einops +flash-attn-4 +flash-linear-attention +gitpython +huggingface-hub +hydra-core +msgpack +numpy +rich +rootutils +safetensors +torch +torchvision +transformers +websockets diff --git a/positronic/vendors/galaxea/server.py b/positronic/vendors/galaxea/server.py new file mode 100644 index 000000000..a4d70dbb4 --- /dev/null +++ b/positronic/vendors/galaxea/server.py @@ -0,0 +1,216 @@ +"""Positronic policy server for internal, non-commercial G0.5-DROID evaluation only.""" + +import os +import socket +import subprocess +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import configuronic as cfn +import msgpack +import pos3 +from websockets.sync.client import connect + +from pimm.logging import init_logging +from positronic.offboard.server import serve +from positronic.offboard.server_utils import wait_for_subprocess_ready +from positronic.policy import Codec, Policy, Session +from positronic.policy import keys as policy_keys +from positronic.policy.base import Answer, Runtime +from positronic.policy.codec import RestrictImageSize +from positronic.policy.layers import ChunkedSchedule, StopOnFault +from positronic.policy.spec import ModelSource, remote +from positronic.utils.serialization import serialize +from positronic.vendors.galaxea import codecs, protocol + + +class _GalaxeaSession(Session): + def __init__(self, url: str, timeout: float, rt: Runtime): + self._rt = rt + self._timeout = timeout + self._answer: Answer | None = None + self._cancelled = False + self._connection = connect(url, compression=None, max_size=None) + try: + handshake = msgpack.unpackb(self._connection.recv(timeout=timeout)) + if handshake.get(protocol.PROTOCOL) != protocol.FULL_CHUNK_V1: + raise ValueError('Expected the Galaxea full-chunk backend; see vendors/galaxea/README.md') + except Exception: + self._connection.close() + raise + + @staticmethod + def infer(connection, obs, timeout: float) -> list[dict[str, Any]]: + try: + connection.send(serialize(obs)) + response = msgpack.unpackb(connection.recv(timeout=timeout)) + except Exception: + connection.close() + raise + if protocol.ERROR in response: + raise RuntimeError(f'G0.5 inference failed: {response[protocol.ERROR]}') + actions = response[protocol.ACTIONS] + if not isinstance(actions, list) or not actions or any(not isinstance(step, dict) for step in actions): + raise ValueError('G0.5 must return a nonempty list of action dictionaries') + return actions + + def __call__(self, obs, time_ns): + if self._answer is None: + self._answer = self._rt.fns['infer'](self._connection, obs, self._timeout) + return None + if not self._answer.done(): + return None + answer, cancelled = self._answer, self._cancelled + self._answer, self._cancelled = None, False + actions = answer.result() + return None if cancelled else actions + + def cancel(self): + self._cancelled = self._answer is not None + + def close(self): + assert self._answer is None or self._answer.done(), 'Close the runtime before its session' + self._connection.close() + + +PYTHONPATH = 'PYTHONPATH' +VIRTUAL_ENV = 'VIRTUAL_ENV' +PATH = 'PATH' + + +class _BackendProcess: + """Own the full-chunk model process in Galaxea's isolated interpreter.""" + + def __init__(self, root: Path, checkpoint: Path, device: str, port: int): + self._root = root.resolve(strict=True) + self._checkpoint = checkpoint.absolute() + self._python = self._root / '.venv/bin/python' + for path in (self._checkpoint, self._python): + if not path.is_file(): + raise FileNotFoundError(path) + self._device = device + self._port = port + self.url = f'ws://127.0.0.1:{port}' + self._process: subprocess.Popen | None = None + + def _ready(self) -> bool: + try: + with connect(self.url, compression=None, open_timeout=1) as connection: + metadata = msgpack.unpackb(connection.recv(timeout=1)) + except (OSError, TimeoutError): + return False + if metadata.get(protocol.PROTOCOL) != protocol.FULL_CHUNK_V1: + raise ValueError('Backend port is occupied by an incompatible server') + return True + + def _crashed(self) -> tuple[bool, int | None]: + assert self._process is not None + code = self._process.poll() + return code is not None, code + + def start(self, on_progress: Callable[[str], None] | None): + # Refuse an occupied port before launching a second model into GPU memory. + with socket.socket() as probe: + probe.bind(('127.0.0.1', self._port)) + env = os.environ.copy() + env[PYTHONPATH] = os.pathsep.join((str(self._root / 'src'), str(self._root), str(Path(__file__).parents[1]))) + env[VIRTUAL_ENV] = str(self._root / '.venv') + env[PATH] = str(self._root / '.venv/bin') + os.pathsep + env.get(PATH, '') + self._process = subprocess.Popen( + [ + str(self._python), + '-m', + 'galaxea.backend', + '--checkpoint', + str(self._checkpoint), + '--device', + self._device, + '--port', + str(self._port), + ], + cwd=self._root, + env=env, + ) + wait_for_subprocess_ready(self._ready, self._crashed, 'Galaxea model', on_progress, max_wait=1800) + + def stop(self): + if self._process is None: + return + self._process.terminate() + try: + self._process.wait(timeout=10) + except subprocess.TimeoutExpired: + self._process.kill() + self._process.wait() + self._process = None + + +class GalaxeaPolicy(Policy): + def __init__(self, backend: _BackendProcess, infer_timeout: float): + self._backend = backend + self._timeout = infer_timeout + + @property + def functions(self): + return {'infer': _GalaxeaSession.infer} + + def new_session(self, context=None, rt=None): + if rt is None: + raise ValueError('GalaxeaPolicy requires a runtime for inference') + return _GalaxeaSession(self._backend.url, self._timeout, rt) + + def close(self): + self._backend.stop() + + +class GalaxeaSource(ModelSource): + """Load G0.5-DROID in its own Python 3.10 process and expose full-chunk inference.""" + + def __init__( + self, + checkpoint_path: str = '/galaxea/checkpoints/g05-droid/checkpoints/model_state_dict.pt', + galaxea_root: str = '/galaxea', + device: str = 'cuda', + backend_port: int = 9000, + infer_timeout: float = 120.0, + ): + self._checkpoint = Path(checkpoint_path) + self._root = Path(galaxea_root) + self._device = device + self._port = backend_port + self._timeout = infer_timeout + + def get_models(self) -> list[str]: + return [protocol.MODEL_ID] + + def load(self, model_id: str, on_progress: Callable[[str], None] | None = None) -> Policy: + if model_id != protocol.MODEL_ID: + raise ValueError(f'Unknown Galaxea model: {model_id}') + backend = _BackendProcess(self._root, self._checkpoint, self._device, self._port) + try: + backend.start(on_progress) + except Exception: + backend.stop() + raise + return GalaxeaPolicy(backend, self._timeout) + + def meta(self, model_id: str) -> dict[str, Any]: + return {policy_keys.CHECKPOINT_PATH: str(self._checkpoint), 'usage': 'internal non-commercial evaluation only'} + + +@cfn.config(codec=codecs.droid, source=cfn.Config(GalaxeaSource)) +def pipeline(codec: Codec, source: ModelSource): + # TODO: Add an opt-in local layer for RoboArena's missing-gripper behavior. Capture the measured + # grip per inference request, with state isolated per session, and fill omitted targets in its chunk. + # Keep preserving the previous target as the default; this state belongs in the layer, not the codec. + return StopOnFault() | ChunkedSchedule() | RestrictImageSize() | remote | codec | source + + +COMMANDS = {name: serve.override(pipeline=pipeline) for name in ('', 'serve', 'droid')} + + +if __name__ == '__main__': + init_logging() + with pos3.mirror(): + cfn.cli(COMMANDS) diff --git a/positronic/vendors/galaxea/tests/__init__.py b/positronic/vendors/galaxea/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/positronic/vendors/galaxea/tests/test_codecs.py b/positronic/vendors/galaxea/tests/test_codecs.py new file mode 100644 index 000000000..b24457c31 --- /dev/null +++ b/positronic/vendors/galaxea/tests/test_codecs.py @@ -0,0 +1,104 @@ +import numpy as np +import pytest + +from positronic import keys +from positronic.cfg.hardware.roboarm import DROID_IMPEDANCE +from positronic.vendors.galaxea import codecs, protocol + + +@pytest.fixture +def observation(): + return { + keys.JOINTS: np.arange(7, dtype=np.float32) / 10, + keys.GRIP: np.array([0.2], dtype=np.float32), + keys.EXTERIOR_IMAGE: np.arange(18, dtype=np.uint8).reshape(2, 3, 3), + keys.WRIST_IMAGE: np.full((4, 5, 3), 123, dtype=np.uint8), + keys.TASK: 'pick up the towel', + } + + +def test_observation_preserves_rgb_pixels_and_maps_state(observation): + encoded = codecs.DroidCodec().encode(observation) + np.testing.assert_array_equal( + encoded[protocol.IMAGES][protocol.EXTERIOR_IMAGE], observation[keys.EXTERIOR_IMAGE].transpose(2, 0, 1) + ) + np.testing.assert_array_equal(encoded[protocol.STATE][protocol.RIGHT_ARM], observation[keys.JOINTS]) + np.testing.assert_allclose(encoded[protocol.STATE][protocol.RIGHT_GRIPPER], [0.8]) + np.testing.assert_allclose(observation[keys.GRIP], [0.2]) + assert encoded[protocol.IMAGES][protocol.DUMMY_WRIST_RIGHT].shape == (3, 224, 224) + assert not encoded[protocol.IMAGES][protocol.DUMMY_WRIST_RIGHT].any() + assert encoded[protocol.TASK] == observation[keys.TASK] + assert encoded[protocol.EMBODIMENT_TYPE] == protocol.DROID_FRANKA + assert encoded[protocol.FREQUENCY] == 15.0 + + +@pytest.mark.parametrize('grip', [0.0, 0.2, 1.0]) +def test_gripper_round_trip_keeps_canonical_endpoints(observation, grip): + codec = codecs.DroidCodec() + observation[keys.GRIP] = grip + state = codec.encode(observation)[protocol.STATE] + action = codec.decode(state) + assert action[keys.TARGET_GRIP] == pytest.approx(grip) + np.testing.assert_array_equal(action[keys.ROBOT_COMMAND].positions, observation[keys.JOINTS]) + + +@pytest.mark.parametrize(('prediction', 'target'), [(-0.2, 1.0), (1.2, 0.0)]) +def test_gripper_predictions_are_clipped_after_inversion(prediction, target): + action = codecs.DroidCodec().decode({protocol.RIGHT_ARM: [0.0] * 7, protocol.RIGHT_GRIPPER: [prediction]}) + assert action[keys.TARGET_GRIP] == target + + +def test_missing_gripper_emits_only_arm_command(): + result = codecs.DroidCodec().decode({protocol.RIGHT_ARM: [0.0] * 7}) + assert set(result) == {keys.ROBOT_COMMAND} + + +def test_missing_arm_is_an_error(): + with pytest.raises(KeyError, match=protocol.RIGHT_ARM): + codecs.DroidCodec().decode({protocol.RIGHT_GRIPPER: [0.5]}) + + +@pytest.mark.parametrize('arm', [[0.0] * 6, [[0.0] * 7], [float('nan')] * 7, [float('inf')] * 7]) +def test_malformed_arm_predictions_fail(arm): + with pytest.raises(ValueError, match='finite'): + codecs.DroidCodec().decode({protocol.RIGHT_ARM: arm}) + + +@pytest.mark.parametrize('grip', [[float('nan')], [float('inf')], [0.0, 1.0]]) +def test_invalid_gripper_is_not_treated_as_missing(grip): + with pytest.raises(ValueError, match='finite'): + codecs.DroidCodec().decode({protocol.RIGHT_ARM: [0.0] * 7, protocol.RIGHT_GRIPPER: grip}) + + +def test_camera_keys_are_configurable(observation): + observation['front'] = observation.pop(keys.EXTERIOR_IMAGE) + observation['hand'] = observation.pop(keys.WRIST_IMAGE) + encoded = codecs.DroidCodec(exterior_camera='front', wrist_camera='hand').encode(observation) + assert encoded[protocol.IMAGES][protocol.WRIST_IMAGE].shape == (3, 4, 5) + + +@pytest.mark.parametrize('image', [np.zeros((2, 2), dtype=np.uint8), np.zeros((2, 2, 3)), np.zeros((0, 2, 3))]) +def test_invalid_camera_input_fails(observation, image): + observation[keys.WRIST_IMAGE] = image + with pytest.raises(ValueError, match='RGB image'): + codecs.DroidCodec().encode(observation) + + +@pytest.mark.parametrize('fps', [0, -1, float('nan'), float('inf')]) +def test_invalid_frequency_fails(fps): + with pytest.raises(ValueError, match='fps'): + codecs.DroidCodec(fps=fps) + + +def test_entire_chunk_is_timed_and_uses_droid_control_mode(observation): + codec = codecs.droid(codec=codecs.DroidCodec(fps=10)) + raw = [{protocol.RIGHT_ARM: [float(i)] * 7, protocol.RIGHT_GRIPPER: [i / 31]} for i in range(32)] + trajectory = codec.decode(raw) + assert len(trajectory) == 33 + for i, action in enumerate(trajectory[:-1]): + assert action[keys.ACTION_TIMESTAMP] == pytest.approx(i / 10) + np.testing.assert_array_equal(action[keys.ROBOT_COMMAND].positions, [i] * 7) + assert action[keys.ROBOT_COMMAND].mode == DROID_IMPEDANCE + assert action[keys.TARGET_GRIP] == pytest.approx(1 - i / 31, abs=1e-7) + assert trajectory[-1] == {keys.ACTION_TIMESTAMP: 3.2} + assert codec.encode(observation)[protocol.FREQUENCY] == 10 diff --git a/positronic/vendors/galaxea/tests/test_protocol.py b/positronic/vendors/galaxea/tests/test_protocol.py new file mode 100644 index 000000000..ec145b8a7 --- /dev/null +++ b/positronic/vendors/galaxea/tests/test_protocol.py @@ -0,0 +1,33 @@ +import msgpack +import numpy as np +import pytest + +from positronic.vendors.galaxea import protocol + + +@pytest.mark.parametrize('length', [1, 16, 32, 48]) +def test_chunk_transport_preserves_every_step(length): + arms = np.arange(length * 7, dtype=np.float32).reshape(length, 7) + grips = np.linspace(0, 1, length, dtype=np.float32).reshape(length, 1) + response = protocol.chunk_response({protocol.RIGHT_ARM: arms, protocol.RIGHT_GRIPPER: grips}) + decoded = msgpack.unpackb(msgpack.packb(response)) + assert len(decoded[protocol.ACTIONS]) == length + np.testing.assert_array_equal([step[protocol.RIGHT_ARM] for step in decoded[protocol.ACTIONS]], arms) + np.testing.assert_array_equal([step[protocol.RIGHT_GRIPPER] for step in decoded[protocol.ACTIONS]], grips) + + +def test_omitted_gripper_remains_omitted_on_every_step(): + response = protocol.chunk_response({protocol.RIGHT_ARM: np.zeros((32, 7))}) + assert len(response[protocol.ACTIONS]) == 32 + assert all(protocol.RIGHT_GRIPPER not in step for step in response[protocol.ACTIONS]) + + +@pytest.mark.parametrize('arms', [np.zeros((0, 7)), np.zeros((1, 32, 7)), np.zeros((32, 6)), np.full((32, 7), np.nan)]) +def test_invalid_arm_chunk_fails(arms): + with pytest.raises(ValueError): + protocol.chunk_response({protocol.RIGHT_ARM: arms}) + + +def test_mismatched_part_lengths_fail_instead_of_repeating_steps(): + with pytest.raises(ValueError, match='Invalid chunk'): + protocol.chunk_response({protocol.RIGHT_ARM: np.zeros((32, 7)), protocol.RIGHT_GRIPPER: np.zeros((16, 1))}) diff --git a/positronic/vendors/galaxea/tests/test_server.py b/positronic/vendors/galaxea/tests/test_server.py new file mode 100644 index 000000000..5fa73f32d --- /dev/null +++ b/positronic/vendors/galaxea/tests/test_server.py @@ -0,0 +1,224 @@ +import runpy +import threading +from dataclasses import dataclass, field +from unittest.mock import Mock + +import msgpack +import numpy as np +import pos3 +import pytest +from websockets.sync.server import serve + +from positronic import keys +from positronic.policy import keys as policy_keys +from positronic.policy.executor import Executor, blocking +from positronic.policy.spec import split +from positronic.utils.serialization import deserialize +from positronic.vendors.galaxea import codecs, protocol, server + + +def test_cli_supports_recording_directory(tmp_path, monkeypatch): + output = tmp_path / 'recordings' + + def record(_commands): + local = pos3.sync(str(output)) + local.mkdir(parents=True, exist_ok=True) + (local / 'recording.txt').write_text('recorded') + + monkeypatch.setattr(server.cfn, 'cli', record) + monkeypatch.setattr('pimm.logging.init_logging', lambda: None) + runpy.run_path(server.__file__, run_name='__main__') + assert (output / 'recording.txt').read_text() == 'recorded' + + +@dataclass +class Backend: + url: str = '' + requests: list = field(default_factory=list) + entered: threading.Event = field(default_factory=threading.Event) + release: threading.Event = field(default_factory=threading.Event) + metadata: dict = field(default_factory=lambda: {protocol.PROTOCOL: protocol.FULL_CHUNK_V1}) + response: dict = field( + default_factory=lambda: protocol.chunk_response({ + protocol.RIGHT_ARM: np.arange(32 * 7).reshape(32, 7), + protocol.RIGHT_GRIPPER: np.zeros((32, 1)), + }) + ) + + def stop(self): + pass + + def handle(self, connection): + connection.send(msgpack.packb(self.metadata)) + for message in connection: + self.requests.append(deserialize(message)) + self.entered.set() + assert self.release.wait(5) + connection.send(msgpack.packb(self.response)) + + +@pytest.fixture +def backend(): + backend = Backend() + backend.release.set() + with serve(backend.handle, '127.0.0.1', 0) as websocket_server: + backend.url = f'ws://127.0.0.1:{websocket_server.socket.getsockname()[1]}' + thread = threading.Thread(target=websocket_server.serve_forever) + thread.start() + try: + yield backend + finally: + backend.release.set() + websocket_server.shutdown() + thread.join(timeout=5) + assert not thread.is_alive() + + +def test_live_adapter_returns_full_chunk_with_server_side_gripper_conversion(backend): + codec = codecs.droid() + policy = codec.wrap(server.GalaxeaPolicy(backend, 5)) + session = blocking(policy).new_session() + obs = { + keys.JOINTS: np.zeros(7), + keys.GRIP: 0.3, + keys.EXTERIOR_IMAGE: np.zeros((2, 3, 3), dtype=np.uint8), + keys.WRIST_IMAGE: np.zeros((2, 3, 3), dtype=np.uint8), + keys.TASK: 'pick towel', + } + try: + assert session.meta == codec.meta + actions = session(obs, 0) + assert len(actions) == 33 + assert actions[-1] == {keys.ACTION_TIMESTAMP: 32 / 15} + np.testing.assert_array_equal(actions[-2][keys.ROBOT_COMMAND].positions, np.arange(217, 224)) + assert all(step[keys.TARGET_GRIP] == 1 for step in actions[:-1]) + assert len(backend.requests) == 1 + np.testing.assert_allclose(backend.requests[0][protocol.STATE][protocol.RIGHT_GRIPPER], [0.7]) + finally: + session.close() + + +def test_stock_step_server_is_rejected(backend): + backend.metadata = {'action_steps': 16} + policy = blocking(server.GalaxeaPolicy(backend, 5)) + with pytest.raises(ValueError, match='full-chunk backend'): + policy.new_session() + + +@pytest.mark.parametrize('response', [{protocol.ERROR: 'missing arm'}, {protocol.ACTIONS: []}, {protocol.ACTIONS: [0]}]) +def test_backend_errors_surface(backend, response): + backend.response = response + session = blocking(server.GalaxeaPolicy(backend, 5)).new_session() + try: + with pytest.raises((RuntimeError, ValueError)): + session({}, 0) + finally: + session.close() + + +def test_cancellation_discards_pending_chunk_and_next_call_recomputes(backend): + policy = server.GalaxeaPolicy(backend, 5) + rt = Executor(policy.functions) + session = policy.new_session(rt=rt) + backend.release.clear() + try: + assert session({protocol.TASK: 'old'}, 0) is None + assert backend.entered.wait(5) + session.cancel() + backend.release.set() + rt.wait(timeout=5) + assert session({protocol.TASK: 'new'}, 1) is None + assert not rt.owes_an_answer + assert session({protocol.TASK: 'new'}, 2) is None + rt.wait(timeout=5) + assert len(session({}, 3)) == 32 + assert [obs[protocol.TASK] for obs in backend.requests] == ['old', 'new'] + finally: + backend.release.set() + rt.close() + session.close() + + +def test_inference_timeout_closes_connection(): + class Connection: + closed = False + + def send(self, data): + pass + + def recv(self, timeout): + raise TimeoutError('late response') + + def close(self): + self.closed = True + + connection = Connection() + with pytest.raises(TimeoutError): + server._GalaxeaSession.infer(connection, {}, 1) + assert connection.closed + + +def test_vendor_codec_stays_on_server_side(): + pipeline = server.pipeline() + local, border, remote_half = split(pipeline) + assert isinstance(pipeline.source, server.GalaxeaSource) + assert remote_half is not None + assert 'DroidCodec' not in str(local.to_spec()) + assert ( + remote_half.encode({ + keys.JOINTS: np.zeros(7), + keys.GRIP: 0, + keys.EXTERIOR_IMAGE: np.zeros((1, 1, 3), dtype=np.uint8), + keys.WRIST_IMAGE: np.zeros((1, 1, 3), dtype=np.uint8), + keys.TASK: '', + })[protocol.STATE][protocol.RIGHT_GRIPPER] + == 1 + ) + + +def test_model_load_failure_stops_child(monkeypatch): + backend = Mock() + backend.start.side_effect = RuntimeError('model failed to load') + monkeypatch.setattr(server, '_BackendProcess', Mock(return_value=backend)) + with pytest.raises(RuntimeError, match='model failed to load'): + server.GalaxeaSource().load(protocol.MODEL_ID) + backend.stop.assert_called_once() + + +def test_loaded_policy_metadata_and_cleanup(tmp_path, monkeypatch): + backend = Mock() + monkeypatch.setattr(server, '_BackendProcess', Mock(return_value=backend)) + progress = Mock() + checkpoint = tmp_path / 'model_state_dict.pt' + source = server.GalaxeaSource(checkpoint_path=str(checkpoint)) + policy = source.load(protocol.MODEL_ID, progress) + assert source.meta(protocol.MODEL_ID)[policy_keys.CHECKPOINT_PATH] == str(checkpoint) + backend.start.assert_called_once_with(progress) + backend.stop.assert_not_called() + policy.close() + backend.stop.assert_called_once() + + +def test_subprocess_preserves_venv_and_checkpoint_symlinks(tmp_path, monkeypatch): + executable = tmp_path / 'python' + executable.touch() + interpreter = tmp_path / '.venv/bin/python' + interpreter.parent.mkdir(parents=True) + interpreter.symlink_to(executable) + weights = tmp_path / 'blob' + weights.touch() + checkpoint = tmp_path / 'model_state_dict.pt' + checkpoint.symlink_to(weights) + popen = Mock() + monkeypatch.setattr(server.subprocess, 'Popen', popen) + monkeypatch.setattr(server, 'wait_for_subprocess_ready', Mock()) + backend = server._BackendProcess(tmp_path, checkpoint, 'cuda', 0) + backend.start(None) + command = popen.call_args.args[0] + assert command[0] == str(interpreter) + assert command[command.index('--checkpoint') + 1] == str(checkpoint) + assert popen.call_args.kwargs['cwd'] == tmp_path + assert popen.call_args.kwargs['env'][server.VIRTUAL_ENV] == str(tmp_path / '.venv') + backend.stop() + popen.return_value.terminate.assert_called_once() + popen.return_value.wait.assert_called_once_with(timeout=10)