diff --git a/fern/versions/latest/pages/agent-server/agent-skills.mdx b/fern/versions/latest/pages/agent-server/agent-skills.mdx index 8431643002..b8c9e4ebc4 100644 --- a/fern/versions/latest/pages/agent-server/agent-skills.mdx +++ b/fern/versions/latest/pages/agent-server/agent-skills.mdx @@ -1,7 +1,7 @@ --- title: "Agent Skills" description: "Evaluate agent skills as a run-level variable, decoupled from the dataset" -position: 3 +position: 4 --- Skills are reusable units of operational knowledge an agent can load at runtime, following the open [Agent Skills standard](https://agentskills.io/specification) used by Claude Code and Codex CLI. A skill is a **directory** containing a `SKILL.md` file (YAML frontmatter + markdown body) plus optional supporting files. diff --git a/fern/versions/latest/pages/agent-server/remote-agent.mdx b/fern/versions/latest/pages/agent-server/remote-agent.mdx new file mode 100644 index 0000000000..71ae4bb803 --- /dev/null +++ b/fern/versions/latest/pages/agent-server/remote-agent.mdx @@ -0,0 +1,284 @@ +--- +title: "Drive a Remote Agent" +description: "Evaluate an agent service you host yourself — a composition of OpenAI Responses-compliant agents, with Gym driving the loop" +position: 3 +--- + +# Drive a Remote Agent + +The `remote_agent` server lets an agent that runs as **its own HTTP service** — in your repo, on +your infrastructure — be driven by standard rollout collection. Your service implements an +endpoint **compliant with the OpenAI `/v1/responses` contract**, and the two servers compose as +Responses-speaking agents: each call your service receives the conversation so far and returns +what it wants to do next. Gym runs the loop, executes environment tools, holds the session, and +verifies. +Your results land in the standard artifacts (`gym eval profile`, aggregation, and training +pipelines all work unchanged). + +``` +collector ──/run──▶ remote_agent (Gym) ──POST /v1/responses──▶ your service ⟲ your model, + │ ▲ your own tools + │ └── Gym-tool results appended, (any number of internal steps, + │ loop repeats then: Gym-tool asks + ▼ or a final answer) + resources server (seed / Gym tools / verify — never exposed to your service) +``` + +## Who does what + +**Gym is responsible for:** + +1. Seeding a fresh environment session per rollout and holding its cookies — session state + and the task's answer key (`verifier_metadata`) never reach your service. +2. Driving the loop: calling your service, executing the tool calls it asks for against the + environment, appending the results, and calling again. +3. All failure handling: bounded connect retries, per-call and whole-rollout timeouts, and + converting every failure into a reward-0 row in the failures sidecar (a bad rollout never + crashes the collection run). +4. Verifying the finished trajectory and reporting the reward. + +**Your service is responsible for:** + +1. Implementing `POST {agent_base_url}/v1/responses` and answering every call with a valid + Responses API object (required fields and types are enforced, unknown extra fields are + tolerated; an invalid object is a terminal, non-retried failure). +2. Deciding what to do next on each call. Within a call your service can do **anything** — + run its own model for any number of turns, execute its own tools, spawn sub-agents. There + are exactly two reasons to return: you need a **Gym-hosted tool** executed (return the ask), + or the rollout is **finished** (return the answer). +3. Being callable N times per rollout. Each call carries the full conversation so far, so a + stateless service needs nothing extra; if you want per-rollout state, set a cookie — Gym + echoes your cookies back on every subsequent call of the same rollout. +4. Reporting `usage` (or omitting it entirely — allowed, but your token metrics are empty). + +## The contract, exactly + +**Every request you receive** is the task's `responses_create_params` with the conversation +accumulated in `input`: + +1. `input` — the task messages, plus (from turn 2 onward) everything so far: your previous + output items and one `function_call_output` per tool call Gym executed for you. +2. `tools` — the tool schemas this environment serves, verbatim from the dataset row. These + are the only tools you may ask Gym to execute. +3. Your own cookies from earlier calls of this rollout, echoed back. + +**Every response you return** is one Responses API object whose `output` decides the next step. +Returning does not mean your agent is done thinking — it means one of two things: "I need a Gym +tool" or "I'm finished." Everything your agent did internally since the last call (its own model +turns, its own tool executions) either stays private or rides along as paired call+output +records: + +| You return | Gym does | +|---|---| +| One or more `function_call` items **without** a matching `function_call_output` (same `call_id`, same response) | Executes each against the resources server, appends each result as a `function_call_output`, and calls you again. | +| `function_call` + `function_call_output` **pairs** (same `call_id`) | Nothing — that's your own internal tool record; it passes into the trajectory untouched. | +| An assistant `message` and no unpaired calls | The rollout is done: Gym merges the full conversation into one trajectory and verifies it. | +| `incomplete_details` set | The loop stops and the trajectory so far is verified. | + +**Expectations and edge semantics:** + +1. Ask only for tools the request's `tools` declared. An unknown tool name is not an error — + Gym sends the resources server's error text (e.g. a 404) back to you as that call's + `function_call_output`, and the rollout continues. +2. Malformed `arguments` (not valid JSON) likewise come back to you as an error output, + never a crash. +3. Never return an unpaired `function_call` for a tool you already executed yourself — + unpaired means "Gym, run this." +4. One rollout = one conversation. Requests of the same rollout share your cookies; + different rollouts (including retries of the same task) start clean. + +## Quickstart + +A complete, realistic service against the in-repo stateful counter environment +(`example_session_state_mgmt`; tasks read "add 1 then add 2 then get the count"). The **agent +brain is the Claude Code CLI**: each call, the service renders the conversation into a prompt, +lets Claude decide, and translates the decision into tool asks or a final answer. Requirements +on the service's machine: `claude` on PATH and authenticated (a logged-in CLI or +`ANTHROPIC_API_KEY` in the environment); pick the model with `CLAUDE_MODEL` (default `haiku`). + +```python +# service.py — the Claude Code CLI as the agent's brain +import json, os, subprocess +from fastapi import FastAPI + +app = FastAPI() +MODEL = os.environ.get("CLAUDE_MODEL", "haiku") + +PROMPT = """You are an agent solving a task by calling tools. You do not execute tools \ +yourself; you ask for them and the results come back in the conversation. + +Task conversation so far: +{conversation} + +Tools you may ask for (JSON schemas): +{tools} + +Reply with ONLY one JSON object, no prose, no code fences: +- to call tools: {{"tool_calls": [{{"name": "", "arguments": {{...}}}}]}} +- to finish: {{"final": ""}} +Finish once the conversation contains enough tool results to answer; the final answer must be \ +exactly what the task asks for. +""" + + +def render(items: list) -> str: + lines = [] + for item in items: + kind = item.get("type", "message") + if kind == "message": + content = item.get("content") + if isinstance(content, list): + content = " ".join(c.get("text", "") for c in content) + lines.append(f"[{item.get('role', 'user')}] {content}") + elif kind == "function_call": + lines.append(f"[you asked for] {item['name']}({item.get('arguments')})") + elif kind == "function_call_output": + lines.append(f"[tool result] {item.get('output')}") + return "\n".join(lines) + + +@app.post("/v1/responses") +def responses(params: dict): # sync handler: FastAPI runs it in a threadpool, so + # concurrent rollouts don't block each other on the subprocess below. + prompt = PROMPT.format(conversation=render(params["input"]), tools=json.dumps(params.get("tools", []))) + # `claude -p` runs Claude's full agentic loop in one invocation: read-only file tools + # (Read/Glob/Grep) work by default; permission-gated tools (WebSearch, Bash, ...) are + # auto-denied headlessly unless pre-approved, e.g. --allowedTools "WebSearch,Bash". + # Either way Claude uses its OWN tools multi-step inside this single call — returning + # to Gym is only for Gym-hosted tools or the final answer. + proc = subprocess.run( + ["claude", "-p", prompt, "--output-format", "json", "--model", MODEL], + capture_output=True, timeout=300, check=True, + ) + # The CLI's print mode returns its own JSON envelope (NOT the Anthropic Messages format): + # the final text sits in "result". The Responses object Gym expects is built by hand below — + # translating whatever your brain speaks into Responses format is the service's job. + payload = json.loads(proc.stdout) + decision = json.loads(payload["result"].strip().strip("`").removeprefix("json").strip()) + turn = sum(1 for i in params["input"] if i.get("type") == "function_call_output") + + if "tool_calls" in decision: # unpaired asks: Gym executes these and calls us again + output = [{"type": "function_call", "id": f"fc_{turn}_{k}", "call_id": f"c_{turn}_{k}", + "name": call["name"], "arguments": json.dumps(call.get("arguments", {}))} + for k, call in enumerate(decision["tool_calls"])] + else: # a final assistant message ends the rollout + output = [{"type": "message", "role": "assistant", "status": "completed", "id": f"m_{turn}", + "content": [{"type": "output_text", "text": str(decision["final"]), "annotations": []}]}] + + usage = payload.get("usage") or {} + return { + "id": "claude-cli-service", "created_at": 0.0, "model": f"claude-{MODEL}", "object": "response", + "output": output, + "parallel_tool_calls": False, "tools": [], "tool_choice": "auto", + "usage": { + "input_tokens": usage.get("input_tokens", 0), + "output_tokens": usage.get("output_tokens", 0), + "total_tokens": usage.get("input_tokens", 0) + usage.get("output_tokens", 0), + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + } +``` + +One config file wires the environment and the agent together (this exact shape drove the live +end-to-end run — the only thing that changes for your own benchmark is the resources-server +block and the ref name): + +```yaml +# counter_remote.yaml +example_session_state_mgmt_resources_server: + resources_servers: + example_session_state_mgmt: + entrypoint: app.py + domain: agent + +remote_agent: + responses_api_agents: + remote_agent: + entrypoint: app.py + # Your service's address — typically another machine, a container, or a cloud box. + # (For a first try with everything on one machine, http://localhost:9000 works too.) + agent_base_url: http://your-agent-host:9000 + resources_server: + type: resources_servers + name: example_session_state_mgmt_resources_server # the top-level key above + concurrency: 4 + max_steps: 8 +``` + +Run the three pieces in **separate terminals** — the first two are long-running servers, the +third is the driver: + +```bash +# Terminal 1 — on your service's machine: +uvicorn service:app --host 0.0.0.0 --port 9000 +``` + +```bash +# Terminal 2 — on the Gym machine: environment + remote_agent (leave running): +gym env start "+config_paths=[counter_remote.yaml]" +``` + +```bash +# Terminal 3 — on the Gym machine, once the servers report ready: +gym eval run --no-serve +agent_name=remote_agent \ + +input_jsonl_fpath=resources_servers/example_session_state_mgmt/data/example.jsonl \ + +output_jsonl_fpath=results/rollouts.jsonl +``` + +Expected: 5 rollouts, `mean/reward: 1.0`, and each trajectory reads +`function_call → function_call_output → ... → message`. A live end-to-end run of exactly this +setup — the service in a container on its own network, Claude deciding every call — scored 5/5. + + +Nothing here is specific to FastAPI or the Claude CLI: any HTTP stack works (the live run used +stdlib `http.server`), and any brain works — render the conversation into your agent's context, +let it decide, translate the decision into unpaired `function_call` items or a final message. +Errors on your side are safe: a 5xx or timeout becomes a reward-0 row in Gym's failures sidecar, +never a crashed run. + + +## Configuration knobs + +| Field | Default | What it does | +|---|---|---| +| `agent_base_url` | required | Your service's base URL. Validated: `http(s)` only, no query string or fragment, no embedded credentials. This is the only network direction — Gym calls you; your service never needs to reach Gym. | +| `resources_server` | required | The environment that seeds, serves tools for, and verifies each rollout. Swap benchmarks by changing this ref — your service doesn't change. | +| `concurrency` | `32` | Maximum rollouts in flight against your service, enforced server-side. | +| `remote_responses_timeout_secs` | `1800` | Wallclock bound on ONE call to your service (a rollout makes one call per loop step). | +| `run_timeout_secs` | `2100` | Bound on a whole rollout (seed + every loop step + verify), started after the concurrency slot is acquired — queue wait doesn't count. | +| `max_steps` | unset | Maximum loop steps per rollout. Unset leaves `run_timeout_secs` as the only bound. | + +## Failure handling and resume + +Failures never crash a collection run. A down service (3 connection attempts, then fail), a +timed-out call, a malformed reply, or a verifier error becomes a reward-0 row with +`_ng_failure_class: "remote_agent_error"` in the failures sidecar (for +`results/rollouts.jsonl` the sidecar is `results/rollouts_failures.jsonl`) — +the main rollouts file stays clean, and `+resume_from_cache=true` retries failed tasks up to the +attempt cap (`NEMO_GYM_MAX_ROLLOUT_ATTEMPTS`, default 3; terminal failures — an invalid response +shape — are not retried). Connection failures and timeouts name the failing URL or the timeout +knob involved. + +## Gotchas + +- **Your service is called multiple times per rollout.** Design for it: everything you need is + in each request's `input`, or in your own cookies. +- **Redirects are rejected, not followed.** Any `3xx` from your service fails the rollout with + the `Location` shown — point `agent_base_url` at the final address. (Followed, a `301/302/303` + would silently re-issue the POST as a body-less GET; a `307/308` would silently re-send the + task to an address you never configured.) +- **Partial `usage` fails validation.** Report the full object — `input_tokens`, + `output_tokens`, `total_tokens`, `input_tokens_details: {cached_tokens}`, + `output_tokens_details: {reasoning_tokens}` — or omit `usage` entirely (allowed; token + metrics are then empty, with a warning). +- **A previous run's output can be reused as the input dataset** (for example, re-running the + tasks in a failures sidecar). Output rows carry the old run's results (`reward`, `response`, + `error`, internal `_ng_*` flags); those fields are stripped from each row on the way in, so a + stale result can't collide with or leak into the new run's output. +- **Exposing your service on the public internet needs network-level trust.** Gym sends no + credential on the `/v1/responses` call (`agent_base_url` rejects embedded credentials), so an + internet-exposed service should sit behind a VPN, private network, IP allowlist, or + authenticating tunnel. Note that proxies configured via `HTTP_PROXY`/`HTTPS_PROXY` environment + variables are ignored (Gym's HTTP client does not read them), so a host that can only reach the + internet through such a proxy cannot reach your service. diff --git a/responses_api_agents/remote_agent/README.md b/responses_api_agents/remote_agent/README.md new file mode 100644 index 0000000000..a9cd4bd111 --- /dev/null +++ b/responses_api_agents/remote_agent/README.md @@ -0,0 +1,46 @@ +# Remote Agent + +An agent server that drives an agent service you host yourself — in your own repo, on your own +infrastructure. Your service implements one endpoint **compliant with the OpenAI +`/v1/responses` contract**, and the two servers compose as Responses-speaking agents: each call +your service receives the conversation so far and returns what it wants to do next. Gym runs +the loop. + +## Who does what + +Gym: seeds a fresh environment session per rollout and holds its cookies (session state and +`verifier_metadata` never reach your service), executes the tool calls your service asks for +against the resources server, appends the results and calls your service again, converts every +failure into a reward-0 sidecar row (never a crashed run), and verifies the finished trajectory. + +Your service: answers each call with a valid Responses API object. Within a call it can do +anything — its own model turns, its own tools, sub-agents; there are exactly two reasons to +return: it needs a Gym-hosted tool executed, or the rollout is finished. It tolerates being +called N times per rollout (each request carries the full conversation; set a cookie if you want +per-rollout state — Gym echoes your cookies back within the rollout) and reports full `usage` or +omits it. + +## The response contract + +- `function_call` items **without** a matching `function_call_output` (same `call_id`, same + response) are asks: Gym executes them on the resources server and feeds the results back as + `function_call_output` items on the next call. +- `function_call` + `function_call_output` **pairs** are your own internal tool records; they + pass into the trajectory untouched. +- An assistant `message` with no unpaired calls finishes the rollout; Gym merges the whole + conversation into one trajectory and verifies it. +- Unknown tool names and malformed arguments are not crashes: the error text comes back to you + as that call's output and the rollout continues. An invalid Responses object is a terminal, + non-retried failure. + +The tool schemas your service may ask for arrive in every request's `tools` field, verbatim from +the dataset row. + +## Run + +```bash +gym env start --resources-server # plus this agent's config +gym eval run --no-serve +agent_name=remote_agent +input_jsonl_fpath=... +output_jsonl_fpath=... +``` + +Knobs and the full contract: see the "Drive a Remote Agent" docs page. diff --git a/responses_api_agents/remote_agent/__init__.py b/responses_api_agents/remote_agent/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/responses_api_agents/remote_agent/app.py b/responses_api_agents/remote_agent/app.py new file mode 100644 index 0000000000..1e61d09ff2 --- /dev/null +++ b/responses_api_agents/remote_agent/app.py @@ -0,0 +1,515 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# 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. +"""Agent server that drives a user-hosted remote agent service through Gym's tool loop. + +The remote service implements ONE endpoint, ``POST {agent_base_url}/v1/responses``, composing +with this server as OpenAI Responses-compliant agents: each call it receives the conversation +so far (the row's +``responses_create_params`` with the accumulated output and tool results appended to +``input``) and returns a Responses API object. To have Gym execute a tool from the +environment, it returns a ``function_call`` item WITHOUT a matching +``function_call_output``; tool calls it already answered itself (its own internal tools) +ride along as paired call+output items and are passed through untouched. Gym runs the +loop: it executes unpaired calls against the resources server and re-posts until the +service returns a final assistant message. + +The resources server is never exposed to the service: tool execution, session cookies, +and ``verifier_metadata`` all stay inside Gym. + +Failures never raise out of ``/run``: every failure (remote endpoint down, timeout, +malformed reply, seed/verify errors) becomes a reward-0 verify response carrying the +``_ng_failure_class`` sentinel, which rollout collection routes to the failures +sidecar and retries on resume. +""" + +import asyncio +import json +from traceback import print_exc +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import urlparse + +import orjson +from aiohttp import ClientOSError, ClientTimeout, ServerDisconnectedError +from fastapi import Body, Request, Response +from pydantic import ConfigDict, PrivateAttr, field_validator +from pydantic import ValidationError as PydanticValidationError + +from nemo_gym.base_resources_server import ( + AggregateMetrics, + AggregateMetricsRequest, + BaseRunRequest, + BaseVerifyResponse, +) +from nemo_gym.base_responses_api_agent import BaseResponsesAPIAgentConfig, SimpleResponsesAPIAgent +from nemo_gym.config_types import ResourcesServerRef +from nemo_gym.global_config import SKILLS_REF_KEY_NAME +from nemo_gym.openai_utils import ( + NeMoGymEasyInputMessage, + NeMoGymFunctionCallOutput, + NeMoGymResponse, + NeMoGymResponseCreateParamsNonStreaming, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseOutputMessage, +) +from nemo_gym.rollout_collection import NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY +from nemo_gym.server_utils import ( + get_global_aiohttp_client, + get_response_json, + is_global_aiohttp_client_request_debug_enabled, + raise_for_status, +) + + +REMOTE_AGENT_FAILURE_CLASS = "remote_agent_error" + +_REMOTE_MAX_TRIES = 3 +_REMOTE_RETRY_SLEEP_SECS = 0.5 +_FAILURE_PRINT_HEAD = 5 +_FAILURE_PRINT_INTERVAL = 100 +_AGGREGATE_PROXY_TIMEOUT_SECS = 600.0 + +# Result/routing keys this server itself produces. Input rows may carry stale copies +# (e.g. a rollouts or failures JSONL re-fed as a dataset); they must never collide with +# the fresh values or leak through the verify echo into the dispatcher's routing. +_RESERVED_RESULT_KEYS = ("reward", "response", "error", NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY) + + +class RemoteAgentError(RuntimeError): + """A rollout-level failure from the remote hop; retryable on resume.""" + + +class RemoteAgentTerminalError(RemoteAgentError): + """A failure that will not fix itself on retry (e.g. an invalid response shape). + + run() reaches responses() over an HTTP self-post, so this class's NAME is the wire + contract: the exception middleware serializes it into the 500 body and run() matches + the name string to set the terminal routing flag. + """ + + +def normalize_remote_url(url: str) -> str: + """Validate the remote service URL and strip any trailing slash.""" + normalized = url.strip().rstrip("/") + parsed = urlparse(normalized) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise ValueError(f"agent_base_url must be an absolute http:// or https:// URL, got {url!r}") + # "/v1/responses" is string-appended; anything after "?" or "#" would swallow it (bare + # delimiters parse as an empty query/fragment, so check the string itself). + if "?" in normalized or "#" in normalized or parsed.params: + raise ValueError( + f"agent_base_url must not carry a query string or fragment, got {url!r}. " + "Pass auth material via your service's own configuration instead." + ) + # Credentials would be stamped into logged configs and error messages; never echo the URL. + if parsed.username or parsed.password: + raise ValueError( + "agent_base_url must not embed credentials (user:pass@host). " + "Pass auth material via your service's own configuration instead." + ) + return normalized + + +class RemoteAgentConfig(BaseResponsesAPIAgentConfig): + agent_base_url: str + resources_server: ResourcesServerRef + concurrency: int = 32 + # Per-call bound on one POST to the remote service; a rollout makes one call per loop step. + remote_responses_timeout_secs: float = 1800.0 + # Bound on the whole /run body (seed + the full agent/tool loop + verify), applied after + # the semaphore is acquired so queue wait does not count against it. The collector's + # named-agent hop carries no timeout of its own; this is the only whole-rollout bound. + run_timeout_secs: float = 2100.0 + # Maximum loop steps (remote calls) per rollout; None leaves run_timeout_secs as the only bound. + max_steps: Optional[int] = None + + @field_validator("agent_base_url") + @classmethod + def _normalize_agent_base_url(cls, value: str) -> str: + return normalize_remote_url(value) + + +class RemoteAgentRunRequest(BaseRunRequest): + model_config = ConfigDict(extra="allow") + + +class RemoteAgentVerifyResponse(BaseVerifyResponse): + model_config = ConfigDict(extra="allow") + + +class RemoteAgent(SimpleResponsesAPIAgent): + config: RemoteAgentConfig + sem: Optional[asyncio.Semaphore] = None + _num_failures: int = PrivateAttr(default=0) + _warn_counts: Dict[str, int] = PrivateAttr(default_factory=dict) + model_config = ConfigDict(arbitrary_types_allowed=True) + + def model_post_init(self, __context: Any) -> None: + self.sem = asyncio.Semaphore(self.config.concurrency) + + async def responses( + self, + request: Request, + response: Response, + body: NeMoGymResponseCreateParamsNonStreaming = Body(), + ) -> NeMoGymResponse: + body = body.model_copy(deep=True) + + if isinstance(body.input, str): + body.input = [NeMoGymEasyInputMessage(role="user", content=body.input)] + + new_outputs = [] + usage = None + step = 0 + agent_server_cookies = None # the service's own cookies, round-tripped so it can keep per-rollout state + resources_server_cookies = request.cookies + + while True: + step += 1 + new_body = body.model_copy(update={"input": body.input + new_outputs}) + + agent_response, agent_server_cookies = await self._post_agent_responses(new_body, agent_server_cookies) + + output = agent_response.output + new_outputs.extend(output) + + if not usage: + usage = agent_response.usage + agent_response.usage = None + + if usage and agent_response.usage: + usage.input_tokens += agent_response.usage.input_tokens + usage.output_tokens += agent_response.usage.output_tokens + usage.total_tokens += agent_response.usage.total_tokens + + # TODO support more advanced token details + usage.input_tokens_details.cached_tokens = 0 + usage.output_tokens_details.reasoning_tokens = 0 + + if agent_response.incomplete_details: + break + + # Execute only unpaired calls: a call the service already answered itself (matching + # function_call_output in the same response) is its own internal-tool record and + # passes through into the trajectory untouched. + answered_call_ids = {o.call_id for o in output if o.type == "function_call_output"} + all_fn_calls: List[NeMoGymResponseFunctionToolCall] = [ + o for o in output if o.type == "function_call" and o.call_id not in answered_call_ids + ] + all_output_messages: List[NeMoGymResponseOutputMessage] = [ + o for o in output if o.type == "message" and o.role == "assistant" + ] + if not all_fn_calls and all_output_messages: + break + + for output_function_call in all_fn_calls: + try: + parsed_arguments = json.loads(output_function_call.arguments) + except (json.JSONDecodeError, TypeError) as e: + # Malformed arguments go back to the service as a tool error output + # instead of crashing the rollout; repr(e) keeps the exception type + # even when str(e) is empty. + tool_response = NeMoGymFunctionCallOutput( + type="function_call_output", + call_id=output_function_call.call_id, + output=json.dumps({"error": f"Invalid tool call arguments: {e!r}"}), + ) + new_outputs.append(tool_response) + continue + + api_response = await self.server_client.post( + server_name=self.config.resources_server.name, + url_path=f"/{output_function_call.name}", + json=parsed_arguments, + cookies=resources_server_cookies, + ) + # No raise_for_status: a tool error (unknown tool, invalid call) is a valid + # result the service should see and react to. + resources_server_cookies = api_response.cookies + + tool_response = NeMoGymFunctionCallOutput( + type="function_call_output", + call_id=output_function_call.call_id, + output=(await api_response.content.read()).decode(), + ) + new_outputs.append(tool_response) + + if self.config.max_steps and step >= self.config.max_steps: + break + + # Resources-server cookies propagate for downstream verification; the service's own + # cookies are its private session and deliberately stay out. + for k, v in resources_server_cookies.items(): + response.set_cookie(k, v) + + agent_response.output = new_outputs + agent_response.usage = usage + return agent_response + + async def _post_agent_responses( + self, new_body: NeMoGymResponseCreateParamsNonStreaming, cookies: Optional[Dict[str, str]] + ) -> Tuple[NeMoGymResponse, Dict[str, str]]: + """One hardened POST to the remote service. Returns (validated response, its cookies).""" + remote_url = f"{self.config.agent_base_url}/v1/responses" + client = get_global_aiohttp_client() + # exclude_unset keeps the wire payload to the fields the dataset row (and the loop) + # actually set, never materialized None defaults. + data = orjson.dumps(new_body.model_dump(exclude_unset=True)) + headers = {"Content-Type": "application/json"} + timeout = ClientTimeout(total=self.config.remote_responses_timeout_secs) + + response = None + last_connect_error: Optional[BaseException] = None + for num_try in range(1, _REMOTE_MAX_TRIES + 1): + try: + # Never follow redirects: aiohttp re-issues 301/302/303 as a body-less GET and + # re-sends 307/308 to an address the user never configured; fail with the 3xx. + response = await client.request( + "POST", + remote_url, + data=data, + headers=headers, + cookies=cookies or {}, + timeout=timeout, + allow_redirects=False, + ) + break + except (ClientOSError, ServerDisconnectedError) as e: + # Refused/reset (ClientOSError) and keepalive races (ServerDisconnectedError) + # are transient connection noise; everything else fails fast. + last_connect_error = e + if num_try < _REMOTE_MAX_TRIES: + await asyncio.sleep(_REMOTE_RETRY_SLEEP_SECS) + except asyncio.TimeoutError: + raise RemoteAgentError( + f"remote /v1/responses timed out after {self.config.remote_responses_timeout_secs}s " + "(remote_responses_timeout_secs; raise it if agent calls legitimately run longer)" + ) from None + except Exception as e: + if is_global_aiohttp_client_request_debug_enabled(): + print_exc() + raise RemoteAgentError(f"{type(e).__name__}: {e}") from e + if response is None: + raise RemoteAgentError( + f"could not reach the remote service after {_REMOTE_MAX_TRIES} tries " + f"({type(last_connect_error).__name__}: {last_connect_error}). " + f"Is your service running at {self.config.agent_base_url}?" + ) + + # client.request() returns once headers arrive; the body read can still fail + # (mid-body disconnect, deadline). + try: + content = await response.read() + except Exception as e: + if is_global_aiohttp_client_request_debug_enabled(): + print_exc() + raise RemoteAgentError(f"reading the response body failed: {type(e).__name__}: {e}") from e + # response.ok is `status < 400`; reject 3xx explicitly (redirects are not followed). + if not response.ok or response.status >= 300: + if is_global_aiohttp_client_request_debug_enabled(): + print( + f"[remote_agent] full HTTP {response.status} body: {content.decode(errors='replace')}", flush=True + ) + location = response.headers.get("Location", "") + raise RemoteAgentError( + f"HTTP {response.status}" + + (f" (redirect to {location}; fix agent_base_url to point at the final address)" if location else "") + + f": {content[:500].decode(errors='replace')}" + ) + try: + result = orjson.loads(content) + except orjson.JSONDecodeError as e: + raise RemoteAgentError(f"response is not valid JSON: {e}") from e + if not isinstance(result, dict): + raise RemoteAgentError(f"expected a JSON object from /v1/responses, got {type(result).__name__}") + + try: + validated = NeMoGymResponse.model_validate(result) + except PydanticValidationError as e: + if is_global_aiohttp_client_request_debug_enabled(): + print(f"[remote_agent] full validation error: {e}", flush=True) + # A shape error will not fix itself on retry. + raise RemoteAgentTerminalError( + f"remote service returned an invalid Responses API object: {str(e)[:500]}" + ) from e + + merged_cookies = dict(cookies or {}) + merged_cookies.update({k: morsel.value for k, morsel in response.cookies.items()}) + return validated, merged_cookies + + async def run(self, request: Request, body: RemoteAgentRunRequest = Body()) -> RemoteAgentVerifyResponse: + record = self._sanitized_record(body) + async with self.sem: + try: + return await asyncio.wait_for( + self._run_once(request, body, record), timeout=self.config.run_timeout_secs + ) + except asyncio.TimeoutError: + return self._failure_response( + record, + f"/run exceeded run_timeout_secs={self.config.run_timeout_secs}s " + "(seed + agent/tool loop + verify)", + ) + except Exception as e: # noqa: BLE001 -- never 500; one task must not abort the whole collection + return self._failure_response(record, f"unexpected error: {type(e).__name__}: {e}") + + def _sanitized_record(self, body: RemoteAgentRunRequest) -> Dict[str, Any]: + record = body.model_dump() + for key in _RESERVED_RESULT_KEYS: + record.pop(key, None) + return record + + async def _run_once( + self, request: Request, body: RemoteAgentRunRequest, record: Dict[str, Any] + ) -> RemoteAgentVerifyResponse: + # body and record are two views of the same row: `record` (sanitized dict, computed + # before run()'s try so failure rows can be built in ANY error state) feeds the Gym + # hops; `body` (typed model) is kept solely because exclude_unset information — which + # fields the dataset actually set — exists only on the model. + if record.get(SKILLS_REF_KEY_NAME): + self._throttled_warn( + "skills_ref", + "WARNING: this run carries a skills_ref, but RemoteAgent cannot stage skills into a " + "remote service; the skills config is ignored.", + ) + + # Seed the session; the cookies key all per-session state on the resources server. + cookies = request.cookies + try: + seed_response = await self.server_client.post( + server_name=self.config.resources_server.name, + url_path="/seed_session", + json=record, + cookies=cookies, + ) + await raise_for_status(seed_response) + cookies = seed_response.cookies + except Exception as e: + return self._failure_response( + record, f"/seed_session on the resources server failed: {type(e).__name__}: {e}" + ) + + try: + loop_response = await self.server_client.post( + server_name=self.config.name, + url_path=self.url_path_for_run("/v1/responses", body), + json=body.responses_create_params, + cookies=cookies, + ) + await raise_for_status(loop_response) + response_json = await get_response_json(loop_response) + cookies = loop_response.cookies + except Exception as e: + content = getattr(e, "response_content", b"") + text = content.decode(errors="replace") if isinstance(content, (bytes, bytearray)) else str(content) + # Terminal classification crosses the HTTP self-post boundary by exception NAME: + # the middleware serialized the raised RemoteAgentTerminalError into the 500 body. + terminal = "RemoteAgentTerminalError" in text + detail = text or f"{type(e).__name__}: {e}" + return self._failure_response(record, f"agent loop failed: {detail[:500]}", terminal=terminal) + + self._warn_on_response_quality(response_json) + + # Verify on the SAME session; the verify response (reward included) is /run's result. + try: + verify_response = await self.server_client.post( + server_name=self.config.resources_server.name, + url_path="/verify", + json=record | {"response": response_json}, + cookies=cookies, + ) + await raise_for_status(verify_response) + verify_json = await get_response_json(verify_response) + except Exception as e: + return self._failure_response(record, f"/verify on the resources server failed: {type(e).__name__}: {e}") + + return RemoteAgentVerifyResponse.model_validate(verify_json) + + def _throttled_warn(self, key: str, message: str) -> None: + """Per-key sampled warning: the first few occurrences, then every 100th. At production + concurrency an unthrottled per-rollout print garbles the collector's progress bar.""" + n = self._warn_counts.get(key, 0) + 1 + self._warn_counts[key] = n + if n <= _FAILURE_PRINT_HEAD or n % _FAILURE_PRINT_INTERVAL == 0: + print(f"{message} (occurrence #{n})", flush=True) + + def _warn_on_response_quality(self, response_json: Dict[str, Any]) -> None: + if not response_json.get("usage"): + self._throttled_warn( + "missing_usage", + "WARNING: the remote response carries no usage; token metrics for this agent will be " + "empty. Have your service report the full usage object: {input_tokens, output_tokens, " + "total_tokens, input_tokens_details: {cached_tokens}, output_tokens_details: " + "{reasoning_tokens}}.", + ) + + def _failure_response( + self, record: Dict[str, Any], error: str, terminal: bool = False + ) -> RemoteAgentVerifyResponse: + self._num_failures += 1 + n = self._num_failures + if n <= _FAILURE_PRINT_HEAD or n % _FAILURE_PRINT_INTERVAL == 0: + print(f"[remote_agent] rollout failed (failure #{n}): {error}", flush=True) + routing: Dict[str, Any] = {NG_FAILURE_CLASS_KEY: REMOTE_AGENT_FAILURE_CLASS, "error": error} + if terminal: + routing[NG_TERMINAL_KEY] = True + # Dict-merge with later keys winning: `record` is sanitized of reserved keys, but merge + # order still guarantees fresh reward/response/routing even if a caller passes a raw dump. + return RemoteAgentVerifyResponse.model_validate( + record | {"reward": 0.0, "response": self._empty_response().model_dump(mode="json")} | routing + ) + + def _empty_response(self) -> NeMoGymResponse: + """Minimal valid response for the failure path, so /run can return 200 with reward 0 + (never 500) even when the remote service produced nothing.""" + return NeMoGymResponse( + id="remote_agent_failure", + created_at=0.0, + model="remote_agent", + object="response", + output=[ + { + "type": "message", + "role": "assistant", + "status": "completed", + "id": "msg_0", + "content": [{"type": "output_text", "text": "", "annotations": []}], + } + ], + parallel_tool_calls=False, + tools=[], + tool_choice="auto", + ) + + async def aggregate_metrics(self, body: AggregateMetricsRequest = Body()) -> AggregateMetrics: + """Proxy aggregate_metrics to the resources server. + + Bounded: the ServerClient hop otherwise retries connection errors forever, and a dead + resources server at end-of-run would hang the collector after all rollouts are on disk. + """ + + async def _proxy() -> AggregateMetrics: + response = await self.server_client.post( + server_name=self.config.resources_server.name, + url_path="/aggregate_metrics", + json=body, + ) + await raise_for_status(response) + return AggregateMetrics.model_validate(await get_response_json(response)) + + return await asyncio.wait_for(_proxy(), timeout=_AGGREGATE_PROXY_TIMEOUT_SECS) + + +if __name__ == "__main__": + RemoteAgent.run_webserver() diff --git a/responses_api_agents/remote_agent/configs/remote_agent.yaml b/responses_api_agents/remote_agent/configs/remote_agent.yaml new file mode 100644 index 0000000000..8ff0354256 --- /dev/null +++ b/responses_api_agents/remote_agent/configs/remote_agent.yaml @@ -0,0 +1,12 @@ +remote_agent: + responses_api_agents: + remote_agent: + entrypoint: app.py + agent_base_url: ??? + resources_server: + type: resources_servers + name: ??? + concurrency: 32 + remote_responses_timeout_secs: 1800.0 + run_timeout_secs: 2100.0 + max_steps: null diff --git a/responses_api_agents/remote_agent/requirements.txt b/responses_api_agents/remote_agent/requirements.txt new file mode 100644 index 0000000000..00ed83213e --- /dev/null +++ b/responses_api_agents/remote_agent/requirements.txt @@ -0,0 +1 @@ +-e nemo-gym[dev] @ ../../ diff --git a/responses_api_agents/remote_agent/tests/__init__.py b/responses_api_agents/remote_agent/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/responses_api_agents/remote_agent/tests/test_app.py b/responses_api_agents/remote_agent/tests/test_app.py new file mode 100644 index 0000000000..974af3ccd5 --- /dev/null +++ b/responses_api_agents/remote_agent/tests/test_app.py @@ -0,0 +1,988 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# 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. +import asyncio +import json +from http.cookies import SimpleCookie +from unittest.mock import AsyncMock, MagicMock + +import orjson +import pytest +from aiohttp import ClientConnectorError, ClientPayloadError, ClientResponseError, ServerDisconnectedError +from fastapi import Response +from pydantic import BaseModel, ValidationError + +import responses_api_agents.remote_agent.app as remote_agent_app +from nemo_gym.config_types import ResourcesServerRef +from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming +from nemo_gym.rollout_collection import NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY +from nemo_gym.server_utils import ServerClient +from responses_api_agents.remote_agent.app import ( + REMOTE_AGENT_FAILURE_CLASS, + RemoteAgent, + RemoteAgentConfig, + RemoteAgentRunRequest, + normalize_remote_url, +) + + +def msg(text: str, item_id: str = "msg_1") -> dict: + return { + "type": "message", + "role": "assistant", + "status": "completed", + "id": item_id, + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + + +def fn_call(call_id: str, name: str, arguments: str) -> dict: + return {"type": "function_call", "call_id": call_id, "name": name, "arguments": arguments, "id": f"fc_{call_id}"} + + +def fn_output(call_id: str, output: str) -> dict: + return {"type": "function_call_output", "call_id": call_id, "output": output} + + +def traj(output: list, usage: dict | None = "default") -> dict: + t = { + "id": "traj_1", + "created_at": 1.0, + "model": "their-model", + "object": "response", + "output": output, + "parallel_tool_calls": False, + "tools": [], + "tool_choice": "auto", + } + if usage == "default": + usage = { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}, + } + if usage is not None: + t["usage"] = usage + return t + + +_MINIMAL_TRAJECTORY = traj([msg("the answer is 42")]) + +_COUNTER_TOOLS = [ + { + "type": "function", + "name": "increment_counter", + "parameters": { + "type": "object", + "properties": {"count": {"type": "integer", "description": ""}}, + "required": ["count"], + "additionalProperties": False, + }, + "strict": True, + "description": "", + } +] + + +def make_config(**overrides) -> RemoteAgentConfig: + fields = dict( + host="0.0.0.0", + port=8080, + entrypoint="", + name="remote_agent", + agent_base_url="http://localhost:9000", + resources_server=ResourcesServerRef(type="resources_servers", name="my_env"), + ) + fields.update(overrides) + return RemoteAgentConfig(**fields) + + +def make_row(tools=None, **extras) -> dict: + row = { + "responses_create_params": {"input": [{"role": "user", "content": "what is 6 x 7?"}]}, + "verifier_metadata": {"expected_answer": "42"}, + } + if tools is not None: + row["responses_create_params"]["tools"] = tools + row.update(extras) + return row + + +def make_request(cookies=None) -> MagicMock: + request = MagicMock() + request.cookies = cookies or {} + return request + + +class FakeRemoteResponse: + """Stands in for an aiohttp ClientResponse from the remote service.""" + + def __init__(self, status: int, content: bytes, headers=None, read_exc=None, set_cookies=None): + self.status = status + self._content = content + self.headers = headers or {} + self._read_exc = read_exc + self.cookies = SimpleCookie() + for k, v in (set_cookies or {}).items(): + self.cookies[k] = v + + @property + def ok(self) -> bool: + return self.status < 400 + + async def read(self) -> bytes: + if self._read_exc is not None: + raise self._read_exc + return self._content + + +class FakeServerClientResponse: + """Stands in for an aiohttp ClientResponse from a Gym server via ServerClient.""" + + def __init__(self, body: dict, cookies=None, status: int = 200): + self._body = body + self.cookies = cookies or {} + self.status = status + + @property + def ok(self) -> bool: + return self.status < 400 + + @property + def content(self): + reader = MagicMock() + + async def _read(): + return orjson.dumps(self._body) + + reader.read = _read + return reader + + def raise_for_status(self): + # Mirror aiohttp so nemo_gym.raise_for_status attaches response_content, the + # channel run() reads middleware-serialized errors (incl. terminal names) from. + if not self.ok: + raise ClientResponseError(request_info=MagicMock(), history=(), status=self.status, message="error") + + async def read(self) -> bytes: + return orjson.dumps(self._body) + + +def mock_remote(monkeypatch: pytest.MonkeyPatch, request_mock: AsyncMock) -> MagicMock: + client = MagicMock() + client.request = request_mock + monkeypatch.setattr(remote_agent_app, "get_global_aiohttp_client", lambda: client) + monkeypatch.setattr(remote_agent_app, "_REMOTE_RETRY_SLEEP_SECS", 0) + return client + + +def scripted_service(*turns): + """AsyncMock remote service returning one canned trajectory per call, recording payloads.""" + received = [] + + async def handler(method, url, data=None, headers=None, cookies=None, **kwargs): + received.append({"payload": orjson.loads(data), "cookies": dict(cookies or {})}) + turn = turns[min(len(received) - 1, len(turns) - 1)] + if isinstance(turn, FakeRemoteResponse): + return turn + return FakeRemoteResponse(200, orjson.dumps(turn)) + + request_mock = AsyncMock(side_effect=handler) + request_mock.received = received + return request_mock + + +def make_agent(server_client=None, **config_overrides) -> RemoteAgent: + return RemoteAgent( + config=make_config(**config_overrides), + server_client=server_client or MagicMock(spec=ServerClient), + ) + + +def wire_gym( + agent: RemoteAgent, + verify_body=None, + seed_cookies=None, + seed_status=200, + verify_status=200, + tool_handler=None, +): + """A ServerClient mock that emulates the Gym side: seed_session and verify on the + resources server, tool routes via tool_handler, and — the load-bearing part — the + /v1/responses self-post routed into the agent's REAL responses() with the exception + middleware emulated (exceptions become a 500 body carrying repr(e)).""" + calls = [] + + async def _post(server_name, url_path, json=None, cookies=None, **kwargs): + calls.append({"server_name": server_name, "url_path": url_path, "json": json, "cookies": cookies}) + if url_path == "/seed_session": + return FakeServerClientResponse({}, cookies=seed_cookies or {"session": "abc123"}, status=seed_status) + if url_path == "/verify": + body = verify_body if verify_body is not None else (json | {"reward": 1.0}) + return FakeServerClientResponse(body, status=verify_status) + if url_path.endswith("/v1/responses"): + wire = json.model_dump(exclude_unset=True) if isinstance(json, BaseModel) else json + params = NeMoGymResponseCreateParamsNonStreaming.model_validate(wire) + fastapi_response = Response() + try: + result = await agent.responses(make_request(dict(cookies or {})), fastapi_response, params) + except Exception as e: # noqa: BLE001 -- emulate SimpleServer's exception middleware + return FakeServerClientResponse({"error": repr(e)}, status=500) + out_cookies = SimpleCookie() + for header_value in fastapi_response.headers.getlist("set-cookie"): + out_cookies.load(header_value) + return FakeServerClientResponse( + result.model_dump(mode="json"), cookies={k: m.value for k, m in out_cookies.items()} + ) + if tool_handler is not None: + return await tool_handler(url_path, json, cookies) + return FakeServerClientResponse({}, status=200) + + server_client = MagicMock(spec=ServerClient) + server_client.post = AsyncMock(side_effect=_post) + server_client.calls = calls + agent.server_client = server_client + return server_client + + +def make_wired_agent(monkeypatch, request_mock, *, tool_handler=None, verify_body=None, **kwargs): + client = mock_remote(monkeypatch, request_mock) + agent = make_agent( + **{k: v for k, v in kwargs.items() if k not in ("seed_cookies", "seed_status", "verify_status")} + ) + server_client = wire_gym( + agent, + verify_body=verify_body, + seed_cookies=kwargs.get("seed_cookies"), + seed_status=kwargs.get("seed_status", 200), + verify_status=kwargs.get("verify_status", 200), + tool_handler=tool_handler, + ) + return agent, client, server_client + + +class TestConfig: + def test_sanity_construct_and_semaphore(self) -> None: + agent = make_agent(concurrency=7) + assert agent.sem._value == 7 + + def test_agent_base_url_normalized(self) -> None: + assert make_config(agent_base_url="http://localhost:9000/").agent_base_url == "http://localhost:9000" + + def test_max_steps_defaults_to_none(self) -> None: + assert make_config().max_steps is None + + @pytest.mark.parametrize( + "bad_url", + [ + "ftp://h:1", + "localhost:9000", + "http://h:1?token=abc", + "http://h:1#frag", + "http://user:pass@h:1", # pragma: allowlist secret + ], + ) + def test_agent_base_url_rejected(self, bad_url: str) -> None: + with pytest.raises(ValidationError): + make_config(agent_base_url=bad_url) + + def test_normalize_remote_url_never_echoes_credentials(self) -> None: + with pytest.raises(ValueError) as exc_info: + normalize_remote_url("http://user:hunter2@h:1") # pragma: allowlist secret + assert "hunter2" not in str(exc_info.value) + + +class TestRunHappyPath: + async def test_seed_then_loop_then_verify(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service(_MINIMAL_TRAJECTORY) + agent, client, server_client = make_wired_agent(monkeypatch, service, seed_cookies={"session": "s1"}) + + row = make_row() + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + paths = [c["url_path"] for c in server_client.calls] + assert paths == ["/seed_session", "/v1/responses", "/verify"] + # Seeded cookies reach the loop and verify; verify carries the trajectory + row keys + assert server_client.calls[1]["cookies"] == {"session": "s1"} + assert server_client.calls[2]["json"]["response"]["id"] == "traj_1" + assert server_client.calls[2]["json"]["verifier_metadata"] == {"expected_answer": "42"} + + # The remote service receives ONLY create-params: no verifier_metadata, no row keys + assert service.received[0]["payload"] == row["responses_create_params"] + args, request_kwargs = client.request.call_args + assert args == ("POST", "http://localhost:9000/v1/responses") + assert request_kwargs["allow_redirects"] is False + assert request_kwargs["timeout"].total == 1800.0 + + dumped = result.model_dump() + assert dumped["reward"] == 1.0 + assert NG_FAILURE_CLASS_KEY not in dumped + assert NG_NO_PERSIST_KEY not in dumped + assert NG_TERMINAL_KEY not in dumped + + async def test_verify_extras_pass_through(self, monkeypatch: pytest.MonkeyPatch) -> None: + row = make_row() + verify_body = row | {"response": _MINIMAL_TRAJECTORY, "reward": 0.5, "grading_notes": "close enough"} + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(_MINIMAL_TRAJECTORY), verify_body=verify_body) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + assert result.model_dump()["grading_notes"] == "close enough" + assert result.reward == 0.5 + + +class TestAgentToolLoop: + """The Gym-driven loop: the service returns unpaired function_calls as asks; Gym + executes them on the resources server and re-posts the grown conversation.""" + + @staticmethod + async def counter_tool_handler(url_path, body, cookies): + if url_path == "/increment_counter": + return FakeServerClientResponse({"success": True}, cookies={"tool_session": "t1"}) + if url_path == "/get_counter_value": + return FakeServerClientResponse({"count": 6}) + return FakeServerClientResponse({"detail": f"Not Found: {url_path}"}, status=404) + + async def test_multi_turn_tool_execution(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service( + traj([fn_call("c1", "increment_counter", '{"count": 3}')]), + traj([msg("done, counter incremented")]), + ) + agent, _, server_client = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + row = make_row(tools=_COUNTER_TOOLS) + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + dumped = result.model_dump() + assert NG_FAILURE_CLASS_KEY not in dumped + assert dumped["reward"] == 1.0 + + # Gym executed the tool: one resources-server POST to /increment_counter with parsed args + tool_calls = [c for c in server_client.calls if c["url_path"] == "/increment_counter"] + assert len(tool_calls) == 1 + assert tool_calls[0]["json"] == {"count": 3} + + # Turn 2 payload = original input + call + tool output + second_input = service.received[1]["payload"]["input"] + assert [i.get("type", "message") for i in second_input] == ["message", "function_call", "function_call_output"] + assert second_input[2]["output"] == '{"success":true}' + + # The final trajectory carries the merged conversation + types = [o["type"] for o in dumped["response"]["output"]] + assert types == ["function_call", "function_call_output", "message"] + + async def test_paired_calls_pass_through_unexecuted(self, monkeypatch: pytest.MonkeyPatch) -> None: + # call_a is the service's own internal tool record (paired); call_b is the ask. + service = scripted_service( + traj( + [ + fn_call("call_a", "web_search", '{"q": "counters"}'), + fn_output("call_a", '{"results": []}'), + fn_call("call_b", "increment_counter", '{"count": 3}'), + ] + ), + traj([msg("done")]), + ) + agent, _, server_client = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + dumped = result.model_dump() + assert dumped["reward"] == 1.0 + executed = [c["url_path"] for c in server_client.calls if c["url_path"].startswith("/increment")] + assert executed == ["/increment_counter"] + # web_search was never sent to the resources server + assert not any(c["url_path"] == "/web_search" for c in server_client.calls) + # The paired record survives in the final trajectory + types = [o["type"] for o in dumped["response"]["output"]] + assert types.count("function_call") == 2 and types.count("function_call_output") == 2 + + async def test_unknown_tool_error_fed_back_not_raised(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service( + traj([fn_call("c1", "web_search", "{}")]), # unpaired ask for a tool the env doesn't serve + traj([msg("recovered")]), + ) + agent, _, _ = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + # The 404 body came back to the service as the tool output + second_input = service.received[1]["payload"]["input"] + assert "Not Found" in second_input[-1]["output"] + + async def test_malformed_arguments_fed_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service( + traj([fn_call("c1", "increment_counter", "not json")]), + traj([msg("ok")]), + ) + agent, _, server_client = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + assert not any(c["url_path"] == "/increment_counter" for c in server_client.calls) + second_input = service.received[1]["payload"]["input"] + assert "Invalid tool call arguments" in second_input[-1]["output"] + + async def test_max_steps_bounds_the_loop(self, monkeypatch: pytest.MonkeyPatch) -> None: + always_ask = traj([fn_call("c1", "increment_counter", '{"count": 1}')]) + service = scripted_service(always_ask, always_ask, always_ask, always_ask) + agent, _, _ = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler, max_steps=2) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + assert len(service.received) == 2 + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + + async def test_service_cookies_round_trip(self, monkeypatch: pytest.MonkeyPatch) -> None: + turn1 = FakeRemoteResponse( + 200, + orjson.dumps(traj([fn_call("c1", "increment_counter", '{"count": 1}')])), + set_cookies={"svc_session": "svc1"}, + ) + service = scripted_service(turn1, traj([msg("done")])) + agent, _, _ = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + assert service.received[0]["cookies"] == {} + assert service.received[1]["cookies"] == {"svc_session": "svc1"} + + async def test_usage_accumulates_across_turns(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service( + traj([fn_call("c1", "increment_counter", '{"count": 1}')]), + traj([msg("done")]), + ) + agent, _, _ = make_wired_agent(monkeypatch, service, tool_handler=self.counter_tool_handler) + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + + usage = result.model_dump()["response"]["usage"] + assert usage["input_tokens"] == 20 and usage["output_tokens"] == 10 and usage["total_tokens"] == 30 + + async def test_string_input_coerced_to_message(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service(_MINIMAL_TRAJECTORY) + agent, _, _ = make_wired_agent(monkeypatch, service) + + row = {"responses_create_params": {"input": "just a string"}} + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + assert service.received[0]["payload"]["input"][0]["content"] == "just a string" + + +class TestRemoteFailuresBecomeSentinelRows: + async def _run(self, monkeypatch, request_mock, **config_overrides): + agent, client, server_client = make_wired_agent(monkeypatch, request_mock, **config_overrides) + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row())) + return client, server_client, result.model_dump() + + async def test_timeout_fails_once_without_retry(self, monkeypatch: pytest.MonkeyPatch) -> None: + client, _, result = await self._run(monkeypatch, AsyncMock(side_effect=asyncio.TimeoutError())) + assert client.request.call_count == 1 + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert NG_TERMINAL_KEY not in result + assert "timed out after 1800.0s" in result["error"] + assert result["reward"] == 0.0 + + async def test_connect_exhaustion_after_bounded_retries(self, monkeypatch: pytest.MonkeyPatch) -> None: + connect_error = ClientConnectorError(MagicMock(), OSError("connection refused")) + client, server_client, result = await self._run(monkeypatch, AsyncMock(side_effect=connect_error)) + assert client.request.call_count == 3 + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "Is your service running at http://localhost:9000?" in result["error"] + # verify is never reached on a failed loop + assert [c["url_path"] for c in server_client.calls] == ["/seed_session", "/v1/responses"] + + async def test_disconnect_then_success_retries(self, monkeypatch: pytest.MonkeyPatch) -> None: + request_mock = AsyncMock( + side_effect=[ServerDisconnectedError(), FakeRemoteResponse(200, orjson.dumps(_MINIMAL_TRAJECTORY))] + ) + client, _, result = await self._run(monkeypatch, request_mock) + assert client.request.call_count == 2 + assert NG_FAILURE_CLASS_KEY not in result + + async def test_http_500_with_body_excerpt(self, monkeypatch: pytest.MonkeyPatch) -> None: + _, _, result = await self._run(monkeypatch, AsyncMock(return_value=FakeRemoteResponse(500, b"kaboom"))) + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "HTTP 500" in result["error"] and "kaboom" in result["error"] + + async def test_redirect_rejected_with_location_hint(self, monkeypatch: pytest.MonkeyPatch) -> None: + response = FakeRemoteResponse(301, b"", headers={"Location": "https://elsewhere"}) + _, _, result = await self._run(monkeypatch, AsyncMock(return_value=response)) + assert "HTTP 301" in result["error"] and "https://elsewhere" in result["error"] + + async def test_invalid_json_body(self, monkeypatch: pytest.MonkeyPatch) -> None: + _, _, result = await self._run(monkeypatch, AsyncMock(return_value=FakeRemoteResponse(200, b"not json"))) + assert "not valid JSON" in result["error"] + + async def test_non_object_body(self, monkeypatch: pytest.MonkeyPatch) -> None: + _, _, result = await self._run(monkeypatch, AsyncMock(return_value=FakeRemoteResponse(200, b"[1, 2]"))) + assert "expected a JSON object" in result["error"] + + @pytest.mark.parametrize( + "read_exc", + [ClientPayloadError("Response payload is not completed"), asyncio.TimeoutError()], + ids=["mid-body disconnect", "deadline during body read"], + ) + async def test_body_read_failure(self, monkeypatch: pytest.MonkeyPatch, read_exc: Exception) -> None: + response = FakeRemoteResponse(200, b"", read_exc=read_exc) + _, _, result = await self._run(monkeypatch, AsyncMock(return_value=response)) + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "reading the response body failed" in result["error"] + + async def test_unexpected_exception_fails_fast_without_retry(self, monkeypatch: pytest.MonkeyPatch) -> None: + client, _, result = await self._run(monkeypatch, AsyncMock(side_effect=RuntimeError("surprise"))) + assert client.request.call_count == 1 + assert "RuntimeError: surprise" in result["error"] + + async def test_invalid_trajectory_shape_is_terminal_and_skips_verify( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + bad = {"id": "x", "object": "response"} # missing required Responses API fields + client, server_client, result = await self._run( + monkeypatch, AsyncMock(return_value=FakeRemoteResponse(200, orjson.dumps(bad))) + ) + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + # Terminal classification survives the HTTP self-post boundary (exception-name match) + assert result[NG_TERMINAL_KEY] is True + assert "invalid Responses API object" in result["error"] + assert [c["url_path"] for c in server_client.calls] == ["/seed_session", "/v1/responses"] + + async def test_mid_loop_failure_becomes_sentinel(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Turn 1 succeeds with a tool ask; turn 2 the service dies — the whole rollout + # must land in the sidecar, not crash the loop. + service = scripted_service( + traj([fn_call("c1", "increment_counter", '{"count": 1}')]), + FakeRemoteResponse(500, b"died mid-rollout"), + ) + agent, _, _ = make_wired_agent(monkeypatch, service, tool_handler=TestAgentToolLoop.counter_tool_handler) + result = ( + await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row(tools=_COUNTER_TOOLS))) + ).model_dump() + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "died mid-rollout" in result["error"] + + +class TestGymSideFailuresBecomeSentinelRows: + async def test_seed_failure_skips_remote_call(self, monkeypatch: pytest.MonkeyPatch) -> None: + service = scripted_service(_MINIMAL_TRAJECTORY) + agent, client, _ = make_wired_agent(monkeypatch, service, seed_status=500) + + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))).model_dump() + + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "/seed_session" in result["error"] + assert client.request.call_count == 0 + + async def test_verify_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(_MINIMAL_TRAJECTORY), verify_status=500) + + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))).model_dump() + + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "/verify" in result["error"] + assert result["reward"] == 0.0 + + async def test_skills_ref_warns_and_continues( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(_MINIMAL_TRAJECTORY)) + + row = make_row(skills_ref={"path": "/skills", "hash": "abc", "skills": []}) + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + assert "skills_ref" in capsys.readouterr().out + + +class TestResponseQualityWarnings: + async def test_missing_usage_warns( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(traj([msg("hi")], usage=None))) + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row())) + assert NG_FAILURE_CLASS_KEY not in result.model_dump() + assert "no usage" in capsys.readouterr().out + + async def test_clean_trajectory_no_warnings( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(_MINIMAL_TRAJECTORY)) + await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row())) + assert "WARNING" not in capsys.readouterr().out + + async def test_quality_warnings_are_throttled( + self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + ) -> None: + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(traj([msg("hi")], usage=None))) + + for _ in range(10): + await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row())) + + # Head of 5, then every 100th: 10 rollouts -> exactly 5 printed warnings + assert capsys.readouterr().out.count("no usage") == 5 + + +class TestRunTimeoutAndSemaphore: + async def test_run_wallclock_bound_becomes_sentinel(self, monkeypatch: pytest.MonkeyPatch) -> None: + async def slow_request(*args, **kwargs): + await asyncio.sleep(30) + + agent, _, _ = make_wired_agent(monkeypatch, AsyncMock(side_effect=slow_request), run_timeout_secs=0.05) + + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))).model_dump() + + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "run_timeout_secs" in result["error"] + + async def test_semaphore_bounds_in_flight_and_releases_on_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + in_flight = 0 + max_in_flight = 0 + release = asyncio.Event() + + async def gated_request(*args, **kwargs): + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + await release.wait() + in_flight -= 1 + return FakeRemoteResponse(500, b"boom") # failure path must release the permit too + + agent, _, _ = make_wired_agent(monkeypatch, AsyncMock(side_effect=gated_request), concurrency=2) + + rows = [RemoteAgentRunRequest.model_validate(make_row()) for _ in range(4)] + tasks = [asyncio.create_task(agent.run(make_request(), row)) for row in rows] + await asyncio.sleep(0.05) + assert max_in_flight == 2 + release.set() + results = await asyncio.gather(*tasks) + + assert all(r.model_dump()[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS for r in results) + assert agent.sem._value == 2 # every permit released despite 4 failures + + async def test_queue_wait_does_not_count_against_run_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + release_first = asyncio.Event() + first_seen = asyncio.Event() + call_count = 0 + + async def gated_request(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + first_seen.set() + await release_first.wait() + return FakeRemoteResponse(200, orjson.dumps(_MINIMAL_TRAJECTORY)) + + agent, _, _ = make_wired_agent( + monkeypatch, AsyncMock(side_effect=gated_request), concurrency=1, run_timeout_secs=0.5 + ) + + first = asyncio.create_task(agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))) + second = asyncio.create_task(agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))) + await first_seen.wait() + # Hold the only permit for most of the second task's would-be budget + await asyncio.sleep(0.4) + release_first.set() + results = [r.model_dump() for r in await asyncio.gather(first, second)] + + # If queue wait counted against run_timeout_secs, the second task would time out + assert all(NG_FAILURE_CLASS_KEY not in r for r in results) + + +class TestRoutes: + def _client(self, monkeypatch, request_mock=None): + from fastapi.testclient import TestClient + + agent, _, _ = make_wired_agent(monkeypatch, request_mock or scripted_service(_MINIMAL_TRAJECTORY)) + return TestClient(agent.setup_webserver(), raise_server_exceptions=False) + + def test_run_route_happy_path(self, monkeypatch: pytest.MonkeyPatch) -> None: + client = self._client(monkeypatch) + response = client.post("/run", json=make_row()) + assert response.status_code == 200 + assert response.json()["reward"] == 1.0 + + def test_run_route_failure_serializes_sentinel_with_http_200(self, monkeypatch: pytest.MonkeyPatch) -> None: + # The sentinel body must survive FastAPI response-model serialization: a 500 here + # would abort the entire collection run instead of routing to the failures sidecar. + client = self._client(monkeypatch, AsyncMock(side_effect=RuntimeError("remote exploded"))) + response = client.post("/run", json=make_row()) + assert response.status_code == 200 + body = response.json() + assert body[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert body["reward"] == 0.0 + assert body["response"]["output"][0]["type"] == "message" + + def test_responses_route_is_live(self, monkeypatch: pytest.MonkeyPatch) -> None: + # /v1/responses is a real route now: create-params in, finished trajectory out. + client = self._client(monkeypatch) + response = client.post("/v1/responses", json={"input": [{"role": "user", "content": "hi"}]}) + assert response.status_code == 200 + assert response.json()["id"] == "traj_1" + + +class TestStatefulToolsEndToEnd: + """The full session contract, in-process: RemoteAgent seeds the real counter environment, + the service asks for tools via unpaired function_calls, GYM executes them against the + counter server on the seeded session, and verify() scores the mutated state.""" + + def _counter_client(self): + from fastapi.testclient import TestClient + + from resources_servers.example_session_state_mgmt.app import ( + StatefulCounterResourcesServer, + StatefulCounterResourcesServerConfig, + ) + + config = StatefulCounterResourcesServerConfig( + host="0.0.0.0", port=8081, entrypoint="", name="counter", domain="agent" + ) + server = StatefulCounterResourcesServer(config=config, server_client=MagicMock(spec=ServerClient)) + return TestClient(server.setup_webserver()) + + async def test_counter_env_reward_through_gym_executed_tools(self, monkeypatch: pytest.MonkeyPatch) -> None: + counter = self._counter_client() + + # The service never sees the counter server: it only returns asks and reads outputs. + def turn2(received): + return traj([fn_call("c3", "get_counter_value", "{}")]) + + service = scripted_service( + traj( + [ + fn_call("c1", "increment_counter", '{"count": 1}'), + fn_call("c2", "increment_counter", '{"count": 2}'), + ] + ), + traj([fn_call("c3", "get_counter_value", "{}")]), + # Final turn: read the count Gym fed back and answer with it + traj([msg("final count is 6")]), + ) + + agent = make_agent() + + async def gym_post(server_name, url_path, json=None, cookies=None, **kwargs): + if url_path.endswith("/v1/responses"): + wire = json.model_dump(exclude_unset=True) if isinstance(json, BaseModel) else json + params = NeMoGymResponseCreateParamsNonStreaming.model_validate(wire) + fastapi_response = Response() + try: + result = await agent.responses(make_request(dict(cookies or {})), fastapi_response, params) + except Exception as e: # noqa: BLE001 + return FakeServerClientResponse({"error": repr(e)}, status=500) + out_cookies = SimpleCookie() + for header_value in fastapi_response.headers.getlist("set-cookie"): + out_cookies.load(header_value) + return FakeServerClientResponse( + result.model_dump(mode="json"), cookies={k: m.value for k, m in out_cookies.items()} + ) + # Everything else — seed, tools, verify — hits the REAL counter server + response = counter.post(url_path, json=json, cookies=dict(cookies or {})) + return FakeServerClientResponse( + response.json(), cookies=dict(response.cookies), status=response.status_code + ) + + server_client = MagicMock(spec=ServerClient) + server_client.post = AsyncMock(side_effect=gym_post) + agent.server_client = server_client + mock_remote(monkeypatch, service) + + row = { + "responses_create_params": { + "input": [{"role": "user", "content": "add 1 then add 2 then get the count"}], + "tools": _COUNTER_TOOLS, + }, + "initial_count": 3, + "expected_count": 6, + } + + result = await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row)) + + dumped = result.model_dump() + assert NG_FAILURE_CLASS_KEY not in dumped + # Reward 1.0 only if seed, both Gym-executed tool calls, and verify shared ONE session + assert dumped["reward"] == 1.0 + # The service really was fed the counter value Gym read back + third_input = service.received[2]["payload"]["input"] + assert any('"count":6' in i.get("output", "") for i in third_input if isinstance(i, dict)) + + +class TestCollectorRoundTrip: + """Drive the real rollout-collection helper against this agent in-process and assert + the sidecar contract end to end: successes to the main jsonl, sentinel rows to the + failures sidecar.""" + + async def test_success_and_failure_routing(self, monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + from fastapi.testclient import TestClient + + import nemo_gym.rollout_collection + from nemo_gym.rollout_collection import RolloutCollectionConfig, RolloutCollectionHelper + + # The collector reads the global config for model-call capture dirs; neutralize the + # Hydra CLI parse it would otherwise attempt under pytest (same as the core tests). + monkeypatch.setattr(nemo_gym.rollout_collection, "get_global_config_dict", MagicMock(return_value={})) + + async def remote_service(method, url, data=None, headers=None, cookies=None, **kwargs): + params = orjson.loads(data) + if "fail" in params["input"][0]["content"]: + return FakeRemoteResponse(500, b"remote exploded") + return FakeRemoteResponse(200, orjson.dumps(_MINIMAL_TRAJECTORY)) + + agent, _, _ = make_wired_agent(monkeypatch, AsyncMock(side_effect=remote_service)) + agent_http = TestClient(agent.setup_webserver(), raise_server_exceptions=False) + + class InProcessHelper(RolloutCollectionHelper): + def setup_server_client(self, *args, **kwargs): + async def _post(server_name, url_path, json=None, **kw): + response = agent_http.post(url_path, json=json) + return FakeServerClientResponse(response.json(), status=response.status_code) + + server_client = MagicMock(spec=ServerClient) + server_client.post = AsyncMock(side_effect=_post) + return server_client + + async def _call_aggregate_metrics(self, results, rows, output_fpath): + return None + + input_fpath = tmp_path / "input.jsonl" + rows = [ + {"responses_create_params": {"input": [{"role": "user", "content": "please succeed"}]}}, + {"responses_create_params": {"input": [{"role": "user", "content": "please fail"}]}}, + ] + input_fpath.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + + config = RolloutCollectionConfig( + input_jsonl_fpath=str(input_fpath), + output_jsonl_fpath=str(tmp_path / "rollouts.jsonl"), + agent_name="remote_agent", + upload_rollouts_to_wandb=False, + ) + await InProcessHelper().run_from_config(config) + + main_rows = [json.loads(line) for line in (tmp_path / "rollouts.jsonl").open()] + assert len(main_rows) == 1 + assert main_rows[0]["reward"] == 1.0 + + sidecar_rows = [json.loads(line) for line in (tmp_path / "rollouts_failures.jsonl").open()] + assert len(sidecar_rows) == 1 + assert sidecar_rows[0][NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "HTTP 500" in sidecar_rows[0]["error"] + + +class TestReviewFindingPins: + """Regression pins for the adversarial-review findings.""" + + _REUSED_ROW_EXTRAS = { + "reward": 0.75, + "response": {"stale": True}, + "error": "stale error", + NG_FAILURE_CLASS_KEY: "stale_class", + NG_NO_PERSIST_KEY: True, + NG_TERMINAL_KEY: True, + } + + async def test_failure_on_reused_rollout_row_still_returns_sentinel(self, monkeypatch: pytest.MonkeyPatch) -> None: + # A rollouts/failures JSONL re-fed as a dataset carries reward/response/error and stale + # routing keys; the failure path must not TypeError on them (the never-raise contract). + agent, _, _ = make_wired_agent(monkeypatch, AsyncMock(side_effect=RuntimeError("remote exploded"))) + + row = make_row(**self._REUSED_ROW_EXTRAS) + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row))).model_dump() + + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert result["reward"] == 0.0 + assert result["response"]["output"][0]["type"] == "message" + assert "remote exploded" in result["error"] + # Stale no-persist/terminal flags from the input row must not survive + assert NG_NO_PERSIST_KEY not in result + assert NG_TERMINAL_KEY not in result + + def test_failure_on_reused_rollout_row_route_level_stays_200(self, monkeypatch: pytest.MonkeyPatch) -> None: + from fastapi.testclient import TestClient + + agent, _, _ = make_wired_agent(monkeypatch, AsyncMock(side_effect=RuntimeError("remote exploded"))) + client = TestClient(agent.setup_webserver(), raise_server_exceptions=False) + + response = client.post("/run", json=make_row(**self._REUSED_ROW_EXTRAS)) + + assert response.status_code == 200 + assert response.json()[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + + async def test_happy_path_reused_row_leaks_no_stale_sentinels(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Stale routing keys on an input row must not echo through verify and misroute a + # SUCCESS into the failures sidecar. + agent, _, _ = make_wired_agent(monkeypatch, scripted_service(_MINIMAL_TRAJECTORY)) + + row = make_row(**self._REUSED_ROW_EXTRAS) + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(row))).model_dump() + + assert result["reward"] == 1.0 + assert NG_FAILURE_CLASS_KEY not in result + assert NG_NO_PERSIST_KEY not in result + assert NG_TERMINAL_KEY not in result + + async def test_run_outer_backstop_never_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + agent = make_agent() + + async def explode(*args, **kwargs): + raise RuntimeError("internal bug") + + monkeypatch.setattr(agent, "_run_once", explode) + result = (await agent.run(make_request(), RemoteAgentRunRequest.model_validate(make_row()))).model_dump() + + assert result[NG_FAILURE_CLASS_KEY] == REMOTE_AGENT_FAILURE_CLASS + assert "internal bug" in result["error"] + + async def test_aggregate_metrics_proxies_to_resources_server(self) -> None: + agg_body = { + "agent_metrics": {"mean/reward": 1.0}, + "key_metrics": {"mean/reward": 1.0}, + "group_level_metrics": [], + } + + async def _post(server_name, url_path, json=None, **kwargs): + assert url_path == "/aggregate_metrics" + assert server_name == "my_env" + return FakeServerClientResponse(agg_body) + + server_client = MagicMock(spec=ServerClient) + server_client.post = AsyncMock(side_effect=_post) + agent = make_agent(server_client=server_client) + + from nemo_gym.base_resources_server import AggregateMetricsRequest + + result = await agent.aggregate_metrics(AggregateMetricsRequest(verify_responses=[])) + assert result.key_metrics == {"mean/reward": 1.0} + + async def test_aggregate_metrics_bounded_when_resources_server_hangs( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + async def hang(*args, **kwargs): + await asyncio.sleep(60) + + server_client = MagicMock(spec=ServerClient) + server_client.post = AsyncMock(side_effect=hang) + agent = make_agent(server_client=server_client) + monkeypatch.setattr(remote_agent_app, "_AGGREGATE_PROXY_TIMEOUT_SECS", 0.05) + + with pytest.raises(asyncio.TimeoutError): + await agent.aggregate_metrics( + __import__( + "nemo_gym.base_resources_server", fromlist=["AggregateMetricsRequest"] + ).AggregateMetricsRequest(verify_responses=[]) + )