RL Sandbox

Use Veris as the environment layer for an RL training run you operate. Your policy and rollout framework keep their existing tool schemas, message formatting, and training loop. The tools call the same vendor SDKs they use in production, but those SDKs point at isolated Veris services instead of live vendors.

This integration works with SkyRL, NeMo Gym and NeMo RL, Tinker Cookbook, or a custom rollout system. It does not require a framework-specific Veris SDK.

Where Veris sits

policy output
    ↓
your framework step / agent loop
    ↓
your tool implementation
    ↓
Stripe, Gmail, Salesforce, or another vendor SDK
    ↓  base URL changed
Veris sandbox

Veris sits at the vendor HTTP boundary. It does not implement a generic step() endpoint.

A tool can make several calls across several vendors, and the meaning of one model action differs between agents and frameworks. Keeping step() in the rollout framework preserves those semantics and avoids an extra network hop around every action.

Create a reusable starting world

An environment names the services in a task and points to a durable baseline. A sandbox is one disposable, mutable copy of that environment.

Create the baseline once, then start every rollout from it:

curl -X POST https://api.veris.ai/v1/environments \
  -H "Authorization: Bearer $VERIS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "billing-operations-rl",
    "services": ["stripe", "google-gmail"]
  }'

curl -X POST \
  "https://api.veris.ai/v1/environments/$VERIS_ENVIRONMENT_ID/sandboxes" \
  -H "Authorization: Bearer $VERIS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"ttl_minutes": 120}'

curl -X POST "$STRIPE_CONTROL_URL/veris/data" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "customers": [{"id": "cus_training", "email": "ops@acme.test"}]
    }
  }'

curl -X POST \
  "https://api.veris.ai/v1/environments/$VERIS_ENVIRONMENT_ID/sandboxes/$CURATOR_SANDBOX_ID/promote" \
  -H "Authorization: Bearer $VERIS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"clock_restore": "frozen"}'

Use clock_restore: "frozen" when identical logical time is part of the task. Frozen time also pauses scheduled callback delivery. Use "rebase" when the episode requires time to advance or webhooks to dispatch: the sandbox starts at the captured instant and then advances normally.

Run an episode

The following helper contains the complete framework-neutral integration. Create one instance for each active trajectory.

import os
import time
import requests
 
VERIS_API = os.getenv("VERIS_API_URL", "https://api.veris.ai")
VERIS_KEY = os.environ["VERIS_API_KEY"]
ENV_ID = os.environ["VERIS_ENVIRONMENT_ID"]
AUTH = {"Authorization": f"Bearer {VERIS_KEY}"}
 
 
class VerisRollout:
    def __init__(self, sandbox):
        self.id = sandbox["id"]
        self.services = {service["name"]: service for service in sandbox["services"]}
 
    @classmethod
    def start(cls, ttl_minutes=120, timeout_seconds=180):
        response = requests.post(
            f"{VERIS_API}/v1/environments/{ENV_ID}/sandboxes",
            headers=AUTH,
            json={"ttl_minutes": ttl_minutes},
            timeout=30,
        )
        response.raise_for_status()
        sandbox = response.json()
 
        try:
            deadline = time.monotonic() + timeout_seconds
            while sandbox["status"] != "ready":
                if sandbox["status"] == "failed":
                    raise RuntimeError(sandbox.get("failure_reason") or "sandbox failed")
                if time.monotonic() >= deadline:
                    raise TimeoutError(f"sandbox {sandbox['id']} did not become ready")
                time.sleep(1)
                response = requests.get(
                    f"{VERIS_API}/v1/environments/{ENV_ID}/sandboxes/{sandbox['id']}",
                    headers=AUTH,
                    timeout=30,
                )
                response.raise_for_status()
                sandbox = response.json()
        except Exception:
            requests.delete(
                f"{VERIS_API}/v1/environments/{ENV_ID}/sandboxes/{sandbox['id']}",
                headers=AUTH,
                timeout=30,
            )
            raise
 
        return cls(sandbox)
 
    def url(self, service):
        return self.services[service]["url"]
 
    def snapshot(self, service):
        control_url = self.services[service]["control_url"]
        response = requests.post(f"{control_url}/veris/snapshot", timeout=30)
        response.raise_for_status()
        return response.json()
 
    def requests(self, service, limit=1000):
        control_url = self.services[service]["control_url"]
        response = requests.get(
            f"{control_url}/veris/requests",
            params={"limit": limit},
            timeout=30,
        )
        response.raise_for_status()
        return response.json()["requests"]
 
    def close(self):
        response = requests.delete(
            f"{VERIS_API}/v1/environments/{ENV_ID}/sandboxes/{self.id}",
            headers=AUTH,
            timeout=30,
        )
        if response.status_code not in (204, 404):
            response.raise_for_status()

Point the existing SDK at the URL returned for that rollout:

import stripe
 
 
def configure_stripe(rollout):
    stripe.api_base = rollout.url("stripe")
    stripe.api_key = "sk_test_veris"
 
 
def refund_charge(charge_id: str):
    return stripe.Refund.create(charge=charge_id)

Any well-formed sandbox credential works in the default permissive auth mode, so your integration can usually keep the credential shape it already uses. Enable strict auth in the starting world when credential behavior is part of the task.

At episode end, read vendor-side state and compute a reward:

def score_refund(rollout, target_charge_id: str) -> float:
    state = rollout.snapshot("stripe")["state"]
    charge = next(c for c in state["charges"] if c["id"] == target_charge_id)
    refunds = [r for r in state["refunds"] if r["charge_id"] == target_charge_id]
 
    outcome = float(charge["refunded"] and len(refunds) == 1)
 
    # Optional process reward: discourage avoidable vendor errors and retries.
    trace = rollout.requests("stripe")
    avoidable_errors = sum(1 for request in trace if (request["status"] or 599) >= 400)
    return max(0.0, outcome - 0.05 * avoidable_errors)

POST /veris/snapshot is a state inspection endpoint, not the durable environment baseline. The promoted environment revision is the reusable starting point; a snapshot is the current state of one service in one live sandbox.

Framework adapters

The lifecycle is the same in every framework: create a sandbox at episode initialization, let your existing tool dispatcher call its service URLs, read state for reward, and delete the sandbox in a finally block.

import json
from skyrl_gym.envs.base_text_env import BaseTextEnv, BaseTextEnvStepOutput
 
from reward import score_refund
from veris_rollout import VerisRollout
 
 
class StripeBillingEnv(BaseTextEnv):
    def __init__(self, env_config, extras):
        super().__init__()
        self.task = extras["task"]
        self.rollout = None
 
    def init(self, prompt):
        self.rollout = VerisRollout.start()
        configure_tools(stripe_base_url=self.rollout.url("stripe"))
        return prompt, {}
 
    def step(self, action: str) -> BaseTextEnvStepOutput:
        call = parse_tool_call(action)
        result = TOOLS[call.name](**call.args)
        reward = score_refund(self.rollout, self.task["target_charge_id"])
        done = task_is_complete(self.rollout, self.task)
        return BaseTextEnvStepOutput(
            observations=[{"role": "tool", "content": json.dumps(result)}],
            reward=reward,
            done=done,
            metadata={"sandbox_id": self.rollout.id},
        )
 
    def close(self):
        if self.rollout:
            self.rollout.close()

import asyncio
 
from fastapi import Request
from pydantic import Field
from nemo_gym.base_resources_server import (
    BaseResourcesServerConfig,
    BaseSeedSessionRequest,
    BaseSeedSessionResponse,
    BaseVerifyRequest,
    BaseVerifyResponse,
    SimpleResourcesServer,
)
from nemo_gym.server_utils import SESSION_ID_KEY
 
from reward import score_refund
from veris_rollout import VerisRollout
 
 
class BillingVerifyRequest(BaseVerifyRequest):
    target_charge_id: str
 
 
class VerisResourcesServer(SimpleResourcesServer):
    config: BaseResourcesServerConfig
    rollouts: dict[str, VerisRollout] = Field(default_factory=dict)
 
    async def seed_session(
        self, request: Request, body: BaseSeedSessionRequest
    ) -> BaseSeedSessionResponse:
        session_id = request.session[SESSION_ID_KEY]
        self.rollouts[session_id] = await asyncio.to_thread(VerisRollout.start)
        return BaseSeedSessionResponse()
 
    async def verify(
        self, request: Request, body: BillingVerifyRequest
    ) -> BaseVerifyResponse:
        session_id = request.session[SESSION_ID_KEY]
        rollout = self.rollouts.pop(session_id)
        try:
            reward = await asyncio.to_thread(
                score_refund, rollout, body.target_charge_id
            )
            return BaseVerifyResponse(**body.model_dump(), reward=reward)
        finally:
            await asyncio.to_thread(rollout.close)

from tinker_cookbook.tool_use import build_agent_tool_env
from tinker_cookbook.tool_use.tools import tool
 
from reward import score_refund
from tools import refund_charge_with_client
from veris_rollout import VerisRollout
 
 
def build_billing_env(task, renderer):
    rollout = VerisRollout.start()
    stripe_client = make_stripe_client(base_url=rollout.url("stripe"))
 
    @tool
    async def refund_charge(charge_id: str):
        """Refund a Stripe charge."""
        return await refund_charge_with_client(stripe_client, charge_id)
 
    async def reward_fn(_history):
        reward = score_refund(rollout, task["target_charge_id"])
        return reward, {"task_success": reward}
 
    env = build_agent_tool_env(
        renderer=renderer,
        initial_messages=[{"role": "user", "content": task["prompt"]}],
        tools=[refund_charge],
        reward_fn=reward_fn,
        max_turns=8,
    )
    env.veris_rollout = rollout  # close from the group/trajectory finalizer
    return env

for batch in dataset:
    rollouts = [VerisRollout.start() for _ in range(group_size)]
    try:
        trajectories = await run_agents(batch, rollouts)
        rewards = [score_refund(r, batch.target_charge_id) for r in rollouts]
        await trainer.update(trajectories, rewards)
    finally:
        for rollout in rollouts:
            rollout.close()

Parallel rollouts

Allocate one sandbox per active trajectory. Never let two trajectories mutate the same sandbox.

For a promoted environment, restoring the episode means deleting the used sandbox and creating another one. POST .../sandboxes/{id}/reset is reserved for profile-backed environments and returns 409 for a promoted baseline so it cannot silently replace the curated world with default seed data.

Do not promote or roll back an environment while a training run is active. Record the environment’s baseline.revision_id with the run configuration so the starting world is auditable alongside the model checkpoint.

Reward signals

Veris supports two complementary reward sources:

• Outcome reward from POST {control_url}/veris/snapshot or paginated GET {control_url}/veris/data?entity_type=....
• Process reward from GET {control_url}/veris/requests, including the vendor route, status, request and response bodies, and state version after each call.

For multi-service tasks, read each relevant service after the final tool call and combine the predicates in your reward code. Keep reward reads outside the agent’s tool surface so the policy cannot inspect ground truth directly.

Evidence for process graders

Veris records the vendor-side evidence from every rollout. Combine it with the trajectory captured by your training framework to evaluate how the agent reached an outcome.

evidence = {
    # Captured by SkyRL, NeMo RL, Tinker, or your rollout loop
    "trajectory": framework_trajectory,
 
    # Captured by Veris
    "environment": {
        "stripe": {
            "requests": rollout.requests("stripe"),
            "final_state": rollout.snapshot("stripe")["state"],
        }
    },
}
 
result = your_behavior_judge(
    behavior="Verify the customer before issuing a refund",
    evidence=evidence,
)
# {"result": "pass", "reason": "Identity was verified before the refund call."}

Process graders can check whether the agent verified identity before a destructive action, recovered appropriately from a vendor error, or claimed success after the vendor rejected a request.

Operational notes

• Sandboxes are TTL-scoped. Delete them explicitly; TTL cleanup is a fallback.
• Treat failed provisioning as terminal and preserve failure_reason in rollout metadata.
• Use control_url for /veris/*. For HTTP services it equals url; for PostgreSQL, url is a database DSN and control_url is the HTTP inspection surface.
• A frozen baseline keeps vendor time fixed and pauses callbacks. Advance or rebase time only when the task defines how time-bearing client protocols are handled.
• Veris resets the dependency world. Your harness must separately reset the agent process, memory, caches, filesystem, and any database not hosted in the Veris sandbox.

API reference