diff --git a/fern/versions/latest/pages/model-server/model-call-capture.mdx b/fern/versions/latest/pages/model-server/model-call-capture.mdx index 8f2a5c0ec3..d73769ab31 100644 --- a/fern/versions/latest/pages/model-server/model-call-capture.mdx +++ b/fern/versions/latest/pages/model-server/model-call-capture.mdx @@ -77,9 +77,10 @@ totals = aggregate_model_call_metrics(store, rollout_id) ``` `ModelCallRecord` is an observability serialization model derived from captured HTTP exchanges. It -contains a unique server-generated `model_call_id`, typed `model_ref`, wall-clock `started_at` and -`completed_at`, a `call_index`, API dialect, token and cache usage, latency, error details, tool -calls, reasoning content, and the captured request and response. `started_at` is recorded immediately +contains a unique server-generated `model_call_id`, the protocol `response_id` when present, typed +`model_ref`, wall-clock `started_at` and `completed_at`, a `call_index`, API dialect, token and cache +usage, latency, error details, tool calls, reasoning content, and the captured request and response. +`started_at` is recorded immediately before invoking the downstream ASGI application; `completed_at` is recorded when that invocation returns or raises, before capture parsing and persistence. Both are UTC Unix seconds for external trace correlation; durations use the monotonic latency fields. `call_index` reflects durable append @@ -113,6 +114,35 @@ for that exact rollout-attempt id, including a kill-shaped attempt being redispa The attachment is additive: it does not replace or rewrite the existing response, reward, `NeMoGymResponse`, token-id, or log-prob fields. Downstream consumers can choose whether to read it. +## Agent observations + +Supported Agent Servers may also attach `ng_agent_observations` when observability is enabled. It +contains agent invocations, tool-call intervals, explicit context-compaction events, and gaps for +facts the integration could not observe. An invocation's `conversation` is the ordered, normalized +set of conversation items supported by that producer. Unsupported or unavailable evidence is +reported in `gaps` instead of being guessed; `agent_transcript_unavailable` means any available +items came from a fallback output rather than a harness transcript. + +Agent observations and model-call capture are separate evidence. Join an invocation's model-call +references by `model_call_id`, or by the exact `(model_ref, response_id)` pair when the harness sees +the protocol response ID. Do not infer ownership from timestamps, text, or list position. The full +model request and response remain in `CaptureStore`; rollout attachments intentionally omit them. + +Tool timestamps are UTC Unix seconds when the source provides wall-clock time. `duration_ms` is the +measured interval, and `timing_source` distinguishes an executor measurement from an artifact-derived +interval. A missing capability is represented in `gaps`, not approximated. + +Model-visible tool calls and results remain in `AgentInvocation.conversation` as `function_call` and +`function_call_output` items. Execution timing and outcome, when observable, live in +`ToolCallObservation` and join through `(invocation_id, tool_call_id)`. + +Producer coverage is integration-specific. Claude Code exports transcript-derived hierarchy, +model-call references, tool intervals, and compaction markers. Hermes exports hierarchy, executor +tool timing, and compaction markers, but not exact model-call ownership. OpenClaw and PinchBench +expose normalized response items and report gaps for hierarchy, timing, compaction, and model-call +ownership. Those items duplicate response output and can make rollout records substantially larger. +Observation payloads are retained in rollout records and excluded from aggregate-metrics requests. + ## Limitations The model-server boundary observes model HTTP requests and responses. It can record tool calls, diff --git a/nemo_gym/base_responses_api_model.py b/nemo_gym/base_responses_api_model.py index 402d28a437..21a36bb854 100644 --- a/nemo_gym/base_responses_api_model.py +++ b/nemo_gym/base_responses_api_model.py @@ -391,6 +391,7 @@ class ModelCallRecord(BaseModel): # Unique server-generated identity for each persisted call. model_call_id: Optional[str] = None + response_id: Optional[str] = None # Durable append order, not a causal or semantic order for concurrent calls. call_index: int @@ -441,6 +442,7 @@ def build_model_call_record(exchange: dict[str, Any], *, call_index: int) -> Mod tool_calls, reasoning_content = _tool_calls_and_reasoning(response) return ModelCallRecord( model_call_id=exchange.get("model_call_id"), + response_id=response.get("id") if isinstance(response.get("id"), str) else None, call_index=call_index, model_ref=exchange.get("model_ref"), dialect=exchange.get("dialect"), @@ -664,11 +666,14 @@ def _reconstruct_chat_sse(events: list[dict[str, Any]]) -> Optional[dict[str, An tool_calls: dict[int, dict[str, Any]] = {} usage: Optional[dict[str, Any]] = None model: Optional[str] = None + response_id: Optional[str] = None role = "assistant" finish_reason: Optional[str] = None saw_choice = False for chunk in events: model = chunk.get("model") or model + if isinstance(chunk.get("id"), str): + response_id = chunk["id"] if chunk.get("usage"): usage = chunk["usage"] for choice in chunk.get("choices") or []: @@ -707,6 +712,8 @@ def _reconstruct_chat_sse(events: list[dict[str, Any]]) -> Optional[dict[str, An "model": model, "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}], } + if response_id is not None: + result["id"] = response_id if usage: result["usage"] = usage return result diff --git a/nemo_gym/rollout_collection.py b/nemo_gym/rollout_collection.py index 999f7ff2f0..ad1dd1a61b 100644 --- a/nemo_gym/rollout_collection.py +++ b/nemo_gym/rollout_collection.py @@ -643,7 +643,11 @@ async def _fetch_agent_metrics(agent_name: str, agent_result_list: List[Dict]) - # Strip heavyweight fields before sending, but preserve response.usage stripped = [] for r in agent_result_list: - entry = {k: v for k, v in r.items() if k not in ("response", "responses_create_params")} + entry = { + k: v + for k, v in r.items() + if k not in ("response", "responses_create_params", "ng_agent_observations") + } usage = (r.get("response") or {}).get("usage") if usage: entry["response"] = {"usage": usage} diff --git a/nemo_gym/rollout_observability.py b/nemo_gym/rollout_observability.py new file mode 100644 index 0000000000..9da863f89a --- /dev/null +++ b/nemo_gym/rollout_observability.py @@ -0,0 +1,93 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Small shared contract for observations exposed by Agent integrations.""" + +from __future__ import annotations + +from typing import Literal, Optional + +from pydantic import BaseModel, Field, model_validator + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.openai_utils import NeMoGymResponse, NeMoGymResponseInputItem + + +class ModelCallRef(BaseModel): + """Stable identifiers an Agent integration can observe for one model call.""" + + model_call_id: Optional[str] = None + model_ref: Optional[ModelServerRef] = None + response_id: Optional[str] = None + + @model_validator(mode="after") + def validate_join_key(self) -> "ModelCallRef": + if not self.model_call_id and not (self.model_ref is not None and self.response_id): + raise ValueError("model_call_id or both model_ref and response_id are required") + return self + + +class AgentInvocation(BaseModel): + """One root Agent or subagent conversation observed by a harness.""" + + invocation_id: str + parent_invocation_id: Optional[str] = None + spawned_by_tool_call_id: Optional[str] = None + status: Literal["completed", "failed", "incomplete", "unknown"] = Field( + default="unknown", description="Harness-reported invocation outcome; unknown when not explicit." + ) + model_calls: list[ModelCallRef] = Field(default_factory=list) + conversation: list[NeMoGymResponseInputItem] = Field( + default_factory=list, + description="Normalized conversation items supported by this producer; gaps describe unavailable evidence.", + ) + + +class ToolCallObservation(BaseModel): + """Timing observed for one tool call at an Agent-owned boundary.""" + + invocation_id: str + tool_call_id: str + tool_name: Optional[str] = None + started_at: Optional[float] = None + completed_at: Optional[float] = None + duration_ms: Optional[float] = None + clock_id: Optional[str] = None + timing_source: Optional[Literal["executor", "artifact"]] = None + status: Literal["completed", "failed", "timeout", "incomplete", "unknown"] = "unknown" + + +class ContextCompactionObservation(BaseModel): + """An explicit context-compaction event reported by the Agent harness.""" + + invocation_id: str + observed_at: Optional[float] = None + trigger: Optional[str] = None + tokens_before: Optional[int] = None + tokens_after: Optional[int] = None + + +class ObservationGap(BaseModel): + """A fact that the selected integration could not observe or join exactly.""" + + code: str + source: str + invocation_id: Optional[str] = None + detail: Optional[str] = None + + +class AgentObservationBundle(BaseModel): + """Normalized observations returned by one Agent Server for one rollout.""" + + source: str + invocations: list[AgentInvocation] = Field(default_factory=list) + tool_calls: list[ToolCallObservation] = Field(default_factory=list) + compactions: list[ContextCompactionObservation] = Field(default_factory=list) + gaps: list[ObservationGap] = Field(default_factory=list) + + +class AgentEpisode(BaseModel): + """An Agent response and the observations available at its execution boundary.""" + + response: NeMoGymResponse + observations: AgentObservationBundle diff --git a/resources_servers/gdpval/app.py b/resources_servers/gdpval/app.py index 47f36db923..111948c53c 100644 --- a/resources_servers/gdpval/app.py +++ b/resources_servers/gdpval/app.py @@ -41,7 +41,7 @@ from pathlib import Path from typing import Any, Dict, List, Literal, Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field from nemo_gym.base_resources_server import ( BaseResourcesServerConfig, @@ -50,7 +50,7 @@ SimpleResourcesServer, ) from nemo_gym.config_types import AggregateMetrics, AggregateMetricsRequest, ModelServerRef -from nemo_gym.server_utils import get_server_url +from nemo_gym.server_utils import apply_rollout_prefix, get_server_url from resources_servers.gdpval.judge_panel import ( ResolvedJudge, dir_contains_audio_video, @@ -234,6 +234,7 @@ class GDPValResourcesServerConfig(BaseResourcesServerConfig): class GDPValVerifyRequest(BaseVerifyRequest): + rollout_id: Optional[str] = Field(default=None, exclude_if=lambda value: value is None) task_id: str sector: Optional[str] = None occupation: Optional[str] = None @@ -324,7 +325,7 @@ def _effective_panel(self) -> List[JudgePanelMember]: JudgePanelMember(create_params_overrides=dict(self.config.judge_responses_create_params_overrides or {})) ] - def _resolve_judges(self) -> List[ResolvedJudge]: + def _resolve_judges(self, rollout_id: Optional[str] = None) -> List[ResolvedJudge]: """Resolve the (always non-empty) panel to concrete upstream coordinates. Every judge — including the single-judge special case (see @@ -337,7 +338,7 @@ def _resolve_judges(self) -> List[ResolvedJudge]: legacy_overrides = dict(self.config.judge_responses_create_params_overrides or {}) def _url(server: ModelServerRef) -> str: - return get_server_url(server.name) + "/v1" + return apply_rollout_prefix(get_server_url(server.name), rollout_id) + "/v1" judges: List[ResolvedJudge] = [] for i, member in enumerate(self._effective_panel()): @@ -373,7 +374,7 @@ async def _verify_rubric(self, body: GDPValVerifyRequest) -> GDPValVerifyRespons invalid_judge_response=True, ) - judges = self._resolve_judges() + judges = self._resolve_judges(body.rollout_id) # Route tasks with audio/video deliverables to the AV-capable judge(s) — # most judges can't read those modalities natively. if dir_contains_audio_video(body.deliverables_dir): @@ -534,7 +535,7 @@ async def _verify_comparison(self, body: GDPValVerifyRequest) -> GDPValVerifyRes # Build the judge panel. Members may share a single proxy server (so we # reuse one OpenAI client per distinct upstream) and differ only by model # + reasoning settings. run_trials samples one member per trial. - resolved_judges = self._resolve_judges() + resolved_judges = self._resolve_judges(body.rollout_id) client_cache: Dict[tuple, Any] = {} def _client_for(judge: ResolvedJudge) -> Any: diff --git a/resources_servers/gdpval/tests/test_app.py b/resources_servers/gdpval/tests/test_app.py index dfbaa08475..ffdc6a4471 100644 --- a/resources_servers/gdpval/tests/test_app.py +++ b/resources_servers/gdpval/tests/test_app.py @@ -103,6 +103,9 @@ def test_missing_dir_returns_empty(self, tmp_path) -> None: class TestApp: + def test_rollout_id_is_absent_when_correlation_is_disabled(self) -> None: + assert "rollout_id" not in _verify_request().model_dump() + def test_sanity_rubric(self) -> None: _server(reward_mode="rubric") @@ -241,7 +244,7 @@ async def fake_score_with_rubric(**kwargs): captured.update(kwargs) return 0.5, {"overall_score": 0.5} - body = _verify_request(rubric_json=[{"criterion": "clarity", "score": 1}]) + body = _verify_request(rubric_json=[{"criterion": "clarity", "score": 1}], rollout_id="7-3") with ( patch("resources_servers.gdpval.scoring.score_with_rubric", side_effect=fake_score_with_rubric), @@ -259,7 +262,7 @@ async def fake_score_with_rubric(**kwargs): assert judges[0].create_overrides == {"reasoning_effort": "medium"} assert judges[2].weight == 2.0 # All share the single proxy base_url. - assert {j.base_url for j in judges} == {"http://localhost:9999/v1"} + assert {j.base_url for j in judges} == {"http://localhost:9999/ng-rollout/7-3/v1"} # A seeded rng is threaded through for reproducible sampling. assert captured["rng"] is not None diff --git a/responses_api_agents/anyterminal_agent/app.py b/responses_api_agents/anyterminal_agent/app.py index b83fd7fdd2..54fdbba0e1 100644 --- a/responses_api_agents/anyterminal_agent/app.py +++ b/responses_api_agents/anyterminal_agent/app.py @@ -37,6 +37,7 @@ from nemo_gym.config_types import ModelServerRef from nemo_gym.global_config import get_first_server_config_dict from nemo_gym.openai_utils import NeMoGymResponse, NeMoGymResponseCreateParamsNonStreaming +from nemo_gym.rollout_observability import AgentObservationBundle, ObservationGap from nemo_gym.server_utils import apply_rollout_prefix @@ -124,6 +125,29 @@ def _safe_config_json(params: "AnyTerminalInstanceConfig", indent: Optional[int] return json.dumps(d, indent=indent) +def _load_agent_observations(path: Path) -> tuple[str, str]: + """Return canonical observation JSON and an error code without raising.""" + try: + raw = path.read_text() + except FileNotFoundError: + return "", "" + except OSError: + return "", "observation_read_failed" + + try: + bundle = AgentObservationBundle.model_validate_json(raw) + except Exception: + return "", "observation_parse_failed" + return bundle.model_dump_json(), "" + + +def _observation_gap(source: str, code: str) -> AgentObservationBundle: + return AgentObservationBundle( + source=source, + gaps=[ObservationGap(code=code, source=source)], + ) + + # Recreates /etc/dpkg in the writable tmpfs overlay so dpkg's rename() calls # don't cross filesystem boundaries (squashfs base → tmpfs overlay = EXDEV). _DPKG_FIX = """\ @@ -153,6 +177,8 @@ def _safe_config_json(params: "AnyTerminalInstanceConfig", indent: Optional[int] INSTRUCTION = Path("/trajectories_mount/instruction.txt").read_text() AGENT_KWARGS = json.loads(os.environ.get("NGTB_AGENT_KWARGS", "{{}}")) SAMPLING = json.loads(os.environ.get("NGTB_SAMPLING", "{{}}")) +OBSERVABILITY = os.environ.get("NGTB_OBSERVABILITY") == "1" +MODEL_SERVER_REF = json.loads(os.environ.get("NGTB_MODEL_SERVER_REF", "null")) from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming, NeMoGymEasyInputMessage from nemo_gym.config_types import ModelServerRef, ResourcesServerRef @@ -164,7 +190,7 @@ def _safe_config_json(params: "AnyTerminalInstanceConfig", indent: Optional[int] _cfg_sampling = {{k: v for k, v in SAMPLING.items() if k in {agent_cfg_class}.model_fields}} -_model_server = ModelServerRef(name="policy_model", type="responses_api_models") if MODEL_URL else None +_model_server = ModelServerRef.model_validate(MODEL_SERVER_REF) if MODEL_URL and MODEL_SERVER_REF else None config = {agent_cfg_class}( host="0.0.0.0", port=0, @@ -190,8 +216,20 @@ def _safe_config_json(params: "AnyTerminalInstanceConfig", indent: Optional[int] model=MODEL_NAME, **SAMPLING, ) -response = asyncio.run(agent.responses(request=None, body=body)) +observed = getattr(agent, "responses_with_observations", None) +if OBSERVABILITY and callable(observed): + episode = asyncio.run(observed(request=None, body=body)) + response = episode.response +else: + response = asyncio.run(agent.responses(request=None, body=body)) Path("/trajectories_mount/response.json").write_text(response.model_dump_json()) +if OBSERVABILITY and callable(observed): + try: + Path("/trajectories_mount/agent_observations.json").write_text( + episode.observations.model_dump_json() + ) + except Exception as exc: + print(f"failed to persist agent observations: {{type(exc).__name__}}", file=sys.stderr, flush=True) print(f"agent finished: {{len(response.output)}} output items", flush=True) """ @@ -300,6 +338,7 @@ class AnyTerminalInstanceConfig(AnyTerminalAgentConfig, AnyTerminalServerConfig) metrics_fpath: Path container: str ray_queue_timestamp: float + observability_enabled: bool = Field(default=False, exclude_if=lambda value: not value) agent_command_str: Optional[str] = None @property @@ -313,6 +352,10 @@ def instance_id(self) -> str: class AnyTerminalVerifyResponse(TerminalBenchMetrics, BaseVerifyResponse): instance_config: Dict[str, Any] + ng_agent_observations: AgentObservationBundle | None = Field( + default=None, + exclude_if=lambda value: value is None, + ) ### Container lifecycle @@ -566,11 +609,18 @@ def _build_agent_cmd(self, params: AnyTerminalInstanceConfig) -> str: if getattr(params.body, k, None) is not None } model_name = params.agent_kwargs.get("model") or params.body.model or "model" + model_server_ref = ( + json.dumps(params.model_server.model_dump(mode="json"), separators=(",", ":")) + if params.model_server is not None + else "" + ) env = ( (f"--env NGTB_MODEL_URL={shlex.quote(params.model_server_url)} " if params.model_server_url else "") + f"--env NGTB_MODEL_NAME={shlex.quote(model_name)} " + + (f"--env NGTB_MODEL_SERVER_REF={shlex.quote(model_server_ref)} " if model_server_ref else "") + f"--env NGTB_AGENT_KWARGS={shlex.quote(json.dumps(params.agent_kwargs))} " + f"--env NGTB_SAMPLING={shlex.quote(json.dumps(sampling))} " + + ("--env NGTB_OBSERVABILITY=1 " if params.observability_enabled else "") ) workdir = params.problem_info.get("workdir") return self._apptainer_exec(params, mounts, "bash /container_scripts/run_script.sh", env=env, workdir=workdir) @@ -618,6 +668,7 @@ def _setup_params( metrics_fpath=persistent_dir / "nemo_gym_metrics.json", container=self._find_container(task_name, problem_info.get("docker_image", "ubuntu:22.04")), ray_queue_timestamp=time.time(), + observability_enabled=rollout_id is not None, ) params.metrics_fpath.write_text("{}") @@ -687,6 +738,20 @@ async def _inner_responses(self, params: AnyTerminalInstanceConfig) -> NeMoGymRe else: output_items, tools = [], [] + metadata = { + "input": json.dumps(params.body.model_dump(mode="json").get("input") or []), + "metrics": params.metrics_fpath.read_text(), + "instance_config": _safe_config_json(params), + } + if params.observability_enabled: + observations_json, observations_error = _load_agent_observations( + params.persistent_dir / "agent_observations.json" + ) + metadata.update( + agent_observations=observations_json, + agent_observations_error=observations_error, + ) + return NeMoGymResponse( id=f"anyterminal-{params.instance_id}", created_at=int(time.time()), @@ -696,11 +761,7 @@ async def _inner_responses(self, params: AnyTerminalInstanceConfig) -> NeMoGymRe parallel_tool_calls=params.body.parallel_tool_calls, tool_choice=params.body.tool_choice, tools=tools, - metadata={ - "input": json.dumps(params.body.model_dump(mode="json").get("input") or []), - "metrics": params.metrics_fpath.read_text(), - "instance_config": _safe_config_json(params), - }, + metadata=metadata, ) async def run(self, body: AnyTerminalRunRequest) -> AnyTerminalVerifyResponse: @@ -711,8 +772,9 @@ async def run(self, body: AnyTerminalRunRequest) -> AnyTerminalVerifyResponse: meta, response.metadata = response.metadata, None metrics = TerminalBenchMetrics.model_validate_json(meta["metrics"]) + instance_config = AnyTerminalInstanceConfig.model_validate_json(meta["instance_config"]) - return AnyTerminalVerifyResponse( + result = dict( responses_create_params=body.responses_create_params.model_dump() | { "input": json.loads(meta["input"]), @@ -722,8 +784,22 @@ async def run(self, body: AnyTerminalRunRequest) -> AnyTerminalVerifyResponse: response=response, reward=1.0 if metrics.resolved else 0.0, **metrics.model_dump(), - instance_config=AnyTerminalInstanceConfig.model_validate_json(meta["instance_config"]).model_dump(), + instance_config=instance_config.model_dump(), ) + observations_json = meta.get("agent_observations") + source = self.config.agent_server_module.split(".")[-2].removesuffix("_agent") + if observations_json: + try: + result["ng_agent_observations"] = AgentObservationBundle.model_validate_json( + observations_json + ).model_dump(mode="json") + except Exception: + result["ng_agent_observations"] = _observation_gap(source, "observation_parse_failed") + elif meta.get("agent_observations_error"): + result["ng_agent_observations"] = _observation_gap(source, meta["agent_observations_error"]) + elif instance_config.observability_enabled: + result["ng_agent_observations"] = _observation_gap(source, "agent_observations_unavailable") + return AnyTerminalVerifyResponse.model_validate(result) if __name__ == "__main__": diff --git a/responses_api_agents/anyterminal_agent/tests/test_app.py b/responses_api_agents/anyterminal_agent/tests/test_app.py index 3441471778..c7615a9eb0 100644 --- a/responses_api_agents/anyterminal_agent/tests/test_app.py +++ b/responses_api_agents/anyterminal_agent/tests/test_app.py @@ -28,16 +28,20 @@ import pytest from nemo_gym import PARENT_DIR -from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming +from nemo_gym.config_types import ModelServerRef +from nemo_gym.openai_utils import NeMoGymResponse, NeMoGymResponseCreateParamsNonStreaming +from nemo_gym.rollout_observability import AgentInvocation, AgentObservationBundle, ModelCallRef from responses_api_agents.anyterminal_agent.app import ( _RUNNER_TEMPLATE, ActiveContainerProcess, AnyTerminalAgent, AnyTerminalAgentConfig, AnyTerminalInstanceConfig, + AnyTerminalRunRequest, GymAgentHarnessProcessor, RunTerminalAgent, _instruction_from_input, + _load_agent_observations, _read_task_meta, _safe_config_json, update_metrics, @@ -88,6 +92,21 @@ def test_sampling_is_forwarded(self) -> None: assert "**SAMPLING," in rendered assert "HermesAgentConfig.model_fields" in rendered + def test_observed_episode_is_written_when_enabled(self) -> None: + rendered = self._render() + assert "responses_with_observations" in rendered + assert "/trajectories_mount/agent_observations.json" in rendered + assert 'os.environ.get("NGTB_OBSERVABILITY") == "1"' in rendered + + def test_uses_configured_model_ref_and_writes_response_first(self) -> None: + rendered = self._render() + assert "ModelServerRef.model_validate(MODEL_SERVER_REF)" in rendered + assert 'ModelServerRef(name="policy_model"' not in rendered + assert rendered.index('Path("/trajectories_mount/response.json").write_text') < rendered.index( + 'Path("/trajectories_mount/agent_observations.json").write_text' + ) + assert "failed to persist agent observations" in rendered + class TestAgentKey: def test_key_from_module(self) -> None: @@ -321,6 +340,39 @@ def test_indent_produces_multiline(self, tmp_path: Path) -> None: cfg = _make_instance_config(tmp_path) assert "\n" in _safe_config_json(cfg, indent=2) + def test_disabled_observability_does_not_change_serialized_config(self, tmp_path: Path) -> None: + serialized = _safe_config_json(_make_instance_config(tmp_path)) + + assert "observability_enabled" not in json.loads(serialized) + assert AnyTerminalInstanceConfig.model_validate_json(serialized).observability_enabled is False + + +class TestLoadAgentObservations: + def test_validates_and_canonicalizes_sidecar(self, tmp_path: Path) -> None: + path = tmp_path / "observations.json" + bundle = AgentObservationBundle(source="claude_code", invocations=[AgentInvocation(invocation_id="root")]) + path.write_text(bundle.model_dump_json()) + + raw, error = _load_agent_observations(path) + + assert error == "" + assert AgentObservationBundle.model_validate_json(raw) == bundle + + @pytest.mark.parametrize( + ("contents", "expected_error"), + [("not json", "observation_parse_failed"), ("{}", "observation_parse_failed")], + ) + def test_rejects_invalid_sidecar(self, tmp_path: Path, contents: str, expected_error: str) -> None: + path = tmp_path / "observations.json" + path.write_text(contents) + assert _load_agent_observations(path) == ("", expected_error) + + def test_read_failure_is_nonfatal(self, tmp_path: Path) -> None: + path = tmp_path / "observations.json" + path.write_text("{}") + with patch.object(Path, "read_text", side_effect=PermissionError): + assert _load_agent_observations(path) == ("", "observation_read_failed") + # ── AnyTerminalInstanceConfig properties ────────────────────────────────────────── @@ -503,14 +555,30 @@ def test_script_written_to_disk(self, tmp_path: Path) -> None: assert "agent_done" in script def test_model_url_env_when_set(self, tmp_path: Path) -> None: - cfg = _make_instance_config(tmp_path, model_server_url="http://model:8000/ng-rollout/2-1") + cfg = _make_instance_config( + tmp_path, + model_server=ModelServerRef(type="responses_api_models", name="custom_policy"), + model_server_url="http://model:8000/ng-rollout/2-1", + ) cmd = AnyTerminalAgent._build_agent_cmd(self._stub(), cfg) assert "NGTB_MODEL_URL=http://model:8000/ng-rollout/2-1" in cmd + assert 'NGTB_MODEL_SERVER_REF=\'{"type":"responses_api_models","name":"custom_policy"}\'' in cmd def test_no_model_url_env_when_empty(self, tmp_path: Path) -> None: cfg = _make_instance_config(tmp_path, model_server_url="") cmd = AnyTerminalAgent._build_agent_cmd(self._stub(), cfg) assert "NGTB_MODEL_URL" not in cmd + assert "NGTB_MODEL_SERVER_REF" not in cmd + + def test_observability_env_is_opt_in(self, tmp_path: Path) -> None: + disabled = AnyTerminalAgent._build_agent_cmd(self._stub(), _make_instance_config(tmp_path / "off")) + enabled = AnyTerminalAgent._build_agent_cmd( + self._stub(), + _make_instance_config(tmp_path / "on", observability_enabled=True), + ) + + assert "NGTB_OBSERVABILITY" not in disabled + assert "NGTB_OBSERVABILITY=1" in enabled def test_workdir_passed_when_in_problem_info(self, tmp_path: Path) -> None: cfg = _make_instance_config( @@ -526,6 +594,123 @@ def test_workdir_passed_when_in_problem_info(self, tmp_path: Path) -> None: assert "--pwd /app" in cmd +class TestObservationRoundTrip: + @staticmethod + def _agent() -> AnyTerminalAgent: + server_client = MagicMock() + server_client.global_config_dict = {"observability_enabled": True} + with patch.object(AnyTerminalAgent, "model_post_init"): + agent = AnyTerminalAgent.model_construct( + config=_config(model_server={"type": "responses_api_models", "name": "custom_policy"}), + server_client=server_client, + ) + agent._sem = asyncio.Semaphore(1) + return agent + + @staticmethod + def _response(instance: AnyTerminalInstanceConfig, observations: str, error: str = "") -> NeMoGymResponse: + return NeMoGymResponse( + id="anyterminal-test", + created_at=1, + model="test-model", + object="response", + output=[], + parallel_tool_calls=True, + tool_choice="auto", + tools=[], + metadata={ + "input": "[]", + "metrics": json.dumps({"resolved": True}), + "instance_config": _safe_config_json(instance), + "agent_observations": observations, + "agent_observations_error": error, + }, + ) + + @staticmethod + def _body() -> AnyTerminalRunRequest: + return AnyTerminalRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + @pytest.mark.asyncio + async def test_preserves_nondefault_model_ref(self, tmp_path: Path) -> None: + model_ref = ModelServerRef(type="responses_api_models", name="custom_policy") + bundle = AgentObservationBundle( + source="claude_code", + invocations=[ + AgentInvocation( + invocation_id="root", + model_calls=[ModelCallRef(model_ref=model_ref, response_id="resp-1")], + ) + ], + ) + instance = _make_instance_config( + tmp_path, + model_server=model_ref, + observability_enabled=True, + ) + agent = self._agent() + + with patch.object( + agent, "_responses", AsyncMock(return_value=self._response(instance, bundle.model_dump_json())) + ): + result = await agent.run(self._body()) + + emitted = result.ng_agent_observations + assert result.reward == 1.0 + assert emitted is not None + assert emitted.invocations[0].model_calls[0].model_ref == model_ref + assert emitted.invocations[0].model_calls[0].response_id == "resp-1" + + @pytest.mark.parametrize( + ("observations", "error", "gap_code"), + [ + ("[]", "", "observation_parse_failed"), + ("", "observation_read_failed", "observation_read_failed"), + ], + ) + @pytest.mark.asyncio + async def test_sidecar_failure_keeps_reward_and_emits_typed_gap( + self, + tmp_path: Path, + observations: str, + error: str, + gap_code: str, + ) -> None: + instance = _make_instance_config(tmp_path, observability_enabled=True) + agent = self._agent() + + with patch.object( + agent, + "_responses", + AsyncMock(return_value=self._response(instance, observations, error)), + ): + result = await agent.run(self._body()) + + assert result.reward == 1.0 + assert result.ng_agent_observations is not None + assert result.ng_agent_observations.invocations == [] + assert [gap.code for gap in result.ng_agent_observations.gaps] == [gap_code] + + @pytest.mark.asyncio + async def test_disabled_observability_does_not_change_response(self, tmp_path: Path) -> None: + instance = _make_instance_config(tmp_path) + agent = self._agent() + + with patch.object(agent, "_responses", AsyncMock(return_value=self._response(instance, ""))): + result = await agent.run(self._body()) + + assert result.reward == 1.0 + assert result.ng_agent_observations is None + assert "ng_agent_observations" not in result.model_dump(mode="json") + assert "observability_enabled" not in result.instance_config + + # ── RunTerminalAgent.process_single_datapoint ──────────────────────────────────── diff --git a/responses_api_agents/claude_code_agent/app.py b/responses_api_agents/claude_code_agent/app.py index 0aaf459b35..5bec884136 100644 --- a/responses_api_agents/claude_code_agent/app.py +++ b/responses_api_agents/claude_code_agent/app.py @@ -24,11 +24,11 @@ from asyncio import Semaphore from pathlib import Path from time import time -from typing import Any, Optional +from typing import Any, Callable, Optional from uuid import uuid4 from fastapi import Request -from pydantic import ConfigDict, PrivateAttr +from pydantic import ConfigDict, Field, PrivateAttr from nemo_gym.base_resources_server import NEMO_GYM_MCP_METADATA_KEY, BaseRunRequest, BaseVerifyResponse from nemo_gym.base_responses_api_agent import BaseResponsesAPIAgentConfig, Body, SimpleResponsesAPIAgent @@ -46,8 +46,10 @@ NeMoGymResponseOutputTokensDetails, NeMoGymResponseUsage, ) +from nemo_gym.rollout_observability import AgentEpisode, AgentObservationBundle, ObservationGap from nemo_gym.server_utils import apply_rollout_prefix, get_response_json, raise_for_status from nemo_gym.skills import stage_skills +from responses_api_agents.claude_code_agent.observability import extract_claude_code_observations from responses_api_agents.claude_code_agent.setup_claude_code import ensure_claude_code @@ -243,6 +245,9 @@ class ClaudeCodeAgentVerifyResponse(BaseVerifyResponse): model_config = ConfigDict(extra="allow") turns_used: int = 0 finished_naturally: bool = False + ng_agent_observations: Optional[AgentObservationBundle] = Field( + default=None, exclude_if=lambda value: value is None + ) class ClaudeCodeAgent(SimpleResponsesAPIAgent): @@ -381,6 +386,7 @@ async def _run_claude_code( mcp_config: Optional[str] = None, skills_path: Optional[str] = None, rollout_id: Optional[str] = None, + observation_collector: Optional[Callable[[Path], None]] = None, ) -> tuple[str, str]: """Run claude -p --output-format=stream-json and return (stdout, model_name). @@ -441,7 +447,14 @@ async def _run_claude_code( return stdout.decode(errors="replace"), model finally: if claude_config_dir is not None: - shutil.rmtree(claude_config_dir, ignore_errors=True) + try: + if observation_collector is not None: + try: + await asyncio.to_thread(observation_collector, claude_config_dir) + except Exception: + LOG.exception("failed to collect Claude Code observations") + finally: + shutil.rmtree(claude_config_dir, ignore_errors=True) def _resources_server_base_url(self) -> str: cfg = get_first_server_config_dict( @@ -509,6 +522,7 @@ async def _create_response( mcp_config: Optional[str] = None, skills_path: Optional[str] = None, rollout_id: Optional[str] = None, + observation_collector: Optional[Callable[[Path], None]] = None, ) -> NeMoGymResponse: body = body.model_copy(deep=True) if isinstance(body.input, str): @@ -524,6 +538,7 @@ async def _create_response( mcp_config=mcp_config, skills_path=skills_path, rollout_id=rollout_id, + observation_collector=observation_collector, ) output_items, usage = parse_stream_json(stdout) @@ -570,6 +585,44 @@ async def responses( ) -> NeMoGymResponse: return await self._create_response(body) + async def responses_with_observations( + self, + request: Optional[Request], + body: NeMoGymResponseCreateParamsNonStreaming, + *, + mcp_config: Optional[str] = None, + skills_path: Optional[str] = None, + rollout_id: Optional[str] = None, + ) -> AgentEpisode: + observations: Optional[AgentObservationBundle] = None + + def collect(config_dir: Path) -> None: + nonlocal observations + try: + observations = extract_claude_code_observations(config_dir, model_ref=self.config.model_server) + if self.config.model_server is None: + observations.gaps.append(ObservationGap(code="model_call_join_unavailable", source="claude_code")) + except Exception: + LOG.exception("failed to extract Claude Code observations") + observations = AgentObservationBundle( + source="claude_code", + gaps=[ObservationGap(code="observation_parse_failed", source="claude_code")], + ) + + response = await self._create_response( + body, + mcp_config=mcp_config, + skills_path=skills_path, + rollout_id=rollout_id, + observation_collector=collect, + ) + if observations is None: + observations = AgentObservationBundle( + source="claude_code", + gaps=[ObservationGap(code="agent_transcript_unavailable", source="claude_code")], + ) + return AgentEpisode(response=response, observations=observations) + async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> ClaudeCodeAgentVerifyResponse: async with self.sem: cookies = request.cookies @@ -593,18 +646,30 @@ async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> Claude with tempfile.TemporaryDirectory(prefix="nemo_gym_claude_mcp_") as mcp_config_dir: mcp_config = self._write_rollout_mcp_config(seed_resp_json, Path(mcp_config_dir)) - agent_resp = await self._create_response( - body.responses_create_params, - mcp_config=mcp_config, - skills_path=skills_path, - rollout_id=rollout_id, - ) + if rollout_id is not None: + episode = await self.responses_with_observations( + request, + body.responses_create_params, + mcp_config=mcp_config, + skills_path=skills_path, + rollout_id=rollout_id, + ) + agent_resp, observations = episode.response, episode.observations + else: + agent_resp = await self._create_response( + body.responses_create_params, + mcp_config=mcp_config, + skills_path=skills_path, + ) + observations = None agent_resp_json = agent_resp.model_dump(mode="json") verify_resp = await self.server_client.post( server_name=self.config.resources_server.name, url_path="/verify", - json=body.model_dump() | {"response": agent_resp_json}, + json=body.model_dump() + | {"response": agent_resp_json} + | ({"rollout_id": rollout_id} if rollout_id is not None else {}), cookies=cookies, ) await raise_for_status(verify_resp) @@ -619,9 +684,10 @@ async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> Claude last = gym_resp.output[-1] if gym_resp.output else None naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant" - return ClaudeCodeAgentVerifyResponse.model_validate( - verify_json | {"turns_used": turns, "finished_naturally": naturally} - ) + result = verify_json | {"turns_used": turns, "finished_naturally": naturally} + if observations is not None: + result["ng_agent_observations"] = observations.model_dump(mode="json") + return ClaudeCodeAgentVerifyResponse.model_validate(result) if __name__ == "__main__": diff --git a/responses_api_agents/claude_code_agent/observability.py b/responses_api_agents/claude_code_agent/observability.py new file mode 100644 index 0000000000..eb1ad39184 --- /dev/null +++ b/responses_api_agents/claude_code_agent/observability.py @@ -0,0 +1,459 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Read Claude Code's per-run transcripts into Gym observability records.""" + +from __future__ import annotations + +import json +import math +from collections import defaultdict +from datetime import datetime +from pathlib import Path +from typing import Any + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.openai_utils import ( + NeMoGymEasyInputMessage, + NeMoGymFunctionCallOutput, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, + NeMoGymResponseReasoningItem, + NeMoGymSummary, +) +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ModelCallRef, + ObservationGap, + ToolCallObservation, +) + + +SOURCE = "claude_code" +TRANSCRIPT_CLOCK = "claude_code_transcript" + + +def _gap(code: str, *, invocation_id: str | None = None, detail: str | None = None) -> ObservationGap: + return ObservationGap(code=code, source=SOURCE, invocation_id=invocation_id, detail=detail) + + +def _timestamp(value: Any) -> float | None: + try: + result = ( + float(value) + if isinstance(value, (int, float)) and not isinstance(value, bool) + else datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + ) + except (AttributeError, TypeError, ValueError): + return None + return result if math.isfinite(result) else None + + +def _text(value: Any) -> str: + if isinstance(value, str): + return value + if isinstance(value, list): + if all(isinstance(item, dict) and item.get("type") == "text" for item in value): + return "".join(str(item.get("text") or "") for item in value) + return json.dumps(value, ensure_ascii=False, sort_keys=True) + if value is None: + return "" + return json.dumps(value, ensure_ascii=False, sort_keys=True) + + +def _status(block: dict[str, Any], result: Any) -> str: + if block.get("is_error") is True: + return "failed" + if isinstance(result, dict): + if result.get("interrupted") is True: + return "incomplete" + value = result.get("status") + if value in {"completed", "failed", "timeout", "incomplete"}: + return value + # A tool_result block is an explicit terminal observation even when Claude Code + # does not attach a separate status object. + return "completed" + + +def _metadata(event: dict[str, Any]) -> dict[str, Any]: + message = event.get("message") + for owner in (event, message if isinstance(message, dict) else {}): + for key in ("compactMetadata", "compact_metadata"): + if isinstance(metadata := owner.get(key), dict): + return metadata + return {} + + +def _integer(metadata: dict[str, Any], *keys: str) -> int | None: + for key in keys: + value = metadata.get(key) + if isinstance(value, int) and not isinstance(value, bool): + return value + return None + + +def _compaction(event: dict[str, Any], invocation_id: str) -> ContextCompactionObservation | None: + message = event.get("message") + is_summary = isinstance(message, dict) and message.get("isCompactSummary") is True + is_boundary = event.get("type") == "system" and event.get("subtype") == "compact_boundary" + metadata = _metadata(event) + if not is_summary and not is_boundary and not metadata: + return None + + trigger = metadata.get("trigger") + return ContextCompactionObservation( + invocation_id=invocation_id, + observed_at=_timestamp(event.get("timestamp")), + trigger=trigger if isinstance(trigger, str) else None, + tokens_before=_integer(metadata, "tokensBefore", "preTokens"), + tokens_after=_integer(metadata, "tokensAfter", "postTokens"), + ) + + +def _message_id(event: dict[str, Any], block_index: int, kind: str) -> str | None: + event_id = event.get("uuid") + if isinstance(event_id, str) and event_id: + return f"{event_id}:{kind}:{block_index}" + message = event.get("message") + response_id = message.get("id") if isinstance(message, dict) else None + if isinstance(response_id, str) and response_id: + return f"{response_id}:{kind}:{block_index}" + return None + + +def _message(item_id: str, text: str) -> NeMoGymResponseOutputMessage: + return NeMoGymResponseOutputMessage(id=item_id, content=[NeMoGymResponseOutputText(text=text, annotations=[])]) + + +def _reasoning(item_id: str, block: dict[str, Any]) -> NeMoGymResponseReasoningItem: + signature = block.get("signature") + return NeMoGymResponseReasoningItem( + id=item_id, + summary=[NeMoGymSummary(text=block["thinking"], type="summary_text")], + encrypted_content=signature if isinstance(signature, str) else None, + ) + + +def _tool_call(tool_call_id: str, block: dict[str, Any]) -> NeMoGymResponseFunctionToolCall: + return NeMoGymResponseFunctionToolCall( + arguments=json.dumps(block.get("input", {}), ensure_ascii=False, sort_keys=True), + call_id=tool_call_id, + name=block.get("name") if isinstance(block.get("name"), str) else "", + id=tool_call_id, + status="completed", + ) + + +def _tool_result(event: dict[str, Any], block: dict[str, Any]) -> NeMoGymFunctionCallOutput: + event_id = event.get("uuid") + return NeMoGymFunctionCallOutput( + call_id=block["tool_use_id"], + output=_text(block.get("content")), + id=event_id if isinstance(event_id, str) else None, + status="completed", + ) + + +def _read_events(config_dir: Path, gaps: list[ObservationGap]) -> list[tuple[int, dict[str, Any]]]: + if not config_dir.is_dir(): + gaps.append(_gap("transcript_dir_missing")) + return [] + + transcript_dir = config_dir / "projects" + if not transcript_dir.is_dir(): + gaps.append(_gap("transcript_dir_missing", detail="projects")) + return [] + + try: + # Claude Code stores session and subagent transcripts below ``projects``. + # Other JSONL files in CLAUDE_CONFIG_DIR may belong to staged skills or + # unrelated CLI state and must not be interpreted as rollout evidence. + paths = sorted(transcript_dir.rglob("*.jsonl")) + except OSError: + gaps.append(_gap("transcript_dir_unreadable")) + return [] + + events: list[tuple[int, dict[str, Any]]] = [] + for path in paths: + try: + lines = path.open(encoding="utf-8", errors="replace") + except OSError: + gaps.append(_gap("transcript_unreadable", detail=path.name)) + continue + with lines: + for line_number, line in enumerate(lines, start=1): + if not line.strip(): + continue + try: + event = json.loads(line) + except (json.JSONDecodeError, UnicodeError): + gaps.append(_gap("malformed_transcript_line", detail=f"{path.name}:{line_number}")) + continue + if not isinstance(event, dict): + gaps.append(_gap("invalid_transcript_record", detail=f"{path.name}:{line_number}")) + continue + if not isinstance(event.get("sessionId"), str): + continue + events.append((len(events), event)) + return events + + +def extract_claude_code_observations( + config_dir: Path, + *, + model_ref: ModelServerRef | None = None, +) -> AgentObservationBundle: + """Extract exact relationships available in one ``CLAUDE_CONFIG_DIR``. + + Transcript IDs and timestamps are used directly. Missing or ambiguous evidence + is reported as a gap; the extractor never joins calls by text or proximity. + """ + + gaps: list[ObservationGap] = [] + raw_events = _read_events(Path(config_dir), gaps) + if not raw_events: + gaps.append(_gap("agent_transcript_unavailable")) + return AgentObservationBundle(source=SOURCE, gaps=gaps) + + events_by_invocation: dict[str, list[tuple[int, dict[str, Any]]]] = defaultdict(list) + first_seen: dict[str, int] = {} + agent_invocations: set[str] = set() + + for ordinal, event in raw_events: + agent_id = event.get("agentId") + invocation_id = agent_id if isinstance(agent_id, str) and agent_id else event["sessionId"] + events_by_invocation[invocation_id].append((ordinal, event)) + first_seen.setdefault(invocation_id, ordinal) + if isinstance(agent_id, str) and agent_id: + agent_invocations.add(invocation_id) + + starts: dict[tuple[str, str], list[tuple[float | None, str]]] = defaultdict(list) + finishes: dict[tuple[str, str], list[tuple[float | None, str]]] = defaultdict(list) + parents: dict[str, tuple[str, str, str, int]] = {} + ambiguous_parents: set[str] = set() + conversations: dict[str, list[Any]] = defaultdict(list) + model_calls: dict[str, list[ModelCallRef]] = defaultdict(list) + compactions: list[ContextCompactionObservation] = [] + + for invocation_id, entries in events_by_invocation.items(): + entries.sort( + key=lambda pair: ( + _timestamp(pair[1].get("timestamp")) is None, + _timestamp(pair[1].get("timestamp")) or 0, + pair[0], + ) + ) + items = conversations[invocation_id] + refs = model_calls[invocation_id] + + def add_gap(code: str, detail: str | None = None) -> None: + gaps.append(_gap(code, invocation_id=invocation_id, detail=detail)) + + seen_response_ids: set[str] = set() + for ordinal, event in entries: + compaction = _compaction(event, invocation_id) + if compaction is not None: + compactions.append(compaction) + if compaction.observed_at is None: + add_gap("compaction_timestamp_missing") + + message = event.get("message") + if not isinstance(message, dict): + continue + role = message.get("role") or event.get("type") + content = message.get("content") + + if role == "assistant": + response_id = message.get("id") + if not isinstance(response_id, str) or not response_id: + add_gap("model_response_id_missing") + elif model_ref is not None and response_id not in seen_response_ids: + refs.append(ModelCallRef(model_ref=model_ref, response_id=response_id)) + seen_response_ids.add(response_id) + + if isinstance(content, list): + blocks = content + elif isinstance(content, str): + blocks = [{"type": "text", "text": content}] + else: + add_gap("unsupported_assistant_content_block", type(content).__name__) + blocks = [] + for block_index, block in enumerate(blocks): + if not isinstance(block, dict): + add_gap("invalid_assistant_content") + continue + block_type = block.get("type") + item_id = _message_id(event, block_index, str(block_type or "content")) + if block_type == "text": + text = block.get("text") + if not isinstance(text, str) or not text: + continue + if item_id is None: + add_gap("assistant_item_id_missing") + continue + items.append(_message(item_id, text)) + elif block_type == "thinking": + thinking = block.get("thinking") + if not isinstance(thinking, str) or not thinking: + continue + if item_id is None: + add_gap("reasoning_item_id_missing") + continue + items.append(_reasoning(item_id, block)) + elif block_type == "tool_use": + tool_call_id = block.get("id") + if not isinstance(tool_call_id, str) or not tool_call_id: + add_gap("tool_call_id_missing") + continue + tool_name = block.get("name") if isinstance(block.get("name"), str) else "" + items.append(_tool_call(tool_call_id, block)) + starts[(invocation_id, tool_call_id)].append((_timestamp(event.get("timestamp")), tool_name)) + else: + add_gap( + "unsupported_assistant_content_block", + block_type if isinstance(block_type, str) else None, + ) + + elif role in {"user", "system", "developer"}: + if isinstance(content, str): + if content: + items.append(NeMoGymEasyInputMessage(role=role, content=content)) + continue + if not isinstance(content, list): + if content is not None: + add_gap("unsupported_user_content_block", type(content).__name__) + continue + + tool_results: list[dict[str, Any]] = [] + result_metadata = event.get("toolUseResult") + for block in content: + if not isinstance(block, dict): + add_gap("invalid_user_content") + continue + if block.get("type") == "tool_result": + tool_results.append(block) + tool_call_id = block.get("tool_use_id") + if not isinstance(tool_call_id, str) or not tool_call_id: + add_gap("tool_result_id_missing") + continue + tool_status = _status(block, result_metadata) + items.append(_tool_result(event, block)) + finishes[(invocation_id, tool_call_id)].append( + (_timestamp(event.get("timestamp")), tool_status) + ) + elif block.get("type") == "text": + if isinstance(block.get("text"), str): + items.append(NeMoGymEasyInputMessage(role=role, content=block["text"])) + else: + add_gap("unsupported_user_content_block", "text") + else: + block_type = block.get("type") + add_gap( + "unsupported_user_content_block", + block_type if isinstance(block_type, str) else None, + ) + + child_id = result_metadata.get("agentId") if isinstance(result_metadata, dict) else None + if isinstance(child_id, str) and child_id: + if len(tool_results) == 1 and isinstance(tool_results[0].get("tool_use_id"), str): + parent = ( + invocation_id, + tool_results[0]["tool_use_id"], + _status(tool_results[0], result_metadata), + ordinal, + ) + if child_id in parents and parents[child_id][:2] != parent[:2]: + parents.pop(child_id) + ambiguous_parents.add(child_id) + gaps.append(_gap("conflicting_subagent_parent", invocation_id=child_id)) + elif child_id not in ambiguous_parents: + parents.setdefault(child_id, parent) + else: + add_gap("ambiguous_subagent_relation") + + tool_calls: list[ToolCallObservation] = [] + for invocation_id, tool_call_id in sorted( + set(starts) | set(finishes), key=lambda key: (first_seen.get(key[0], math.inf), key[1]) + ): + call_starts = starts.get((invocation_id, tool_call_id), []) + call_finishes = finishes.get((invocation_id, tool_call_id), []) + + def add_tool_gap(code: str) -> None: + gaps.append(_gap(code, invocation_id=invocation_id, detail=tool_call_id)) + + if len(call_starts) > 1 or len(call_finishes) > 1: + add_tool_gap("ambiguous_tool_artifact") + continue + started_at, tool_name = call_starts[0] if call_starts else (None, "") + completed_at, tool_status = call_finishes[0] if call_finishes else (None, "incomplete") + if not call_starts: + add_tool_gap("tool_start_missing") + if not call_finishes: + add_tool_gap("tool_result_missing") + if call_starts and started_at is None: + add_tool_gap("tool_start_timestamp_missing") + if call_finishes and completed_at is None: + add_tool_gap("tool_result_timestamp_missing") + duration_ms = None + if started_at is not None and completed_at is not None and completed_at >= started_at: + duration_ms = (completed_at - started_at) * 1000 + tool_calls.append( + ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=tool_call_id, + tool_name=tool_name or None, + started_at=started_at, + completed_at=completed_at, + duration_ms=duration_ms, + clock_id=TRANSCRIPT_CLOCK if started_at is not None or completed_at is not None else None, + timing_source="artifact" if started_at is not None or completed_at is not None else None, + status=tool_status, + ) + ) + + parent_by_invocation = {child_id: parent[:3] for child_id, parent in parents.items()} + for child_id, parent in parents.items(): + if child_id not in events_by_invocation: + first_seen[child_id] = parent[3] + gaps.append(_gap("subagent_transcript_missing", invocation_id=child_id)) + + for invocation_id in agent_invocations - set(parent_by_invocation): + gaps.append(_gap("subagent_parent_unavailable", invocation_id=invocation_id)) + + all_invocation_ids = set(events_by_invocation) | set(parent_by_invocation) + invocations_by_id: dict[str, AgentInvocation] = {} + for invocation_id in all_invocation_ids: + parent = parent_by_invocation.get(invocation_id) + invocations_by_id[invocation_id] = AgentInvocation( + invocation_id=invocation_id, + parent_invocation_id=parent[0] if parent else None, + spawned_by_tool_call_id=parent[1] if parent else None, + status=parent[2] if parent else "unknown", + model_calls=model_calls[invocation_id], + conversation=conversations[invocation_id], + ) + + def order_key(invocation_id: str) -> tuple[tuple[float, str], ...]: + path: list[tuple[float, str]] = [] + seen: set[str] = set() + while invocation_id not in seen: + seen.add(invocation_id) + path.append((first_seen.get(invocation_id, math.inf), invocation_id)) + parent = parent_by_invocation.get(invocation_id) + if parent is None: + break + invocation_id = parent[0] + return tuple(reversed(path)) + + ordered_ids = sorted(all_invocation_ids, key=order_key) + + return AgentObservationBundle( + source=SOURCE, + invocations=[invocations_by_id[invocation_id] for invocation_id in ordered_ids], + tool_calls=tool_calls, + compactions=compactions, + gaps=gaps, + ) diff --git a/responses_api_agents/claude_code_agent/tests/test_app.py b/responses_api_agents/claude_code_agent/tests/test_app.py index 8209eafbb6..ed2b5093b4 100644 --- a/responses_api_agents/claude_code_agent/tests/test_app.py +++ b/responses_api_agents/claude_code_agent/tests/test_app.py @@ -15,6 +15,7 @@ import asyncio import json +import threading from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -41,6 +42,7 @@ _extract_instruction, parse_stream_json, ) +from responses_api_agents.claude_code_agent.observability import extract_claude_code_observations def _write_skill_dir(root: Path, name: str = "cot_enhanced") -> Path: @@ -260,26 +262,22 @@ def _gym_response(text: str = "done") -> dict: } -class TestRunForwardsSkillsPath: - """run() reads skills_ref off the request's model_extra (extra='allow') and forwards its path - directly to _create_response/_run_claude_code.""" +def _seed_and_verify_post(): + async def _post(server_name, url_path, json=None, cookies=None, **kw): + if url_path == "/verify": + return _FakeHttpResp( + {"responses_create_params": {"input": []}, "response": _gym_response(), "reward": 1.0} + ) + return _FakeHttpResp({}) - def _seed_and_verify_post(self): - async def _post(server_name, url_path, json=None, cookies=None, **kw): - if url_path == "/verify": - return _FakeHttpResp( - {"responses_create_params": {"input": []}, "response": _gym_response(), "reward": 1.0} - ) - return _FakeHttpResp({}) + return AsyncMock(side_effect=_post) - return AsyncMock(side_effect=_post) +class TestRunForwardsSkillsPath: def _run(self, agent: ClaudeCodeAgent, body: ClaudeCodeAgentRunRequest, run_claude_code: AsyncMock): - agent.server_client.post = self._seed_and_verify_post() + agent.server_client.post = _seed_and_verify_post() req = MagicMock() req.cookies = {} - # Stub the CLI invocation; _create_response still runs for real, so we exercise the full - # run() -> _create_response -> _run_claude_code argument threading. with patch.object( ClaudeCodeAgent, "_run_claude_code", @@ -306,9 +304,56 @@ def test_no_skills_ref_forwards_none(self) -> None: run_claude_code = AsyncMock(return_value=("", "claude-sonnet-4-6")) body = ClaudeCodeAgentRunRequest.model_validate({"responses_create_params": {"input": []}}) - self._run(agent, body, run_claude_code) + result = self._run(agent, body, run_claude_code) assert run_claude_code.call_args.kwargs["skills_path"] is None + assert "ng_agent_observations" not in result.model_dump(mode="json") + + +class TestObservability: + def test_run_returns_observations_when_enabled(self, tmp_path: Path) -> None: + agent = _make_agent(model_server=ModelServerRef(type="responses_api_models", name="policy")) + agent.server_client.global_config_dict = {"observability_enabled": True} + agent.server_client.post = _seed_and_verify_post() + + async def run_claude_code(*args, observation_collector=None, **kwargs): + transcript = tmp_path / "projects" / "session.jsonl" + transcript.parent.mkdir(parents=True) + transcript.write_text( + json.dumps( + { + "type": "assistant", + "sessionId": "session-1", + "timestamp": "2026-07-22T10:00:00Z", + "uuid": "event-1", + "message": { + "role": "assistant", + "id": "msg-1", + "content": [{"type": "text", "text": "done"}], + }, + } + ) + ) + observation_collector(tmp_path) + return _event("assistant", message={"content": [{"type": "text", "text": "done"}]}), "model" + + request = MagicMock() + request.cookies = {} + body = ClaudeCodeAgentRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + with patch.object(ClaudeCodeAgent, "_run_claude_code", run_claude_code): + result = asyncio.run(agent.run(request, body)) + + observations = result.ng_agent_observations + assert observations is not None + assert observations.invocations[0].invocation_id == "session-1" + assert observations.invocations[0].model_calls[0].response_id == "msg-1" + assert agent.server_client.post.await_args_list[-1].kwargs["json"]["rollout_id"] == "1-2" class TestRunClaudeCode: @@ -422,6 +467,53 @@ async def fake_wait_for(coro, timeout): assert killed["called"] is True assert model == "claude-sonnet-4-6" + def test_collects_observations_before_cleanup(self, tmp_path: Path) -> None: + agent = _make_agent() + captured: dict = {} + event_loop_thread = threading.get_ident() + + class FakeProc: + returncode = 0 + + async def communicate(self): + return b'{"type":"result","usage":{}}\n', b"" + + async def fake_exec(*cmd, **kwargs): + config_dir = Path(kwargs["env"]["CLAUDE_CONFIG_DIR"]) + transcript = config_dir / "projects" / "run.jsonl" + transcript.parent.mkdir(parents=True) + transcript.write_text( + json.dumps( + { + "type": "assistant", + "sessionId": "session-1", + "timestamp": "2026-07-22T10:00:00Z", + "uuid": "event-1", + "message": { + "role": "assistant", + "id": "msg-1", + "content": [{"type": "text", "text": "done"}], + }, + } + ) + ) + captured["config_dir"] = config_dir + return FakeProc() + + def collect(config_dir: Path) -> None: + captured["collector_thread"] = threading.get_ident() + captured["observations"] = extract_claude_code_observations(config_dir) + + with ( + patch("responses_api_agents.claude_code_agent.app.Path.home", return_value=tmp_path), + patch("responses_api_agents.claude_code_agent.app.asyncio.create_subprocess_exec", fake_exec), + ): + asyncio.run(agent._run_claude_code("hello", observation_collector=collect)) + + assert captured["observations"].invocations[0].invocation_id == "session-1" + assert captured["collector_thread"] != event_loop_thread + assert not captured["config_dir"].exists() + class TestRolloutMCPConfig: def test_no_metadata_preserves_static_config(self, tmp_path: Path) -> None: diff --git a/responses_api_agents/claude_code_agent/tests/test_observability.py b/responses_api_agents/claude_code_agent/tests/test_observability.py new file mode 100644 index 0000000000..bb09ae223d --- /dev/null +++ b/responses_api_agents/claude_code_agent/tests/test_observability.py @@ -0,0 +1,275 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import json +from pathlib import Path + +import pytest + +from nemo_gym.config_types import ModelServerRef +from responses_api_agents.claude_code_agent.observability import extract_claude_code_observations + + +MODEL_REF = ModelServerRef(type="responses_api_models", name="policy") + + +def _write(path: Path, *events: dict | str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(event if isinstance(event, str) else json.dumps(event) for event in events)) + + +def _event( + session: str, + role: str, + timestamp: str, + content: str | list[dict], + *, + agent: str | None = None, + message_id: str | None = None, + message_extra: dict | None = None, + **extra: object, +) -> dict: + message = {"role": role, "content": content, **(message_extra or {})} + if message_id: + message["id"] = message_id + event = { + "type": role, + "sessionId": session, + "timestamp": timestamp, + "message": message, + **extra, + } + if agent: + event["agentId"] = agent + return event + + +def _assistant(session: str, timestamp: str, message_id: str, *content: dict, agent: str | None = None) -> dict: + return _event( + session, + "assistant", + timestamp, + list(content), + agent=agent, + message_id=message_id, + uuid=f"{message_id}-event", + ) + + +def _tool_result( + session: str, + timestamp: str, + tool_call_id: str, + *, + agent: str | None = None, + child_id: str | None = None, + status: str = "completed", + is_error: bool = False, +) -> dict: + content = [{"type": "tool_result", "tool_use_id": tool_call_id, "content": "result", "is_error": is_error}] + event = _event(session, "user", timestamp, content, agent=agent, uuid=f"{tool_call_id}-result") + if child_id is not None: + event["toolUseResult"] = {"agentId": child_id, "status": status} + return event + + +def test_extracts_nested_tree_model_refs_and_parallel_tool_timing(tmp_path: Path) -> None: + session = "session-root" + child = "agent-child" + grandchild = "agent-grandchild" + root = tmp_path / "projects" / "work" / f"{session}.jsonl" + subagents = root.parent / session / "subagents" + + _write( + root, + _event(session, "user", "2026-07-22T10:00:00Z", "solve", uuid="root-user"), + _assistant( + session, + "2026-07-22T10:00:01Z", + "msg-root", + {"type": "thinking", "thinking": "plan", "signature": "sig"}, + ), + _assistant( + session, + "2026-07-22T10:00:02Z", + "msg-root", + {"type": "tool_use", "id": "tool-fast", "name": "Read", "input": {"path": "a"}}, + {"type": "tool_use", "id": "tool-child", "name": "Agent", "input": {"prompt": "delegate"}}, + ), + _tool_result(session, "2026-07-22T10:00:03Z", "tool-fast"), + _tool_result( + session, + "2026-07-22T10:00:05Z", + "tool-child", + child_id=child, + ), + ) + _write( + subagents / f"{child}.jsonl", + _event(session, "user", "2026-07-22T10:00:02.100Z", "child task", agent=child, uuid="child-user"), + _assistant( + session, + "2026-07-22T10:00:03Z", + "msg-child", + {"type": "tool_use", "id": "tool-grandchild", "name": "Agent", "input": {}}, + agent=child, + ), + _tool_result( + session, + "2026-07-22T10:00:04Z", + "tool-grandchild", + agent=child, + child_id=grandchild, + ), + ) + _write( + subagents / f"{grandchild}.jsonl", + _assistant( + session, + "2026-07-22T10:00:03.100Z", + "msg-grandchild", + {"type": "text", "text": "done"}, + agent=grandchild, + ), + ) + + bundle = extract_claude_code_observations(tmp_path, model_ref=MODEL_REF) + + assert [invocation.invocation_id for invocation in bundle.invocations] == [session, child, grandchild] + root_invocation, child_invocation, grandchild_invocation = bundle.invocations + assert child_invocation.parent_invocation_id == session + assert child_invocation.spawned_by_tool_call_id == "tool-child" + assert grandchild_invocation.parent_invocation_id == child + assert grandchild_invocation.spawned_by_tool_call_id == "tool-grandchild" + assert [reference.response_id for reference in root_invocation.model_calls] == ["msg-root"] + assert [reference.response_id for reference in child_invocation.model_calls] == ["msg-child"] + assert [reference.response_id for reference in grandchild_invocation.model_calls] == ["msg-grandchild"] + assert all( + reference.model_ref == MODEL_REF for invocation in bundle.invocations for reference in invocation.model_calls + ) + assert [item.type for item in root_invocation.conversation] == [ + "message", + "reasoning", + "function_call", + "function_call", + "function_call_output", + "function_call_output", + ] + + timings = {tool.tool_call_id: tool for tool in bundle.tool_calls} + assert timings["tool-fast"].duration_ms == pytest.approx(1000) + assert timings["tool-child"].duration_ms == pytest.approx(3000) + assert timings["tool-grandchild"].duration_ms == pytest.approx(1000) + assert all(tool.timing_source == "artifact" for tool in timings.values()) + assert bundle.gaps == [] + + +def test_extracts_explicit_compaction_markers(tmp_path: Path) -> None: + _write( + tmp_path / "projects" / "work" / "session.jsonl", + _event( + "root", + "user", + "2026-07-22T10:00:00Z", + "summary", + message_extra={ + "isCompactSummary": True, + "compactMetadata": {"tokensBefore": 1000, "tokensAfter": 200, "trigger": "auto"}, + }, + ), + _event( + "root", + "system", + "2026-07-22T10:00:01Z", + "", + subtype="compact_boundary", + compact_metadata={"preTokens": 900, "postTokens": 180}, + ), + ) + + bundle = extract_claude_code_observations(tmp_path) + + assert [event.trigger for event in bundle.compactions] == ["auto", None] + assert bundle.compactions[0].tokens_before == 1000 + assert bundle.compactions[0].tokens_after == 200 + assert bundle.compactions[1].tokens_before == 900 + assert bundle.compactions[1].tokens_after == 180 + + +def test_malformed_and_incomplete_artifacts_produce_sanitized_gaps(tmp_path: Path) -> None: + secret = "do-not-leak-this-line" + _write( + tmp_path / "projects" / "work" / "root.jsonl", + f'{{"secret":"{secret}"', + _assistant( + "root", + "bad-timestamp", + "msg-root", + {"type": "tool_use", "id": "pending", "name": "Bash", "input": {}}, + ), + _tool_result("root", "2026-07-22T10:00:03Z", "orphan"), + ) + _write( + tmp_path / "projects" / "work" / "subagents" / "agent-orphan.jsonl", + _assistant( + "root", + "2026-07-22T10:00:01Z", + "msg-orphan", + {"type": "text", "text": "answer"}, + agent="agent-orphan", + ), + ) + + bundle = extract_claude_code_observations(tmp_path) + codes = {gap.code for gap in bundle.gaps} + + assert { + "malformed_transcript_line", + "subagent_parent_unavailable", + "tool_result_missing", + "tool_start_timestamp_missing", + "tool_start_missing", + } <= codes + assert all(not invocation.model_calls for invocation in bundle.invocations) + assert secret not in bundle.model_dump_json() + + +def test_ignores_non_transcript_jsonl_and_reports_no_usable_transcript(tmp_path: Path) -> None: + _write( + tmp_path / "skills" / "fixture.jsonl", + _assistant("unrelated", "2026-07-22T10:00:00Z", "msg-unrelated", {"type": "text", "text": "x"}), + ) + (tmp_path / "projects").mkdir() + + bundle = extract_claude_code_observations(tmp_path, model_ref=MODEL_REF) + + assert bundle.invocations == [] + assert "agent_transcript_unavailable" in {gap.code for gap in bundle.gaps} + + +def test_reports_missing_response_id_and_unsupported_content_blocks(tmp_path: Path) -> None: + _write( + tmp_path / "projects" / "work" / "session.jsonl", + _event( + "root", + "assistant", + "2026-07-22T10:00:00Z", + [{"type": "image", "source": "omitted"}], + uuid="assistant-event", + ), + _event( + "root", + "user", + "2026-07-22T10:00:01Z", + [{"type": "image", "source": "omitted"}], + uuid="user-event", + ), + ) + + bundle = extract_claude_code_observations(tmp_path, model_ref=MODEL_REF) + codes = {gap.code for gap in bundle.gaps} + + assert "model_response_id_missing" in codes + assert "unsupported_assistant_content_block" in codes + assert "unsupported_user_content_block" in codes + assert bundle.invocations[0].model_calls == [] diff --git a/responses_api_agents/hermes_agent/app.py b/responses_api_agents/hermes_agent/app.py index b828a52de4..70ba3c8b1a 100644 --- a/responses_api_agents/hermes_agent/app.py +++ b/responses_api_agents/hermes_agent/app.py @@ -21,12 +21,12 @@ import tempfile from asyncio import Semaphore from time import time -from typing import Any, Optional +from typing import Any, Callable, Optional from uuid import uuid4 import model_tools # noqa: F401 # fail-fast if hermes-agent isn't installed # pyright: ignore[reportMissingImports] from fastapi import Request -from pydantic import ConfigDict +from pydantic import ConfigDict, Field from nemo_gym.base_resources_server import BaseRunRequest, BaseVerifyResponse from nemo_gym.base_responses_api_agent import ( @@ -47,7 +47,9 @@ NeMoGymResponseOutputTokensDetails, NeMoGymResponseUsage, ) +from nemo_gym.rollout_observability import AgentEpisode, AgentObservationBundle, ObservationGap from nemo_gym.server_utils import get_response_json, raise_for_status +from responses_api_agents.hermes_agent.observability import HermesAgentObserver def _trajectory_to_output_items(messages, n_input): @@ -175,6 +177,10 @@ class HermesAgentVerifyResponse(BaseVerifyResponse): model_config = ConfigDict(extra="allow") turns_used: int = 0 finished_naturally: bool = False + ng_agent_observations: AgentObservationBundle | None = Field( + default=None, + exclude_if=lambda value: value is None, + ) class HermesAgent(SimpleResponsesAPIAgent): @@ -251,10 +257,12 @@ def model_post_init(self, __context: Any) -> None: _f.write(self._build_config()) os.environ["HERMES_HOME"] = hermes_home - async def responses( + async def _create_response( self, - request: Request, - body: NeMoGymResponseCreateParamsNonStreaming = Body(), + body: NeMoGymResponseCreateParamsNonStreaming, + *, + rollout_id: Optional[str] = None, + observation_collector: Optional[Callable[[AgentObservationBundle], None]] = None, ) -> NeMoGymResponse: from run_agent import AIAgent # from hermes-agent on path # pyright: ignore[reportMissingImports] @@ -265,8 +273,6 @@ async def responses( user_message, history, input_system = _split_input_to_user_and_history(body.input) system_message = self.config.system_prompt or input_system - # A prefixed self-call carries the rollout id into the model-server base URL. - rollout_id = request.path_params.get("rollout_id") if request is not None else None base_url = self.resolve_model_base_url(self.config.model_server.name, rollout_id) model_name = str(self.config.model_server.name) @@ -296,6 +302,12 @@ def _patched_build_api_kwargs(api_messages): return kw agent._build_api_kwargs = _patched_build_api_kwargs + observer = None + if observation_collector is not None: + try: + observer = HermesAgentObserver().instrument(agent) + except Exception: + LOG.exception("failed to initialize Hermes observability") # Interrupt the agent cleanly on SIGTERM so run_conversation returns with partial messages # instead of being killed mid-turn (which would leave response.json unwritten). A single @@ -303,6 +315,8 @@ def _patched_build_api_kwargs(api_messages): self._ensure_sigterm_handler() self.active_agents.add(agent) + result = None + agent_error: Optional[BaseException] = None try: result = await asyncio.to_thread( agent.run_conversation, @@ -310,8 +324,31 @@ def _patched_build_api_kwargs(api_messages): system_message, history, ) + except BaseException as exc: + agent_error = exc + raise finally: self.active_agents.discard(agent) + if observation_collector is not None: + try: + observations = ( + observer.finish(result, error=agent_error) + if observer is not None + else AgentObservationBundle( + source="hermes", + gaps=[ObservationGap(code="observation_capture_failed", source="hermes")], + ) + ) + except Exception: + LOG.exception("failed to finish Hermes observability") + observations = AgentObservationBundle( + source="hermes", + gaps=[ObservationGap(code="observation_capture_failed", source="hermes")], + ) + try: + observation_collector(observations) + except Exception: + LOG.exception("failed to return Hermes observations") messages = result.get("messages") or [] # aiagent omits system from returned messages @@ -372,6 +409,39 @@ def _patched_build_api_kwargs(api_messages): ), ) + async def responses( + self, + request: Request, + body: NeMoGymResponseCreateParamsNonStreaming = Body(), + ) -> NeMoGymResponse: + rollout_id = request.path_params.get("rollout_id") if request is not None else None + return await self._create_response(body, rollout_id=rollout_id) + + async def responses_with_observations( + self, + request: Optional[Request], + body: NeMoGymResponseCreateParamsNonStreaming, + *, + rollout_id: Optional[str] = None, + ) -> AgentEpisode: + observations: Optional[AgentObservationBundle] = None + + def collect(bundle: AgentObservationBundle) -> None: + nonlocal observations + observations = bundle + + response = await self._create_response( + body, + rollout_id=rollout_id, + observation_collector=collect, + ) + if observations is None: + observations = AgentObservationBundle( + source="hermes", + gaps=[ObservationGap(code="observation_capture_failed", source="hermes")], + ) + return AgentEpisode(response=response, observations=observations) + async def run(self, request: Request, body: HermesAgentRunRequest) -> HermesAgentVerifyResponse: async with self.sem: cookies = request.cookies @@ -385,20 +455,33 @@ async def run(self, request: Request, body: HermesAgentRunRequest) -> HermesAgen await raise_for_status(seed_resp) cookies = seed_resp.cookies - agent_resp = 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(agent_resp) - cookies = agent_resp.cookies - agent_resp_json = await get_response_json(agent_resp) + rollout_id = self.rollout_id_from_run(body) + if rollout_id is not None: + episode = await self.responses_with_observations( + request, + body.responses_create_params, + rollout_id=rollout_id, + ) + agent_resp, observations = episode.response, episode.observations + agent_resp_json = agent_resp.model_dump(mode="json") + else: + agent_resp = 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(agent_resp) + cookies = agent_resp.cookies + agent_resp_json = await get_response_json(agent_resp) + observations = None verify_resp = await self.server_client.post( server_name=self.config.resources_server.name, url_path="/verify", - json=body.model_dump() | {"response": agent_resp_json}, + json=body.model_dump() + | {"response": agent_resp_json} + | ({"rollout_id": rollout_id} if rollout_id is not None else {}), cookies=cookies, ) await raise_for_status(verify_resp) @@ -413,9 +496,10 @@ async def run(self, request: Request, body: HermesAgentRunRequest) -> HermesAgen last = gym_resp.output[-1] if gym_resp.output else None naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant" - return HermesAgentVerifyResponse.model_validate( - verify_json | {"turns_used": turns, "finished_naturally": naturally} - ) + result = verify_json | {"turns_used": turns, "finished_naturally": naturally} + if observations is not None: + result["ng_agent_observations"] = observations.model_dump(mode="json") + return HermesAgentVerifyResponse.model_validate(result) if __name__ == "__main__": diff --git a/responses_api_agents/hermes_agent/observability.py b/responses_api_agents/hermes_agent/observability.py new file mode 100644 index 0000000000..e7a178a942 --- /dev/null +++ b/responses_api_agents/hermes_agent/observability.py @@ -0,0 +1,381 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Observability hooks for the pinned Hermes ``AIAgent`` integration.""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Callable, Iterable +from time import monotonic, time +from typing import Any + +from nemo_gym.openai_utils import ( + NeMoGymEasyInputMessage, + NeMoGymFunctionCallOutput, + NeMoGymResponseFunctionToolCall, + NeMoGymResponseInputItem, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, +) +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ObservationGap, + ToolCallObservation, +) + + +_SOURCE = "hermes" + + +def _text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join( + part.get("text", "") if isinstance(part, dict) else str(getattr(part, "text", "")) for part in content + ) + return "" if content is None else str(content) + + +def normalize_hermes_messages(messages: Iterable[Any], *, id_prefix: str = "hermes") -> list[NeMoGymResponseInputItem]: + """Convert a Hermes conversation to ordered Gym Responses items.""" + output: list[NeMoGymResponseInputItem] = [] + for index, message in enumerate(messages): + if not isinstance(message, dict): + continue + role, content = message.get("role"), _text(message.get("content")) + if role in {"system", "user", "developer"}: + output.append(NeMoGymEasyInputMessage(role=role, content=content)) + elif role == "assistant": + output.append( + NeMoGymResponseOutputMessage( + id=str(message.get("id") or f"{id_prefix}-message-{index}"), + content=[NeMoGymResponseOutputText(text=content, annotations=[])], + ) + ) + for call in message.get("tool_calls") or []: + function = call.get("function") if isinstance(call, dict) else None + if not isinstance(function, dict): + continue + arguments = function.get("arguments", "") + if not isinstance(arguments, str): + try: + arguments = json.dumps(arguments, ensure_ascii=False) + except (TypeError, ValueError): + arguments = "{}" + call_id = str(call.get("id") or "") + output.append( + NeMoGymResponseFunctionToolCall( + arguments=arguments, + call_id=call_id, + id=call_id or None, + name=str(function.get("name") or ""), + status="completed", + ) + ) + elif role == "tool": + output.append( + NeMoGymFunctionCallOutput( + call_id=str(message.get("tool_call_id") or ""), + output=content, + status="completed", + ) + ) + return output + + +class _ObservedChildren(list): + def __init__(self, values: Iterable[Any], observer: "HermesAgentObserver", parent_id: str): + super().__init__(values) + self.observer, self.parent_id = observer, parent_id + + def append(self, child: Any) -> None: + super().append(child) + self.observer._child_added(child, self.parent_id) + + +class HermesAgentObserver: + """Instrument one Hermes agent tree without modifying global Hermes state.""" + + def __init__(self, *, root_invocation_id: str = "root") -> None: + self._lock = threading.RLock() + self._current = threading.local() + self._root_id = root_invocation_id + self._child_index = 0 + self._agents: set[int] = set() + self._tools_by_args: dict[int, tuple[str, str]] = {} + self._tools: dict[tuple[str, str], ToolCallObservation] = {} + self._started_ticks: dict[tuple[str, str], float] = {} + self._invocations = {root_invocation_id: AgentInvocation(invocation_id=root_invocation_id)} + self._compactions: list[ContextCompactionObservation] = [] + self._gaps: list[ObservationGap] = [] + + def instrument(self, agent: Any) -> "HermesAgentObserver": + self._instrument_safely(agent, self._root_id, wrap_conversation=False) + return self + + def finish( + self, result: dict[str, Any] | None = None, *, error: BaseException | None = None + ) -> AgentObservationBundle: + self._record_conversation(self._root_id, result, error) + with self._lock: + for tool in self._tools.values(): + if tool.status == "unknown": + tool.status = "incomplete" + if tool.timing_source is None: + self._gap( + "tool_execution_boundary_unavailable", + tool.invocation_id, + f"No matching _invoke_tool execution was observed for {tool.tool_call_id}.", + ) + for invocation in self._invocations.values(): + self._gap( + "model_call_ownership_unavailable", + invocation.invocation_id, + "Hermes messages do not expose outer model response IDs.", + ) + return AgentObservationBundle( + source=_SOURCE, + invocations=list(self._invocations.values()), + tool_calls=list(self._tools.values()), + compactions=self._compactions, + gaps=self._gaps, + ) + + def _instrument_safely(self, agent: Any, invocation_id: str, *, wrap_conversation: bool) -> None: + try: + self._instrument(agent, invocation_id, wrap_conversation) + except Exception as exc: + self._gap("hermes_observer_error", invocation_id, f"instrument: {type(exc).__name__}") + + def _instrument(self, agent: Any, invocation_id: str, wrap_conversation: bool) -> None: + with self._lock: + agent_id = id(agent) + if agent_id in self._agents: + return + self._agents.add(agent_id) + + self._chain_callback(agent, "tool_start_callback", self._tool_started, invocation_id) + self._chain_callback(agent, "tool_complete_callback", self._tool_completed, invocation_id) + self._wrap_invoke(agent, invocation_id) + self._wrap_compaction(agent, invocation_id) + + children = getattr(agent, "_active_children", None) + if isinstance(children, list): + setattr(agent, "_active_children", _ObservedChildren(children, self, invocation_id)) + else: + self._gap("hermes_hook_unavailable", invocation_id, "_active_children") + + if wrap_conversation: + original = getattr(agent, "run_conversation", None) + if callable(original): + + def run(*args: Any, **kwargs: Any) -> Any: + try: + child_result = original(*args, **kwargs) + except BaseException as exc: + self._record_conversation(invocation_id, None, exc) + raise + self._record_conversation(invocation_id, child_result, None) + return child_result + + setattr(agent, "run_conversation", run) + else: + self._gap("hermes_hook_unavailable", invocation_id, "run_conversation") + + def _chain_callback(self, agent: Any, name: str, observer: Callable[..., None], invocation_id: str) -> None: + if not hasattr(agent, name): + self._gap("hermes_hook_unavailable", invocation_id, name) + return + previous = getattr(agent, name) + + def callback(*args: Any, **kwargs: Any) -> None: + try: + observer(invocation_id, *args, **kwargs) + except Exception: + self._gap("hermes_observer_error", invocation_id, name) + if callable(previous): + try: + previous(*args, **kwargs) + except Exception: + pass # Hermes callbacks are explicitly non-fatal. + + setattr(agent, name, callback) + + def _wrap_invoke(self, agent: Any, invocation_id: str) -> None: + original = getattr(agent, "_invoke_tool", None) + if not callable(original): + self._gap("hermes_hook_unavailable", invocation_id, "_invoke_tool") + return + + def invoke(*args: Any, **kwargs: Any) -> Any: + key, previous = None, getattr(self._current, "tool", None) + try: + name = args[0] if args else kwargs.get("function_name") + call_args = args[1] if len(args) > 1 else kwargs.get("function_args") + with self._lock: + key = self._tools_by_args.get(id(call_args)) + if key is not None and key[0] == invocation_id: + self._current.tool = (*key, name) + self._start_execution(key) + else: + key = None + except Exception: + self._gap("hermes_observer_error", invocation_id, "_invoke_tool") + failed = False + try: + return original(*args, **kwargs) + except BaseException: + failed = True + raise + finally: + if key is not None: + self._end_execution(key, failed=failed) + self._current.tool = previous + + setattr(agent, "_invoke_tool", invoke) + + def _wrap_compaction(self, agent: Any, invocation_id: str) -> None: + original = getattr(agent, "_compress_context", None) + if not callable(original): + self._gap("hermes_hook_unavailable", invocation_id, "_compress_context") + return + + def compact(*args: Any, **kwargs: Any) -> Any: + result = original(*args, **kwargs) + try: + before = kwargs.get("approx_tokens") + after = getattr(getattr(agent, "context_compressor", None), "last_prompt_tokens", None) + with self._lock: + self._compactions.append( + ContextCompactionObservation( + invocation_id=invocation_id, + observed_at=time(), + trigger="context_pressure", + tokens_before=before if type(before) is int else None, + tokens_after=after if type(after) is int else None, + ) + ) + except Exception: + self._gap("hermes_observer_error", invocation_id, "_compress_context") + return result + + setattr(agent, "_compress_context", compact) + + def _child_added(self, child: Any, parent_id: str) -> None: + try: + with self._lock: + self._child_index += 1 + invocation_id = f"{parent_id}.child-{self._child_index}" + current = getattr(self._current, "tool", None) + call_id = current[1] if current and current[0] == parent_id and current[2] == "delegate_task" else None + self._invocations[invocation_id] = AgentInvocation( + invocation_id=invocation_id, + parent_invocation_id=parent_id, + spawned_by_tool_call_id=call_id, + ) + if call_id is None: + self._gap("subagent_spawn_unattributed", invocation_id, "No active delegate_task matched.") + self._instrument_safely(child, invocation_id, wrap_conversation=True) + except Exception as exc: + self._gap("hermes_observer_error", parent_id, f"child: {type(exc).__name__}") + + def _tool_started(self, invocation_id: str, call_id: Any, name: Any, args: Any) -> None: + key = (invocation_id, str(call_id or "")) + with self._lock: + if key in self._tools: + self._gap("duplicate_tool_call", invocation_id, key[1]) + self._tools[key] = ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=key[1], + tool_name=str(name or "") or None, + ) + if isinstance(args, dict): + self._tools_by_args[id(args)] = key + self._current.tool = (*key, str(name or "")) + + def _tool_completed(self, invocation_id: str, call_id: Any, name: Any, args: Any, result: Any) -> None: + key = (invocation_id, str(call_id or "")) + with self._lock: + if key not in self._tools: + self._tool_started(invocation_id, call_id, name, args) + tool = self._tools[key] + failed = self._failed_result(result) + if tool.timing_source is None: + tool.status = "failed" if failed else "completed" + elif failed: + tool.status = "failed" + if isinstance(args, dict): + self._tools_by_args.pop(id(args), None) + current = getattr(self._current, "tool", None) + if current and current[:2] == key: + self._current.tool = None + + def _start_execution(self, key: tuple[str, str]) -> None: + with self._lock: + tool = self._tools[key] + tool.started_at = time() + tool.clock_id = "unix_epoch" + tool.timing_source = "executor" + self._started_ticks[key] = monotonic() + + def _end_execution(self, key: tuple[str, str], *, failed: bool) -> None: + try: + with self._lock: + tool = self._tools[key] + if tool.completed_at is None: + tool.completed_at = time() + tool.duration_ms = max(0.0, (monotonic() - self._started_ticks.pop(key)) * 1000) + tool.status = "failed" if failed else "completed" + elif failed: + tool.status = "failed" + except Exception: + self._gap("hermes_observer_error", key[0], "_invoke_tool") + + def _record_conversation( + self, + invocation_id: str, + result: dict[str, Any] | None, + error: BaseException | None, + ) -> None: + try: + messages = result.get("messages") if isinstance(result, dict) else [] + conversation = normalize_hermes_messages(messages or [], id_prefix=invocation_id) + status = "failed" if error or (result and result.get("error")) else "unknown" + if isinstance(result, dict) and status != "failed": + if result.get("interrupted"): + status = "incomplete" + elif result.get("completed") or result.get("final_response"): + status = "completed" + elif messages: + status = "incomplete" + with self._lock: + self._invocations[invocation_id].conversation = conversation + self._invocations[invocation_id].status = status + except Exception as exc: + self._gap("hermes_observer_error", invocation_id, f"conversation: {type(exc).__name__}") + + @staticmethod + def _failed_result(result: Any) -> bool: + if not isinstance(result, str): + return False + value = result.lstrip() + if value.lower().startswith("error executing tool"): + return True + try: + payload = json.loads(value) + except (json.JSONDecodeError, TypeError): + return False + return isinstance(payload, dict) and ( + payload.get("status") in {"error", "failed"} or bool(payload.get("error")) + ) + + def _gap(self, code: str, invocation_id: str | None, detail: str | None = None) -> None: + with self._lock: + gap = ObservationGap(code=code, source=_SOURCE, invocation_id=invocation_id, detail=detail) + if gap not in self._gaps: + self._gaps.append(gap) diff --git a/responses_api_agents/hermes_agent/tests/test_app.py b/responses_api_agents/hermes_agent/tests/test_app.py index c59a702090..cc2eb98968 100644 --- a/responses_api_agents/hermes_agent/tests/test_app.py +++ b/responses_api_agents/hermes_agent/tests/test_app.py @@ -13,23 +13,41 @@ # See the License for the specific language governing permissions and # limitations under the License. import asyncio -from unittest.mock import MagicMock +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest from nemo_gym.openai_utils import ( NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, + NeMoGymResponse, NeMoGymResponseFunctionToolCall, NeMoGymResponseOutputMessageForTraining, ) +from nemo_gym.rollout_observability import AgentEpisode, AgentObservationBundle from nemo_gym.server_utils import ServerClient from responses_api_agents.hermes_agent.app import ( HermesAgent, HermesAgentConfig, + HermesAgentRunRequest, ModelServerRef, ResourcesServerRef, _split_input_to_user_and_history, _trajectory_to_output_items, ) +from responses_api_agents.hermes_agent.observability import HermesAgentObserver + + +class _FakeResponse: + ok = True + + def __init__(self, payload: dict, cookies: dict | None = None) -> None: + self.payload = payload + self.cookies = cookies or {} + + async def read(self) -> bytes: + return json.dumps(self.payload).encode() def _config(**kwargs) -> HermesAgentConfig: @@ -248,8 +266,6 @@ def test_skips_non_dict_items(self) -> None: class TestRolloutCorrelation: - """The prefixed self-call path makes responses() build AIAgent with the same model URL prefix.""" - def test_responses_applies_rollout_prefix(self, monkeypatch) -> None: from fastapi.testclient import TestClient @@ -282,3 +298,132 @@ def run_conversation(self, *args, **kwargs) -> dict: asyncio.run(agent.responses(request=None, body=NeMoGymResponseCreateParamsNonStreaming(input="hi"))) assert seen["base_url"] == "http://h:1/v1" + + episode = asyncio.run( + agent.responses_with_observations( + request=None, + body=NeMoGymResponseCreateParamsNonStreaming(input="hi"), + rollout_id="rid", + ) + ) + assert seen["base_url"] == "http://h:1/ng-rollout/rid/v1" + assert episode.observations.source == "hermes" + assert episode.observations.invocations[0].invocation_id == "root" + + +class TestObservability: + def test_observation_failure_does_not_change_response(self, monkeypatch) -> None: + import nemo_gym.base_responses_api_agent as base_agent + from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming + + monkeypatch.setattr(base_agent, "get_first_server_config_dict", lambda _gc, _name: {"host": "h", "port": 1}) + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = {} + server_client._build_server_base_url = lambda _cfg: "http://h:1" + agent = HermesAgent(config=_config(), server_client=server_client) + monkeypatch.setattr(agent, "_ensure_sigterm_handler", lambda: None) + + class _StubAIAgent: + def __init__(self, **kwargs) -> None: + self._build_api_kwargs = lambda _messages: {} + + def run_conversation(self, *args, **kwargs) -> dict: + return { + "completed": True, + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "ok"}, + ], + } + + monkeypatch.setattr("run_agent.AIAgent", _StubAIAgent) + body = NeMoGymResponseCreateParamsNonStreaming(input="hi") + baseline = asyncio.run(agent.responses(request=None, body=body)) + + def fail_finish(*args, **kwargs): + raise RuntimeError("observer failed") + + monkeypatch.setattr(HermesAgentObserver, "finish", fail_finish) + episode = asyncio.run(agent.responses_with_observations(request=None, body=body, rollout_id="rid")) + + assert episode.response.output == baseline.output + assert episode.response.usage == baseline.usage + assert [gap.code for gap in episode.observations.gaps] == ["observation_capture_failed"] + + def test_run_passes_rollout_id_to_verifier(self) -> None: + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = {"observability_enabled": True} + agent = HermesAgent(config=_config(), server_client=server_client) + response = NeMoGymResponse.model_validate( + { + "id": "resp-1", + "created_at": 1, + "model": "model", + "object": "response", + "output": [], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + } + ) + observed_response = AsyncMock( + return_value=AgentEpisode( + response=response, + observations=AgentObservationBundle(source="hermes"), + ) + ) + + async def post(server_name, url_path, json=None, cookies=None, **kwargs): + if url_path == "/seed_session": + return _FakeResponse({}, {"session": "1"}) + return _FakeResponse(json | {"reward": 1.0}) + + server_client.post = AsyncMock(side_effect=post) + request = MagicMock() + request.cookies = {} + body = HermesAgentRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + with patch.object(HermesAgent, "responses_with_observations", observed_response): + asyncio.run(agent.run(request, body)) + + assert server_client.post.await_args_list[-1].kwargs["json"]["rollout_id"] == "1-2" + + def test_observer_failure_does_not_mask_agent_exception(self, monkeypatch) -> None: + import nemo_gym.base_responses_api_agent as base_agent + from nemo_gym.openai_utils import NeMoGymResponseCreateParamsNonStreaming + + monkeypatch.setattr(base_agent, "get_first_server_config_dict", lambda _gc, _name: {"host": "h", "port": 1}) + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = {} + server_client._build_server_base_url = lambda _cfg: "http://h:1" + agent = HermesAgent(config=_config(), server_client=server_client) + monkeypatch.setattr(agent, "_ensure_sigterm_handler", lambda: None) + + class _FailingAIAgent: + def __init__(self, **kwargs) -> None: + self._build_api_kwargs = lambda _messages: {} + + def run_conversation(self, *args, **kwargs) -> dict: + raise ValueError("agent failed") + + monkeypatch.setattr("run_agent.AIAgent", _FailingAIAgent) + monkeypatch.setattr( + HermesAgentObserver, + "finish", + MagicMock(side_effect=RuntimeError("observer failed")), + ) + + with pytest.raises(ValueError, match="agent failed"): + asyncio.run( + agent.responses_with_observations( + request=None, + body=NeMoGymResponseCreateParamsNonStreaming(input="hi"), + rollout_id="rid", + ) + ) diff --git a/responses_api_agents/hermes_agent/tests/test_observability.py b/responses_api_agents/hermes_agent/tests/test_observability.py new file mode 100644 index 0000000000..eb0969f608 --- /dev/null +++ b/responses_api_agents/hermes_agent/tests/test_observability.py @@ -0,0 +1,285 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import threading +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from responses_api_agents.hermes_agent.observability import ( + HermesAgentObserver, + normalize_hermes_messages, +) + + +class _FakeAgent: + def __init__(self, conversation=None): + self.tool_start_callback = None + self.tool_complete_callback = None + self._active_children = [] + self.context_compressor = SimpleNamespace(last_prompt_tokens=0) + self._conversation = conversation or {"completed": True, "messages": []} + + def _invoke_tool(self, function_name, function_args, effective_task_id): + return function_args.get("result", function_name) + + def _compress_context(self, messages, system_message, **kwargs): + self.context_compressor.last_prompt_tokens = 7 + return messages[-1:], system_message + + def run_conversation(self, *args, **kwargs): + return self._conversation + + +def _invocation(bundle, invocation_id): + return next(item for item in bundle.invocations if item.invocation_id == invocation_id) + + +def _tool(bundle, tool_call_id): + return next(item for item in bundle.tool_calls if item.tool_call_id == tool_call_id) + + +def test_normalize_hermes_messages_preserves_conversation_order(): + items = normalize_hermes_messages( + [ + {"role": "user", "content": "inspect"}, + { + "role": "assistant", + "content": "working", + "tool_calls": [ + { + "id": "call-1", + "function": {"name": "terminal", "arguments": {"command": "pwd"}}, + } + ], + }, + {"role": "tool", "tool_call_id": "call-1", "content": "/workspace"}, + { + "role": "assistant", + "content": [{"type": "text", "text": "done"}], + }, + ], + id_prefix="root", + ) + + assert [item.type for item in items] == [ + "message", + "message", + "function_call", + "function_call_output", + "message", + ] + assert items[2].arguments == '{"command": "pwd"}' + assert items[3].call_id == "call-1" + assert items[4].content[0].text == "done" + + +def test_matching_invoke_tool_records_executor_interval(): + calls = [] + agent = _FakeAgent() + agent.tool_start_callback = lambda *args: calls.append(("start", *args)) + agent.tool_complete_callback = lambda *args: calls.append(("complete", *args)) + observer = HermesAgentObserver().instrument(agent) + args = {"command": "pwd"} + + agent.tool_start_callback("call-1", "terminal", args) + agent._invoke_tool("terminal", args, "task") + agent.tool_complete_callback("call-1", "terminal", args, "/workspace") + bundle = observer.finish({"completed": True, "messages": []}) + + tool = _tool(bundle, "call-1") + assert tool.status == "completed" + assert tool.started_at is not None + assert tool.completed_at >= tool.started_at + assert tool.duration_ms >= 0 + assert [call[0] for call in calls] == ["start", "complete"] + + +@pytest.mark.parametrize( + "result", + [ + "Error executing tool: boom", + '{"error":"boom"}', + '{"status":"error","message":"boom"}', + ], +) +def test_tool_error_payloads_are_reported_as_failed(result): + agent = _FakeAgent() + observer = HermesAgentObserver().instrument(agent) + args = {"command": "pwd"} + + agent.tool_start_callback("call-1", "terminal", args) + agent.tool_complete_callback("call-1", "terminal", args, result) + + assert _tool(observer.finish({"completed": True, "messages": []}), "call-1").status == "failed" + + +def test_callback_only_tool_keeps_status_without_claiming_executor_timing(): + agent = _FakeAgent() + observer = HermesAgentObserver().instrument(agent) + callback_args = {"command": "pwd"} + + agent.tool_start_callback("call-1", "terminal", callback_args) + agent._invoke_tool("terminal", {"command": "pwd"}, "task") + agent.tool_complete_callback("call-1", "terminal", callback_args, "/workspace") + bundle = observer.finish({"completed": True, "messages": []}) + + tool = _tool(bundle, "call-1") + assert tool.status == "completed" + assert tool.started_at is None + assert tool.completed_at is None + assert tool.duration_ms is None + assert tool.timing_source is None + assert any( + gap.code == "tool_execution_boundary_unavailable" and gap.invocation_id == "root" for gap in bundle.gaps + ) + + +def test_concurrent_tool_intervals_end_at_each_worker_not_completion_callback(): + from run_agent import AIAgent + + agent = _FakeAgent() + release = {"fast": threading.Event(), "slow": threading.Event()} + entered = {"fast": threading.Event(), "slow": threading.Event()} + + def invoke(function_name, function_args, effective_task_id): + entered[function_name].set() + assert release[function_name].wait(timeout=2) + return function_name + + agent._invoke_tool = invoke + agent._interrupt_requested = False + agent.quiet_mode = False + agent.verbose_logging = False + agent.log_prefix_chars = 100 + agent.tool_progress_callback = None + agent._checkpoint_mgr = SimpleNamespace(enabled=False) + agent._get_budget_warning = lambda _count: None + observer = HermesAgentObserver().instrument(agent) + calls = [ + SimpleNamespace(id=f"{name}-call", function=SimpleNamespace(name=name, arguments="{}")) + for name in ("fast", "slow") + ] + execute = AIAgent._execute_tool_calls_concurrent.__get__(agent, _FakeAgent) + execution = threading.Thread(target=execute, args=(SimpleNamespace(tool_calls=calls), [], "task")) + execution.start() + assert entered["fast"].wait(timeout=2) + assert entered["slow"].wait(timeout=2) + release["fast"].set() + assert not threading.Event().wait(timeout=0.02) + release["slow"].set() + execution.join(timeout=2) + assert not execution.is_alive() + bundle = observer.finish({"completed": True, "messages": []}) + + fast = _tool(bundle, "fast-call") + slow = _tool(bundle, "slow-call") + assert fast.completed_at < slow.completed_at + assert fast.duration_ms < slow.duration_ms + + +def test_delegate_children_retain_exact_tree_and_full_conversations(): + root = _FakeAgent() + observer = HermesAgentObserver(root_invocation_id="root").instrument(root) + delegate_args = {"tasks": [{"goal": "one"}]} + root.tool_start_callback("delegate-1", "delegate_task", delegate_args) + + child = _FakeAgent( + { + "completed": True, + "messages": [ + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "child answer"}, + ], + } + ) + root._active_children.append(child) + child.run_conversation("one") + child_delegate_args = {"goal": "nested"} + child.tool_start_callback("delegate-2", "delegate_task", child_delegate_args) + grandchild = _FakeAgent( + { + "completed": True, + "messages": [{"role": "assistant", "content": "nested answer"}], + } + ) + child._active_children.append(grandchild) + grandchild.run_conversation("nested") + child.tool_complete_callback("delegate-2", "delegate_task", child_delegate_args, "ok") + root.tool_complete_callback("delegate-1", "delegate_task", delegate_args, "ok") + + bundle = observer.finish( + { + "completed": True, + "messages": [ + {"role": "user", "content": "root task"}, + {"role": "assistant", "content": "root answer"}, + ], + } + ) + + root_invocation = _invocation(bundle, "root") + child_invocation = _invocation(bundle, "root.child-1") + grandchild_invocation = _invocation(bundle, "root.child-1.child-2") + assert root_invocation.status == "completed" + assert child_invocation.parent_invocation_id == "root" + assert child_invocation.spawned_by_tool_call_id == "delegate-1" + assert [item.type for item in child_invocation.conversation] == ["message", "message"] + assert grandchild_invocation.parent_invocation_id == "root.child-1" + assert grandchild_invocation.spawned_by_tool_call_id == "delegate-2" + assert all(not invocation.model_calls for invocation in bundle.invocations) + ownership_gaps = [gap for gap in bundle.gaps if gap.code == "model_call_ownership_unavailable"] + assert {gap.invocation_id for gap in ownership_gaps} == { + "root", + "root.child-1", + "root.child-1.child-2", + } + + +def test_compaction_is_explicit_and_hook_failures_do_not_break_execution(): + agent = _FakeAgent() + agent.tool_start_callback = MagicMock(side_effect=RuntimeError("callback")) + observer = HermesAgentObserver().instrument(agent) + + assert agent._invoke_tool("terminal", {"result": "ok"}, "task") == "ok" + compressed, prompt = agent._compress_context([{"role": "user", "content": "x"}], "system", approx_tokens=42) + assert compressed == [{"role": "user", "content": "x"}] + assert prompt == "system" + bundle = observer.finish({"interrupted": True, "messages": [{"role": "user", "content": "x"}]}) + + assert _invocation(bundle, "root").status == "incomplete" + assert len(bundle.compactions) == 1 + assert bundle.compactions[0].tokens_before == 42 + assert bundle.compactions[0].tokens_after == 7 + + +def test_missing_private_hooks_are_reported_without_raising(): + class MinimalAgent: + pass + + observer = HermesAgentObserver().instrument(MinimalAgent()) + bundle = observer.finish({"completed": True, "messages": []}) + + unavailable = {gap.detail for gap in bundle.gaps if gap.code == "hermes_hook_unavailable"} + assert unavailable == { + "tool_start_callback", + "tool_complete_callback", + "_invoke_tool", + "_compress_context", + "_active_children", + } + + +@pytest.mark.parametrize( + ("result", "expected"), + [ + ({"completed": True, "messages": []}, "completed"), + ({"final_response": "ok", "messages": []}, "completed"), + ({"interrupted": True, "messages": []}, "incomplete"), + ({"error": "boom", "messages": []}, "failed"), + ], +) +def test_root_status(result, expected): + observer = HermesAgentObserver().instrument(_FakeAgent()) + assert _invocation(observer.finish(result), "root").status == expected diff --git a/responses_api_agents/openclaw_agent/app.py b/responses_api_agents/openclaw_agent/app.py index 266db68727..2043f338e7 100644 --- a/responses_api_agents/openclaw_agent/app.py +++ b/responses_api_agents/openclaw_agent/app.py @@ -23,7 +23,7 @@ from asyncio import Semaphore from pathlib import Path from time import time -from typing import Any, ClassVar, Optional +from typing import Any, Callable, ClassVar, Optional from uuid import uuid4 from fastapi import Request @@ -48,7 +48,9 @@ NeMoGymResponseOutputTokensDetails, NeMoGymResponseUsage, ) +from nemo_gym.rollout_observability import AgentEpisode, AgentObservationBundle, ObservationGap from nemo_gym.server_utils import get_response_json, raise_for_status +from responses_api_agents.openclaw_agent.observability import build_openclaw_observations from responses_api_agents.openclaw_agent.setup_openclaw import ensure_openclaw @@ -245,6 +247,10 @@ class OpenClawAgentVerifyResponse(BaseVerifyResponse): model_config = ConfigDict(extra="allow") turns_used: int = 0 finished_naturally: bool = False + ng_agent_observations: AgentObservationBundle | None = Field( + default=None, + exclude_if=lambda value: value is None, + ) class OpenClawAgent(SimpleResponsesAPIAgent): @@ -332,7 +338,10 @@ def _session_file(envelope: Optional[dict[str, Any]]) -> Optional[Path]: return Path(session_file) if isinstance(session_file, str) and session_file else None async def _run_openclaw( - self, instruction: str, system_prompt: Optional[str] + self, + instruction: str, + system_prompt: Optional[str], + observation_collector: Optional[Callable[[str], None]] = None, ) -> tuple[list[Any], dict[str, int], str]: """setup and run agent. returns (output_items, usage, model_name).""" prompt = instruction if not system_prompt else f"{system_prompt}\n\n{instruction}" @@ -383,17 +392,23 @@ async def _run_openclaw( output_items: list[Any] = [] session_path = self._session_file(envelope) if session_path and session_path.is_file(): - output_items = parse_openclaw_session(session_path.read_text(errors="replace")) + session_text = session_path.read_text(errors="replace") + output_items = parse_openclaw_session(session_text) + if output_items and observation_collector is not None: + try: + observation_collector(session_path.stem) + except Exception: + LOG.exception("failed to record OpenClaw session identity") if not output_items: output_items = fallback_items return output_items, usage, self.config.model finally: shutil.rmtree(work_dir, ignore_errors=True) - async def responses( + async def _create_response( self, - request: Request, - body: NeMoGymResponseCreateParamsNonStreaming = Body(), + body: NeMoGymResponseCreateParamsNonStreaming, + observation_collector: Optional[Callable[[str], None]] = None, ) -> NeMoGymResponse: body = body.model_copy(deep=True) if isinstance(body.input, str): @@ -404,7 +419,11 @@ async def responses( system_prompt = "\n\n".join(system_parts) if system_parts else None try: - output_items, usage, model_name = await self._run_openclaw(user_message, system_prompt) + output_items, usage, model_name = await self._run_openclaw( + user_message, + system_prompt, + observation_collector=observation_collector, + ) except TimeoutError: LOG.warning("OpenClaw timed out, padding empty output so the rollout scores instead of erroring") output_items, usage, model_name = [], {"input_tokens": 0, "output_tokens": 0}, self.config.model @@ -442,6 +461,39 @@ async def responses( ), ) + async def responses( + self, + request: Request, + body: NeMoGymResponseCreateParamsNonStreaming = Body(), + ) -> NeMoGymResponse: + return await self._create_response(body) + + async def responses_with_observations( + self, + request: Optional[Request], + body: NeMoGymResponseCreateParamsNonStreaming, + ) -> AgentEpisode: + session_id: Optional[str] = None + + def collect(value: str) -> None: + nonlocal session_id + session_id = value + + response = await self._create_response(body, observation_collector=collect) + try: + observations = build_openclaw_observations( + session_id or response.id, + response.output, + transcript_available=session_id is not None, + ) + except Exception: + LOG.exception("failed to build OpenClaw observations") + observations = AgentObservationBundle( + source="openclaw", + gaps=[ObservationGap(code="observation_capture_failed", source="openclaw")], + ) + return AgentEpisode(response=response, observations=observations) + async def run(self, request: Request, body: OpenClawAgentRunRequest) -> OpenClawAgentVerifyResponse: async with self.sem: cookies = request.cookies @@ -455,20 +507,29 @@ async def run(self, request: Request, body: OpenClawAgentRunRequest) -> OpenClaw await raise_for_status(seed_resp) cookies = seed_resp.cookies - agent_resp = await self.server_client.post( - server_name=self.config.name, - url_path="/v1/responses", - json=body.responses_create_params, - cookies=cookies, - ) - await raise_for_status(agent_resp) - cookies = agent_resp.cookies - agent_resp_json = await get_response_json(agent_resp) + rollout_id = self.rollout_id_from_run(body) + if rollout_id is not None: + episode = await self.responses_with_observations(request, body.responses_create_params) + agent_resp, observations = episode.response, episode.observations + agent_resp_json = agent_resp.model_dump(mode="json") + else: + agent_resp = await self.server_client.post( + server_name=self.config.name, + url_path="/v1/responses", + json=body.responses_create_params, + cookies=cookies, + ) + await raise_for_status(agent_resp) + cookies = agent_resp.cookies + agent_resp_json = await get_response_json(agent_resp) + observations = None verify_resp = await self.server_client.post( server_name=self.config.resources_server.name, url_path="/verify", - json=body.model_dump() | {"response": agent_resp_json}, + json=body.model_dump() + | {"response": agent_resp_json} + | ({"rollout_id": rollout_id} if rollout_id is not None else {}), cookies=cookies, ) await raise_for_status(verify_resp) @@ -483,9 +544,10 @@ async def run(self, request: Request, body: OpenClawAgentRunRequest) -> OpenClaw last = gym_resp.output[-1] if gym_resp.output else None naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant" - return OpenClawAgentVerifyResponse.model_validate( - verify_json | {"turns_used": turns, "finished_naturally": naturally} - ) + result = verify_json | {"turns_used": turns, "finished_naturally": naturally} + if observations is not None: + result["ng_agent_observations"] = observations.model_dump(mode="json") + return OpenClawAgentVerifyResponse.model_validate(result) if __name__ == "__main__": diff --git a/responses_api_agents/openclaw_agent/observability.py b/responses_api_agents/openclaw_agent/observability.py new file mode 100644 index 0000000000..40fb01c8c7 --- /dev/null +++ b/responses_api_agents/openclaw_agent/observability.py @@ -0,0 +1,70 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Observability exposed by Gym's existing OpenClaw response parser.""" + +from collections.abc import Iterable +from typing import Any + +from nemo_gym.openai_utils import NeMoGymResponseInputItem +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ObservationGap, + ToolCallObservation, +) + + +def _field(item: Any, name: str) -> Any: + return item.get(name) if isinstance(item, dict) else getattr(item, name, None) + + +def build_openclaw_observations( + invocation_id: str, + conversation: Iterable[NeMoGymResponseInputItem], + *, + transcript_available: bool, + source: str = "openclaw", +) -> AgentObservationBundle: + """Describe signals present in Gym's normalized OpenClaw response.""" + items = list(conversation) + result_ids = { + call_id + for item in items + if _field(item, "type") == "function_call_output" and isinstance((call_id := _field(item, "call_id")), str) + } + + def tool_status(call_id: str) -> str: + # OpenClaw's normalized output marks result presence, not execution + # success. Do not turn that transport status into a success claim. + return "unknown" if call_id in result_ids else "incomplete" + + tools = [ + ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=call_id, + tool_name=_field(item, "name"), + status=tool_status(call_id), + ) + for item in items + if _field(item, "type") == "function_call" + and isinstance((call_id := _field(item, "call_id")), str) + and call_id + ] + + gaps = [ + ObservationGap(code="subagent_hierarchy_unavailable", source=source), + ObservationGap(code="model_call_ownership_unavailable", source=source), + ObservationGap(code="context_compaction_unavailable", source=source), + ] + if not transcript_available: + gaps.append(ObservationGap(code="agent_transcript_unavailable", source=source)) + if tools: + gaps.append(ObservationGap(code="tool_timing_unavailable", source=source, invocation_id=invocation_id)) + + return AgentObservationBundle( + source=source, + invocations=[AgentInvocation(invocation_id=invocation_id, conversation=items)], + tool_calls=tools, + gaps=gaps, + ) diff --git a/responses_api_agents/openclaw_agent/tests/test_app.py b/responses_api_agents/openclaw_agent/tests/test_app.py index b0322c5728..396682ba96 100644 --- a/responses_api_agents/openclaw_agent/tests/test_app.py +++ b/responses_api_agents/openclaw_agent/tests/test_app.py @@ -16,7 +16,7 @@ import asyncio import json from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import yaml @@ -31,6 +31,7 @@ from responses_api_agents.openclaw_agent.app import ( OpenClawAgent, OpenClawAgentConfig, + OpenClawAgentRunRequest, ResourcesServerRef, _decode_last_json_dict_suffix, _extract_instruction, @@ -40,6 +41,17 @@ ) +class _FakeResponse: + ok = True + + def __init__(self, payload: dict, cookies: dict | None = None) -> None: + self.payload = payload + self.cookies = cookies or {} + + async def read(self) -> bytes: + return json.dumps(self.payload).encode() + + def _config(**kwargs) -> OpenClawAgentConfig: kwargs.setdefault("openclaw_version", "2026.6.11") return OpenClawAgentConfig( @@ -260,6 +272,84 @@ def test_env_passthrough(self) -> None: assert "EMPTY" not in env +class TestObservability: + def test_run_returns_observations_when_enabled(self) -> None: + agent = _make_agent() + agent.server_client.global_config_dict = {"observability_enabled": True} + session = "\n".join( + [ + json.dumps({"type": "session", "id": "session-1"}), + json.dumps( + { + "type": "message", + "message": {"role": "assistant", "content": [{"type": "text", "text": "done"}]}, + } + ), + ] + ) + + async def run_openclaw(*args, observation_collector=None, **kwargs): + observation_collector("session-1") + return parse_openclaw_session(session), {"input_tokens": 1, "output_tokens": 1}, "model" + + async def post(server_name, url_path, json=None, cookies=None, **kwargs): + if url_path == "/seed_session": + return _FakeResponse({}, {"session": "1"}) + return _FakeResponse(json | {"reward": 1.0}) + + agent.server_client.post = AsyncMock(side_effect=post) + request = MagicMock() + request.cookies = {} + body = OpenClawAgentRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + with patch.object(agent, "_run_openclaw", run_openclaw): + result = asyncio.run(agent.run(request, body)) + + observations = result.ng_agent_observations + assert observations is not None + assert observations.invocations[0].invocation_id == "session-1" + assert observations.invocations[0].conversation + assert agent.server_client.post.await_args_list[-1].kwargs["json"]["rollout_id"] == "1-2" + + def test_observation_failure_does_not_change_response(self) -> None: + agent = _make_agent() + + async def run_openclaw(*args, observation_collector=None, **kwargs): + if observation_collector is not None: + observation_collector("session-1") + return ( + [ + NeMoGymResponseOutputMessage( + id="msg-1", + content=[{"type": "output_text", "text": "done", "annotations": []}], + ) + ], + {"input_tokens": 1, "output_tokens": 1}, + "model", + ) + + body = NeMoGymResponseCreateParamsNonStreaming(input="solve") + with patch.object(agent, "_run_openclaw", run_openclaw): + baseline = asyncio.run(agent.responses(MagicMock(), body)) + with ( + patch.object(agent, "_run_openclaw", run_openclaw), + patch( + "responses_api_agents.openclaw_agent.app.build_openclaw_observations", + side_effect=RuntimeError("observer failed"), + ), + ): + episode = asyncio.run(agent.responses_with_observations(None, body)) + + assert episode.response.output == baseline.output + assert episode.response.usage == baseline.usage + assert [gap.code for gap in episode.observations.gaps] == ["observation_capture_failed"] + + class TestDeepMerge: def test_nested_merge(self) -> None: base = {"a": {"b": 1, "c": 2}} diff --git a/responses_api_agents/openclaw_agent/tests/test_observability.py b/responses_api_agents/openclaw_agent/tests/test_observability.py new file mode 100644 index 0000000000..e74556da9f --- /dev/null +++ b/responses_api_agents/openclaw_agent/tests/test_observability.py @@ -0,0 +1,68 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from nemo_gym.openai_utils import NeMoGymFunctionCallOutput, NeMoGymResponseFunctionToolCall +from responses_api_agents.openclaw_agent.observability import build_openclaw_observations + + +def test_builds_root_observation_and_pairs_tools() -> None: + bundle = build_openclaw_observations( + "session-1", + [ + NeMoGymResponseFunctionToolCall( + arguments='{"query":"AAPL"}', + call_id="call-1", + name="web_search", + id="call-1", + status="completed", + ), + NeMoGymFunctionCallOutput(call_id="call-1", output="result", status="completed"), + ], + transcript_available=True, + ) + + assert bundle.invocations[0].invocation_id == "session-1" + assert bundle.invocations[0].conversation + assert bundle.tool_calls[0].tool_call_id == "call-1" + assert bundle.tool_calls[0].status == "unknown" + assert {gap.code for gap in bundle.gaps} == { + "subagent_hierarchy_unavailable", + "model_call_ownership_unavailable", + "context_compaction_unavailable", + "tool_timing_unavailable", + } + + +def test_marks_missing_transcript_and_incomplete_tool() -> None: + bundle = build_openclaw_observations( + "fallback", + [ + NeMoGymResponseFunctionToolCall( + arguments="{}", + call_id="call-1", + name="tool", + id="call-1", + status="completed", + ) + ], + transcript_available=False, + ) + + assert bundle.tool_calls[0].status == "incomplete" + assert "agent_transcript_unavailable" in {gap.code for gap in bundle.gaps} + + +def test_result_presence_does_not_claim_success_and_preserves_source() -> None: + bundle = build_openclaw_observations( + "run-1", + [ + NeMoGymResponseFunctionToolCall(arguments="{}", call_id="call-1", name="tool"), + NeMoGymFunctionCallOutput(call_id="call-1", output="partial", status="incomplete"), + ], + transcript_available=True, + source="pinchbench", + ) + + assert bundle.source == "pinchbench" + assert bundle.tool_calls[0].status == "unknown" + assert {gap.source for gap in bundle.gaps} == {"pinchbench"} diff --git a/responses_api_agents/pinchbench/app.py b/responses_api_agents/pinchbench/app.py index 11f3ce6237..e3ebdde305 100644 --- a/responses_api_agents/pinchbench/app.py +++ b/responses_api_agents/pinchbench/app.py @@ -38,6 +38,7 @@ import asyncio import glob import json +import logging import shutil import tarfile import textwrap @@ -46,7 +47,7 @@ from typing import Any, Literal, Optional from fastapi import Request, Response -from pydantic import ConfigDict +from pydantic import ConfigDict, Field from nemo_gym.base_resources_server import BaseRunRequest, BaseVerifyResponse from nemo_gym.base_responses_api_agent import ( @@ -67,7 +68,12 @@ NeMoGymResponseUsage, NeMoGymSummary, ) +from nemo_gym.rollout_observability import AgentObservationBundle, ObservationGap from nemo_gym.sandbox import AsyncSandbox, SandboxResources, SandboxSpec +from responses_api_agents.openclaw_agent.observability import build_openclaw_observations + + +LOG = logging.getLogger(__name__) class PinchBenchAgentConfig(BaseResponsesAPIAgentConfig): @@ -150,6 +156,10 @@ class PinchBenchVerifyResponse(BaseVerifyResponse): grading_notes: str status: str raw_rollout: dict # transcript archive location + compact metadata + ng_agent_observations: AgentObservationBundle | None = Field( + default=None, + exclude_if=lambda value: value is None, + ) class PinchBenchAgent(SimpleResponsesAPIAgent): @@ -662,6 +672,8 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None response = self._empty_response(task_id) transcript_events: list = [] archive_path = "" + observations: Optional[AgentObservationBundle] = None + observe = self.rollout_id_from_run(body) is not None try: async with self._sem: await self._run_in_sandbox(task_id, out_dir) # one sandbox per task @@ -684,6 +696,22 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None elif failure_class == "timeout_exceeded": routing[NG_TERMINAL_KEY] = True finally: + if observe: + try: + observations = build_openclaw_observations( + run_id, + response.output, + transcript_available=any( + isinstance(event, dict) and event.get("type") == "message" for event in transcript_events + ), + source="pinchbench", + ) + except Exception: + LOG.exception("failed to build PinchBench observations") + observations = AgentObservationBundle( + source="pinchbench", + gaps=[ObservationGap(code="observation_capture_failed", source="pinchbench")], + ) shutil.rmtree(out_dir, ignore_errors=True) return PinchBenchVerifyResponse( @@ -700,6 +728,7 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None "archived_to": archive_path, "run_id": run_id, }, + **({"ng_agent_observations": observations.model_dump(mode="json")} if observations is not None else {}), **routing, ) diff --git a/responses_api_agents/pinchbench/tests/test_app.py b/responses_api_agents/pinchbench/tests/test_app.py index 8c62967c0b..4346f76354 100644 --- a/responses_api_agents/pinchbench/tests/test_app.py +++ b/responses_api_agents/pinchbench/tests/test_app.py @@ -30,6 +30,7 @@ NG_TERMINAL_KEY, PinchBenchAgent, PinchBenchAgentConfig, + PinchBenchRunRequest, SandboxKilledError, _classify_task_failure, ) @@ -334,10 +335,110 @@ async def ok(task_id, out_dir): resp = await agent.run(body=_run_body()) dumped = resp.model_dump() assert dumped["reward"] == 1.0 + assert "ng_agent_observations" not in dumped for key in (NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY): assert key not in dumped +@pytest.mark.asyncio +async def test_run_returns_openclaw_observations_when_enabled(tmp_path, monkeypatch): + agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "archive")) + agent.server_client.global_config_dict = {"observability_enabled": True} + + async def run_in_sandbox(task_id, out_dir): + transcript_dir = out_dir / "0001_transcripts" + transcript_dir.mkdir(parents=True) + (transcript_dir / f"{task_id}.jsonl").write_text( + "\n".join( + [ + json.dumps({"type": "session", "id": "session-1"}), + json.dumps( + { + "type": "message", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "done"}], + }, + } + ), + ] + ) + ) + + monkeypatch.setattr(agent, "_run_in_sandbox", run_in_sandbox) + monkeypatch.setattr( + agent, + "_parse_result", + lambda *args: { + "reward": 1.0, + "grading_type": "automated", + "breakdown": {}, + "notes": "ok", + "status": "success", + }, + ) + body = PinchBenchRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "verifier_metadata": {"task_id": "task_x"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + result = await agent.run(body=body) + + observations = result.ng_agent_observations + assert observations is not None + assert observations.source == "pinchbench" + assert observations.invocations[0].invocation_id == result.raw_rollout["run_id"] + assert observations.invocations[0].conversation + + +@pytest.mark.asyncio +async def test_observation_failure_does_not_change_result(tmp_path, monkeypatch): + agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "archive")) + agent.server_client.global_config_dict = {"observability_enabled": True} + + async def run_in_sandbox(task_id, out_dir): + return None + + monkeypatch.setattr(agent, "_run_in_sandbox", run_in_sandbox) + monkeypatch.setattr( + agent, + "_parse_result", + lambda *args: { + "reward": 1.0, + "grading_type": "automated", + "breakdown": {}, + "notes": "ok", + "status": "success", + }, + ) + monkeypatch.setattr( + "responses_api_agents.pinchbench.app.build_openclaw_observations", + MagicMock(side_effect=RuntimeError("observer failed")), + ) + body = PinchBenchRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "verifier_metadata": {"task_id": "task_x"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + result = await agent.run(body=body) + + assert result.reward == 1.0 + assert result.status == "success" + assert result.response.output[0].content[0].text == "" + observations = result.ng_agent_observations + assert observations is not None + assert observations.source == "pinchbench" + assert [gap.code for gap in observations.gaps] == ["observation_capture_failed"] + + def test_classify_task_failure_mapping(): assert _classify_task_failure(SandboxKilledError("rc=-15")) == "kill_shaped" assert _classify_task_failure(TimeoutError("timed out")) == "timeout_exceeded" diff --git a/responses_api_agents/stirrup_agent/app.py b/responses_api_agents/stirrup_agent/app.py index 6d08dfaa86..8ca8aa632f 100644 --- a/responses_api_agents/stirrup_agent/app.py +++ b/responses_api_agents/stirrup_agent/app.py @@ -49,7 +49,7 @@ NeMoGymResponseOutputMessage, NeMoGymResponseOutputText, ) -from nemo_gym.server_utils import get_response_json, raise_for_status +from nemo_gym.server_utils import apply_rollout_prefix, get_response_json, raise_for_status from responses_api_agents.stirrup_agent.task_strategy import TaskSampleSkipError, TaskStrategy @@ -936,7 +936,9 @@ class StirrupRunRequest(BaseRunRequest): "rubric_json", "rubric_pretty", "instance_id", + "_ng_task_index", "_ng_rollout_index", + "_ng_attempt_index", ) @@ -1036,20 +1038,21 @@ def model_post_init(self, __context: Any) -> None: # -- helpers ---------------------------------------------------------- - def _get_model_base_url(self) -> str: + def _get_model_base_url(self, rollout_id: Optional[str] = None) -> str: from nemo_gym.global_config import get_first_server_config_dict from nemo_gym.server_utils import ServerClient global_config_dict = ServerClient.load_from_global_config().global_config_dict model_server_config = get_first_server_config_dict(global_config_dict, self.config.model_server.name) - return f"http://{model_server_config['host']}:{model_server_config['port']}/v1" + base_url = f"http://{model_server_config['host']}:{model_server_config['port']}" + return f"{apply_rollout_prefix(base_url, rollout_id)}/v1" # -- /v1/responses ---------------------------------------------------- async def responses(self, body: NeMoGymResponseCreateParamsNonStreaming = Body()) -> NeMoGymResponse: task_info = self.task_strategy.extract_task_info(body.metadata) - model_base_url = self._get_model_base_url() + model_base_url = self._get_model_base_url(self.rollout_id_from_run(body.metadata)) if self.config.task == "gdpval": system_prompt = None @@ -1180,10 +1183,16 @@ async def run(self, request: Request, body: StirrupRunRequest): for key in _TASK_METADATA_FIELDS: top_value = body_dict.get(key) meta_value = existing_metadata.get(key) + if key in ("_ng_task_index", "_ng_attempt_index") and not self._model_call_capture_enabled(): + continue + if key.startswith("_ng_") and top_value is not None: + existing_metadata[key] = str(top_value) + continue if top_value is not None and meta_value is None: existing_metadata[key] = top_value elif meta_value is not None and top_value is None: body_dict[key] = meta_value + rollout_id = self.rollout_id_from_run(body_dict) update: Dict[str, Any] = {"metadata": existing_metadata} if fixed_params.tool_choice is None: update["tool_choice"] = "auto" @@ -1388,6 +1397,8 @@ async def run(self, request: Request, body: StirrupRunRequest): verify_request_body = dict(body_dict) verify_request_body["response"] = response_clean.model_dump(mode="json") + if rollout_id is not None: + verify_request_body["rollout_id"] = rollout_id if deliverables_dir is not None: verify_request_body["deliverables_dir"] = deliverables_dir # Surface the agent's runtime metadata for downstream logging. diff --git a/responses_api_agents/stirrup_agent/tests/test_app.py b/responses_api_agents/stirrup_agent/tests/test_app.py index ad8ff6913a..d66c239720 100644 --- a/responses_api_agents/stirrup_agent/tests/test_app.py +++ b/responses_api_agents/stirrup_agent/tests/test_app.py @@ -125,6 +125,51 @@ def test_sanity(self) -> None: ) StirrupAgentWrapper(config=config, server_client=MagicMock(spec=ServerClient)) + def test_model_base_url_accepts_rollout_correlation(self) -> None: + wrapper = StirrupAgentWrapper(config=_make_config(), server_client=MagicMock(spec=ServerClient)) + loaded = MagicMock(global_config_dict={"policy_model": {}}) + + with ( + patch("nemo_gym.server_utils.ServerClient.load_from_global_config", return_value=loaded), + patch( + "nemo_gym.global_config.get_first_server_config_dict", + return_value={"host": "model-host", "port": 8000}, + ), + ): + assert wrapper._get_model_base_url("7-3") == "http://model-host:8000/ng-rollout/7-3/v1" + assert wrapper._get_model_base_url() == "http://model-host:8000/v1" + + @pytest.mark.asyncio + async def test_run_correlates_policy_and_judge_calls(self) -> None: + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = {"observability_enabled": True} + server_client.post = AsyncMock(return_value=MagicMock()) + wrapper = StirrupAgentWrapper(config=_make_config(), server_client=server_client) + body = StirrupRunRequest( + responses_create_params=NeMoGymResponseCreateParamsNonStreaming( + input="ignored", + metadata={"task_id": "task-1", "prompt": "do the thing", "_ng_rollout_index": "99"}, + ), + task_id="task-1", + prompt="do the thing", + _ng_task_index=7, + _ng_rollout_index=3, + ) + request = MagicMock(cookies={}) + responses_mock = AsyncMock(return_value=_fake_response()) + + with ( + patch.object(StirrupAgentWrapper, "responses", responses_mock), + patch("responses_api_agents.stirrup_agent.app.raise_for_status", AsyncMock()), + patch("responses_api_agents.stirrup_agent.app.get_response_json", AsyncMock(return_value={"reward": 1.0})), + ): + await wrapper.run(request, body) + + policy_params = responses_mock.await_args.args[0] + assert wrapper.rollout_id_from_run(policy_params.metadata) == "7-3" + verify_calls = [call for call in server_client.post.await_args_list if call.kwargs["url_path"] == "/verify"] + assert verify_calls[0].kwargs["json"]["rollout_id"] == "7-3" + def test_output_history_preserves_nemo_user_tool_results(self) -> None: """Run-history export should keep NeMo user-role tool results as tool outputs.""" history = [ @@ -229,6 +274,7 @@ async def test_run_judge_only_scores_cached_deliverables(self, tmp_path) -> None config = _make_config(judge_only=True, persist_deliverables_dir=str(tmp_path)) server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = {"observability_enabled": True} server_client.post = AsyncMock(return_value=MagicMock()) wrapper = StirrupAgentWrapper(config=config, server_client=server_client) @@ -236,7 +282,13 @@ async def test_run_judge_only_scores_cached_deliverables(self, tmp_path) -> None input="ignored", metadata={"task_id": "task-1", "prompt": "do the thing", "_ng_rollout_index": "0"}, ) - body = StirrupRunRequest(responses_create_params=params, task_id="task-1", prompt="do the thing") + body = StirrupRunRequest( + responses_create_params=params, + task_id="task-1", + prompt="do the thing", + _ng_task_index=7, + _ng_rollout_index=0, + ) request = MagicMock() request.cookies = {} @@ -258,6 +310,7 @@ async def test_run_judge_only_scores_cached_deliverables(self, tmp_path) -> None assert len(verify_calls) == 1 verify_json = verify_calls[0].kwargs["json"] assert verify_json["deliverables_dir"].endswith(str(Path("task_task-1") / "repeat_0")) + assert verify_json["rollout_id"] == "7-0" assert result == {"reward": 0.9, "judge_response": "ok"} @pytest.mark.asyncio diff --git a/responses_api_agents/swe_agents/app.py b/responses_api_agents/swe_agents/app.py index f492ba156a..1a2bf1b786 100644 --- a/responses_api_agents/swe_agents/app.py +++ b/responses_api_agents/swe_agents/app.py @@ -62,7 +62,7 @@ NeMoGymResponseCreateParamsNonStreaming, ) from nemo_gym.profiling import Profiler -from nemo_gym.server_utils import get_first_server_config_dict +from nemo_gym.server_utils import apply_rollout_prefix, get_first_server_config_dict from responses_api_models.vllm_model.app import VLLMConverter, split_responses_input_output_items @@ -212,6 +212,7 @@ class ExecuteContainerCommandArgs(BaseModel): class SWEBenchWrapperInstanceConfig(SWEBenchWrapperServerConfig, SWEBenchWrapperConfig): + rollout_id: Optional[str] = Field(default=None, exclude_if=lambda value: value is None) metrics_fpath: Path problem_info: Dict[str, Any] body: NeMoGymResponseCreateParamsNonStreaming @@ -301,6 +302,10 @@ class SWEBenchVerifyResponse(SWEBenchMetrics, BaseVerifyResponse): subagent_trajectories: Optional[List[Dict[str, Any]]] = None +class SWEBenchRunRequest(BaseRunRequest): + model_config = ConfigDict(extra="allow") + + ######################################## # START Dataset and harness handling ######################################## @@ -1953,7 +1958,9 @@ def get_run_command(self) -> ExecuteContainerCommandArgs: # openai_model.yaml uses `openai_model`; vllm_model.yaml uses `model`. try: model_server_cfg = get_first_server_config_dict(get_global_config_dict(), self.config.model_server_name) - model_server_base_url = f"http://{model_server_cfg.host}:{model_server_cfg.port}" + model_server_base_url = apply_rollout_prefix( + f"http://{model_server_cfg.host}:{model_server_cfg.port}", self.config.rollout_id + ) default_model_name = ( getattr(model_server_cfg, "openai_model", None) or getattr(model_server_cfg, "model", None) or "" ) @@ -3374,7 +3381,7 @@ def _item_type(item) -> Optional[str]: return json.dumps(chat_messages) def _setup_params( - self, body: NeMoGymResponseCreateParamsNonStreaming + self, body: NeMoGymResponseCreateParamsNonStreaming, rollout_id: Optional[str] = None ) -> Tuple[SWEBenchWrapperInstanceConfig, BaseDatasetHarnessProcessor]: problem_info = body.metadata | {"container_formatter": self.config.container_formatter} instance_id = problem_info.get("instance_id", "unknown") @@ -3448,6 +3455,7 @@ def _setup_params( params: SWEBenchWrapperInstanceConfig = SWEBenchWrapperInstanceConfig( **self.config.model_dump(), **self._swe_bench_wrapper_server_config.model_dump(), + rollout_id=rollout_id, problem_info=problem_info, body=body, persistent_dir=persistent_dir, @@ -3522,7 +3530,12 @@ def _setup_params( return params, dataset_processor async def responses(self, body: NeMoGymResponseCreateParamsNonStreaming = Body()) -> NeMoGymResponse: - params, dataset_processor = self._setup_params(body) + return await self._responses(body) + + async def _responses( + self, body: NeMoGymResponseCreateParamsNonStreaming, rollout_id: Optional[str] = None + ) -> NeMoGymResponse: + params, dataset_processor = self._setup_params(body, rollout_id) with (params.eval_private_dir / "params.json").open("w") as f: f.write(params.model_dump_json(indent=4)) @@ -3668,12 +3681,12 @@ def _item_field(item, name: str): metadata=metadata, ) - async def run(self, body: BaseRunRequest) -> SWEBenchVerifyResponse: + async def run(self, body: SWEBenchRunRequest) -> SWEBenchVerifyResponse: async with self._sem: body.responses_create_params.parallel_tool_calls = True body.responses_create_params.tool_choice = "auto" - response = await self.responses(body.responses_create_params) + response = await self._responses(body.responses_create_params, self.rollout_id_from_run(body)) metadata, response.metadata = response.metadata, None responses_create_params = body.responses_create_params.model_dump() | { diff --git a/responses_api_agents/swe_agents/tests/test_app.py b/responses_api_agents/swe_agents/tests/test_app.py index 6f1402f24f..02712208ea 100644 --- a/responses_api_agents/swe_agents/tests/test_app.py +++ b/responses_api_agents/swe_agents/tests/test_app.py @@ -43,6 +43,7 @@ SweBenchDatasetProcessor, SWEBenchMetrics, SweBenchMultilingualDatasetProcessor, + SWEBenchRunRequest, SWEBenchVerifyResponse, SWEBenchWrapper, SWEBenchWrapperConfig, @@ -340,6 +341,7 @@ def test_resolved_defaults(self) -> None: assert config.resolved_agent_cls == "CodeActAgent" assert config.resolved_diversify_tool_names is False assert config.resolved_camel_case_tool_names is False + assert "rollout_id" not in config.model_dump() class TestSWEBenchMetrics: @@ -1076,6 +1078,14 @@ def test_get_run_command_writes_script(self, _stub_model_server_lookup) -> None: assert "--max-turns" not in script # max_turns is positional assert str(config.agent_max_turns) in script + def test_get_run_command_prefixes_observed_rollout(self, _stub_model_server_lookup) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + config = self._opencode_config(tmpdir, rollout_id="7-2") + OpenCodeHarnessProcessor(config=config).get_run_command() + + script = self._read_agent_script(config) + assert "NEMO_GYM_MODEL_SERVER_BASE_URL=http://test-host:12345/ng-rollout/7-2" in script + def test_get_run_command_subagents_disabled_by_default(self, _stub_model_server_lookup) -> None: with tempfile.TemporaryDirectory() as tmpdir: config = self._opencode_config(tmpdir) @@ -2379,6 +2389,7 @@ class TestSWEBenchWrapperRun: @pytest.mark.asyncio async def test_run_resolved(self, monkeypatch) -> None: wrapper = _create_wrapper(monkeypatch) + wrapper.server_client.global_config_dict = {"observability_enabled": True} mock_response = NeMoGymResponse( id="swebench-test", @@ -2396,10 +2407,10 @@ async def test_run_resolved(self, monkeypatch) -> None: }, ) - with patch.object(SWEBenchWrapper, "responses", new_callable=AsyncMock, return_value=mock_response): - from nemo_gym.base_resources_server import BaseRunRequest - - body = BaseRunRequest( + with patch.object( + SWEBenchWrapper, "_responses", new_callable=AsyncMock, return_value=mock_response + ) as responses_mock: + body = SWEBenchRunRequest( responses_create_params=NeMoGymResponseCreateParamsNonStreaming( model="test-model", input=[], @@ -2411,12 +2422,15 @@ async def test_run_resolved(self, monkeypatch) -> None: "split": "test", "instance_dict": "{}", }, - ) + ), + _ng_task_index=7, + _ng_rollout_index=2, ) result = await wrapper.run(body) assert isinstance(result, SWEBenchVerifyResponse) assert result.reward == 1.0 + assert responses_mock.await_args.args[1] == "7-2" @pytest.mark.asyncio async def test_run_not_resolved(self, monkeypatch) -> None: @@ -2438,10 +2452,8 @@ async def test_run_not_resolved(self, monkeypatch) -> None: }, ) - with patch.object(SWEBenchWrapper, "responses", new_callable=AsyncMock, return_value=mock_response): - from nemo_gym.base_resources_server import BaseRunRequest - - body = BaseRunRequest( + with patch.object(SWEBenchWrapper, "_responses", new_callable=AsyncMock, return_value=mock_response): + body = SWEBenchRunRequest( responses_create_params=NeMoGymResponseCreateParamsNonStreaming( model="test-model", input=[], diff --git a/responses_api_agents/tau2/tests/test_app.py b/responses_api_agents/tau2/tests/test_app.py index 53dfd9afb0..be12e96b0f 100644 --- a/responses_api_agents/tau2/tests/test_app.py +++ b/responses_api_agents/tau2/tests/test_app.py @@ -17,7 +17,9 @@ from typing import Tuple from unittest.mock import AsyncMock, MagicMock, patch +import pytest from fastapi.testclient import TestClient +from tau2.data_model.simulation import RewardInfo, SimulationRun, TerminationReason from nemo_gym.base_responses_api_agent import AggregateMetricsRequest from nemo_gym.server_utils import ServerClient @@ -114,6 +116,46 @@ def _clean(d): assert _clean(expected_response_dict) == _clean(actual_response_dict) + @pytest.mark.parametrize( + ("observability_enabled", "url_suffix"), + [(True, "/ng-rollout/7-2/v1"), (False, "/v1")], + ) + def test_policy_and_user_model_calls_share_rollout_correlation( + self, observability_enabled: bool, url_suffix: str + ) -> None: + example_jsonl = Path(__file__).parent.parent / "data" / "example.jsonl" + request_body = json.loads(example_jsonl.read_text().splitlines()[0]) + request_body |= {"_ng_task_index": 7, "_ng_rollout_index": 2} + + config, server = self._dummy_server() + config.model_server.name = "policy" + config.user_model_server.name = "user" + server.server_client.global_config_dict = {"observability_enabled": observability_enabled} + with patch("responses_api_agents.tau2.app.ensure_tau2_data_dir"): + client = TestClient(server.setup_webserver()) + + result = SimulationRun( + id="run-1", + task_id="task-1", + start_time="2026-07-22T00:00:00Z", + end_time="2026-07-22T00:00:00Z", + duration=0, + termination_reason=TerminationReason.AGENT_STOP, + reward_info=RewardInfo(reward=1), + messages=[], + ) + model_urls = {"policy": "http://policy:8000", "user": "http://user:8001"} + with ( + patch("responses_api_agents.tau2.app.get_server_url", side_effect=model_urls.__getitem__), + patch("responses_api_agents.tau2.app.run_single_task", AsyncMock(return_value=result)), + ): + response = client.post("/run", json=request_body) + + assert response.status_code == 200 + response_config = response.json()["config"] + assert response_config["llm_args_agent"]["api_base"] == model_urls["policy"] + url_suffix + assert response_config["llm_args_user"]["api_base"] == model_urls["user"] + url_suffix + async def test_compute_metrics(self) -> None: example_rollouts_fpath = Path(__file__).parent.parent / "data" / "example_rollouts.jsonl" with example_rollouts_fpath.open() as f: diff --git a/tests/unit_tests/test_base_responses_api_model.py b/tests/unit_tests/test_base_responses_api_model.py index c1fd00304d..94c0c003ae 100644 --- a/tests/unit_tests/test_base_responses_api_model.py +++ b/tests/unit_tests/test_base_responses_api_model.py @@ -135,6 +135,7 @@ def test_build_model_call_record_from_exchange(): "latency_ms": 18.4, "request": {"input": "hi"}, "response": { + "id": "resp-1", "model": "m", "usage": { "input_tokens": 10, @@ -152,6 +153,7 @@ def test_build_model_call_record_from_exchange(): } rec = build_model_call_record(exchange, call_index=3) assert rec.model_call_id == "call-1" + assert rec.response_id == "resp-1" assert rec.call_index == 3 assert rec.model_ref is not None and rec.model_ref.name == "srv" assert rec.dialect == "responses" @@ -161,8 +163,10 @@ def test_build_model_call_record_from_exchange(): assert rec.reasoning_content == "thinking..." assert rec.tool_calls == [{"call_id": "c1", "name": "calc", "arguments": {"x": 1}}] assert rec.latency_total_ms == 18.4 + assert build_model_call_record({"response": {"id": 123}}, call_index=0).response_id is None assert { "model_call_id", + "response_id", "call_index", "model_ref", "dialect", @@ -840,6 +844,7 @@ async def gen(): assert len(calls) == 1 call = calls[0] assert call.model_call_id + assert call.response_id == "msg_1" assert call.model_ref is not None and call.model_ref.name == "srv" assert call.started_at is not None and call.completed_at is not None assert call.started_at <= call.completed_at @@ -856,7 +861,11 @@ def test_reconstruct_chat_sse(): from nemo_gym.base_responses_api_model import _reconstruct_streamed_response chunks = [ - {"model": "m", "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hel"}}]}, + { + "id": "chatcmpl-1", + "model": "m", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hel"}}], + }, {"choices": [{"index": 0, "delta": {"content": "lo", "reasoning": "hmm"}}]}, # vLLM `reasoning` alias { "choices": [ @@ -882,6 +891,7 @@ def test_reconstruct_chat_sse(): raw = (b"".join(_sse("", c) for c in chunks) + b"data: [DONE]\n\n").replace(b"\n", b"\r\n") resp = _reconstruct_streamed_response(raw, "chat") msg = resp["choices"][0]["message"] + assert resp["id"] == "chatcmpl-1" assert msg["content"] == "Hello" and msg["reasoning_content"] == "hmm" assert msg["tool_calls"][0]["function"] == {"name": "f", "arguments": '{"a":1}'} assert resp["usage"]["total_tokens"] == 8 @@ -963,7 +973,12 @@ def test_merge_capture_attaches_metrics_without_raw_payloads(tmp_path): "responses", "A", {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}, - {"output": [{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "ok"}]}]}, + { + "id": "resp-A", + "output": [ + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "ok"}]} + ], + }, ), ) @@ -976,6 +991,7 @@ def test_merge_capture_attaches_metrics_without_raw_payloads(tmp_path): assert capture["metrics"]["num_calls"] == 1 attached_call = capture["calls"][0] assert attached_call["model_call_id"] == "call-A" + assert attached_call["response_id"] == "resp-A" assert attached_call["model_ref"] == {"type": "responses_api_models", "name": "A"} assert attached_call["started_at"] == 100.0 and attached_call["completed_at"] == 100.01 assert attached_call["tokens_in"] == 3 diff --git a/tests/unit_tests/test_rollout_collection.py b/tests/unit_tests/test_rollout_collection.py index 294bd5a4ae..647130b2ea 100644 --- a/tests/unit_tests/test_rollout_collection.py +++ b/tests/unit_tests/test_rollout_collection.py @@ -877,7 +877,13 @@ def setup_server_client(self): {AGENT_REF_KEY_NAME: {"name": "my_agent"}, TASK_INDEX_KEY_NAME: 1, ROLLOUT_INDEX_KEY_NAME: 1}, ] results = [ - {TASK_INDEX_KEY_NAME: 0, ROLLOUT_INDEX_KEY_NAME: 0, "reward": 1.0, "response": {"usage": {"tokens": 10}}}, + { + TASK_INDEX_KEY_NAME: 0, + ROLLOUT_INDEX_KEY_NAME: 0, + "reward": 1.0, + "response": {"usage": {"tokens": 10}}, + "ng_agent_observations": {"invocations": [{"conversation": ["large"]}]}, + }, {TASK_INDEX_KEY_NAME: 0, ROLLOUT_INDEX_KEY_NAME: 1, "reward": 0.0, "response": {"usage": {"tokens": 12}}}, {TASK_INDEX_KEY_NAME: 1, ROLLOUT_INDEX_KEY_NAME: 0, "reward": 1.0, "response": {"usage": {"tokens": 8}}}, {TASK_INDEX_KEY_NAME: 1, ROLLOUT_INDEX_KEY_NAME: 1, "reward": 0.0, "response": {"usage": {"tokens": 15}}}, @@ -906,6 +912,7 @@ def setup_server_client(self): ) for item in sent_data: assert "responses_create_params" not in item + assert "ng_agent_observations" not in item assert "usage" in item["response"] async def test_call_aggregate_metrics_multiple_agents(self, tmp_path: Path) -> None: diff --git a/tests/unit_tests/test_rollout_observability.py b/tests/unit_tests/test_rollout_observability.py new file mode 100644 index 0000000000..2dec357148 --- /dev/null +++ b/tests/unit_tests/test_rollout_observability.py @@ -0,0 +1,24 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest +from pydantic import ValidationError + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.rollout_observability import ModelCallRef + + +@pytest.mark.parametrize( + "value", + ({}, {"response_id": "resp-1"}, {"model_ref": {"name": "policy", "type": "responses_api_models"}}), +) +def test_model_call_ref_rejects_incomplete_join_keys(value: dict) -> None: + with pytest.raises(ValidationError, match="model_call_id or both model_ref and response_id"): + ModelCallRef.model_validate(value) + + +def test_model_call_ref_accepts_supported_join_keys() -> None: + model_ref = ModelServerRef(name="policy", type="responses_api_models") + + assert ModelCallRef(model_call_id="call-1").model_call_id == "call-1" + assert ModelCallRef(model_ref=model_ref, response_id="resp-1").model_ref == model_ref