Reach daemons inside an environment: ate-env-api port proxy, NeMo Gym endpoint()

README.md

@@ -266,6 +266,31 @@ Provides chunked streaming file reading and writing within the environment conta
 | `WriteFile` | Streams raw binary or text chunks directly to a target file |
 
 
+## Reaching a service inside an environment
+
+A daemon that listens on a port inside an environment, such as one a runtime layer starts next to
+the guest ([docs/rootfs-images/RUNTIMES.md](docs/rootfs-images/RUNTIMES.md)), is reachable through
+the API's listener. A plain HTTP request that carries `X-Env-Id` and `X-Env-Port` (and
+`X-Env-Atespace` outside `ate-env`) is forwarded through the Substrate router to that port, whatever
+its path; the API strips the three headers. The router resumes a suspended environment first.
+
+```bash
+curl -H 'X-Env-Id: dev1' -H 'X-Env-Port: 6000' localhost:7777/health
+```
+
+Clients that cannot set headers, such as WebSocket clients and browsers, put the target in the path
+instead: `/v1alpha/envs/{id}/ports/{port}/...`, or `/v1alpha/atespaces/{atespace}/envs/{id}/ports/{port}/...`
+outside `ate-env`. The API strips that prefix and forwards the rest, so a client given
+`http://localhost:7777/v1alpha/envs/dev1/ports/6000` as its base URL works unchanged. WebSocket upgrades pass
+through either form; gRPC to the daemon is not supported yet.
+
+```bash
+curl localhost:7777/v1alpha/envs/dev1/ports/6000/health
+```
+
+`--atenet-connect` names the router's CONNECT listener (default
+`atenet-router.ate-system.svc.cluster.local:8081`); an empty value turns the forwarding off.
+
 ## Built-in MCP Server
 
 The API exposes a streamable MCP endpoint at `POST /v1alpha/envs/{id}/mcp` serving built-in tools.

cmd/ate-env-api/main.go

@@ -29,6 +29,7 @@ import (
 	"github.com/agent-substrate/env/internal/apiservice"
 	"github.com/agent-substrate/env/internal/ate"
 	"github.com/agent-substrate/env/internal/mcp"
+	"github.com/agent-substrate/env/internal/portproxy"
 	ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha"
 	"google.golang.org/grpc"
 )
@@ -38,6 +39,7 @@ func main() {
 		listen     = flag.String("listen", "0.0.0.0:7777", "address to serve the API on")
 		ateapi     = flag.String("ateapi", ate.DefaultControlAddr, "address of the ateapi gRPC control plane")
 		atenet     = flag.String("atenet", ate.DefaultRouterAddr, "address of the atenet HTTP router")
+		atenetConn = flag.String("atenet-connect", ate.DefaultRouterConnectAddr, "address of the atenet router's plaintext CONNECT listener; HTTP requests carrying X-Env-Id and X-Env-Port are forwarded through it to that port in the environment. Empty disables the port proxy")
 		hostSuffix = flag.String("host-suffix", ate.DefaultHostSuffix, "atenet router host suffix for actor routing")
 		skipVerify = flag.Bool("skip-verify", true, "skip TLS certificate verification on the control plane connection")
 		tokenFile  = flag.String("ateapi-token-file", "", "file with a bearer token for the control plane, e.g. a projected ServiceAccount token with the ateapi audience")
@@ -82,6 +84,19 @@ func main() {
 	mux.Handle("/v1alpha/envs/{id}/mcp", mcp.NewHandler(client))
 	mux.Handle("/", grpcServer)
 
+	// Requests that name an environment port go to the port proxy before
+	// anything else, whatever their path: the path belongs to the daemon.
+	var handler http.Handler = mux
+	if *atenetConn != "" {
+		pp := portproxy.New(portproxy.Options{
+			ConnectAddr:     *atenetConn,
+			HostSuffix:      *hostSuffix,
+			DefaultAtespace: ate.DefaultAtespace,
+		})
+		pp.Register(mux) // the path form, /v1alpha/envs/{id}/ports/{port}/...
+		handler = portproxy.Dispatch(pp, mux)
+	}
+
 	// Enable both HTTP/1.1 (for /healthz and /v1alpha/ HTTP endpoints) and
 	// unencrypted HTTP/2 (h2c, for gRPC services) on the same plaintext TCP listener.
 	var protocols http.Protocols
@@ -90,7 +105,7 @@ func main() {
 
 	srv := &http.Server{
 		Addr:      *listen,
-		Handler:   mux,
+		Handler:   handler,
 		Protocols: &protocols,
 	}
 
@@ -103,7 +118,7 @@ func main() {
 		_ = srv.Shutdown(context.Background())
 	}()
 
-	log.Printf("ate-env-api listening on %s (ateapi %s, atenet %s)", *listen, *ateapi, *atenet)
+	log.Printf("ate-env-api listening on %s (ateapi %s, atenet %s, atenet CONNECT %q)", *listen, *ateapi, *atenet, *atenetConn)
 	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		log.Fatalf("server error: %v", err)
 	}

docs/rootfs-images/DESIGN.md

@@ -193,10 +193,10 @@ one golden snapshot per image. Neither is reclaimed automatically.
   with distinct images, not with environments. The plan is a last-used
   annotation updated on each create, collection by age or by last use, and
   the golden snapshot removed alongside.
-- **Reaching a second runtime from outside.** Layers and sidecars put a
-  runtime such as OpenSandbox's `execd` inside the actor and gate readiness on
-  it; an endpoint lookup on `ate-env-api` that names the runtime's port
-  through the router is what an external SDK still needs.
+- **gRPC and raw TCP to a second runtime.** `ate-env-api` forwards HTTP to a
+  runtime's port through the router's CONNECT listener (see
+  [RUNTIMES.md](RUNTIMES.md#who-can-reach-the-second-runtime)); a runtime that
+  speaks gRPC or a non-HTTP protocol needs an h2c upstream or a raw tunnel.
 - **Per-create overrides** for resources and workspace, so one base can serve
   images with different needs.
 - **Image validation at create time**, so a bad reference fails the call

docs/rootfs-images/RUNTIMES.md

@@ -84,14 +84,29 @@ sidecar, run under two different rootfs images.
   away. An agent framework whose tool calls run inside the sandbox is served.
 - **The guest's own APIs** are unaffected: exec, files and MCP keep going
   through `ate-env-api` on port 80.
-- **Clients outside the environment** currently reach only port 80. The atenet
-  router forwards to the actor's port 80 unless the request is a CONNECT whose
-  authority names another port, and `ate-env-api` proxies only the guest. An
-  SDK that speaks HTTP to the second runtime therefore needs one of two things,
-  neither of which is in this branch: an endpoint lookup on `ate-env-api` that
-  returns a router address naming the runtime's port, or a reverse proxy. This
-  is the next step for the OpenSandbox integration, whose lifecycle server
-  resolves sandbox endpoints exactly this way.
+- **Clients outside the environment** reach it through `ate-env-api`. A plain
+  HTTP request to the API's listener that carries `X-Env-Id`, `X-Env-Port` and
+  optionally `X-Env-Atespace` goes to that port in that environment, whatever
+  its path: `ate-env-api` opens a CONNECT tunnel through the atenet router's
+  CONNECT listener (`--atenet-connect`, port 8081 on a stock install), the
+  router resumes the actor if needed and forwards to the port, and the request
+  and its streamed response pass through unchanged apart from the routing
+  headers, which are stripped. Idle tunnels are reused, one pool for each
+  environment and port.
+
+  ```bash
+  curl -H 'X-Env-Id: py1' -H 'X-Env-Port: 44772' localhost:7777/ready
+  curl localhost:7777/v1alpha/envs/py1/ports/44772/ready   # the same, for clients without headers
+  ```
+
+  The path form, `/v1alpha/envs/{id}/ports/{port}/...` (or
+  `/v1alpha/atespaces/{atespace}/envs/...`), is for clients that cannot set
+  headers: the prefix is stripped, so it works as a base URL. The tunnel
+  carries HTTP/1.1, not raw TCP. Streamed responses are flushed as they arrive
+  and WebSocket upgrades pass through; gRPC to the runtime is not supported
+  yet. The NeMo Gym
+  provider hands this out as its `endpoint(handle, port)`
+  ([integrations/nemo-gym](../../integrations/nemo-gym/README.md)).
 
 ## Suspend, resume and restarts
 

integrations/nemo-gym/README.md

@@ -204,6 +204,75 @@ To share one `ate_env.Client` between the provider and other code, pass it in:
 
 Unknown keys are rejected at `create()` time, as are unknown keys anywhere in the config block.
 
+### Services inside the sandbox: an injected daemon
+
+Some NeMo components talk to a service inside the sandbox instead of running commands through
+`exec`: an exec server, a code-execution sandbox server, a tool daemon. The task images do not
+contain it, and rebuilding hundreds of thousands of SWE-bench images to add it is not an option.
+ate-env attaches such a daemon at runtime the same way it attaches its own guest: as a read-only
+OCI image volume, a *runtime layer*, that the guest starts and supervises next to itself
+([docs/rootfs-images/RUNTIMES.md](../../docs/rootfs-images/RUNTIMES.md)). The provider implements
+nemo-gym 0.7's `SupportsSandboxEndpoint`, so Gym reaches the daemon from outside the cluster's
+actor network.
+
+1. **Put the daemon on the base template, once.** Every environment created from this base with a
+   task image (`rootfs_image`) gets the daemon; the task images stay unmodified.
+
+   ```bash
+   ate-env manifest template --template gym-base \
+     --layer          nemo=<registry>/nemo-daemon@sha256:<digest>=/opt/nemo \
+     --sidecar        '/opt/nemo/daemon --port 6000' \
+     --sidecar-readyz http://127.0.0.1:6000/health \
+     --snapshots-bucket gs://<bucket>/ate-env/ | kubectl-ate create actor-template -f -
+   ```
+
+   The actor reports ready only once the daemon answers its readiness URL, so `create()` returns
+   a sandbox whose daemon is up.
+
+2. **Select the base and declare the port.**
+
+   ```yaml
+   sandbox:
+     substrate:
+       connection:
+         api_url: ate-env-api.ate-env:7777
+         # endpoint_url: http://ate-env-api.ate-env:7777   # default: api_url
+       create:
+         template: gym-base
+   ```
+
+   ```python
+   handle = await provider.create(SandboxSpec(image=task_image, ports=(6000,)))
+   ```
+
+3. **Resolve the endpoint and call the daemon.**
+
+   ```python
+   ep = await provider.endpoint(handle, 6000)
+   async with httpx.AsyncClient(base_url=ep.endpoint, headers=ep.headers) as http:
+       r = await http.post("/execute", json={"generated_code": "print(1)", "language": "python"})
+   ```
+
+   `ep.endpoint` is ate-env-api, and `ep.headers` (`X-Env-Id`, `X-Env-Atespace`, `X-Env-Port`)
+   route each request through the Substrate router to port 6000 in this sandbox. `endpoint()`
+   wakes a parked sandbox before it returns. A port that `SandboxSpec.ports` did not declare is
+   a `ValueError`, as in Gym's Docker provider. Set `connection.endpoint_url` when the code that
+   calls the daemon reaches ate-env-api at another address than the provider does.
+
+Limits:
+
+- **HTTP/1.1 only.** The router tunnels HTTP to the port, not raw TCP. Streamed responses are
+  flushed as they arrive and WebSocket upgrades pass through; gRPC to the daemon is not supported
+  yet.
+- **The daemon must be self-contained.** A layer cannot rely on the task image's libraries: ship a
+  static binary, or put the interpreter and its packages in the layer. NeMo Skills' sandbox server
+  (Flask on CPython, port 6000, `POST /execute`, `GET /health`) fits only with a relocatable Python
+  inside its layer; Gym's ECS exec server needs `python3` in the task image.
+- Port 80 belongs to the guest. The daemon shares the task image's user, environment and
+  filesystem.
+- Each distinct task image yields one derived ActorTemplate on its base, created on first use and
+  reused after; the daemon adds no templates.
+
 ## Testing
 
 ```bash

integrations/nemo-gym/src/nemo_gym_substrate/_compat.py

@@ -82,10 +82,41 @@ class SandboxCreateVerificationError(SandboxCreateError):
         pass
 
 
+# SandboxEndpoint is newer than the rest (nemo-gym 0.7): an installed 0.5
+# has the base module without it, so it falls back on its own.
+try:  # pragma: no cover - exercised only with nemo-gym >= 0.7
+    from nemo_gym.sandbox.providers.base import SandboxEndpoint  # type: ignore
+except ImportError:  # pragma: no cover - the in-repo test path and nemo-gym 0.5
+    from collections.abc import Mapping as _Mapping
+    from dataclasses import dataclass as _dataclass
+    from dataclasses import field as _field
+    from urllib.parse import urlsplit as _urlsplit
+
+    @_dataclass(frozen=True)
+    class SandboxEndpoint:  # type: ignore[no-redef]
+        endpoint: str
+        headers: dict[str, str] = _field(default_factory=dict)
+
+        def __post_init__(self) -> None:
+            if not isinstance(self.endpoint, str) or not self.endpoint.strip():
+                raise ValueError("Sandbox endpoint must be a non-empty absolute URL")
+            endpoint = self.endpoint.strip()
+            parsed = _urlsplit(endpoint)
+            if not parsed.scheme or not parsed.netloc:
+                raise ValueError("Sandbox endpoint must be a non-empty absolute URL")
+            if not isinstance(self.headers, _Mapping):
+                raise TypeError("Sandbox endpoint headers must be a mapping")
+            object.__setattr__(self, "endpoint", endpoint)
+            object.__setattr__(
+                self, "headers", {str(k): str(v) for k, v in self.headers.items()}
+            )
+
+
 __all__ = [
     "NEMO_GYM_AVAILABLE",
     "SandboxCreateError",
     "SandboxCreateVerificationError",
+    "SandboxEndpoint",
     "SandboxExecResult",
     "SandboxHandle",
     "SandboxSpec",

integrations/nemo-gym/src/nemo_gym_substrate/provider.py

@@ -46,6 +46,7 @@
 from ._compat import (
     SandboxCreateError,
     SandboxCreateVerificationError,
+    SandboxEndpoint,
     SandboxExecResult,
     SandboxHandle,
     SandboxSpec,
@@ -69,6 +70,12 @@
 # Exit code the guest reports for a process group it killed with SIGKILL.
 _SIGKILL_EXIT = 137
 
+# Routing headers of ate-env-api's port proxy (internal/portproxy): a plain
+# HTTP request carrying them is forwarded to that port inside the environment.
+_ENV_ID_HEADER = "X-Env-Id"
+_ENV_ATESPACE_HEADER = "X-Env-Atespace"
+_ENV_PORT_HEADER = "X-Env-Port"
+
 # Env lifecycle status -> Gym status. Parked (suspended / paused) actors report
 # RUNNING: the router resumes them on the next guest call, which is what a
 # rollout loop wants to see (see README, "Resource mapping").
@@ -116,17 +123,29 @@ class ConnectionConfig:
     api_url: str = "http://127.0.0.1:7777"
     # Deadline for each lifecycle RPC (create / get / delete).
     request_timeout_s: float = 30.0
+    # Base URL that endpoint() hands out for services inside a sandbox. Unset
+    # means api_url as an http(s) URL: ate-env-api serves its port proxy on the
+    # same listener. Set it when the code that calls the service reaches
+    # ate-env-api at another address than the provider does.
+    endpoint_url: str | None = None
 
-    _ALLOWED = frozenset({"api_url", "request_timeout_s"})
+    _ALLOWED = frozenset({"api_url", "request_timeout_s", "endpoint_url"})
 
     @classmethod
     def from_mapping(cls, options: Mapping[str, Any]) -> "ConnectionConfig":
         _require_keys(options, cls._ALLOWED, "sandbox.substrate.connection")
+        endpoint_url = options.get("endpoint_url")
         return cls(
             api_url=str(options.get("api_url", cls.api_url)).rstrip("/"),
             request_timeout_s=float(options.get("request_timeout_s", cls.request_timeout_s)),
+            endpoint_url=None if endpoint_url is None else str(endpoint_url).rstrip("/"),
         )
 
+    def endpoint_base(self) -> str:
+        if self.endpoint_url:
+            return self.endpoint_url
+        return self.api_url if "://" in self.api_url else f"http://{self.api_url}"
+
 
 @dataclass(frozen=True)
 class CreateConfig:
@@ -256,6 +275,9 @@ async def create(self, spec: SandboxSpec) -> SandboxHandle:
                 "atespace": atespace,
                 "workdir": spec.workdir,
                 "env": dict(spec.env or {}),
+                # Ports a service inside the sandbox listens on (nemo-gym
+                # >= 0.7); endpoint() resolves only these.
+                "ports": [int(p) for p in (getattr(spec, "ports", None) or ())],
             },
         )
 
@@ -347,6 +369,35 @@ async def status(self, handle: SandboxHandle) -> SandboxStatus:
             return SandboxStatus.UNKNOWN
         return _STATUS_MAP.get(info.status, SandboxStatus.UNKNOWN)
 
+    async def endpoint(self, handle: SandboxHandle, port: int) -> SandboxEndpoint:
+        """Resolve a declared service port to a caller-reachable endpoint
+        (nemo-gym's ``SupportsSandboxEndpoint``).
+
+        The service is a daemon inside the sandbox, typically one a runtime
+        layer on the base template mounts into the task image and the guest
+        starts next to itself (see README). The endpoint is ate-env-api's port
+        proxy; its headers route each plain HTTP request through the Substrate
+        router to ``port`` in this sandbox. The sandbox is woken first, so the
+        endpoint answers at once even after Substrate parked it.
+        """
+        raw = handle.raw or {}
+        declared = list(raw.get("ports") or ())
+        if port not in declared:
+            raise ValueError(
+                f"Sandbox port {port} was not declared in SandboxSpec.ports; "
+                f"declared ports: {declared!r}"
+            )
+        env = self._env(handle)
+        await self._wait_ready(env, self._create.ready_timeout_s, error=RuntimeError)
+        return SandboxEndpoint(
+            endpoint=self._connection.endpoint_base(),
+            headers={
+                _ENV_ID_HEADER: handle.sandbox_id,
+                _ENV_ATESPACE_HEADER: self._atespace_of(raw),
+                _ENV_PORT_HEADER: str(port),
+            },
+        )
+
     async def close(self, handle: SandboxHandle) -> None:
         await self._best_effort_delete(handle.sandbox_id, self._atespace_of(handle.raw or {}))
 
@@ -401,10 +452,17 @@ def _resolve_template(
             return self._create.template, spec.image
         return self._create.template, None
 
-    async def _wait_ready(self, env: Env, timeout_s: float) -> None:
+    async def _wait_ready(
+        self,
+        env: Env,
+        timeout_s: float,
+        *,
+        error: type[Exception] = SandboxCreateVerificationError,
+    ) -> None:
         """Poll a trivial command until the guest answers it. A fresh actor may
         still be booting, and a guest call made before it serves fails with a
-        transport-level error; both are retried until the deadline."""
+        transport-level error; both are retried until the deadline, then
+        ``error`` is raised. The first call also resumes a parked actor."""
         deadline = time.monotonic() + timeout_s
         last_error = "no probe attempted"
         while True:
@@ -420,9 +478,7 @@ async def _wait_ready(self, env: Env, timeout_s: float) -> None:
             if time.monotonic() >= deadline:
                 break
             await asyncio.sleep(self._create.ready_poll_interval_s)
-        raise SandboxCreateVerificationError(
-            f"substrate env {env.id!r} not ready after {timeout_s}s: {last_error}"
-        )
+        raise error(f"substrate env {env.id!r} not ready after {timeout_s}s: {last_error}")
 
     @staticmethod
     async def _collect(process: Process) -> SandboxExecResult: