diff --git a/Makefile b/Makefile
index 67f348d..ef2c1f4 100644
--- a/Makefile
+++ b/Makefile
@@ -43,6 +43,12 @@ python-protos:
 python-test:
 	cd clients/python && python3 -m pytest
 
+# Run the NeMo Gym provider tests (requires an environment with the client and
+# the provider's test extra installed: pip install clients/python -e
+# 'integrations/nemo-gym[test]').
+nemo-gym-test:
+	cd integrations/nemo-gym && python3 -m pytest
+
 images:
 	@echo "Building and pushing container images to $(ATE_ENV_IMAGE_REPO)..."
 	@guest_img=$$(KO_DOCKER_REPO=$(ATE_ENV_IMAGE_REPO)/ate-env-guest ko build --bare ./cmd/ate-env-guest | tail -n 1); \
diff --git a/README.md b/README.md
index 67978ef..477b88d 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,7 @@ while this project adds the environment-shaped API on top.
 - **`cmd/ate-env-guest`** — The daemon server running inside each actor serving command executions, file read/write, and built-in MCP tools.
 - **`clients/go`** — The Go client library to manage environments, run commands, and perform file operations.
 - **`clients/python`** — The async Python client library ([README](clients/python/README.md)).
+- **`integrations/nemo-gym`** — A [NeMo Gym](https://github.com/NVIDIA-NeMo/Gym) sandbox provider that runs rollout sandboxes as environments, built on the Python client ([README](integrations/nemo-gym/README.md)).
 
 ## Installation
 
@@ -251,6 +252,8 @@ For complete runnable Go programs:
 - **MCP**: See [mcp](examples/mcp/main.go) to connect to an environment's MCP endpoint, discover tools, and execute tool calls.
 - **Guest Daemon**: See [guest-daemon](examples/guest-daemon/main.go) to run a standalone in-actor gRPC service for asynchronous process execution and chunked file transfer.
 
+For using environments from a framework, see [integrations/nemo-gym](integrations/nemo-gym/README.md): a NeMo Gym sandbox provider registered through the `nemo_gym.sandbox_providers` entry point.
+
 ## Cleanup
 
 ```bash
diff --git a/integrations/nemo-gym/.gitignore b/integrations/nemo-gym/.gitignore
new file mode 100644
index 0000000..e05d4ab
--- /dev/null
+++ b/integrations/nemo-gym/.gitignore
@@ -0,0 +1,6 @@
+*.egg-info/
+__pycache__/
+.pytest_cache/
+.venv/
+build/
+dist/
diff --git a/integrations/nemo-gym/README.md b/integrations/nemo-gym/README.md
new file mode 100644
index 0000000..1498d89
--- /dev/null
+++ b/integrations/nemo-gym/README.md
@@ -0,0 +1,296 @@
+# NeMo Gym sandbox provider for Agent Substrate
+
+Runs [NeMo Gym](https://github.com/NVIDIA-NeMo/Gym) rollout sandboxes as **Substrate actors**,
+fronted by the `ate-env` API of this repo and reached through its Python client,
+[`ate_env`](../../clients/python).
+Registers as provider `substrate` next to the built-ins (Docker, Daytona, ECS Fargate, Enroot,
+OpenShell, OpenSandbox, Apptainer) via the `nemo_gym.sandbox_providers` entry point — **no changes
+to NeMo Gym required**, and no changes to your resource servers beyond the `sandbox:` config block.
+
+Why run rollouts on Substrate:
+
+- **Idle sandboxes hold no worker.** A sandbox that is waiting on model inference is suspended by
+  Substrate; the atenet router resumes it transparently on the next command. A rollout fleet is
+  mostly waiting, so effective density is bounded by the *active* set, not the fleet size.
+- **Repeat sandboxes start from golden snapshots.** The first sandbox on an image pays the boot;
+  later ones restore a memory snapshot instead of booting.
+- **Prepared states fork.** A sandbox that finished expensive task setup can be snapshotted and
+  branched into N independent rollouts (best-of-N from a mid-episode state).
+
+Verified against **nemo-gym 0.5.0**: the registry resolves this provider through the entry point
+(`create_provider({"substrate": {...}})`) and the full unit suite passes against the real
+`nemo_gym.sandbox.providers.base` types.
+
+## How it maps onto ate-env
+
+One gRPC endpoint, `ate-env-api`, carries everything (see the
+[client README](../../clients/python/README.md#how-it-works)):
+
+| Provider call | `ate_env` call | Where it lands |
+|---|---|---|
+| `create(spec)` | `Client.create(id, template_name=…, template_atespace=…)`, then `Env.shell("true")` until it answers, then `Env.write_file` for `spec.files` | `EnvironmentService` → Substrate control plane; the probe and the files go to the guest |
+| `exec(handle, cmd, cwd=, env=, timeout_s=)` | `Env.start_process(["sh","-c",cmd], cwd, env)` + `stream_outputs(follow=True)` + `wait` | `ProcessService` in `ate-env-guest`, proxied through the atenet router |
+| `upload_file` / `download_file` | `Env.write_file` / `Env.read_file_bytes` | `FileSystemService` in the guest |
+| `status(handle)` | `Client.get(id)` → lifecycle status | `EnvironmentService`; does **not** wake a parked sandbox |
+| `close(handle)` | `Client.delete(id)`, NOT_FOUND-safe | `EnvironmentService` |
+
+Environment ids are `gym-<10 hex>`; each handle remembers the atespace it was created in.
+
+## Prerequisites
+
+On the cluster:
+
+1. **Agent Substrate installed** (`ate-system` namespace healthy — api-server, controller,
+   atelet, atenet-router, valkey). See `hack/install-ate.sh` in the
+   [substrate repo](https://github.com/agent-substrate/substrate). The guest data
+   plane is gRPC through the router, which needs a substrate at or after
+   [substrate#1183](https://github.com/agent-substrate/substrate/pull/1183) (h2 on the ingress,
+   protocol mirrored to actors). Without it every exec and file call fails with
+   `server closed the stream without sending trailers`.
+2. **The `ate-env` system deployed** from this repo at or after the `ateenv.v1alpha` proto
+   move: namespace, WorkerPool, ActorTemplate(s), and the `ate-env-api` service (see the
+   [root README](../../README.md#quickstart)).
+
+   ```bash
+   ate-env manifest \
+     --guest-image  <ate-env-guest image> \
+     --api-image    <ate-env-api image> \
+     --ateom-image  <ateom-gvisor image matching your substrate build> \
+     --snapshots-bucket gs://<bucket>/ate-env/ | kubectl apply -f -
+   kubectl get pods -n ate-env   # api + warm workers Running
+   ```
+
+   If your `ateapi` requires authentication (it does on any standard install), the
+   `ate-env-api` Deployment needs a projected ServiceAccount token with audience
+   `api.ate-system.svc` and the `-ateapi-token-file` flag pointing at it.
+3. **An ActorTemplate per task image**, pre-provisioned in the `ate-env` atespace (the substrate
+   analog of "the image is available"). `ate-env manifest` creates `default-template`; add one
+   template per additional image and map them under `create.image_templates` below.
+
+   > **Every template must run `ate-env-guest` inside the actor.** Readiness, exec, file
+   > transfer, even `create()` completing — all of it goes through the guest's gRPC services
+   > behind the router. A template built from an unmodified task image (no guest) fails
+   > *completely*: the actor starts, but every guest call fails and `create()` times out with
+   > `SandboxCreateVerificationError`. See "Task images" for the two ways to get the guest in.
+   > Substrate also requires template images to be digest-pinned.
+4. **Network path from wherever Gym's rollout workers run** to `ate-env-api`: in-cluster DNS
+   (`ate-env-api.ate-env:7777`) or, for a workstation, a port-forward:
+
+   ```bash
+   kubectl port-forward -n ate-env svc/ate-env-api 7777:7777
+   ```
+
+Locally:
+
+5. **Python ≥ 3.10** and the `ate-env-client` package from this repo's `clients/python` (not on
+   PyPI yet). `nemo-gym` itself is only needed on the machine running the Gym resource servers —
+   this package's tests run without it (see Testing).
+
+## Task images
+
+The guest is a static Go binary. Two ways to run it on top of an arbitrary task image:
+
+**Mode B — mount the guest as an OCI image volume (preferred).** Substrate mounts a second image
+into the actor, so the task image stays unmodified and digest-pinned as published. One
+`ActorTemplate` per task image:
+
+```yaml
+apiVersion: ate.dev/v1alpha1
+kind: ActorTemplate
+metadata: { name: py-task, namespace: ate-env }
+spec:
+  containers:
+  - name: main
+    image: python@sha256:<task image digest>
+    command: ["/bin/sh", "-c", "exec /ate/ko-app/ate-env-guest -listen :80 -workspace /workspace"]
+    wakeupProbe: { httpGet: { path: /readyz, port: 80 } }
+    volumeMounts: [{ name: guest, mountPath: /ate }]
+  volumes:
+  - name: guest
+    image: { reference: <ate-env-guest image>@sha256:<digest> }
+  sandboxClass: gvisor
+  snapshotsConfig: { location: "gs://$BUCKET/ate-env/" }
+  workerSelector: { matchLabels: { workload: default-env } }
+```
+
+This is the shape the SWE-500 benchmark uses for 500 unmodified SWE-bench images
+(`sandbox-rl-performance/substrate/drivers/run_swe500.py`).
+
+**Mode A — bake the guest into the image.** For clusters without image volumes,
+`hack/bake-task-image.sh` builds the guest as a layer on top of any base image with `ko` — no
+Dockerfile — and prints the digest-pinned ref:
+
+```bash
+# needs: ko on PATH (or $KO); builds the guest from this repo (override with $ATE_ENV_REPO)
+IMAGE=$(integrations/nemo-gym/hack/bake-task-image.sh python:3.12-slim gcr.io/$PROJECT/py-task-guest)
+```
+
+Then the template's container runs `/ko-app/ate-env-guest` directly, with no volume.
+
+In both modes the task image needs a `timeout` binary (coreutils or busybox — present in the slim
+Debian and Alpine bases, absent from distroless): `exec(timeout_s=…)` is enforced guest-side with a
+`timeout` wrapper. Verified: `python:3.12-slim` and `node:22-slim` bases both run real `python3` /
+`node` tasks through the provider.
+
+## Installation
+
+```bash
+# from a checkout of this repo:
+pip install ./clients/python                       # ate-env-client (import name ate_env)
+pip install ./integrations/nemo-gym                # nemo-gym-substrate
+# with NeMo Gym in the same environment:
+pip install './integrations/nemo-gym[gym]'
+```
+
+The entry point makes the provider discoverable immediately:
+
+```python
+from nemo_gym.sandbox.providers.registry import create_provider
+provider = create_provider({"substrate": {"connection": {"api_url": "127.0.0.1:7777"}}})
+```
+
+## How to use
+
+### From NeMo Gym config (the normal path)
+
+Add the `sandbox:` block to your resource-server config and select the `substrate` provider —
+nothing else in the Gym setup changes:
+
+```yaml
+sandbox:
+  default_metadata:
+    sandbox-api: substrate
+  substrate:
+    connection:
+      api_url: ate-env-api.ate-env:7777    # host:port or http(s):// URL; port-forward for dev
+      request_timeout_s: 30                # deadline per lifecycle RPC (create/get/delete)
+    create:
+      template: default-template           # ActorTemplate when the spec names none
+      atespace: ate-env                    # atespace the templates live in and envs are created in
+      ready_timeout_s: 120
+      ready_poll_interval_s: 1.0
+      image_templates:                     # optional SandboxSpec.image → template map
+        python:3.12-slim: gym-py312
+```
+
+`namespace` is still accepted as an alias for `atespace` (the name this provider used before
+ate-env adopted substrate's vocabulary); setting both is an error. `api_url` is plain gRPC
+(h2c) like the Go client and CLI; an `https://` URL opens a TLS channel instead.
+
+Then run your resource server / `gym env start` as usual. Sandboxes created by Gym's resource
+lifecycle now appear as Substrate actors; `kubectl get pods -n ate-env` shows the warm workers
+hosting them.
+
+### Directly from Python (debugging, scripts)
+
+```python
+import asyncio
+from nemo_gym_substrate import SubstrateSandboxProvider
+from nemo_gym.sandbox.providers.base import SandboxSpec   # or nemo_gym_substrate._compat
+
+async def main():
+    provider = SubstrateSandboxProvider({"connection": {"api_url": "127.0.0.1:7777"}})
+    handle = await provider.create(SandboxSpec(
+        files={"/task/hello.sh": "echo hello from substrate"},
+        workdir="/task",
+        env={"EPISODE": "1"},
+    ))
+    result = await provider.exec(handle, "sh /task/hello.sh && echo episode=$EPISODE")
+    print(result.stdout, result.return_code)
+    await provider.close(handle)
+    await provider.aclose()
+
+asyncio.run(main())
+```
+
+To share one `ate_env.Client` between the provider and other code, pass it in:
+`SubstrateSandboxProvider(config, client=client)`. The provider then never closes it.
+
+### Per-sandbox `provider_options` (on `SandboxSpec`)
+
+| Option | Meaning |
+|---|---|
+| `template` | ActorTemplate for this sandbox (overrides `create.template` and `image_templates`) |
+| `atespace` | Atespace of that template, and of the environment (`namespace` accepted as alias) |
+
+Unknown keys are rejected at `create()` time, as are unknown keys anywhere in the config block.
+
+## Testing
+
+```bash
+cd integrations/nemo-gym
+python3 -m venv .venv
+.venv/bin/pip install ../../clients/python -e '.[test]'
+
+# Unit/contract tests — hermetic: an in-process fake ate-env-api (tests/fakes.py)
+# behind the real ate_env client; no cluster, no nemo-gym needed (base types fall
+# back to structural mirrors in _compat.py; with nemo-gym installed the same
+# tests run against the real types):
+.venv/bin/python -m pytest tests/test_provider.py -q      # or: make nemo-gym-test (repo root)
+
+# End-to-end against a live cluster (full lifecycle on a real actor):
+kubectl port-forward -n ate-env svc/ate-env-api 7777:7777 &
+SUBSTRATE_E2E_API_URL=127.0.0.1:7777 .venv/bin/python -m pytest tests/test_e2e.py -q
+# SUBSTRATE_E2E_TEMPLATE=<name> picks the ActorTemplate (default: default-template)
+```
+
+## Benchmark
+
+`benchmarks/rollout_bench.py` shapes load like a NeMo-RL rollout batch: N parallel rollouts,
+each *create → seed task file → T agent turns (exec + mocked model think time) → close*, on real
+actors:
+
+```bash
+python benchmarks/rollout_bench.py --api-url 127.0.0.1:7777 --rollouts 5 --turns 3 --think-s 2
+```
+
+`benchmarks/driver_sim.py` is the start-time / time-to-first-exec probe over a list of templates
+(one task per template per rollout), the same shape as the SWE-500 campaign's eval arm.
+
+First numbers on a small test cluster (5 warm workers, gVisor, cold creates; measured with the
+earlier HTTP guest proxy, 2026-09-03 — rerun on the gRPC path before quoting):
+
+| Phase | p50 | max | Note |
+|---|---|---|---|
+| create | 4.95 s | 18.1 s | cold resume; max shows 5-way contention for 5 workers |
+| exec (turn) | 0.14 s | 0.15 s | router + guest round trip |
+| close | 3.06 s | 4.0 s | delete path |
+
+Turn overhead is already negligible against model think time; create/close dominate — which is
+exactly what golden-snapshot starts and fork-from-golden creates are for.
+
+## Resource mapping and isolation
+
+- One Gym sandbox = one Substrate **actor**: a gVisor (or micro-VM) sandbox multiplexed onto a
+  warm WorkerPool worker. Isolation is the sandbox class; actors share worker nodes.
+- `SandboxSpec.image` resolves to a pre-provisioned ActorTemplate via `image_templates` — there
+  is no dynamic image pull per create; that is what makes creates fast and repeatable.
+- `spec.files` are seeded through the guest filesystem service after readiness. `spec.env` /
+  `spec.workdir` are applied per exec (`StartProcess` takes `env` and `cwd` natively).
+- `exec(timeout_s=…)` is enforced guest-side (`timeout(1)` wrapper) and client-side; on the
+  client-side deadline the process is killed and a `return_code=-1` sentinel result is returned
+  rather than raising. Transport failures (env deleted mid-exec, router down) raise
+  `ate_env.EnvError` subclasses.
+- `status()` reads the environment's lifecycle state from `ate-env-api` without touching the
+  guest, so polling it never wakes a parked sandbox. A **suspended or paused actor reports
+  RUNNING** — the router auto-resumes it on the next command, which is the behavior a rollout
+  loop wants. `RESUMING` → `STARTING`, `CRASHED` → `ERROR`, deleted or `DELETING` → `STOPPED`.
+
+Known limits (tracked in the RL-on-Substrate proposal):
+
+- `spec.ttl_s` is **not enforced** — substrate has no server-side TTL yet; clean up via
+  `close()`. A crashed harness leaks actors until deleted.
+- `exec(user=…)` raises: commands run as the actor's configured user.
+- `spec.ports` / `SandboxEndpoint` exposure is not implemented in this demo.
+
+## Troubleshooting
+
+| Symptom | Cause / fix |
+|---|---|
+| `SandboxCreateError: no ActorTemplate mapped for image …` | The spec named an image with no template. Pre-provision a template and add it to `create.image_templates`, or set `provider_options.template` |
+| `SandboxCreateError: … UNAVAILABLE` at once | `api_url` unreachable — port-forward died, or wrong kubectl context (`kubectl config current-context`) |
+| Create times out (`SandboxCreateVerificationError: … UNAVAILABLE` / `INTERNAL`) | Check warm workers exist (`kubectl get pods -n ate-env`) and the template is Ready (`kubectl get actortemplate -n ate-env`). More parallel creates than warm workers will queue. If the message says `server closed the stream without sending trailers`, the router is not carrying gRPC to actors (see Prerequisites #1) |
+| Create times out and the actor is `RUNNING` | The template's image does not run `ate-env-guest` — the actor is up but nothing serves the guest API. See "Task images" |
+| Every exec fails with `NOT_FOUND` | The environment is gone (deleted by another party, or the `ate-env-api` restarted with a different atespace default). `status()` returns `STOPPED`; create a new sandbox |
+| Every exec returns `UNAVAILABLE` with `CERTIFICATE_EXPIRED` | The atenet router's pod certificate expired and wasn't hot-reloaded (seen on routers running > ~1 day): `kubectl -n ate-system rollout restart deploy/atenet-router` |
+| Actor stuck `RESUMING` forever (`status()` stays `STARTING`) | Restore landed on a CPU-incompatible node (mixed Intel/AMD pools) or raced a router restart. Delete the env and recreate; long-term fix is CPU-aware placement |
diff --git a/integrations/nemo-gym/benchmarks/driver_sim.py b/integrations/nemo-gym/benchmarks/driver_sim.py
new file mode 100644
index 0000000..5c5bef3
--- /dev/null
+++ b/integrations/nemo-gym/benchmarks/driver_sim.py
@@ -0,0 +1,161 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Substrate rollout-driver simulator: start-time / time-to-first-command bench.
+
+Mirrors the agent-sandbox `asrl-driver` bench convention (per task: create the
+sandbox, deliver the environment, run a first `exec` probe — probe defaults to
+`true`, isolating start latency). Here each task is a Substrate actor driven
+through the substrate NeMo Gym provider (create -> wait ready -> exec -> delete).
+
+Phases measured per task:
+  * ready_s     — create actor + wait until it serves commands (env delivery)
+  * first_cmd_s — the first exec (probe) round-trip
+  * ttfe_s      — ready_s + first_cmd_s (time to first execution)
+
+Templates must be pre-provisioned (image + ate-env-guest, one per image). See
+hack/bake-task-image.sh. Input file lists one substrate template name per line
+(optionally `template<TAB>display-image`); blank lines and #comments ignored.
+
+Usage:
+  kubectl port-forward -n ate-env svc/ate-env-api 7777:7777 &
+  python driver_sim.py --templates-file r2e_templates.txt \
+      --rollouts 2 --concurrency 5 --probe true --out results.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import statistics
+import time
+
+from nemo_gym_substrate._compat import SandboxSpec
+from nemo_gym_substrate.provider import SubstrateSandboxProvider
+
+
+def _pct(values: list[float], p: float) -> float | None:
+    if not values:
+        return None
+    if len(values) == 1:
+        return round(values[0], 3)
+    q = statistics.quantiles(values, n=100, method="inclusive")
+    return round(q[min(p, len(q)) - 1], 3)
+
+
+def _summary(name: str, values: list[float]) -> dict:
+    if not values:
+        return {"phase": name, "n": 0}
+    return {
+        "phase": name, "n": len(values),
+        "p50": _pct(values, 50), "p90": _pct(values, 90),
+        "max": round(max(values), 3), "mean": round(statistics.fmean(values), 3),
+    }
+
+
+async def _task(provider, template, namespace, rollout, probe, ready_timeout, sem, out):
+    async with sem:
+        opts = {"template": template}
+        if namespace:
+            opts["namespace"] = namespace
+        rec = {"template": template, "rollout": rollout, "ok": False}
+        handle = None
+        try:
+            t0 = time.monotonic()
+            handle = await provider.create(
+                SandboxSpec(provider_options=opts, ready_timeout_s=ready_timeout)
+            )
+            rec["ready_s"] = time.monotonic() - t0
+            t1 = time.monotonic()
+            res = await provider.exec(handle, probe)
+            rec["first_cmd_s"] = time.monotonic() - t1
+            rec["ttfe_s"] = rec["ready_s"] + rec["first_cmd_s"]
+            rec["exit_code"] = res.return_code
+            rec["ok"] = res.return_code == 0
+        except Exception as exc:  # noqa: BLE001 - record, keep the sweep going
+            rec["error"] = f"{type(exc).__name__}: {str(exc)[:200]}"
+        finally:
+            if handle is not None:
+                try:
+                    await provider.close(handle)
+                except Exception:  # noqa: BLE001
+                    pass
+        status = "ok" if rec["ok"] else f"ERR {rec.get('error', rec.get('exit_code'))}"
+        print(f"  [{template.split('/')[-1][:48]:48} r{rollout}] "
+              f"ttfe={rec.get('ttfe_s', float('nan')):.2f}s "
+              f"(ready={rec.get('ready_s', float('nan')):.2f} "
+              f"cmd={rec.get('first_cmd_s', float('nan')):.2f}) {status}", flush=True)
+        out.append(rec)
+
+
+def _load_templates(path: str, limit: int) -> list[str]:
+    templates = []
+    for line in open(path):
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        templates.append(line.split("\t")[0].split()[0])
+    return templates[:limit] if limit else templates
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--api-url", default="http://127.0.0.1:7777")
+    ap.add_argument("--templates-file", required=True)
+    ap.add_argument("--rollouts", type=int, default=1, help="tasks per template")
+    ap.add_argument("--concurrency", type=int, default=5)
+    ap.add_argument("--probe", default="true", help="first command (default no-op)")
+    ap.add_argument("--problems", type=int, default=0, help="limit templates (0=all)")
+    ap.add_argument("--ready-timeout-s", type=float, default=300.0)
+    ap.add_argument("--namespace", default=None)
+    ap.add_argument("--out", default=None)
+    a = ap.parse_args()
+
+    templates = _load_templates(a.templates_file, a.problems)
+    tasks_plan = [(t, g) for t in templates for g in range(a.rollouts)]
+    print(f"driver: {len(templates)} templates x {a.rollouts} = {len(tasks_plan)} tasks; "
+          f"concurrency={a.concurrency} probe={a.probe!r}", flush=True)
+
+    provider = SubstrateSandboxProvider({"connection": {"api_url": a.api_url}})
+    sem = asyncio.Semaphore(a.concurrency)
+    out: list[dict] = []
+
+    wall0 = time.monotonic()
+    await asyncio.gather(*(
+        _task(provider, t, a.namespace, g, a.probe, a.ready_timeout_s, sem, out)
+        for t, g in tasks_plan
+    ))
+    wall = time.monotonic() - wall0
+    await provider.aclose()
+
+    ok = [r for r in out if r["ok"]]
+    ready = [r["ready_s"] for r in ok]
+    cmd = [r["first_cmd_s"] for r in ok]
+    ttfe = [r["ttfe_s"] for r in ok]
+    summary = {
+        "tasks": len(out), "ok": len(ok), "err": len(out) - len(ok),
+        "wall_s": round(wall, 1),
+        "throughput_tasks_per_s": round(len(out) / wall, 3) if wall else None,
+        "phases": [_summary("ready", ready), _summary("first_cmd", cmd), _summary("ttfe", ttfe)],
+    }
+    print("\n=== RESULT ===")
+    print(json.dumps(summary, indent=2))
+    if a.out:
+        json.dump({"summary": summary, "tasks": out}, open(a.out, "w"), indent=2)
+        print(f"wrote {a.out}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/integrations/nemo-gym/benchmarks/rollout_bench.py b/integrations/nemo-gym/benchmarks/rollout_bench.py
new file mode 100644
index 0000000..20bad5b
--- /dev/null
+++ b/integrations/nemo-gym/benchmarks/rollout_bench.py
@@ -0,0 +1,134 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Mock NeMo-RL rollout benchmark for the substrate sandbox provider.
+
+Shapes the load like a NeMo Gym rollout batch: N parallel rollouts, each
+create -> seed task file -> T agent turns (exec + simulated model "think"
+time) -> artifact download -> close. The LLM is mocked with asyncio.sleep;
+the sandboxes are real Substrate actors.
+
+Usage:
+  kubectl port-forward -n ate-env svc/ate-env-api 7777:7777 &
+  python benchmarks/rollout_bench.py --rollouts 5 --turns 3 --think-s 2
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import statistics
+import time
+
+from nemo_gym_substrate._compat import SandboxSpec
+from nemo_gym_substrate.provider import SubstrateSandboxProvider
+
+# Pure-sh turn work: append the turn marker to the episode log, emit a
+# checksum of the log so far. Runs in any guest with a POSIX shell.
+TURN_CMD = 'echo "turn {turn} $(date +%s%N)" >> /task/episode.log && wc -c < /task/episode.log'
+
+
+async def rollout(
+    provider: SubstrateSandboxProvider,
+    index: int,
+    turns: int,
+    think_s: float,
+    template: str | None,
+    metrics: dict[str, list[float]],
+    sem: asyncio.Semaphore,
+) -> None:
+    async with sem:
+        opts = {"template": template} if template else {}
+        t0 = time.monotonic()
+        handle = await provider.create(
+            SandboxSpec(
+                files={"/task/episode.log": f"rollout {index}\n"},
+                workdir="/task",
+                provider_options=opts,
+                ready_timeout_s=180,
+            )
+        )
+        metrics["create_s"].append(time.monotonic() - t0)
+        try:
+            for turn in range(turns):
+                t1 = time.monotonic()
+                result = await provider.exec(handle, TURN_CMD.format(turn=turn))
+                metrics["exec_s"].append(time.monotonic() - t1)
+                if result.return_code != 0:
+                    raise RuntimeError(
+                        f"rollout {index} turn {turn} failed rc={result.return_code}: "
+                        f"{result.stderr}"
+                    )
+                await asyncio.sleep(think_s)  # mock model inference / eval
+        finally:
+            t2 = time.monotonic()
+            await provider.close(handle)
+            metrics["close_s"].append(time.monotonic() - t2)
+
+
+def summarize(name: str, values: list[float]) -> str:
+    if not values:
+        return f"{name:>9}: n=0"
+    q = statistics.quantiles(values, n=20) if len(values) >= 2 else [values[0]] * 19
+    return (
+        f"{name:>9}: n={len(values):<4} p50={statistics.median(values):6.2f}s "
+        f"p95={q[18]:6.2f}s max={max(values):6.2f}s"
+    )
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--api-url", default="http://127.0.0.1:7777")
+    ap.add_argument("--rollouts", type=int, default=5)
+    ap.add_argument("--turns", type=int, default=3)
+    ap.add_argument("--think-s", type=float, default=2.0)
+    ap.add_argument("--max-concurrency", type=int, default=5,
+                    help="parallel rollouts (bound by warm workers on small clusters)")
+    ap.add_argument("--template", default=None)
+    args = ap.parse_args()
+
+    provider = SubstrateSandboxProvider({"connection": {"api_url": args.api_url}})
+    metrics: dict[str, list[float]] = {"create_s": [], "exec_s": [], "close_s": []}
+    sem = asyncio.Semaphore(args.max_concurrency)
+
+    wall0 = time.monotonic()
+    results = await asyncio.gather(
+        *(
+            rollout(provider, i, args.turns, args.think_s, args.template, metrics, sem)
+            for i in range(args.rollouts)
+        ),
+        return_exceptions=True,
+    )
+    wall = time.monotonic() - wall0
+    await provider.aclose()
+
+    failures = [r for r in results if isinstance(r, BaseException)]
+    execs = len(metrics["exec_s"])
+    print(f"\nrollouts={args.rollouts} turns={args.turns} think={args.think_s}s "
+          f"concurrency={args.max_concurrency}")
+    print(summarize("create", metrics["create_s"]))
+    print(summarize("exec", metrics["exec_s"]))
+    print(summarize("close", metrics["close_s"]))
+    ideal = args.turns * args.think_s  # pure think time per rollout
+    print(f"{'wall':>9}: {wall:6.2f}s   execs/s={execs / wall:5.2f}   "
+          f"think-only floor per rollout={ideal:.1f}s")
+    if failures:
+        print(f"FAILURES: {len(failures)}")
+        for f in failures[:3]:
+            print("  -", type(f).__name__, str(f)[:200])
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/integrations/nemo-gym/hack/bake-task-image.sh b/integrations/nemo-gym/hack/bake-task-image.sh
new file mode 100755
index 0000000..24d35e0
--- /dev/null
+++ b/integrations/nemo-gym/hack/bake-task-image.sh
@@ -0,0 +1,70 @@
+#!/usr/bin/env bash
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Bake ate-env-guest onto an arbitrary base image (Mode A: runtime baked in).
+#
+# ko builds the guest Go binary as a layer on top of BASE and pushes the result
+# to REPO, printing the digest-pinned image ref (which substrate requires for
+# ActorTemplates). The guest is a static binary, so BASE can be any distro.
+#
+# Usage:
+#   bake-task-image.sh <base-image> <output-repo> [env-repo-path]
+#
+# Example:
+#   integrations/nemo-gym/hack/bake-task-image.sh python:3.12-slim gcr.io/PROJECT/py-task-guest
+#
+# Env:
+#   ATE_ENV_REPO   path to an agent-substrate/env checkout (default: this repo)
+#   KO             ko binary (default: ko on PATH, else ~/go/bin/ko)
+set -o errexit -o nounset -o pipefail
+
+if [[ $# -lt 2 ]]; then
+  sed -n '18,32p' "$0" >&2
+  exit 2
+fi
+
+BASE="$1"
+REPO="$2"
+ENV_REPO="${3:-${ATE_ENV_REPO:-$(cd "$(dirname "$0")/../../.." && pwd)}}"
+KO="${KO:-$(command -v ko || echo "${HOME}/go/bin/ko")}"
+GUEST_PKG="github.com/agent-substrate/env/cmd/ate-env-guest"
+
+[[ -x "${KO}" ]] || { echo "ko not found (set \$KO or install to ~/go/bin)" >&2; exit 1; }
+[[ -d "${ENV_REPO}/cmd/ate-env-guest" ]] || {
+  echo "env repo not found at ${ENV_REPO} (set \$ATE_ENV_REPO)" >&2; exit 1; }
+
+# A ko config that overrides the guest's base image to BASE. This overrides the
+# repo's own .ko.yaml (which pins bash:latest), so the guest lands on BASE's
+# rootfs instead.
+# ko infers the config format from the file extension, so the path must end in
+# .yaml (macOS mktemp -t appends a random suffix, which breaks that) — use a
+# temp dir holding a real .ko.yaml.
+CFG_DIR="$(mktemp -d -t ko-bake-XXXX)"
+CFG="${CFG_DIR}/.ko.yaml"
+trap 'rm -rf "${CFG_DIR}"' EXIT
+cat > "${CFG}" <<EOF
+baseImageOverrides:
+  ${GUEST_PKG}: ${BASE}
+EOF
+
+echo "baking ${GUEST_PKG} onto ${BASE} -> ${REPO} ..." >&2
+cd "${ENV_REPO}"
+IMAGE="$(
+  KO_DOCKER_REPO="${REPO}" \
+  KO_CONFIG_PATH="${CFG}" \
+  KO_DEFAULTPLATFORMS="${KO_DEFAULTPLATFORMS:-linux/amd64}" \
+  "${KO}" build --bare ./cmd/ate-env-guest
+)"
+echo "${IMAGE}"
diff --git a/integrations/nemo-gym/pyproject.toml b/integrations/nemo-gym/pyproject.toml
new file mode 100644
index 0000000..5e0f171
--- /dev/null
+++ b/integrations/nemo-gym/pyproject.toml
@@ -0,0 +1,40 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "nemo-gym-substrate"
+version = "0.2.0"
+description = "NeMo Gym sandbox provider backed by Agent Substrate actors (via ate-env)"
+readme = "README.md"
+license = { text = "Apache-2.0" }
+requires-python = ">=3.10"
+# ate-env-client is this repo's Python client (clients/python); it is not on
+# PyPI yet, so install it from the checkout first: pip install ./clients/python
+dependencies = ["ate-env-client>=0.1.0", "grpcio>=1.83"]
+
+[project.optional-dependencies]
+# The real deployment target. Without it the package falls back to local
+# structural mirrors of the nemo_gym base types (tests only).
+gym = ["nemo-gym"]
+test = ["pytest>=8"]
+
+[project.entry-points."nemo_gym.sandbox_providers"]
+substrate = "nemo_gym_substrate.provider:SubstrateSandboxProvider"
+
+[tool.setuptools.packages.find]
+where = ["src"]
diff --git a/integrations/nemo-gym/src/nemo_gym_substrate/__init__.py b/integrations/nemo-gym/src/nemo_gym_substrate/__init__.py
new file mode 100644
index 0000000..ef50338
--- /dev/null
+++ b/integrations/nemo-gym/src/nemo_gym_substrate/__init__.py
@@ -0,0 +1,29 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""NeMo Gym sandbox provider for Agent Substrate."""
+
+from .provider import (
+    ConnectionConfig,
+    CreateConfig,
+    SubstrateProviderOptions,
+    SubstrateSandboxProvider,
+)
+
+__all__ = [
+    "ConnectionConfig",
+    "CreateConfig",
+    "SubstrateProviderOptions",
+    "SubstrateSandboxProvider",
+]
diff --git a/integrations/nemo-gym/src/nemo_gym_substrate/_compat.py b/integrations/nemo-gym/src/nemo_gym_substrate/_compat.py
new file mode 100644
index 0000000..2d21be8
--- /dev/null
+++ b/integrations/nemo-gym/src/nemo_gym_substrate/_compat.py
@@ -0,0 +1,93 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""NeMo Gym base types, with hermetic fallbacks.
+
+When ``nemo_gym`` is installed (any real deployment), its types are used so
+handles and results are the genuine articles. The fallback dataclasses below
+are structural mirrors of ``nemo_gym.sandbox.providers.base`` that let this
+demo's contract tests run inside the substrate repo without the nemo-gym
+dependency. They must be kept shape-identical to the upstream types.
+"""
+
+from __future__ import annotations
+
+try:  # pragma: no cover - exercised only when nemo_gym is installed
+    from nemo_gym.sandbox.providers.base import (  # type: ignore
+        SandboxCreateError,
+        SandboxCreateVerificationError,
+        SandboxExecResult,
+        SandboxHandle,
+        SandboxSpec,
+        SandboxStatus,
+    )
+
+    NEMO_GYM_AVAILABLE = True
+except ImportError:  # pragma: no cover - the in-repo test path
+    from dataclasses import dataclass, field
+    from enum import Enum
+    from typing import Any
+
+    NEMO_GYM_AVAILABLE = False
+
+    class SandboxStatus(str, Enum):
+        STARTING = "starting"
+        RUNNING = "running"
+        STOPPED = "stopped"
+        ERROR = "error"
+        UNKNOWN = "unknown"
+
+    @dataclass(frozen=True)
+    class SandboxHandle:
+        sandbox_id: str
+        provider_name: str
+        raw: Any
+
+    @dataclass(frozen=True)
+    class SandboxExecResult:
+        stdout: str | None
+        stderr: str | None
+        return_code: int
+        error_type: str | None = None
+
+    @dataclass(frozen=True)
+    class SandboxSpec:
+        image: str | None = None
+        ttl_s: int | float | None = None
+        ready_timeout_s: int | float | None = None
+        workdir: str | None = None
+        env: dict[str, str] = field(default_factory=dict)
+        files: dict[str, str] = field(default_factory=dict)
+        metadata: dict[str, str] = field(default_factory=dict)
+        resources: dict[str, Any] = field(default_factory=dict)
+        entrypoint: list[str] | None = None
+        provider_options: dict[str, Any] = field(default_factory=dict)
+        ports: tuple[int, ...] = ()
+
+    class SandboxCreateError(RuntimeError):
+        pass
+
+    class SandboxCreateVerificationError(SandboxCreateError):
+        pass
+
+
+__all__ = [
+    "NEMO_GYM_AVAILABLE",
+    "SandboxCreateError",
+    "SandboxCreateVerificationError",
+    "SandboxExecResult",
+    "SandboxHandle",
+    "SandboxSpec",
+    "SandboxStatus",
+]
diff --git a/integrations/nemo-gym/src/nemo_gym_substrate/provider.py b/integrations/nemo-gym/src/nemo_gym_substrate/provider.py
new file mode 100644
index 0000000..cd1540b
--- /dev/null
+++ b/integrations/nemo-gym/src/nemo_gym_substrate/provider.py
@@ -0,0 +1,424 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""NeMo Gym sandbox provider backed by Agent Substrate.
+
+Each sandbox is a Substrate actor fronted by the ``ate-env`` API
+(https://github.com/agent-substrate/env), reached through this repo's Python
+client, ``ate_env`` (``clients/python``). Environment lifecycle (create, get, delete) goes to
+``ateenv.v1alpha.EnvironmentService`` on ``ate-env-api``; command execution and
+file transfer go to the ``ProcessService`` / ``FileSystemService`` served by
+the ``ate-env-guest`` daemon inside every actor, which ``ate-env-api`` proxies
+through the atenet router. One gRPC endpoint carries both. Idle sandboxes can
+be suspended by Substrate and are resumed transparently on the next guest
+call, so a fleet of mostly-waiting rollout sandboxes holds no workers.
+
+Contract: https://docs.nvidia.com/nemo/gym/main/infrastructure/sandbox/adding-a-provider/
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import math
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping
+from urllib.parse import urlparse
+
+import grpc
+import grpc.aio
+from ate_env import Client, Env, EnvError, EnvironmentStatus, NotFoundError, OutputSource
+
+from ._compat import (
+    SandboxCreateError,
+    SandboxCreateVerificationError,
+    SandboxExecResult,
+    SandboxHandle,
+    SandboxSpec,
+    SandboxStatus,
+)
+
+logger = logging.getLogger(__name__)
+
+_PROVIDER_NAME = "substrate"
+
+# Client-side slack on top of exec(timeout_s=...): the guest enforces the
+# deadline itself (see exec), so the client only has to outlive it.
+_EXEC_GRACE_S = 5.0
+# Deadline for one readiness probe. A guest that is still booting may hold a
+# call open; bounding it keeps the poll loop honest.
+_PROBE_TIMEOUT_S = 10.0
+
+# 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").
+_STATUS_MAP = {
+    EnvironmentStatus.RESUMING: SandboxStatus.STARTING,
+    EnvironmentStatus.RUNNING: SandboxStatus.RUNNING,
+    EnvironmentStatus.SUSPENDING: SandboxStatus.RUNNING,
+    EnvironmentStatus.SUSPENDED: SandboxStatus.RUNNING,
+    EnvironmentStatus.PAUSING: SandboxStatus.RUNNING,
+    EnvironmentStatus.PAUSED: SandboxStatus.RUNNING,
+    EnvironmentStatus.CRASHED: SandboxStatus.ERROR,
+    EnvironmentStatus.DELETING: SandboxStatus.STOPPED,
+}
+
+
+def _require_keys(options: Mapping[str, Any], allowed: frozenset[str], where: str) -> None:
+    unknown = sorted(set(options) - allowed)
+    if unknown:
+        raise ValueError(f"{where}: unknown option(s) {unknown}; allowed: {sorted(allowed)}")
+
+
+def _describe(exc: BaseException) -> str:
+    """``CODE: details`` for ate_env errors, so messages name the gRPC status."""
+    code = getattr(exc, "code", None)
+    if isinstance(code, grpc.StatusCode):
+        return f"{code.name}: {exc}"
+    if isinstance(exc, EnvError):
+        return f"{type(exc).__name__}: {exc}"
+    return str(exc) or type(exc).__name__
+
+
+def _atespace_option(options: Mapping[str, Any], where: str) -> str | None:
+    """``atespace`` is the ate-env name; ``namespace`` is accepted as the older alias."""
+    if "atespace" in options and "namespace" in options:
+        raise ValueError(f"{where}: set either atespace or namespace, not both")
+    value = options.get("atespace", options.get("namespace"))
+    return None if value is None else str(value)
+
+
+@dataclass(frozen=True)
+class ConnectionConfig:
+    """`connection:` block of the provider config."""
+
+    # host:port or an http(s):// URL of ate-env-api. Plain gRPC (h2c) unless https.
+    api_url: str = "http://127.0.0.1:7777"
+    # Deadline for each lifecycle RPC (create / get / delete).
+    request_timeout_s: float = 30.0
+
+    _ALLOWED = frozenset({"api_url", "request_timeout_s"})
+
+    @classmethod
+    def from_mapping(cls, options: Mapping[str, Any]) -> "ConnectionConfig":
+        _require_keys(options, cls._ALLOWED, "sandbox.substrate.connection")
+        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)),
+        )
+
+
+@dataclass(frozen=True)
+class CreateConfig:
+    """`create:` block of the provider config."""
+
+    # ActorTemplate when the spec names none; ate-env's own default.
+    template: str = "default-template"
+    # Atespace the templates live in and the environments are created in.
+    atespace: str = "ate-env"
+    ready_timeout_s: float = 120.0
+    ready_poll_interval_s: float = 1.0
+    # Optional mapping from SandboxSpec.image to an ActorTemplate name, for
+    # workloads that select sandboxes by image reference. Templates must be
+    # pre-provisioned on the cluster (see README).
+    image_templates: Mapping[str, str] = None  # type: ignore[assignment]
+
+    _ALLOWED = frozenset(
+        {
+            "template",
+            "atespace",
+            "namespace",
+            "ready_timeout_s",
+            "ready_poll_interval_s",
+            "image_templates",
+        }
+    )
+
+    @classmethod
+    def from_mapping(cls, options: Mapping[str, Any]) -> "CreateConfig":
+        _require_keys(options, cls._ALLOWED, "sandbox.substrate.create")
+        return cls(
+            template=str(options.get("template", cls.template)),
+            atespace=_atespace_option(options, "sandbox.substrate.create") or cls.atespace,
+            ready_timeout_s=float(options.get("ready_timeout_s", cls.ready_timeout_s)),
+            ready_poll_interval_s=float(
+                options.get("ready_poll_interval_s", cls.ready_poll_interval_s)
+            ),
+            image_templates=dict(options.get("image_templates") or {}),
+        )
+
+
+@dataclass(frozen=True)
+class SubstrateProviderOptions:
+    """Per-sandbox options carried in ``SandboxSpec.provider_options``."""
+
+    template: str | None = None
+    atespace: str | None = None
+
+    _ALLOWED = frozenset({"template", "atespace", "namespace"})
+
+    @classmethod
+    def from_mapping(cls, options: Mapping[str, Any]) -> "SubstrateProviderOptions":
+        _require_keys(options, cls._ALLOWED, "SandboxSpec.provider_options")
+        return cls(
+            template=options.get("template"),
+            atespace=_atespace_option(options, "SandboxSpec.provider_options"),
+        )
+
+
+class SubstrateSandboxProvider:
+    """NeMo Gym sandbox provider running sandboxes as Substrate actors."""
+
+    name = _PROVIDER_NAME
+
+    def __init__(
+        self,
+        config: Mapping[str, Any] | None = None,
+        *,
+        client: Client | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """``config`` is the ``sandbox.substrate`` block. Pass ``client`` to reuse a
+        caller-owned ``ate_env.Client`` (the provider then never closes it)."""
+        config = dict(config or {})
+        config.update(kwargs)
+        _require_keys(config, frozenset({"connection", "create"}), "sandbox.substrate")
+        self._connection = ConnectionConfig.from_mapping(config.get("connection") or {})
+        self._create = CreateConfig.from_mapping(config.get("create") or {})
+        # The channel is opened lazily so constructing a provider never dials.
+        self._client: Client | None = client
+        self._owns_client = client is None
+        self._channel: grpc.aio.Channel | None = None
+
+    # -- provider contract -------------------------------------------------
+
+    async def create(self, spec: SandboxSpec) -> SandboxHandle:
+        opts = SubstrateProviderOptions.from_mapping(spec.provider_options or {})
+        template = self._resolve_template(spec, opts)
+        atespace = opts.atespace or self._create.atespace
+        env_id = f"gym-{uuid.uuid4().hex[:10]}"
+
+        try:
+            env = await asyncio.wait_for(
+                self._ate().create(
+                    env_id,
+                    atespace=atespace,
+                    template_name=template,
+                    template_atespace=atespace,
+                ),
+                timeout=self._connection.request_timeout_s,
+            )
+        except EnvError as exc:
+            raise SandboxCreateError(
+                f"creating substrate env {env_id!r} (template {atespace}/{template}): "
+                f"{_describe(exc)}"
+            ) from exc
+        except asyncio.TimeoutError as exc:
+            raise SandboxCreateError(
+                f"creating substrate env {env_id!r}: no answer from "
+                f"{self._connection.api_url} within {self._connection.request_timeout_s}s"
+            ) from exc
+
+        try:
+            await self._wait_ready(env, spec.ready_timeout_s or self._create.ready_timeout_s)
+            for target_path, content in (spec.files or {}).items():
+                await env.write_file(target_path, content.encode(), mode=0o644)
+        except Exception:
+            await self._best_effort_delete(env_id, atespace)
+            raise
+
+        return SandboxHandle(
+            sandbox_id=env_id,
+            provider_name=self.name,
+            raw={
+                "env_id": env_id,
+                "atespace": atespace,
+                "workdir": spec.workdir,
+                "env": dict(spec.env or {}),
+            },
+        )
+
+    async def exec(
+        self,
+        handle: SandboxHandle,
+        command: str,
+        *,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        timeout_s: int | float | None = None,
+        user: str | int | None = None,
+    ) -> SandboxExecResult:
+        if user is not None:
+            raise ValueError(
+                "the substrate provider does not support per-exec `user`: "
+                "the guest daemon runs commands as the actor's configured user"
+            )
+        raw = handle.raw or {}
+        merged_env = {**(raw.get("env") or {}), **(env or {})}
+        effective_cwd = cwd or raw.get("workdir") or ""
+
+        argv = ["sh", "-c", command]
+        # `is not None` so an explicit 0 means "shortest allowed deadline"
+        # (1s guest-side), not "unlimited".
+        if timeout_s is not None:
+            # Enforce the deadline guest-side, so a runaway process does not
+            # outlive the call that started it. Round up to whole seconds with
+            # a floor of 1: `timeout 0` disables the limit in coreutils, so a
+            # sub-second timeout_s must not floor to zero.
+            argv = ["timeout", str(max(1, math.ceil(timeout_s))), *argv]
+
+        sandbox = self._env(handle)
+        process_id = await sandbox.start_process(argv, cwd=effective_cwd, env=merged_env)
+        try:
+            return await asyncio.wait_for(
+                self._collect(sandbox, process_id),
+                timeout=None if timeout_s is None else timeout_s + _EXEC_GRACE_S,
+            )
+        except asyncio.TimeoutError:
+            with contextlib.suppress(EnvError):
+                await sandbox.kill_process(process_id)
+            return SandboxExecResult(
+                stdout=None,
+                stderr=f"substrate provider: exec timed out after {timeout_s}s",
+                return_code=-1,
+            )
+
+    async def upload_file(self, handle: SandboxHandle, source_path: Path, target_path: str) -> None:
+        await self._env(handle).write_file(
+            target_path, Path(source_path).read_bytes(), mode=0o644
+        )
+
+    async def download_file(self, handle: SandboxHandle, source_path: str, target_path: Path) -> None:
+        content = await self._env(handle).read_file_bytes(source_path)
+        target = Path(target_path)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_bytes(content)
+
+    async def status(self, handle: SandboxHandle) -> SandboxStatus:
+        # GetEnvironment reads the actor's state without touching the guest,
+        # so asking does not wake a parked sandbox.
+        raw = handle.raw or {}
+        try:
+            info = await asyncio.wait_for(
+                self._ate().get(handle.sandbox_id, atespace=self._atespace_of(raw)),
+                timeout=self._connection.request_timeout_s,
+            )
+        except NotFoundError:
+            return SandboxStatus.STOPPED
+        except (EnvError, asyncio.TimeoutError):
+            return SandboxStatus.UNKNOWN
+        return _STATUS_MAP.get(info.status, SandboxStatus.UNKNOWN)
+
+    async def close(self, handle: SandboxHandle) -> None:
+        await self._best_effort_delete(handle.sandbox_id, self._atespace_of(handle.raw or {}))
+
+    async def aclose(self) -> None:
+        if self._client is not None and self._owns_client:
+            await self._client.close()
+        if self._channel is not None:
+            await self._channel.close()
+        self._client = None if self._owns_client else self._client
+        self._channel = None
+
+    # -- internals ----------------------------------------------------------
+
+    def _ate(self) -> Client:
+        if self._client is None:
+            self._client = self._open_client(self._connection.api_url)
+        return self._client
+
+    def _open_client(self, api_url: str) -> Client:
+        url = urlparse(api_url if "://" in api_url else f"//{api_url}")
+        if url.scheme == "https":
+            # ate_env.Client only dials h2c itself; hand it a TLS channel.
+            target = f"{url.hostname}:{url.port or 443}"
+            self._channel = grpc.aio.secure_channel(target, grpc.ssl_channel_credentials())
+            return Client(channel=self._channel)
+        return Client(api_url)
+
+    def _atespace_of(self, raw: Mapping[str, Any]) -> str:
+        return str(raw.get("atespace") or self._create.atespace)
+
+    def _env(self, handle: SandboxHandle) -> Env:
+        return self._ate().env(handle.sandbox_id, atespace=self._atespace_of(handle.raw or {}))
+
+    def _resolve_template(self, spec: SandboxSpec, opts: SubstrateProviderOptions) -> str:
+        if opts.template:
+            return opts.template
+        if spec.image:
+            mapped = (self._create.image_templates or {}).get(spec.image)
+            if mapped:
+                return mapped
+            raise SandboxCreateError(
+                f"no ActorTemplate mapped for image {spec.image!r}: add it to "
+                "sandbox.substrate.create.image_templates or set "
+                "provider_options.template (templates are pre-provisioned; see README)"
+            )
+        return self._create.template
+
+    async def _wait_ready(self, env: Env, timeout_s: float) -> 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."""
+        deadline = time.monotonic() + timeout_s
+        last_error = "no probe attempted"
+        while True:
+            try:
+                result = await asyncio.wait_for(env.shell("true"), timeout=_PROBE_TIMEOUT_S)
+                if result.exit_code == 0:
+                    return
+                last_error = f"probe exited {result.exit_code}: {result.stderr.strip()[:200]}"
+            except EnvError as exc:
+                last_error = _describe(exc)
+            except asyncio.TimeoutError:
+                last_error = f"probe did not answer within {_PROBE_TIMEOUT_S}s"
+            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}"
+        )
+
+    @staticmethod
+    async def _collect(env: Env, process_id: str) -> SandboxExecResult:
+        stdout = bytearray()
+        stderr = bytearray()
+        async for chunk in env.stream_outputs(process_id, follow=True):
+            if chunk.source == OutputSource.STDOUT:
+                stdout.extend(chunk.data)
+            elif chunk.source == OutputSource.STDERR:
+                stderr.extend(chunk.data)
+        # The follow stream can end before the process record's status
+        # flips; wait() polls for the final state and exit code.
+        proc = await env.wait(process_id)
+        return SandboxExecResult(
+            stdout=stdout.decode("utf-8", errors="replace"),
+            stderr=stderr.decode("utf-8", errors="replace"),
+            return_code=proc.exit_code,
+        )
+
+    async def _best_effort_delete(self, env_id: str, atespace: str) -> None:
+        try:
+            await asyncio.wait_for(
+                self._ate().delete(env_id, atespace=atespace),
+                timeout=self._connection.request_timeout_s,
+            )
+        except NotFoundError:
+            pass  # already gone: close() is idempotent
+        except (EnvError, asyncio.TimeoutError) as exc:
+            logger.warning("deleting substrate env %s: %s", env_id, _describe(exc))
diff --git a/integrations/nemo-gym/tests/__init__.py b/integrations/nemo-gym/tests/__init__.py
new file mode 100644
index 0000000..58d482e
--- /dev/null
+++ b/integrations/nemo-gym/tests/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
diff --git a/integrations/nemo-gym/tests/fakes.py b/integrations/nemo-gym/tests/fakes.py
new file mode 100644
index 0000000..a3639e4
--- /dev/null
+++ b/integrations/nemo-gym/tests/fakes.py
@@ -0,0 +1,237 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""In-process fake of ate-env-api for the provider's contract tests.
+
+Serves the three gRPC services the real ``ate-env-api`` exposes on one port:
+``EnvironmentService`` (lifecycle) and, proxied to the guest in production,
+``ProcessService`` and ``FileSystemService``. The guest services require the
+``x-env-id`` routing metadata the ``ate_env`` client attaches, and answer
+NOT_FOUND for an unknown environment the way the proxy does. Behavior is
+scripted through :class:`FakeAteEnv` (create/get/delete failures, readiness
+probes to fail first, commands that never exit).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from dataclasses import dataclass, field
+
+import grpc
+import grpc.aio
+from ate_env import EnvironmentStatus, OutputSource, ProcessStatus
+from ate_env._gen.ateenv.v1alpha import env_pb2, env_pb2_grpc, guest_pb2, guest_pb2_grpc
+
+DEFAULT_ATESPACE = "ate-env"
+DEFAULT_TEMPLATE = "default-template"
+
+
+@dataclass
+class FakeProc:
+    env_id: str
+    argv: list[str]
+    stdout: bytes = b""
+    stderr: bytes = b""
+    exit_code: int | None = None  # None while running
+    killed: bool = False
+
+    @property
+    def status(self) -> int:
+        if self.exit_code is None:
+            return int(ProcessStatus.RUNNING)
+        if self.killed:
+            return int(ProcessStatus.TERMINATED)
+        return int(ProcessStatus.COMPLETED if self.exit_code == 0 else ProcessStatus.FAILED)
+
+
+@dataclass
+class FakeAteEnv:
+    """State shared by the fake services, plus the knobs tests script."""
+
+    envs: dict[str, dict] = field(default_factory=dict)
+    files: dict[tuple[str, str], bytes] = field(default_factory=dict)
+    modes: dict[tuple[str, str], int] = field(default_factory=dict)
+    starts: list[dict] = field(default_factory=list)
+    procs: dict[str, FakeProc] = field(default_factory=dict)
+    create_code: grpc.StatusCode | None = None
+    get_code: grpc.StatusCode | None = None
+    delete_code: grpc.StatusCode | None = None
+    ready_after_probes: int = 0  # fail this many `true` probes with UNAVAILABLE first
+    hang: set[str] = field(default_factory=set)  # shell commands whose process never exits
+    _probes: int = 0
+    _seq: int = 0
+
+    def env_pb(self, env_id: str) -> env_pb2.Environment:
+        e = self.envs[env_id]
+        return env_pb2.Environment(
+            id=env_id,
+            atespace=e["atespace"],
+            template=env_pb2.Template(name=e["template"], atespace=e["template_atespace"]),
+            status=e["status"],
+        )
+
+    def set_status(self, env_id: str, status: EnvironmentStatus) -> None:
+        self.envs[env_id]["status"] = int(status)
+
+
+async def _env_from_metadata(state: FakeAteEnv, context) -> str:
+    md = dict(context.invocation_metadata())
+    env_id = md.get("x-env-id", "")
+    if not env_id:
+        await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "missing x-env-id metadata")
+    if env_id not in state.envs:
+        await context.abort(grpc.StatusCode.NOT_FOUND, f"environment {env_id!r} not found")
+    return env_id
+
+
+class FakeEnvironmentService(env_pb2_grpc.EnvironmentServiceServicer):
+    def __init__(self, state: FakeAteEnv) -> None:
+        self.state = state
+
+    async def CreateEnvironment(self, request, context):
+        s = self.state
+        if s.create_code is not None:
+            await context.abort(s.create_code, "create refused")
+        if request.id in s.envs:
+            await context.abort(grpc.StatusCode.ALREADY_EXISTS, "exists")
+        s.envs[request.id] = {
+            "atespace": request.atespace or DEFAULT_ATESPACE,
+            "template": request.template.name or DEFAULT_TEMPLATE,
+            "template_atespace": request.template.atespace or DEFAULT_ATESPACE,
+            "status": int(EnvironmentStatus.RUNNING),
+        }
+        return env_pb2.CreateEnvironmentResponse(environment=s.env_pb(request.id))
+
+    async def GetEnvironment(self, request, context):
+        s = self.state
+        if s.get_code is not None:
+            await context.abort(s.get_code, "get refused")
+        if request.id not in s.envs:
+            await context.abort(grpc.StatusCode.NOT_FOUND, "no such env")
+        return env_pb2.GetEnvironmentResponse(environment=s.env_pb(request.id))
+
+    async def SuspendEnvironment(self, request, context):
+        if request.id not in self.state.envs:
+            await context.abort(grpc.StatusCode.NOT_FOUND, "no such env")
+        self.state.set_status(request.id, EnvironmentStatus.SUSPENDED)
+        return env_pb2.SuspendEnvironmentResponse()
+
+    async def DeleteEnvironment(self, request, context):
+        s = self.state
+        if s.delete_code is not None:
+            await context.abort(s.delete_code, "delete refused")
+        if s.envs.pop(request.id, None) is None:
+            await context.abort(grpc.StatusCode.NOT_FOUND, "no such env")
+        return env_pb2.DeleteEnvironmentResponse()
+
+
+class FakeProcessService(guest_pb2_grpc.ProcessServiceServicer):
+    def __init__(self, state: FakeAteEnv) -> None:
+        self.state = state
+
+    async def StartProcess(self, request, context):
+        s = self.state
+        env_id = await _env_from_metadata(s, context)
+        argv = list(request.command)
+        shell_cmd = argv[-1] if argv else ""
+        s.starts.append(
+            {"env_id": env_id, "command": argv, "cwd": request.cwd, "env": dict(request.env)}
+        )
+        if shell_cmd == "true" and s._probes < s.ready_after_probes:
+            s._probes += 1
+            await context.abort(grpc.StatusCode.UNAVAILABLE, "guest not up yet")
+        s._seq += 1
+        pid = f"p{s._seq}"
+        proc = FakeProc(env_id=env_id, argv=argv)
+        if shell_cmd in s.hang:
+            proc.exit_code = None
+        elif "exit 7" in shell_cmd:
+            proc.stderr, proc.exit_code = b"boom", 7
+        else:
+            proc.stdout, proc.exit_code = b"ok", 0
+        s.procs[pid] = proc
+        return guest_pb2.StartProcessResponse(process_id=pid)
+
+    async def _lookup(self, process_id: str, context) -> FakeProc:
+        await _env_from_metadata(self.state, context)
+        proc = self.state.procs.get(process_id)
+        if proc is None:
+            await context.abort(grpc.StatusCode.NOT_FOUND, f"process {process_id!r} not found")
+        return proc
+
+    async def GetProcess(self, request, context):
+        proc = await self._lookup(request.process_id, context)
+        return guest_pb2.Process(
+            process_id=request.process_id, status=proc.status, exit_code=proc.exit_code or 0
+        )
+
+    async def StreamProcessOutputs(self, request, context):
+        proc = await self._lookup(request.process_id, context)
+        if proc.stdout:
+            yield guest_pb2.OutputChunk(source=int(OutputSource.STDOUT), data=proc.stdout)
+        if proc.stderr:
+            yield guest_pb2.OutputChunk(source=int(OutputSource.STDERR), data=proc.stderr)
+        if request.follow:
+            while proc.exit_code is None:
+                await asyncio.sleep(0.01)
+
+    async def KillProcess(self, request, context):
+        proc = await self._lookup(request.process_id, context)
+        if proc.exit_code is None:
+            proc.exit_code, proc.killed = 137, True
+        return guest_pb2.KillProcessResponse(exit_code=proc.exit_code)
+
+
+class FakeFileSystemService(guest_pb2_grpc.FileSystemServiceServicer):
+    def __init__(self, state: FakeAteEnv) -> None:
+        self.state = state
+
+    async def ReadFile(self, request, context):
+        env_id = await _env_from_metadata(self.state, context)
+        data = self.state.files.get((env_id, request.path))
+        if data is None:
+            await context.abort(grpc.StatusCode.NOT_FOUND, f"{request.path}: no such file")
+        # Small chunks so the client's streaming reassembly is exercised.
+        for i in range(0, len(data), 4):
+            yield guest_pb2.FileChunk(data=data[i : i + 4])
+
+    async def WriteFile(self, request_iterator, context):
+        env_id = await _env_from_metadata(self.state, context)
+        path, mode, buf = None, 0, bytearray()
+        async for req in request_iterator:
+            if path is None:
+                path, mode = req.path, req.mode
+            buf.extend(req.chunk)
+        if not path:
+            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "path is required")
+        self.state.files[(env_id, path)] = bytes(buf)
+        self.state.modes[(env_id, path)] = mode
+        return guest_pb2.WriteFileResponse(bytes_written=len(buf))
+
+
+@contextlib.asynccontextmanager
+async def serve(state: FakeAteEnv | None = None):
+    """Run the fake ate-env-api on a loopback port; yields ``(target, state)``."""
+    state = state or FakeAteEnv()
+    server = grpc.aio.server()
+    env_pb2_grpc.add_EnvironmentServiceServicer_to_server(FakeEnvironmentService(state), server)
+    guest_pb2_grpc.add_ProcessServiceServicer_to_server(FakeProcessService(state), server)
+    guest_pb2_grpc.add_FileSystemServiceServicer_to_server(FakeFileSystemService(state), server)
+    port = server.add_insecure_port("127.0.0.1:0")
+    await server.start()
+    try:
+        yield f"127.0.0.1:{port}", state
+    finally:
+        await server.stop(None)
diff --git a/integrations/nemo-gym/tests/test_e2e.py b/integrations/nemo-gym/tests/test_e2e.py
new file mode 100644
index 0000000..d560536
--- /dev/null
+++ b/integrations/nemo-gym/tests/test_e2e.py
@@ -0,0 +1,88 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""End-to-end test against a live ate-env deployment.
+
+Skipped unless SUBSTRATE_E2E_API_URL points at a reachable ate-env-api (e.g.
+`kubectl port-forward -n ate-env svc/ate-env-api 7777:7777` then
+`SUBSTRATE_E2E_API_URL=127.0.0.1:7777 pytest tests/test_e2e.py`; an http://
+URL works too). SUBSTRATE_E2E_TEMPLATE picks the ActorTemplate (default
+`default-template`, ate-env's own default).
+Exercises the full provider lifecycle on a real Substrate actor:
+create → exec (incl. nonzero exit, env/cwd) → file round-trip → status →
+close → status STOPPED.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from pathlib import Path
+
+import pytest
+
+from nemo_gym_substrate._compat import SandboxSpec, SandboxStatus
+from nemo_gym_substrate.provider import SubstrateSandboxProvider
+
+API_URL = os.environ.get("SUBSTRATE_E2E_API_URL")
+TEMPLATE = os.environ.get("SUBSTRATE_E2E_TEMPLATE", "default-template")
+
+pytestmark = pytest.mark.skipif(
+    not API_URL, reason="SUBSTRATE_E2E_API_URL not set; e2e needs a live ate-env-api"
+)
+
+
+def test_full_lifecycle(tmp_path: Path) -> None:
+    async def scenario() -> None:
+        provider = SubstrateSandboxProvider({"connection": {"api_url": API_URL}})
+        handle = None
+        try:
+            handle = await provider.create(
+                SandboxSpec(
+                    files={"/task/hello.txt": "hello from nemo-gym-substrate e2e\n"},
+                    workdir="/task",
+                    env={"E2E_MARK": "42"},
+                    provider_options={"template": TEMPLATE},
+                    ready_timeout_s=180,
+                )
+            )
+
+            # exec: env + cwd honored, output captured.
+            result = await provider.exec(handle, "echo -n $E2E_MARK; pwd")
+            assert result.return_code == 0
+            assert "42" in (result.stdout or "")
+            assert "/task" in (result.stdout or "")
+
+            # nonzero exit is a result, not an exception.
+            result = await provider.exec(handle, "exit 7")
+            assert result.return_code == 7
+
+            # file round-trip: the seeded file comes back, an uploaded file too.
+            out = tmp_path / "hello.txt"
+            await provider.download_file(handle, "/task/hello.txt", out)
+            assert out.read_text().startswith("hello from nemo-gym-substrate")
+            src = tmp_path / "up.bin"
+            src.write_bytes(bytes(range(64)))
+            await provider.upload_file(handle, src, "/task/up.bin")
+            result = await provider.exec(handle, "wc -c < /task/up.bin")
+            assert (result.stdout or "").strip() == "64"
+
+            assert await provider.status(handle) == SandboxStatus.RUNNING
+        finally:
+            if handle is not None:
+                await provider.close(handle)
+                assert await provider.status(handle) == SandboxStatus.STOPPED
+            await provider.aclose()
+
+    asyncio.run(scenario())
diff --git a/integrations/nemo-gym/tests/test_provider.py b/integrations/nemo-gym/tests/test_provider.py
new file mode 100644
index 0000000..c7fa7ad
--- /dev/null
+++ b/integrations/nemo-gym/tests/test_provider.py
@@ -0,0 +1,418 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Contract tests for the substrate NeMo Gym provider.
+
+Runs against an in-process fake of ate-env-api (tests/fakes.py) reached
+through the real ``ate_env`` client over loopback gRPC, asserting the rules
+from the adding-a-provider contract: create returns only when the sandbox
+executes commands, exec never raises on nonzero exits, close is
+cleanup-safe, and provider_options are validated strictly.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from pathlib import Path
+
+import grpc
+import pytest
+from ate_env import Client, EnvironmentStatus, NotFoundError
+
+from nemo_gym_substrate import provider as provider_mod
+from nemo_gym_substrate._compat import (
+    SandboxCreateError,
+    SandboxCreateVerificationError,
+    SandboxSpec,
+    SandboxStatus,
+)
+from nemo_gym_substrate.provider import SubstrateSandboxProvider
+
+from .fakes import FakeAteEnv, serve
+
+
+@contextlib.asynccontextmanager
+async def stack(create: dict | None = None, *, state: FakeAteEnv | None = None):
+    """A provider dialing a fake ate-env-api; yields ``(provider, fake)``."""
+    async with serve(state) as (target, fake):
+        config = {
+            "connection": {"api_url": target, "request_timeout_s": 5.0},
+            "create": {
+                "ready_poll_interval_s": 0.01,
+                "image_templates": {"img:1": "tpl-img1"},
+                **(create or {}),
+            },
+        }
+        provider = SubstrateSandboxProvider(config)
+        try:
+            yield provider, fake
+        finally:
+            await provider.aclose()
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+def _probe_count(fake: FakeAteEnv) -> int:
+    return sum(1 for s in fake.starts if s["command"][-1:] == ["true"])
+
+
+def test_create_returns_ready_handle_and_seeds_files():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(
+                SandboxSpec(files={"/task/input.json": '{"n": 1}'}, workdir="/task")
+            )
+            assert handle.provider_name == "substrate"
+            assert handle.sandbox_id.startswith("gym-")
+            env = fake.envs[handle.sandbox_id]
+            assert (env["template"], env["template_atespace"], env["atespace"]) == (
+                "default-template",
+                "ate-env",
+                "ate-env",
+            )
+            assert fake.files[(handle.sandbox_id, "/task/input.json")] == b'{"n": 1}'
+            assert fake.modes[(handle.sandbox_id, "/task/input.json")] == 0o644
+
+    run(t())
+
+
+def test_create_waits_for_readiness():
+    async def t():
+        async with stack() as (provider, fake):
+            fake.ready_after_probes = 2
+            handle = await provider.create(SandboxSpec())
+            assert handle.sandbox_id in fake.envs
+            assert _probe_count(fake) == 3  # two UNAVAILABLE + one success
+
+    run(t())
+
+
+def test_create_failure_raises_create_error():
+    async def t():
+        async with stack() as (provider, fake):
+            fake.create_code = grpc.StatusCode.RESOURCE_EXHAUSTED
+            with pytest.raises(SandboxCreateError, match="RESOURCE_EXHAUSTED"):
+                await provider.create(SandboxSpec())
+            assert fake.envs == {}
+
+    run(t())
+
+
+def test_create_readiness_timeout_cleans_up():
+    async def t():
+        async with stack() as (provider, fake):
+            fake.ready_after_probes = 10_000
+            with pytest.raises(SandboxCreateVerificationError, match="UNAVAILABLE"):
+                await provider.create(SandboxSpec(ready_timeout_s=0.05))
+            assert fake.envs == {}  # the half-created env was deleted
+
+    run(t())
+
+
+def test_image_resolves_via_mapping_or_fails():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec(image="img:1"))
+            assert fake.envs[handle.sandbox_id]["template"] == "tpl-img1"
+            with pytest.raises(SandboxCreateError, match="no ActorTemplate mapped"):
+                await provider.create(SandboxSpec(image="img:unmapped"))
+
+    run(t())
+
+
+def test_exec_nonzero_exit_returns_result_not_raise():
+    async def t():
+        async with stack() as (provider, _):
+            handle = await provider.create(SandboxSpec())
+            result = await provider.exec(handle, "exit 7")
+            assert (result.return_code, result.stderr, result.stdout) == (7, "boom", "")
+
+    run(t())
+
+
+def test_exec_merges_spec_env_and_cwd():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec(workdir="/w", env={"A": "1", "B": "spec"}))
+            await provider.exec(handle, "echo hi", env={"B": "call"})
+            sent = fake.starts[-1]
+            assert sent["command"] == ["sh", "-c", "echo hi"]
+            assert sent["env"] == {"A": "1", "B": "call"}
+            assert sent["cwd"] == "/w"
+            await provider.exec(handle, "echo hi", cwd="/other")
+            assert fake.starts[-1]["cwd"] == "/other"
+
+    run(t())
+
+
+@pytest.mark.parametrize(
+    ("timeout_s", "guest_deadline"),
+    [(3, "3"), (0.5, "1"), (0, "1"), (2.2, "3")],
+)
+def test_exec_timeout_wraps_command_guest_side(timeout_s, guest_deadline):
+    # `timeout 0` disables the limit in coreutils, so a sub-second or zero
+    # deadline must round up to 1s rather than floor to 0.
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            await provider.exec(handle, "sleep 100", timeout_s=timeout_s)
+            assert fake.starts[-1]["command"] == ["timeout", guest_deadline, "sh", "-c", "sleep 100"]
+
+    run(t())
+
+
+def test_exec_without_timeout_is_unwrapped():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            await provider.exec(handle, "sleep 1")
+            assert fake.starts[-1]["command"] == ["sh", "-c", "sleep 1"]
+
+    run(t())
+
+
+def test_exec_rejects_user():
+    async def t():
+        async with stack() as (provider, _):
+            handle = await provider.create(SandboxSpec())
+            with pytest.raises(ValueError, match="user"):
+                await provider.exec(handle, "id", user="root")
+
+    run(t())
+
+
+def test_exec_client_timeout_returns_sentinel_and_kills(monkeypatch):
+    monkeypatch.setattr(provider_mod, "_EXEC_GRACE_S", 0.05)
+
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            fake.hang.add("sleepy")
+            result = await provider.exec(handle, "sleepy", timeout_s=0.01)
+            assert result.return_code == -1
+            assert result.stdout is None
+            assert "timed out" in (result.stderr or "")
+            hung = [p for p in fake.procs.values() if p.argv[-1] == "sleepy"]
+            assert hung and hung[0].killed
+
+    run(t())
+
+
+def test_file_roundtrip(tmp_path: Path):
+    async def t():
+        async with stack() as (provider, _):
+            handle = await provider.create(SandboxSpec())
+            src = tmp_path / "up.bin"
+            src.write_bytes(b"\x00\x01payload-longer-than-one-chunk")
+            await provider.upload_file(handle, src, "/data/up.bin")
+            dst = tmp_path / "down" / "up.bin"
+            await provider.download_file(handle, "/data/up.bin", dst)
+            assert dst.read_bytes() == b"\x00\x01payload-longer-than-one-chunk"
+
+    run(t())
+
+
+def test_download_missing_file_raises(tmp_path: Path):
+    async def t():
+        async with stack() as (provider, _):
+            handle = await provider.create(SandboxSpec())
+            with pytest.raises(NotFoundError):
+                await provider.download_file(handle, "/nope", tmp_path / "out")
+
+    run(t())
+
+
+def test_status_running_stopped_unknown():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            assert await provider.status(handle) == SandboxStatus.RUNNING
+            fake.get_code = grpc.StatusCode.UNAVAILABLE
+            assert await provider.status(handle) == SandboxStatus.UNKNOWN
+            fake.get_code = None
+            await provider.close(handle)
+            assert await provider.status(handle) == SandboxStatus.STOPPED
+
+    run(t())
+
+
+@pytest.mark.parametrize(
+    ("env_status", "expected"),
+    [
+        (EnvironmentStatus.RESUMING, SandboxStatus.STARTING),
+        (EnvironmentStatus.SUSPENDED, SandboxStatus.RUNNING),  # parked is transparent
+        (EnvironmentStatus.PAUSED, SandboxStatus.RUNNING),
+        (EnvironmentStatus.CRASHED, SandboxStatus.ERROR),
+        (EnvironmentStatus.DELETING, SandboxStatus.STOPPED),
+        (EnvironmentStatus.UNSPECIFIED, SandboxStatus.UNKNOWN),
+    ],
+)
+def test_status_maps_env_lifecycle(env_status, expected):
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            fake.set_status(handle.sandbox_id, env_status)
+            assert await provider.status(handle) == expected
+
+    run(t())
+
+
+def test_status_does_not_touch_the_guest():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            before = len(fake.starts)
+            await provider.status(handle)
+            assert len(fake.starts) == before  # no probe exec, so no wake-up
+
+    run(t())
+
+
+def test_close_is_idempotent():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            await provider.close(handle)
+            await provider.close(handle)  # second delete hits NOT_FOUND; must not raise
+            assert handle.sandbox_id not in fake.envs
+
+    run(t())
+
+
+def test_close_logs_but_does_not_raise_on_delete_failure(caplog):
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(SandboxSpec())
+            fake.delete_code = grpc.StatusCode.INTERNAL
+            with caplog.at_level(logging.WARNING, logger="nemo_gym_substrate.provider"):
+                await provider.close(handle)  # best-effort: never raises
+            assert any(
+                "deleting substrate env" in r.message and "INTERNAL" in r.message
+                for r in caplog.records
+            )
+
+    run(t())
+
+
+def test_provider_options_override_template_and_atespace():
+    async def t():
+        async with stack() as (provider, fake):
+            handle = await provider.create(
+                SandboxSpec(provider_options={"template": "tpl-x", "atespace": "as-x"})
+            )
+            env = fake.envs[handle.sandbox_id]
+            assert (env["template"], env["template_atespace"], env["atespace"]) == (
+                "tpl-x",
+                "as-x",
+                "as-x",
+            )
+            # The handle remembers its atespace, so later calls route there.
+            assert await provider.status(handle) == SandboxStatus.RUNNING
+            await provider.close(handle)
+            assert handle.sandbox_id not in fake.envs
+
+    run(t())
+
+
+def test_namespace_is_accepted_as_alias_for_atespace():
+    async def t():
+        async with stack({"namespace": "legacy"}) as (provider, fake):
+            handle = await provider.create(SandboxSpec(provider_options={"namespace": "per-sb"}))
+            assert fake.envs[handle.sandbox_id]["atespace"] == "per-sb"
+            handle2 = await provider.create(SandboxSpec())
+            assert fake.envs[handle2.sandbox_id]["atespace"] == "legacy"
+
+    run(t())
+
+
+def test_atespace_and_namespace_together_rejected():
+    with pytest.raises(ValueError, match="either atespace or namespace"):
+        SubstrateSandboxProvider({"create": {"atespace": "a", "namespace": "b"}})
+
+
+def test_unknown_provider_option_rejected():
+    async def t():
+        async with stack() as (provider, _):
+            with pytest.raises(ValueError, match="unknown option"):
+                await provider.create(SandboxSpec(provider_options={"tempalte": "oops"}))
+
+    run(t())
+
+
+def test_unknown_config_key_rejected():
+    with pytest.raises(ValueError, match="unknown option"):
+        SubstrateSandboxProvider({"connection": {"api_urll": "typo"}})
+
+
+def test_create_timeout_on_unreachable_api():
+    async def t():
+        # A closed port refuses at once (UNAVAILABLE); either way it must not hang.
+        provider = SubstrateSandboxProvider(
+            {"connection": {"api_url": "127.0.0.1:9", "request_timeout_s": 0.5}}
+        )
+        try:
+            with pytest.raises(SandboxCreateError):
+                await provider.create(SandboxSpec())
+        finally:
+            await provider.aclose()
+
+    run(t())
+
+
+def test_aclose_is_safe_to_repeat_and_drops_the_client():
+    async def t():
+        async with stack() as (provider, _):
+            await provider.create(SandboxSpec())
+            assert provider._client is not None  # noqa: SLF001 - test seam
+            await provider.aclose()
+            await provider.aclose()
+            assert provider._client is None  # noqa: SLF001
+
+    run(t())
+
+
+def test_caller_owned_client_is_not_closed():
+    async def t():
+        async with serve() as (target, _):
+            client = Client(target)
+            try:
+                provider = SubstrateSandboxProvider(client=client)
+                handle = await provider.create(SandboxSpec())
+                await provider.aclose()
+                # Still usable: aclose() left the caller's client open.
+                info = await client.get(handle.sandbox_id)
+                assert info.id == handle.sandbox_id
+            finally:
+                await client.close()
+
+    run(t())
+
+
+def test_https_api_url_uses_a_tls_channel():
+    async def t():
+        provider = SubstrateSandboxProvider(
+            {"connection": {"api_url": "https://ate-env.example:443"}}
+        )
+        try:
+            assert provider._ate() is not None  # noqa: SLF001 - opens the channel, never dials
+            assert provider._channel is not None  # noqa: SLF001
+        finally:
+            await provider.aclose()
+        assert provider._channel is None  # noqa: SLF001
+
+    run(t())