diff --git a/.github/workflows/config/.secrets.baseline b/.github/workflows/config/.secrets.baseline index 4421169de1..4b1caf1e83 100644 --- a/.github/workflows/config/.secrets.baseline +++ b/.github/workflows/config/.secrets.baseline @@ -139,21 +139,21 @@ "filename": "responses_api_agents/pinchbench/tests/test_app.py", "hashed_secret": "7af5fc59b6cdbb45a6f1fde3f2f1900529c12bb0", "is_verified": false, - "line_number": 45 + "line_number": 49 }, { "type": "Secret Keyword", "filename": "responses_api_agents/pinchbench/tests/test_app.py", "hashed_secret": "a06371a099b2c32a96b7d9e690b713c165ce5fa8", "is_verified": false, - "line_number": 49 + "line_number": 53 }, { "type": "Secret Keyword", "filename": "responses_api_agents/pinchbench/tests/test_app.py", "hashed_secret": "15cd4504edaccefbda7199b58ae6dac9bd32cb30", "is_verified": false, - "line_number": 50 + "line_number": 54 } ] }, diff --git a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx index a3421b1523..22ba064a90 100644 --- a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx +++ b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx @@ -59,11 +59,10 @@ For C1, C2, and C4, `V` evaluates the correlated Gym Model Server path; direct-p | `mini_swe_agent` | X | X | X | X | X | X | X | | `mini_swe_agent_2` | V | V | X | V | X | X | X | | `non_executing_simple_agent` | V | V | X | V | X | X | X | -| `openclaw_agent` | X | X | X | X | X | X | X | +| `openclaw_agent` | V | V | X | O | V | V | V | | `opencode_agent` | X | X | X | X | X | X | X | | `osworld_agent` | O | O | X | O | X | X | X | | `pi_agent` | X | X | X | X | X | X | X | -| `pinchbench` | X | X | X | X | X | X | X | | `proof_refinement_agent` | V | V | X | V | X | X | X | | `remote_agent` | X | O | X | X | X | X | X | | `scicode_agent` | X | X | X | X | X | X | X | @@ -80,4 +79,5 @@ Simple and LabBench C3 support is partial because verifier responses may omit re CVDP Simple path, Finance and Remote aggregate usage, LabBench image redaction, OSWorld M3 and Pointer direct-provider routes, omitted raised failures in Simple-derived agents, and Stirrup calls outside its policy path. The matrix reports producer output, not schema capacity. +OpenClaw coverage includes its standalone resource-server path and the PinchBench sandbox benchmark path. C7 requires standardized agent-side trajectory evidence; model HTTP capture alone does not satisfy it. diff --git a/responses_api_agents/openclaw_agent/app.py b/responses_api_agents/openclaw_agent/app.py index 4044a7bb40..7b0a6142bd 100644 --- a/responses_api_agents/openclaw_agent/app.py +++ b/responses_api_agents/openclaw_agent/app.py @@ -21,9 +21,10 @@ import shlex import shutil from asyncio import Semaphore +from collections.abc import Mapping 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 @@ -36,7 +37,6 @@ SimpleResponsesAPIAgent, ) from nemo_gym.config_types import ModelServerRef, ResourcesServerRef -from nemo_gym.global_config import get_first_server_config_dict from nemo_gym.openai_utils import ( NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, @@ -47,13 +47,28 @@ NeMoGymResponseOutputMessage, NeMoGymResponseOutputText, NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, NeMoGymResponseUsage, + NeMoGymSummary, +) +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 ( + OPENCLAW_OBSERVATION_SOURCE, + OpenClawSessionTree, + build_openclaw_observation_tree, + build_openclaw_observations, + discover_openclaw_session_tree, +) from responses_api_agents.openclaw_agent.setup_openclaw import ensure_openclaw LOG = logging.getLogger(__name__) +_INTERNAL_OBSERVATIONS_KEY = "_ng_agent_observations" def _decode_last_json_dict_suffix(raw: str) -> Optional[dict[str, Any]]: @@ -93,7 +108,7 @@ def _text_from_openclaw_payloads(envelope: dict[str, Any]) -> str: def parse_openclaw_output(stdout: str) -> tuple[list[Any], dict[str, int]]: envelope = _decode_last_json_dict_suffix(stdout) if not envelope: - return [], {"input_tokens": 0, "output_tokens": 0} + return [], {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} text = _text_from_openclaw_payloads(envelope) output_items: list[Any] = [] @@ -114,30 +129,85 @@ def parse_openclaw_output(stdout: str) -> tuple[list[Any], dict[str, int]]: cache_read = int(usage.get("cacheRead") or 0) input_tokens = int(usage.get("input") or 0) + cache_read output_tokens = int(usage.get("output") or 0) - return output_items, {"input_tokens": input_tokens, "output_tokens": output_tokens} + return output_items, { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "cached_tokens": cache_read, + } -def parse_openclaw_session(session_text: str) -> list[Any]: - """Convert an OpenClaw session .jsonl into Gym output items, including tool calls""" +def parse_openclaw_session_items(events: list[dict[str, Any]], *, include_input: bool = False) -> list[Any]: + """Convert OpenClaw session events into Gym conversation items.""" output_items: list[Any] = [] - for line in session_text.splitlines(): - line = line.strip() - if not line: - continue - try: - event = json.loads(line) - except json.JSONDecodeError: - continue + for event in events: + event_id = event.get("id") if event.get("type") != "message": continue message = event.get("message") or {} + if not isinstance(message, dict): + continue role = message.get("role") content = message.get("content") - if not isinstance(content, list): + + if include_input and role in {"user", "system", "developer"}: + text = content if isinstance(content, str) else "" + if isinstance(content, list): + text = "\n".join( + block.get("text", "") + for block in content + if isinstance(block, dict) and isinstance(block.get("text"), str) and block["text"] + ) + if text: + output_items.append(NeMoGymEasyInputMessage(role=role, content=text)) + continue + if not include_input and not isinstance(content, list): continue if role == "assistant": - texts = [b["text"] for b in content if isinstance(b, dict) and (b.get("text") or "").strip()] + reasoning = [] + for key in ("reasoning_content", "reasoning_text", "thinking"): + value = message.get(key) + if isinstance(value, str) and value: + reasoning.append(value) + message_reasoning = message.get("reasoning") + if isinstance(message_reasoning, str) and message_reasoning: + reasoning.append(message_reasoning) + elif isinstance(message_reasoning, dict): + for key in ("content", "text", "summary"): + value = message_reasoning.get(key) + if isinstance(value, str) and value: + reasoning.append(value) + if isinstance(content, list): + reasoning.extend( + text + for block in content + if isinstance(block, dict) + and block.get("type") in {"thinking", "reasoning"} + and isinstance((text := block.get("thinking") or block.get("text") or block.get("reasoning")), str) + and text + ) + if include_input and reasoning: + output_items.append( + NeMoGymResponseReasoningItem( + id=f"rs_{event_id or len(output_items)}", + summary=[NeMoGymSummary(text="\n".join(reasoning), type="summary_text")], + ) + ) + + texts = [content] if include_input and isinstance(content, str) and content else [] + if isinstance(content, list): + texts = [ + block["text"] for block in content if isinstance(block, dict) and (block.get("text") or "").strip() + ] + if include_input: + texts = [ + block["text"] + for block in content + if isinstance(block, dict) + and block.get("type") not in {"thinking", "reasoning", "toolCall"} + and isinstance(block.get("text"), str) + and block["text"].strip() + ] if texts: output_items.append( NeMoGymResponseOutputMessage( @@ -148,10 +218,12 @@ def parse_openclaw_session(session_text: str) -> list[Any]: type="message", ) ) - for block in content: + for block in content if isinstance(content, list) else []: if not isinstance(block, dict) or block.get("type") != "toolCall": continue args = block.get("arguments") + if include_input and args is None: + args = block.get("partialArgs") arguments = json.dumps(args) if isinstance(args, (dict, list)) else str(args or "") call_id = block.get("id") or f"call-{uuid4().hex[:8]}" output_items.append( @@ -166,10 +238,16 @@ def parse_openclaw_session(session_text: str) -> list[Any]: ) elif role == "toolResult": - call_id = message.get("toolCallId", "") - result_text = "".join( - b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text" - ) + call_id = message.get("toolCallId") or (message.get("tool_call_id") if include_input else "") or "" + result_text = content if include_input and isinstance(content, str) else "" + if isinstance(content, list): + result_text = "".join( + block.get("text", "") + for block in content + if isinstance(block, dict) and block.get("type") == "text" + ) + if include_input and not result_text and message.get("details") is not None: + result_text = json.dumps(message["details"], ensure_ascii=False) output_items.append( NeMoGymFunctionCallOutput( type="function_call_output", @@ -182,6 +260,51 @@ def parse_openclaw_session(session_text: str) -> list[Any]: return output_items +def openclaw_session_conversation( + events: list[dict[str, Any]], + *, + input_items: list[Any] | None = None, + fallback_output: list[Any] | None = None, +) -> list[Any]: + """Prefer retained transcript items and fill only evidence missing from the artifact.""" + conversation = parse_openclaw_session_items(events, include_input=True) + inputs = input_items or [] + fallback = fallback_output or [] + if not conversation: + return [*inputs, *fallback] + retained_roles = { + role for item in conversation if (role := getattr(item, "role", None)) in {"user", "system", "developer"} + } + missing_inputs = ( + [item for item in inputs if getattr(item, "role", None) not in retained_roles] if retained_roles else inputs + ) + if missing_inputs: + conversation = [*missing_inputs, *conversation] + if fallback and not any( + getattr(item, "role", None) == "assistant" + or getattr(item, "type", None) in {"reasoning", "function_call", "function_call_output"} + for item in conversation + ): + conversation.extend(fallback) + return conversation + + +def parse_openclaw_session(session_text: str) -> list[Any]: + """Convert an OpenClaw session .jsonl into Gym output items, including tool calls.""" + return parse_openclaw_session_items(parse_openclaw_session_events(session_text)) + + +def parse_openclaw_session_events(session_text: str) -> list[dict[str, Any]]: + events = [] + for line in session_text.splitlines(): + try: + event = json.loads(line) + except (json.JSONDecodeError, RecursionError): + event = {"raw": line} + events.append(event if isinstance(event, dict) else {"raw": line}) + return events + + def _extract_instruction(body_input) -> tuple[str, Optional[str]]: """Return (user_message, system_message) from a responses body input list.""" items = list(body_input) @@ -249,6 +372,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): @@ -285,7 +412,7 @@ def _merge_headless_tool_denies(self, cfg: dict[str, Any]) -> None: merged = list(dict.fromkeys([item for item in deny if isinstance(item, str)] + list(self._HEADLESS_TOOL_DENY))) tools["deny"] = merged - def _build_openclaw_config(self, base: dict[str, Any]) -> dict[str, Any]: + def _build_openclaw_config(self, base: dict[str, Any], rollout_id: Optional[str] = None) -> dict[str, Any]: cfg = copy.deepcopy(base) self._deep_merge(cfg, copy.deepcopy(self.config.openclaw_config)) if self.config.model_server: @@ -294,7 +421,7 @@ def _build_openclaw_config(self, base: dict[str, Any]) -> dict[str, Any]: nemo.update( { "api": "openai-completions", - "baseUrl": self._resolve_model_base_url(), + "baseUrl": self._resolve_model_base_url(rollout_id), "apiKey": "EMPTY", # pragma: allowlist secret "models": [ { @@ -312,15 +439,10 @@ def _build_openclaw_config(self, base: dict[str, Any]) -> dict[str, Any]: self._merge_headless_tool_denies(cfg) return cfg - def _resolve_model_base_url(self) -> str: + def _resolve_model_base_url(self, rollout_id: Optional[str] = None) -> str: if self.config.model_server is None: return "" - config = get_first_server_config_dict( - self.server_client.global_config_dict, - self.config.model_server.name, - ) - base_url = self.server_client._build_server_base_url(config).rstrip("/") - return base_url if base_url.endswith("/v1") else f"{base_url}/v1" + return self.resolve_model_base_url(self.config.model_server.name, rollout_id) def _effective_model(self) -> str: return f"nemo/{self.config.model}" if self.config.model_server else self.config.model @@ -370,7 +492,13 @@ 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], + rollout_id: Optional[str] = None, + observation_collector: Optional[ + Callable[[str, list[dict[str, Any]], OpenClawSessionTree, list[ObservationGap]], 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}" @@ -393,7 +521,7 @@ async def _run_openclaw( if not config_path.is_file(): raise RuntimeError(f"openclaw setup did not produce a config at {config_path}: {stderr}") base_cfg = json.loads(config_path.read_text()) - config_path.write_text(json.dumps(self._build_openclaw_config(base_cfg), indent=2) + "\n") + config_path.write_text(json.dumps(self._build_openclaw_config(base_cfg, rollout_id), indent=2) + "\n") cmd = [ *self.config.command_parts, @@ -421,17 +549,40 @@ 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 observation_collector is not None: + try: + session_events = parse_openclaw_session_events(session_text) + native_session_id = next( + ( + event.get("id") + for event in session_events + if event.get("type") == "session" and isinstance(event.get("id"), str) + ), + session_path.stem, + ) + session_tree, tree_gaps = discover_openclaw_session_tree( + home / ".openclaw" / "agents", + native_session_id, + ) + observation_collector(native_session_id, session_events, session_tree, tree_gaps) + except Exception: + LOG.exception("failed to record OpenClaw session artifact") 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, + rollout_id: Optional[str] = None, + observation_collector: Optional[ + Callable[[str, list[dict[str, Any]], OpenClawSessionTree, list[ObservationGap]], None] + ] = None, + output_collector: Optional[Callable[[list[Any]], None]] = None, ) -> NeMoGymResponse: body = body.model_copy(deep=True) if isinstance(body.input, str): @@ -442,11 +593,18 @@ 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, + rollout_id=rollout_id, + 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 + if output_collector is not None: + output_collector(list(output_items)) if not output_items: LOG.warning("OpenClaw produced no assistant message. Padding empty output") output_items.append( @@ -461,6 +619,7 @@ async def responses( input_tokens = usage.get("input_tokens", 0) output_tokens = usage.get("output_tokens", 0) + cached_tokens = usage.get("cached_tokens", 0) return NeMoGymResponse( id=f"resp_{uuid4().hex}", @@ -473,13 +632,106 @@ async def responses( parallel_tool_calls=body.parallel_tool_calls, usage=NeMoGymResponseUsage( input_tokens=input_tokens, - input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=0), + input_tokens_details=NeMoGymResponseInputTokensDetails(cached_tokens=cached_tokens), output_tokens=output_tokens, output_tokens_details=NeMoGymResponseOutputTokensDetails(reasoning_tokens=0), total_tokens=input_tokens + output_tokens, ), ) + async def responses( + self, + request: Request, + body: NeMoGymResponseCreateParamsNonStreaming = Body(), + ) -> NeMoGymResponse: + path_params = getattr(request, "path_params", None) + rollout_id = path_params.get("rollout_id") if isinstance(path_params, Mapping) else None + if not isinstance(rollout_id, str): + return await self._create_response(body) + episode = await self._create_episode(body, rollout_id=rollout_id) + return episode.response.model_copy( + update={_INTERNAL_OBSERVATIONS_KEY: episode.observations.model_dump(mode="json")} + ) + + async def _create_episode( + self, + body: NeMoGymResponseCreateParamsNonStreaming, + *, + rollout_id: str, + ) -> AgentEpisode: + session_id: Optional[str] = None + session_events: list[dict[str, Any]] = [] + session_tree: OpenClawSessionTree = [] + tree_gaps: list[ObservationGap] = [] + input_items: list[Any] = ( + [NeMoGymEasyInputMessage(role="user", content=body.input)] + if isinstance(body.input, str) + else list(body.input) + ) + observed_output: list[Any] = [] + + def collect( + value: str, + events: list[dict[str, Any]], + tree: OpenClawSessionTree, + gaps: list[ObservationGap], + ) -> None: + nonlocal session_id, session_events, session_tree, tree_gaps + session_id = value + session_events = events + session_tree = tree + tree_gaps = gaps + + def collect_output(value: list[Any]) -> None: + observed_output.extend(value) + + response = await self._create_response( + body, + rollout_id=rollout_id, + observation_collector=collect, + output_collector=collect_output, + ) + try: + if session_tree: + tree_inputs = [] + for invocation_id, parent_id, events in session_tree: + conversation = openclaw_session_conversation( + events, + input_items=input_items if parent_id is None else None, + fallback_output=observed_output if parent_id is None else None, + ) + tree_inputs.append((invocation_id, parent_id, conversation, events)) + observations = build_openclaw_observation_tree( + tree_inputs, + model_ref=self.config.model_server, + ) + observations.gaps.extend(tree_gaps) + else: + transcript_available = any(event.get("type") == "message" for event in session_events) + observations = build_openclaw_observations( + session_id or response.id, + openclaw_session_conversation( + session_events, + input_items=input_items, + fallback_output=observed_output, + ), + session_events, + transcript_available=transcript_available, + model_ref=self.config.model_server, + ) + if any(gap.code == "subagent_hierarchy_unavailable" for gap in tree_gaps): + observations.gaps = [ + gap for gap in observations.gaps if gap.code != "subagent_hierarchy_unavailable" + ] + observations.gaps.extend(tree_gaps) + except Exception: + LOG.exception("failed to build OpenClaw observations") + observations = AgentObservationBundle( + source=OPENCLAW_OBSERVATION_SOURCE, + gaps=[ObservationGap(code="observation_capture_failed")], + ) + return AgentEpisode(response=response, observations=observations) + async def run(self, request: Request, body: OpenClawAgentRunRequest) -> OpenClawAgentVerifyResponse: async with self.sem: cookies = request.cookies @@ -493,15 +745,22 @@ async def run(self, request: Request, body: OpenClawAgentRunRequest) -> OpenClaw await raise_for_status(seed_resp) cookies = seed_resp.cookies + rollout_id = self.rollout_id_from_run(body) agent_resp = await self.server_client.post( server_name=self.config.name, - url_path="/v1/responses", + 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) + raw_observations = ( + agent_resp_json.pop(_INTERNAL_OBSERVATIONS_KEY, None) if rollout_id is not None else None + ) + observations = ( + AgentObservationBundle.model_validate(raw_observations) if isinstance(raw_observations, dict) else None + ) verify_resp = await self.server_client.post( server_name=self.config.resources_server.name, @@ -521,9 +780,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..c8e6e74403 --- /dev/null +++ b/responses_api_agents/openclaw_agent/observability.py @@ -0,0 +1,444 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import json +from collections import Counter +from collections.abc import Iterable +from datetime import datetime, timezone +from math import isfinite +from pathlib import Path +from typing import Any + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.openai_utils import NeMoGymResponseInputItem +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ModelCallRef, + ObservationGap, + ToolCallObservation, +) + + +OpenClawSessionTree = list[tuple[str, str | None, list[dict[str, Any]]]] +OPENCLAW_OBSERVATION_SOURCE = "openclaw" +_MILLISECOND_EPOCH_THRESHOLD = 100_000_000_000 + + +def _read_events(path: Path) -> list[dict[str, Any]]: + events = [] + for line in path.read_text(errors="replace").splitlines(): + try: + event = json.loads(line) + except (json.JSONDecodeError, RecursionError): + event = {"raw": line} + events.append(event if isinstance(event, dict) else {"raw": line}) + return events + + +def discover_openclaw_session_tree( + agents_root: Path, + root_session_id: str, +) -> tuple[OpenClawSessionTree, list[ObservationGap]]: + """Read exact retained-session lineage from OpenClaw's session stores.""" + agents_root = agents_root.resolve() + stores = sorted( + store + for store in agents_root.glob("*/sessions/sessions.json") + if not store.is_symlink() and store.resolve().is_relative_to(agents_root) + ) + if not stores: + return [], [ObservationGap(code="subagent_hierarchy_unavailable")] + + entries: dict[str, tuple[dict[str, Any], Path]] = {} + duplicate_keys: set[str] = set() + gaps: list[ObservationGap] = [] + incomplete = False + for store in stores: + try: + data = json.loads(store.read_text()) + except (OSError, UnicodeError, json.JSONDecodeError, RecursionError): + incomplete = True + gaps.append(ObservationGap(code="agent_session_store_unreadable")) + continue + if not isinstance(data, dict): + incomplete = True + gaps.append(ObservationGap(code="agent_session_store_unreadable")) + continue + for session_key, entry in data.items(): + if not isinstance(session_key, str) or not isinstance(entry, dict): + incomplete = True + gaps.append(ObservationGap(code="agent_session_entry_unparseable")) + continue + if session_key in entries or session_key in duplicate_keys: + incomplete = True + gaps.append(ObservationGap(code="agent_session_identity_ambiguous", detail=session_key)) + entries.pop(session_key, None) + duplicate_keys.add(session_key) + continue + entries[session_key] = (entry, store.parent) + + roots = [key for key, (entry, _) in entries.items() if entry.get("sessionId") == root_session_id] + if len(roots) != 1: + gaps.append( + ObservationGap( + code="subagent_hierarchy_unavailable", + detail="root_session_not_found" if not roots else "root_session_ambiguous", + ) + ) + return [], gaps + + root = roots[0] + selected = {root} + reported_parent_conflicts: set[str] = set() + parents: dict[str, str | None] = {root: None} + changed = True + while changed: + changed = False + for key, (entry, _) in entries.items(): + if key in selected or key in reported_parent_conflicts: + continue + spawned_by = entry.get("spawnedBy") + parent_key = entry.get("parentSessionKey") + if spawned_by and parent_key and spawned_by != parent_key: + if spawned_by in selected or parent_key in selected: + incomplete = True + gaps.append( + ObservationGap( + code="subagent_parent_ambiguous", + invocation_id=key, + ) + ) + reported_parent_conflicts.add(key) + continue + parent = spawned_by or parent_key + if isinstance(parent, str) and parent in selected: + selected.add(key) + parents[key] = parent + changed = True + + ordered = [root] + while len(ordered) < len(selected): + children = sorted(key for key in selected - set(ordered) if parents.get(key) in ordered) + if not children: + incomplete = True + break + ordered.extend(children) + + sessions: OpenClawSessionTree = [] + for key in ordered: + entry, directory = entries[key] + session_id = entry.get("sessionId") + candidates: list[Path] = [] + session_file = entry.get("sessionFile") + if isinstance(session_file, str) and session_file: + candidates.append(directory / Path(session_file).name) + if isinstance(session_id, str) and session_id: + candidates.append(directory / f"{session_id}.jsonl") + path = next( + ( + candidate + for candidate in candidates + if candidate.is_file() + and not candidate.is_symlink() + and candidate.resolve().is_relative_to(agents_root) + ), + None, + ) + events = _read_events(path) if path is not None else [] + if path is None: + incomplete = True + gaps.append( + ObservationGap( + code="agent_transcript_unavailable", + invocation_id=key, + detail=session_id if isinstance(session_id, str) else None, + ) + ) + sessions.append((key, parents[key], events)) + + if incomplete: + gaps.append(ObservationGap(code="subagent_hierarchy_incomplete")) + return sessions, gaps + + +def build_openclaw_observation_tree( + sessions: Iterable[ + tuple[ + str, + str | None, + Iterable[NeMoGymResponseInputItem], + Iterable[dict[str, Any]], + ] + ], + *, + model_ref: ModelServerRef | None = None, +) -> AgentObservationBundle: + """Combine per-session observations after exact store-based lineage discovery.""" + combined = AgentObservationBundle(source=OPENCLAW_OBSERVATION_SOURCE) + for invocation_id, parent_id, conversation, events in sessions: + events = list(events) + bundle = build_openclaw_observations( + invocation_id, + conversation, + events, + transcript_available=any(event.get("type") == "message" for event in events), + prefer_native_session_id=False, + model_ref=model_ref, + ) + combined.gaps.extend(gap for gap in bundle.gaps if gap.code != "subagent_hierarchy_unavailable") + invocation = next((record for record in bundle.records if isinstance(record, AgentInvocation)), None) + if invocation is None: + combined.gaps.append( + ObservationGap( + code="agent_invocation_unavailable", + invocation_id=invocation_id, + ) + ) + continue + invocation.parent_invocation_id = parent_id + combined.records.extend(bundle.records) + if parent_id is not None: + combined.gaps.append( + ObservationGap( + code="subagent_spawn_tool_unavailable", + invocation_id=invocation_id, + ) + ) + return combined + + +def _timestamp(value: Any) -> float | None: + if isinstance(value, bool): + return None + if isinstance(value, (int, float)) and isfinite(value) and value >= 0: + # Current Unix timestamps are ~1e9 seconds or ~1e12 milliseconds. + return float(value) / 1000 if value >= _MILLISECOND_EPOCH_THRESHOLD else float(value) + if isinstance(value, str): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed.replace(tzinfo=parsed.tzinfo or timezone.utc).timestamp() + except ValueError: + return None + return None + + +def _model_visible_tool_calls( + conversation: Iterable[NeMoGymResponseInputItem], +) -> list[tuple[str, str | None]]: + """Return model-visible call IDs and tool names.""" + + def field(item: Any, name: str) -> Any: + return item.get(name) if isinstance(item, dict) else getattr(item, name, None) + + return [ + (call_id, field(item, "name")) + for item in conversation + if field(item, "type") == "function_call" and isinstance((call_id := field(item, "call_id")), str) and call_id + ] + + +def _has_conversation_branches(events: Iterable[dict[str, Any]]) -> bool: + children: dict[str, int] = {} + for event in events: + parent_id = event.get("parentId") + if isinstance(parent_id, str) and parent_id: + children[parent_id] = children.get(parent_id, 0) + 1 + return any(count > 1 for count in children.values()) + + +def build_openclaw_observations( + invocation_id: str, + conversation: Iterable[NeMoGymResponseInputItem], + events: Iterable[dict[str, Any]], + *, + transcript_available: bool, + prefer_native_session_id: bool = True, + model_ref: ModelServerRef | None = None, +) -> AgentObservationBundle: + events = list(events) + native_session_id = next( + ( + event.get("id") + for event in events + if isinstance(event, dict) + and event.get("type") == "session" + and isinstance(event.get("id"), str) + and event["id"] + ), + None, + ) + if prefer_native_session_id: + invocation_id = native_session_id or invocation_id + conversation = list(conversation) + invocation = AgentInvocation(invocation_id=invocation_id, conversation=conversation) + bundle = AgentObservationBundle( + source=OPENCLAW_OBSERVATION_SOURCE, + records=[invocation], + gaps=[ + ObservationGap(code="subagent_hierarchy_unavailable"), + ObservationGap(code="model_call_ownership_unavailable"), + ObservationGap(code="context_compaction_unavailable"), + ], + ) + if not transcript_available: + bundle.gaps.append(ObservationGap(code="agent_transcript_unavailable")) + return bundle + + response_ids = list( + dict.fromkeys( + response_id + for event in events + if event.get("type") == "message" + and isinstance((message := event.get("message")), dict) + and message.get("role") == "assistant" + and isinstance((response_id := message.get("responseId")), str) + and response_id + ) + ) + if model_ref is not None and response_ids: + invocation.model_calls = [ + ModelCallRef(model_ref=model_ref, response_id=response_id) for response_id in response_ids + ] + bundle.gaps = [gap for gap in bundle.gaps if gap.code != "model_call_ownership_unavailable"] + + if _has_conversation_branches(events): + bundle.gaps.append( + ObservationGap( + code="agent_conversation_branching_unavailable", + invocation_id=invocation_id, + ) + ) + + bundle.gaps = [gap for gap in bundle.gaps if gap.code != "context_compaction_unavailable"] + visible_tools = _model_visible_tool_calls(conversation) + tool_counts = Counter(call_id for call_id, _ in visible_tools) + tool_metadata = {call_id: tool_name for call_id, tool_name in visible_tools if tool_counts[call_id] == 1} + for call_id, count in tool_counts.items(): + if count > 1: + bundle.gaps.append( + ObservationGap( + code="tool_call_identity_ambiguous", + invocation_id=invocation_id, + detail=call_id, + ) + ) + tools: dict[str, ToolCallObservation] = {} + + for event in events: + if not isinstance(event, dict) or "raw" in event: + bundle.gaps.append(ObservationGap(code="agent_artifact_record_unparseable")) + continue + + message = event.get("message") if isinstance(event.get("message"), dict) else {} + if event.get("type") == "message" and message.get("role") == "toolResult": + call_id = message.get("toolCallId") or message.get("tool_call_id") + if not isinstance(call_id, str) or call_id not in tool_metadata or call_id in tools: + bundle.gaps.append( + ObservationGap( + code="tool_result_unowned", + invocation_id=invocation_id, + detail=call_id if isinstance(call_id, str) else None, + ) + ) + continue + + tool_name = tool_metadata[call_id] + tool = ToolCallObservation( + invocation_id=invocation_id, + tool_call_id=call_id, + tool_name=tool_name, + ) + details = message.get("details") if isinstance(message.get("details"), dict) else {} + duration = details.get("durationMs") + completed_at = _timestamp(message.get("timestamp")) + if completed_at is None: + completed_at = _timestamp(event.get("timestamp")) + if ( + isinstance(duration, (int, float)) + and not isinstance(duration, bool) + and isfinite(duration) + and duration >= 0 + and completed_at is not None + ): + tool.duration_ms = float(duration) + tool.completed_at = completed_at + tool.started_at = completed_at - float(duration) / 1000 + tool.timing_source = "artifact" + + native_status = details.get("status") + if message.get("isError") is True or native_status in {"error", "failed"}: + tool.status = "failed" + elif native_status in {"completed", "success", "ok"}: + tool.status = "completed" + elif native_status == "timeout": + tool.status = "timeout" + elif message.get("isError") is False: + tool.status = "completed" + else: + tool.status = "unknown" + tools[call_id] = tool + bundle.records.append(tool) + continue + + if event.get("type") != "compaction": + continue + tokens_before = event.get("tokensBefore") + tokens_after = event.get("tokensAfter") + summary = event.get("summary") + compaction = ContextCompactionObservation( + invocation_id=invocation_id, + observed_at=_timestamp(event.get("timestamp")), + trigger=event.get("reason") if isinstance(event.get("reason"), str) else None, + tokens_before=tokens_before if type(tokens_before) is int and tokens_before >= 0 else None, + tokens_after=tokens_after if type(tokens_after) is int and tokens_after >= 0 else None, + outcome="completed", + summary=summary if isinstance(summary, str) else None, + first_kept_item_id=( + event.get("firstKeptEntryId") if isinstance(event.get("firstKeptEntryId"), str) else None + ), + ) + bundle.records.append(compaction) + bundle.gaps.append( + ObservationGap( + code="compaction_model_call_boundary_unavailable", + invocation_id=invocation_id, + ) + ) + if compaction.summary is None: + bundle.gaps.append(ObservationGap(code="compaction_summary_unavailable", invocation_id=invocation_id)) + if compaction.tokens_after is None: + bundle.gaps.append( + ObservationGap( + code="compaction_tokens_after_unavailable", + invocation_id=invocation_id, + ) + ) + + for call_id in tool_metadata.keys() - tools.keys(): + bundle.gaps.append( + ObservationGap( + code="tool_timing_unavailable", + invocation_id=invocation_id, + detail=call_id, + ) + ) + for tool in tools.values(): + if tool.timing_source is None: + bundle.gaps.append( + ObservationGap( + code="tool_timing_unavailable", + invocation_id=invocation_id, + detail=tool.tool_call_id, + ) + ) + if tool.status == "unknown": + bundle.gaps.append( + ObservationGap( + code="tool_outcome_unavailable", + invocation_id=invocation_id, + detail=tool.tool_call_id, + ) + ) + return bundle diff --git a/responses_api_agents/openclaw_agent/tests/test_app.py b/responses_api_agents/openclaw_agent/tests/test_app.py index f85b54e7bd..d92c971df2 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 @@ -28,19 +28,39 @@ NeMoGymResponseFunctionToolCall, NeMoGymResponseOutputMessage, ) +from nemo_gym.rollout_observability import AgentInvocation, ObservationGap from nemo_gym.server_utils import ServerClient from responses_api_agents.openclaw_agent.app import ( OpenClawAgent, OpenClawAgentConfig, + OpenClawAgentRunRequest, ResourcesServerRef, _decode_last_json_dict_suffix, _extract_instruction, _text_from_openclaw_payloads, + openclaw_session_conversation, parse_openclaw_output, parse_openclaw_session, + parse_openclaw_session_events, + parse_openclaw_session_items, ) +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 _invocations(bundle): + return [record for record in bundle.records if isinstance(record, AgentInvocation)] + + def _config(**kwargs) -> OpenClawAgentConfig: kwargs.setdefault("openclaw_version", "2026.6.11") return OpenClawAgentConfig( @@ -135,7 +155,7 @@ class TestParseOpenclawOutput: def test_empty(self) -> None: items, usage = parse_openclaw_output("") assert items == [] - assert usage == {"input_tokens": 0, "output_tokens": 0} + assert usage == {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} def test_text_message_and_usage(self) -> None: raw = _envelope( @@ -148,6 +168,7 @@ def test_text_message_and_usage(self) -> None: assert items[0].content[0].text == "the answer is 4" assert usage["input_tokens"] == 105 assert usage["output_tokens"] == 20 + assert usage["cached_tokens"] == 5 def test_no_text_no_items(self) -> None: raw = _envelope([], usage={"input": 1, "output": 0}) @@ -202,10 +223,66 @@ def test_user_messages_ignored(self) -> None: assert parse_openclaw_session(line) == [] def test_malformed_lines_skipped(self) -> None: - line = "not-json\n" + self._msg("assistant", [{"type": "text", "text": "ok"}]) + line = "not-json\nnull\n[]\n" + self._msg("assistant", [{"type": "text", "text": "ok"}]) items = parse_openclaw_session(line) assert len(items) == 1 + def test_preserves_string_input_and_reasoning(self) -> None: + events = [ + {"type": "message", "message": {"role": "user", "content": "solve this"}}, + { + "type": "message", + "message": { + "role": "assistant", + "content": [ + {"type": "reasoning", "text": "need a tool"}, + {"type": "toolCall", "id": "c1", "name": "search", "arguments": {}}, + ], + }, + }, + ] + + items = parse_openclaw_session_items(events, include_input=True) + + assert [item.type for item in items] == ["message", "reasoning", "function_call"] + assert items[0].content == "solve this" + assert items[1].summary[0].text == "need a tool" + + def test_input_only_transcript_uses_fallback_output(self) -> None: + fallback = NeMoGymResponseOutputMessage( + id="fallback", + content=[{"type": "output_text", "text": "done", "annotations": []}], + ) + + conversation = openclaw_session_conversation( + [{"type": "message", "message": {"role": "user", "content": "retained input"}}], + input_items=[NeMoGymEasyInputMessage(role="user", content="request input")], + fallback_output=[fallback], + ) + + assert conversation == [NeMoGymEasyInputMessage(role="user", content="retained input"), fallback] + + def test_preserves_conflicting_duplicate_event_ids(self) -> None: + events = [ + {"id": "same", "type": "message", "message": {"role": "assistant", "content": "one"}}, + {"id": "same", "type": "message", "message": {"role": "assistant", "content": "two"}}, + ] + + items = parse_openclaw_session_items(events, include_input=True) + + assert [item.content[0].text for item in items] == ["one", "two"] + + def test_restores_missing_known_system_input(self) -> None: + system = NeMoGymEasyInputMessage(role="system", content="system rules") + user = NeMoGymEasyInputMessage(role="user", content="solve this") + + conversation = openclaw_session_conversation( + [{"type": "message", "message": {"role": "user", "content": "solve this"}}], + input_items=[system, user], + ) + + assert conversation == [system, user] + class TestBuildOpenclawConfig: def test_headless_message_tool_denied(self) -> None: @@ -234,6 +311,30 @@ def test_user_openclaw_config_merged(self) -> None: assert cfg["extra"] == {"k": "v"} assert cfg["gateway"]["mode"] == "local" + def test_model_server_replaces_provider_base_url_for_rollout(self) -> None: + agent = _make_agent( + model_server=ModelServerRef(type="responses_api_models", name="policy"), + ) + with patch.object( + OpenClawAgent, + "resolve_model_base_url", + return_value="http://policy/ng-rollout/7-2/v1", + ): + cfg = agent._build_openclaw_config({}, "7-2") + + assert cfg["models"]["providers"]["nemo"]["baseUrl"] == "http://policy/ng-rollout/7-2/v1" + + def test_responses_propagates_rollout_path(self) -> None: + agent = _make_agent() + + async def run_openclaw(*args, **kwargs): + assert kwargs["rollout_id"] == "7-2" + return [], {"input_tokens": 0, "output_tokens": 0}, "model" + + request = MagicMock(path_params={"rollout_id": "7-2"}) + with patch.object(agent, "_run_openclaw", run_openclaw): + asyncio.run(agent.responses(request, NeMoGymResponseCreateParamsNonStreaming(input="solve"))) + def test_user_deny_cannot_drop_headless_deny(self) -> None: agent = _make_agent(openclaw_config={"tools": {"deny": ["custom"]}}) cfg = agent._build_openclaw_config({}) @@ -275,6 +376,267 @@ def test_env_passthrough(self) -> None: assert "EMPTY" not in env +class TestObservability: + def test_collects_session_artifact_before_workspace_cleanup(self, tmp_path: Path) -> None: + agent = _make_agent(workspace_root=str(tmp_path)) + work_dir = tmp_path / "run" + config_path = work_dir / ".openclaw-home" / ".openclaw" / "openclaw.json" + config_path.parent.mkdir(parents=True) + config_path.write_text("{}") + sessions_dir = config_path.parent / "agents" / "main" / "sessions" + sessions_dir.mkdir(parents=True) + session_path = sessions_dir / "session-1.jsonl" + session_path.write_text( + "\n".join( + [ + json.dumps({"type": "session", "id": "session-1"}), + json.dumps( + { + "type": "message", + "message": {"role": "assistant", "content": [{"type": "text", "text": "done"}]}, + } + ), + ] + ) + ) + child_path = sessions_dir / "child-transcript.jsonl" + child_path.write_text( + "\n".join( + [ + json.dumps({"type": "session", "id": "child-transcript"}), + json.dumps( + { + "type": "message", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "child done"}], + }, + } + ), + ] + ) + ) + (sessions_dir / "sessions.json").write_text( + json.dumps( + { + "agent:main:main": {"sessionId": "session-1", "sessionFile": str(session_path)}, + "agent:main:subagent:child": { + "sessionId": "child-transcript", + "spawnedBy": "agent:main:main", + }, + } + ) + ) + stdout = json.dumps({"payloads": [], "meta": {"agentMeta": {"sessionFile": str(session_path)}}}) + collector = MagicMock() + + with ( + patch.object(agent, "_workspace_root", return_value=work_dir), + patch.object(agent, "_run_exec", AsyncMock(side_effect=[(0, "", ""), (0, stdout, "")])), + ): + output, _, _ = asyncio.run(agent._run_openclaw("solve", None, observation_collector=collector)) + + assert output[0].content[0].text == "done" + assert collector.call_args.args[0] == "session-1" + assert collector.call_args.args[1][1]["type"] == "message" + assert [(item[0], item[1]) for item in collector.call_args.args[2]] == [ + ("agent:main:main", None), + ("agent:main:subagent:child", "agent:main:main"), + ] + assert not work_dir.exists() + + def test_scoring_padding_is_not_reported_as_agent_output(self) -> None: + agent = _make_agent() + + async def run_openclaw(*args, observation_collector=None, **kwargs): + observation_collector( + "session-without-output", + [{"type": "session", "id": "session-without-output"}], + [], + [], + ) + return [], {"input_tokens": 0, "output_tokens": 0}, "model" + + body = NeMoGymResponseCreateParamsNonStreaming(input="solve") + with patch.object(agent, "_run_openclaw", run_openclaw): + episode = asyncio.run(agent._create_episode(body, rollout_id="1-2")) + + assert episode.response.output[0].content[0].text == "" + [invocation] = _invocations(episode.observations) + assert invocation.invocation_id == "session-without-output" + assert invocation.conversation == [NeMoGymEasyInputMessage(role="user", content="solve")] + assert "agent_transcript_unavailable" in {gap.code for gap in episode.observations.gaps} + + def test_fallback_output_is_preserved_without_a_session_transcript(self) -> None: + agent = _make_agent() + output = NeMoGymResponseOutputMessage( + id="fallback", + content=[{"type": "output_text", "text": "done", "annotations": []}], + ) + + async def run_openclaw(*args, **kwargs): + return [output], {"input_tokens": 1, "output_tokens": 1}, "model" + + body = NeMoGymResponseCreateParamsNonStreaming(input="solve") + with patch.object(agent, "_run_openclaw", run_openclaw): + episode = asyncio.run(agent._create_episode(body, rollout_id="1-2")) + + [invocation] = _invocations(episode.observations) + assert invocation.conversation == [ + NeMoGymEasyInputMessage(role="user", content="solve"), + output, + ] + assert "agent_transcript_unavailable" in {gap.code for gap in episode.observations.gaps} + + def test_hierarchy_discovery_gap_replaces_generic_fallback(self) -> None: + agent = _make_agent() + + async def run_openclaw(*args, observation_collector=None, **kwargs): + observation_collector( + "session-1", + [], + [], + [ObservationGap(code="subagent_hierarchy_unavailable", detail="root_session_not_found")], + ) + return [], {"input_tokens": 0, "output_tokens": 0}, "model" + + with patch.object(agent, "_run_openclaw", run_openclaw): + episode = asyncio.run( + agent._create_episode( + NeMoGymResponseCreateParamsNonStreaming(input="solve"), + rollout_id="1-2", + ) + ) + + hierarchy_gaps = [gap for gap in episode.observations.gaps if gap.code == "subagent_hierarchy_unavailable"] + assert [(gap.code, gap.detail) for gap in hierarchy_gaps] == [ + ("subagent_hierarchy_unavailable", "root_session_not_found") + ] + + def test_root_observation_uses_retained_transcript(self) -> None: + agent = _make_agent() + scored_output = NeMoGymResponseOutputMessage( + id="scored", + content=[{"type": "output_text", "text": "scoring view", "annotations": []}], + ) + events = [ + {"type": "session", "id": "session-1"}, + { + "type": "message", + "message": { + "role": "user", + "content": [{"type": "text", "text": "retained input"}], + }, + }, + { + "type": "message", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "from transcript"}], + }, + }, + ] + + async def run_openclaw(*args, observation_collector=None, **kwargs): + observation_collector("session-1", events, [("root", None, events)], []) + return [scored_output], {"input_tokens": 1, "output_tokens": 1}, "model" + + with patch.object(agent, "_run_openclaw", run_openclaw): + episode = asyncio.run( + agent._create_episode( + NeMoGymResponseCreateParamsNonStreaming(input="solve"), + rollout_id="1-2", + ) + ) + + [invocation] = _invocations(episode.observations) + conversation = invocation.conversation + assert conversation[0] == NeMoGymEasyInputMessage(role="user", content="retained input") + assert conversation[1].content[0].text == "from transcript" + assert "scoring view" not in str(conversation) + + 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", parse_openclaw_session_events(session), [], []) + 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"}) + if url_path.endswith("/v1/responses"): + response = await agent.responses(MagicMock(path_params={"rollout_id": "1-2"}), json) + return _FakeResponse(response.model_dump(mode="json"), cookies) + 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 + [invocation] = _invocations(observations) + assert invocation.invocation_id == "session-1" + assert invocation.conversation + verify_json = agent.server_client.post.await_args_list[-1].kwargs["json"] + assert "_ng_agent_observations" not in verify_json["response"] + + 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._create_episode(body, rollout_id="1-2")) + + 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..a5b0d9c54e --- /dev/null +++ b/responses_api_agents/openclaw_agent/tests/test_observability.py @@ -0,0 +1,303 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import json +from pathlib import Path + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.openai_utils import NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, NeMoGymResponseFunctionToolCall +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ToolCallObservation, +) +from responses_api_agents.openclaw_agent.observability import ( + build_openclaw_observation_tree, + build_openclaw_observations, + discover_openclaw_session_tree, +) + + +def _records(bundle, record_type): + return [record for record in bundle.records if isinstance(record, record_type)] + + +def _session(path: Path, session_id: str, text: str = "done") -> None: + path.write_text( + "\n".join( + [ + json.dumps({"type": "session", "id": session_id}), + json.dumps( + { + "type": "message", + "message": {"role": "assistant", "content": [{"type": "text", "text": text}]}, + } + ), + ] + ) + ) + + +def test_discovers_cross_agent_session_tree(tmp_path: Path) -> None: + main = tmp_path / "main" / "sessions" + worker = tmp_path / "worker" / "sessions" + main.mkdir(parents=True) + worker.mkdir(parents=True) + _session(main / "root.jsonl", "root") + _session(main / "child.jsonl", "child") + _session(worker / "grandchild.jsonl", "grandchild") + (main / "sessions.json").write_text( + json.dumps( + { + "agent:main:main": {"sessionId": "root"}, + "agent:main:subagent:child": {"sessionId": "child", "spawnedBy": "agent:main:main"}, + } + ) + ) + (worker / "sessions.json").write_text( + json.dumps( + { + "agent:worker:subagent:grandchild": { + "sessionId": "grandchild", + "parentSessionKey": "agent:main:subagent:child", + } + } + ) + ) + + sessions, gaps = discover_openclaw_session_tree(tmp_path, "root") + + assert [(item[0], item[1]) for item in sessions] == [ + ("agent:main:main", None), + ("agent:main:subagent:child", "agent:main:main"), + ("agent:worker:subagent:grandchild", "agent:main:subagent:child"), + ] + assert gaps == [] + + bundle = build_openclaw_observation_tree( + ( + (invocation_id, parent_id, [NeMoGymEasyInputMessage(role="user", content=invocation_id)], events) + for invocation_id, parent_id, events in sessions + ) + ) + assert [item.parent_invocation_id for item in _records(bundle, AgentInvocation)] == [ + None, + "agent:main:main", + "agent:main:subagent:child", + ] + assert "subagent_hierarchy_unavailable" not in {gap.code for gap in bundle.gaps} + + +def test_missing_child_transcript_is_explicit(tmp_path: Path) -> None: + sessions_dir = tmp_path / "main" / "sessions" + sessions_dir.mkdir(parents=True) + _session(sessions_dir / "root.jsonl", "root") + (sessions_dir / "sessions.json").write_text( + json.dumps( + { + "root-key": {"sessionId": "root"}, + "child-key": {"sessionId": "deleted", "spawnedBy": "root-key"}, + } + ) + ) + + sessions, gaps = discover_openclaw_session_tree(tmp_path, "root") + + assert [item[0] for item in sessions] == ["root-key", "child-key"] + assert {(gap.code, gap.invocation_id) for gap in gaps} >= { + ("agent_transcript_unavailable", "child-key"), + ("subagent_hierarchy_incomplete", None), + } + + +def test_conflicting_parent_is_reported_once_across_tree_passes(tmp_path: Path) -> None: + sessions_dir = tmp_path / "main" / "sessions" + sessions_dir.mkdir(parents=True) + _session(sessions_dir / "root.jsonl", "root") + _session(sessions_dir / "child.jsonl", "child") + (sessions_dir / "sessions.json").write_text( + json.dumps( + { + "root-key": {"sessionId": "root"}, + "ambiguous-key": { + "sessionId": "ambiguous", + "spawnedBy": "root-key", + "parentSessionKey": "child-key", + }, + "child-key": {"sessionId": "child", "spawnedBy": "root-key"}, + } + ) + ) + + _, gaps = discover_openclaw_session_tree(tmp_path, "root") + + ambiguous = [gap for gap in gaps if gap.code == "subagent_parent_ambiguous"] + assert [(gap.invocation_id, gap.detail) for gap in ambiguous] == [("ambiguous-key", None)] + + +def test_tree_reports_missing_invocation(monkeypatch) -> None: + monkeypatch.setattr( + "responses_api_agents.openclaw_agent.observability.build_openclaw_observations", + lambda *args, **kwargs: AgentObservationBundle(source="openclaw"), + ) + + bundle = build_openclaw_observation_tree([("session-1", None, [], [])]) + + assert [(gap.code, gap.invocation_id) for gap in bundle.gaps] == [("agent_invocation_unavailable", "session-1")] + + +def test_three_duplicate_session_keys_are_reported_without_crashing(tmp_path: Path) -> None: + for agent in ("one", "two", "three"): + sessions_dir = tmp_path / agent / "sessions" + sessions_dir.mkdir(parents=True) + (sessions_dir / "sessions.json").write_text(json.dumps({"duplicate": {"sessionId": agent}})) + + sessions, gaps = discover_openclaw_session_tree(tmp_path, "one") + + assert sessions == [] + assert "agent_session_identity_ambiguous" in {gap.code for gap in gaps} + + +def test_session_file_cannot_escape_the_session_archive(tmp_path: Path) -> None: + agents_root = tmp_path / "agents" + sessions_dir = agents_root / "main" / "sessions" + sessions_dir.mkdir(parents=True) + outside = tmp_path / "outside-session.jsonl" + _session(outside, "root", "secret") + (sessions_dir / outside.name).symlink_to(outside) + (sessions_dir / "sessions.json").write_text( + json.dumps({"root-key": {"sessionId": "root", "sessionFile": str(outside)}}) + ) + + sessions, gaps = discover_openclaw_session_tree(agents_root, "root") + + assert sessions == [("root-key", None, [])] + assert "agent_transcript_unavailable" in {gap.code for gap in gaps} + + +def test_builds_root_observation_and_reports_missing_tool_timing() -> None: + model_ref = ModelServerRef(type="responses_api_models", name="policy") + 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"), + ], + [ + { + "type": "message", + "message": {"role": "assistant", "responseId": "response-1"}, + } + ], + transcript_available=True, + model_ref=model_ref, + ) + + [invocation] = _records(bundle, AgentInvocation) + assert invocation.invocation_id == "session-1" + assert invocation.conversation + assert invocation.model_calls[0].response_id == "response-1" + assert invocation.model_calls[0].model_ref == model_ref + assert _records(bundle, ToolCallObservation) == [] + assert {gap.code for gap in bundle.gaps} == { + "subagent_hierarchy_unavailable", + "tool_timing_unavailable", + } + + +def test_marks_missing_transcript() -> None: + bundle = build_openclaw_observations( + "fallback", + [], + [], + transcript_available=False, + ) + + assert "agent_transcript_unavailable" in {gap.code for gap in bundle.gaps} + + +def test_duplicate_tool_ids_are_ambiguous_and_do_not_imply_success() -> None: + bundle = build_openclaw_observations( + "session-1", + [ + NeMoGymResponseFunctionToolCall(arguments="{}", call_id="duplicate", name="tool"), + NeMoGymResponseFunctionToolCall(arguments="{}", call_id="duplicate", name="tool"), + NeMoGymFunctionCallOutput(call_id="duplicate", output="result", status="completed"), + ], + [], + transcript_available=True, + ) + + assert _records(bundle, ToolCallObservation) == [] + assert "tool_call_identity_ambiguous" in {gap.code for gap in bundle.gaps} + + +def test_extracts_parallel_tool_intervals_and_compaction() -> None: + conversation = [ + NeMoGymResponseFunctionToolCall(arguments="{}", call_id="call-1", name="tool"), + NeMoGymFunctionCallOutput(call_id="call-1", output="result", status="completed"), + NeMoGymResponseFunctionToolCall(arguments="{}", call_id="call-2", name="tool"), + NeMoGymFunctionCallOutput(call_id="call-2", output="error", status="completed"), + ] + events = [ + { + "type": "message", + "message": { + "role": "toolResult", + "toolCallId": "call-1", + "timestamp": 1_750_000_002_000, + "details": {"durationMs": 1000, "status": "completed"}, + }, + }, + { + "type": "message", + "message": { + "role": "toolResult", + "toolCallId": "call-2", + "timestamp": 1_750_000_002_500, + "details": {"durationMs": 1000, "status": "failed"}, + "isError": True, + }, + }, + { + "type": "compaction", + "timestamp": "2025-06-15T15:06:43Z", + "summary": "condensed context", + "firstKeptEntryId": "entry-7", + "tokensBefore": 120_000, + }, + ] + + bundle = build_openclaw_observations("session-1", conversation, events, transcript_available=True) + + first, second = _records(bundle, ToolCallObservation) + assert first.started_at < second.started_at < first.completed_at < second.completed_at + assert first.duration_ms == second.duration_ms == 1000 + assert first.timing_source == second.timing_source == "artifact" + assert first.status == "completed" + assert second.status == "failed" + [compaction] = _records(bundle, ContextCompactionObservation) + assert compaction.summary == "condensed context" + assert compaction.first_kept_item_id == "entry-7" + assert compaction.tokens_before == 120_000 + assert {"compaction_tokens_after_unavailable", "compaction_model_call_boundary_unavailable"} <= { + gap.code for gap in bundle.gaps + } + + +def test_reports_intra_session_branching() -> None: + events = [ + {"type": "message", "id": "first", "parentId": "root", "message": {"role": "assistant", "content": []}}, + {"type": "message", "id": "second", "parentId": "root", "message": {"role": "assistant", "content": []}}, + ] + + bundle = build_openclaw_observations("session-1", [], events, transcript_available=True) + + assert "agent_conversation_branching_unavailable" in {gap.code for gap in bundle.gaps} diff --git a/responses_api_agents/pinchbench/README.md b/responses_api_agents/pinchbench/README.md index 8dbcf27804..961d355d8e 100644 --- a/responses_api_agents/pinchbench/README.md +++ b/responses_api_agents/pinchbench/README.md @@ -56,12 +56,14 @@ small NVIDIA integration patch: | `task_timeout_s` | per-task exec timeout | | `openclaw_provider_timeout_seconds` | optional OpenClaw model idle/request timeout in seconds; written to both the provider timeout and agent timeout ceiling so values above OpenClaw's 120s default are effective | | `model_base_url` / `model_api_key` / `model_name` | policy model OpenClaw runs against | +| `model_server` | optional Gym Model Server reference for correlated policy calls; otherwise `model_base_url` is used directly | | `judge_model` / `judge_base_url` / `judge_api_key` | judge for hybrid / `llm_judge` tasks | +| `judge_model_server` | optional Gym Model Server reference for correlated judge calls; otherwise `judge_base_url` is used directly | | `max_tokens`, `context_window`, `max_concurrent`, `timeout_multiplier` | run tuning | -> **Model wiring:** OpenClaw must point at a **streaming-capable** endpoint directly — *not* a Gym -> model server, which is non-streaming (`stream: Literal[False]`) and would 422 OpenClaw's streamed -> requests. So the policy/judge endpoints are passed straight through to OpenClaw. +> **Model wiring:** Set `model_server` and `judge_model_server` to route calls through correlated Gym +> Model Servers. Gym supplies the streaming SSE envelope OpenClaw expects. Leave either reference unset +> to use its configured direct endpoint instead. ## Setup diff --git a/responses_api_agents/pinchbench/app.py b/responses_api_agents/pinchbench/app.py index 636e40a136..81c4ee27dd 100644 --- a/responses_api_agents/pinchbench/app.py +++ b/responses_api_agents/pinchbench/app.py @@ -20,6 +20,7 @@ import asyncio import glob import json +import logging import shutil import tarfile import textwrap @@ -28,7 +29,7 @@ from typing import Any, Literal, Optional from fastapi import Request, Response -from pydantic import ConfigDict, model_validator +from pydantic import ConfigDict, Field, model_validator from nemo_gym.base_resources_server import BaseRunRequest, BaseVerifyResponse from nemo_gym.base_responses_api_agent import ( @@ -36,7 +37,9 @@ Body, SimpleResponsesAPIAgent, ) +from nemo_gym.config_types import ModelServerRef from nemo_gym.openai_utils import ( + NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, NeMoGymResponse, NeMoGymResponseCreateParamsNonStreaming, @@ -50,20 +53,31 @@ NeMoGymSummary, ) from nemo_gym.rollout_collection import NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY +from nemo_gym.rollout_observability import AgentObservationBundle, ObservationGap from nemo_gym.sandbox import AsyncSandbox, SandboxResources, SandboxSpec, get_provider_class +from responses_api_agents.openclaw_agent.app import openclaw_session_conversation +from responses_api_agents.openclaw_agent.observability import ( + OPENCLAW_OBSERVATION_SOURCE, + build_openclaw_observation_tree, + build_openclaw_observations, + discover_openclaw_session_tree, +) _SEARCH_KEY_MAP: dict[str, str] = {"brave": "brave_api_key", "tavily": "tavily_api_key"} +LOG = logging.getLogger(__name__) class PinchBenchAgentConfig(BaseResponsesAPIAgentConfig): model_base_url: str model_api_key: str model_name: str + model_server: Optional[ModelServerRef] = None judge_model: str judge_base_url: str judge_api_key: str + judge_model_server: Optional[ModelServerRef] = None openclaw_mode: Literal["gateway"] = "gateway" gateway_token: str = "pinchbench-local" @@ -159,6 +173,10 @@ class PinchBenchVerifyResponse(BaseVerifyResponse): grading_notes: str status: str raw_rollout: dict + ng_agent_observations: AgentObservationBundle | None = Field( + default=None, + exclude_if=lambda value: value is None, + ) class PinchBenchAgent(SimpleResponsesAPIAgent): @@ -176,14 +194,24 @@ async def responses( ) -> NeMoGymResponse: raise NotImplementedError("PinchBench is an external benchmark; use /run.") - def _task_env(self, task_id: str) -> dict: + def _task_env(self, task_id: str, rollout_id: Optional[str] = None) -> dict: + model_base_url = ( + self.resolve_model_base_url(self.config.model_server.name, rollout_id) + if self.config.model_server is not None + else self.config.model_base_url + ) + judge_base_url = ( + self.resolve_model_base_url(self.config.judge_model_server.name, rollout_id) + if self.config.judge_model_server is not None + else self.config.judge_base_url + ) env = { "TASK_ID": task_id, "MODEL_NAME": self.config.model_name, - "MODEL_BASE_URL": self.config.model_base_url, + "MODEL_BASE_URL": model_base_url, "MODEL_API_KEY": self.config.model_api_key, "JUDGE_MODEL": self.config.judge_model, - "JUDGE_BASE_URL": self.config.judge_base_url, + "JUDGE_BASE_URL": judge_base_url, "JUDGE_API_KEY": self.config.judge_api_key, "OPENAI_API_KEY": self.config.model_api_key, "PINCHBENCH_WEB_SEARCH_PROVIDER": self.config.web_search_provider, @@ -199,9 +227,11 @@ def _task_env(self, task_id: str) -> dict: env["BRAVE_API_KEY"] = self.config.brave_api_key if self.config.tavily_api_key: env["TAVILY_API_KEY"] = self.config.tavily_api_key + if rollout_id is not None: + env["NEMO_GYM_OBSERVABILITY_ENABLED"] = "1" return env - def _build_spec(self, task_id: str) -> SandboxSpec: + def _build_spec(self, task_id: str, rollout_id: Optional[str] = None) -> SandboxSpec: cfg = dict(self.config.sandbox_spec) return SandboxSpec( image=cfg.get("image"), @@ -210,11 +240,16 @@ def _build_spec(self, task_id: str) -> SandboxSpec: workdir=cfg.get("workdir"), resources=SandboxResources.from_mapping(cfg.get("resources", {})), provider_options=cfg.get("provider_options", {}), - env=self._task_env(task_id), + env=self._task_env(task_id, rollout_id), metadata={"task_id": task_id}, ) - async def _run_in_sandbox(self, task_id: str, out_dir: Path) -> int | None: + async def _run_in_sandbox( + self, + task_id: str, + out_dir: Path, + rollout_id: Optional[str] = None, + ) -> int | None: """Run one PinchBench task and pull its /out archive back. Returns the apptainer exit code when the direct_exec path exits non-zero but @@ -223,15 +258,20 @@ async def _run_in_sandbox(self, task_id: str, out_dir: Path) -> int | None: provider = self.config.sandbox_provider or {} apptainer_cfg = provider.get("apptainer") if isinstance(provider, dict) else None if isinstance(apptainer_cfg, dict) and apptainer_cfg.get("direct_exec"): - return await self._run_in_apptainer_direct(task_id, out_dir, apptainer_cfg) + return await self._run_in_apptainer_direct(task_id, out_dir, apptainer_cfg, rollout_id=rollout_id) if not self.config.sandbox_provider: raise ValueError("pinchbench requires sandbox_provider (see configs/pinchbench.yaml)") archive = f"{self.config.sandbox_work_base.rstrip('/')}/out/out.tgz" sb = AsyncSandbox(self.config.sandbox_provider) try: - await sb.start(self._build_spec(task_id)) - await sb.exec("bash /opt/run_task.sh", timeout_s=self.config.task_timeout_s) + await sb.start(self._build_spec(task_id, rollout_id)) + exec_result = await sb.exec("bash /opt/run_task.sh", timeout_s=self.config.task_timeout_s) + if exec_result.error_type == "timeout": + raise TimeoutError("PinchBench sandbox execution timed out") + if exec_result.error_type: + detail = exec_result.stderr or exec_result.stdout or "unknown sandbox error" + raise RuntimeError(f"PinchBench sandbox execution failed ({exec_result.error_type}): {detail}") await sb.download(archive, out_dir / "out.tgz") finally: await sb.stop() @@ -348,7 +388,13 @@ def _write_direct_exec_wrapper(self, staging_dir: Path) -> Path: wrapper_path.chmod(0o755) return wrapper_path - async def _run_in_apptainer_direct(self, task_id: str, out_dir: Path, apptainer_cfg: dict[str, Any]) -> int | None: + async def _run_in_apptainer_direct( + self, + task_id: str, + out_dir: Path, + apptainer_cfg: dict[str, Any], + rollout_id: Optional[str] = None, + ) -> int | None: image = self.config.sandbox_spec.get("image") if not image: raise ValueError("pinchbench sandbox_spec.image is required for direct Apptainer exec") @@ -365,7 +411,7 @@ async def _run_in_apptainer_direct(self, task_id: str, out_dir: Path, apptainer_ elif isinstance(direct_args, str): direct_args = direct_args.split() - task_env = self._task_env(task_id) + task_env = self._task_env(task_id, rollout_id) argv = ["apptainer", "exec", *[str(arg) for arg in direct_args]] argv += ["--bind", f"{staging_dir}:{work_base}"] for key, value in task_env.items(): @@ -450,6 +496,8 @@ def _content_text(content) -> str: parts.append(item) elif isinstance(item, dict): parts.append(item.get("text") or item.get("output") or "") + else: + parts.append(getattr(item, "text", None) or getattr(item, "output", None) or "") return "\n".join(p for p in parts if p) @staticmethod @@ -480,12 +528,13 @@ def _reasoning_text(message: dict) -> str: @staticmethod def _tool_call_arguments(block: dict) -> str: - partial_args = block.get("partialArgs") - if isinstance(partial_args, str): - return partial_args args = block.get("arguments") if isinstance(args, str): return args + if args is None: + partial_args = block.get("partialArgs") + if isinstance(partial_args, str): + return partial_args if args is None: args = {} return json.dumps(args, ensure_ascii=False) @@ -504,17 +553,25 @@ def as_int(value) -> int: return 0 for event in events: - message = event.get("message") or {} + message = event.get("message") + if not isinstance(message, dict): + continue if message.get("role") != "assistant": continue - usage = message.get("usage") or {} + usage = message.get("usage") + if not isinstance(usage, dict): + continue input_tokens += as_int(usage.get("input") or usage.get("input_tokens") or usage.get("prompt_tokens")) output_tokens += as_int( usage.get("output") or usage.get("output_tokens") or usage.get("completion_tokens") ) cached_tokens += as_int(usage.get("cacheRead")) - input_details = usage.get("input_tokens_details") or {} - output_details = usage.get("output_tokens_details") or {} + input_details = usage.get("input_tokens_details") + if not isinstance(input_details, dict): + input_details = {} + output_details = usage.get("output_tokens_details") + if not isinstance(output_details, dict): + output_details = {} cached_tokens += as_int(input_details.get("cached_tokens")) reasoning_tokens += as_int(usage.get("reasoning") or output_details.get("reasoning_tokens")) @@ -533,8 +590,9 @@ def _read_transcript_events(task_id: str, out_dir: Path) -> list[dict]: if tpath.exists(): for line in tpath.read_text().splitlines(): try: - events.append(json.loads(line)) - except json.JSONDecodeError: + event = json.loads(line) + events.append(event if isinstance(event, dict) else {"raw": line}) + except (json.JSONDecodeError, RecursionError): events.append({"raw": line}) return events @@ -552,7 +610,9 @@ def _response_from_transcript_events(self, task_id: str, events: list[dict]) -> if event.get("type") != "message": continue - message = event.get("message") or {} + message = event.get("message") + if not isinstance(message, dict): + continue role = message.get("role") if role == "assistant": reasoning = self._reasoning_text(message) @@ -648,6 +708,9 @@ def _collect_transcript(self, task_id: str, out_dir: Path, run_id: str) -> tuple dest.parent.mkdir(parents=True, exist_ok=True) try: shutil.copytree(tdir, dest, dirs_exist_ok=True) + session_store = out_dir / "openclaw_sessions" + if session_store.exists(): + shutil.copytree(session_store, dest / "openclaw_sessions", dirs_exist_ok=True) archive = str(dest) except OSError: pass @@ -674,6 +737,81 @@ def _empty_response(self, task_id: str) -> NeMoGymResponse: tool_choice="auto", ) + def _build_observations( + self, + body: PinchBenchRunRequest, + response: NeMoGymResponse, + transcript_events: list[dict[str, Any]], + out_dir: Path, + run_id: str, + ) -> AgentObservationBundle: + try: + transcript_available = any(event.get("type") == "message" for event in transcript_events) + request_input = body.responses_create_params.input + request_items = ( + [NeMoGymEasyInputMessage(role="user", content=request_input)] + if isinstance(request_input, str) + else list(request_input) + ) + observed_output: list[Any] = [] + if transcript_available: + observed_output = list(response.output) + if ( + len(observed_output) == 1 + and getattr(observed_output[0], "type", None) == "message" + and not self._content_text(getattr(observed_output[0], "content", None)) + ): + observed_output = [] + + root_session_id = next( + ( + event.get("id") + for event in transcript_events + if event.get("type") == "session" and isinstance(event.get("id"), str) + ), + run_id, + ) + session_tree, tree_gaps = discover_openclaw_session_tree( + out_dir / "openclaw_sessions" / "agents", + root_session_id, + ) + if session_tree: + tree_inputs = [] + for invocation_id, parent_id, events in session_tree: + conversation = openclaw_session_conversation( + events, + input_items=request_items if parent_id is None else None, + fallback_output=observed_output if parent_id is None else None, + ) + tree_inputs.append((invocation_id, parent_id, conversation, events)) + observations = build_openclaw_observation_tree( + tree_inputs, + model_ref=self.config.model_server, + ) + else: + conversation = openclaw_session_conversation( + transcript_events, + input_items=request_items, + fallback_output=observed_output, + ) + observations = build_openclaw_observations( + root_session_id, + conversation, + transcript_events, + transcript_available=transcript_available, + model_ref=self.config.model_server, + ) + if any(gap.code == "subagent_hierarchy_unavailable" for gap in tree_gaps): + observations.gaps = [gap for gap in observations.gaps if gap.code != "subagent_hierarchy_unavailable"] + observations.gaps.extend(tree_gaps) + except Exception: + LOG.exception("failed to build OpenClaw observations") + observations = AgentObservationBundle( + source=OPENCLAW_OBSERVATION_SOURCE, + gaps=[ObservationGap(code="observation_capture_failed")], + ) + return observations + async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None) -> PinchBenchVerifyResponse: record = body.model_dump() meta = record.get("verifier_metadata") or {} @@ -690,12 +828,15 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None transcript_events: list = [] archive_path = "" non_clean_exit_rc: int | None = None + observations: Optional[AgentObservationBundle] = None + rollout_id = self.rollout_id_from_run(body) + observe = rollout_id is not None try: async with self._sem: - non_clean_exit_rc = await self._run_in_sandbox(task_id, out_dir) + non_clean_exit_rc = await self._run_in_sandbox(task_id, out_dir, rollout_id=rollout_id) result = self._parse_result(task_id, out_dir) - response = self._response_from_transcript(task_id, out_dir) transcript_events, archive_path = self._collect_transcript(task_id, out_dir, run_id) + response = self._response_from_transcript_events(task_id, transcript_events) except Exception as exc: # noqa: BLE001 -- one task error must not abort the batch failure_class = _classify_task_failure(exc) print(f"[pinchbench-{failure_class}] {task_id}: {type(exc).__name__}: {exc}", flush=True) @@ -712,6 +853,14 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None elif failure_class == "timeout_exceeded": routing[NG_TERMINAL_KEY] = True finally: + if observe: + observations = self._build_observations( + body, + response, + transcript_events, + out_dir, + run_id, + ) shutil.rmtree(out_dir, ignore_errors=True) raw_rollout: dict = { @@ -732,6 +881,7 @@ async def run(self, body: PinchBenchRunRequest = Body(), request: Request = None grading_notes=result["notes"], status=result["status"], raw_rollout=raw_rollout, + **({"ng_agent_observations": observations.model_dump(mode="json")} if observations is not None else {}), **routing, ) diff --git a/responses_api_agents/pinchbench/configs/pinchbench.yaml b/responses_api_agents/pinchbench/configs/pinchbench.yaml index 6c89d78ccd..9a3ec6462f 100644 --- a/responses_api_agents/pinchbench/configs/pinchbench.yaml +++ b/responses_api_agents/pinchbench/configs/pinchbench.yaml @@ -31,10 +31,12 @@ pinchbench_agent: model_base_url: ${model_base_url} model_api_key: ${model_api_key} model_name: ${model_name} + model_server: null # Judge for hybrid / llm_judge tasks (OpenAI-compatible endpoint). judge_model: ${judge_model} judge_base_url: ${judge_base_url} judge_api_key: ${judge_api_key} + judge_model_server: null web_search_provider: brave brave_api_key: ${brave_api_key} tavily_api_key: ${tavily_api_key} diff --git a/responses_api_agents/pinchbench/run_task.sh b/responses_api_agents/pinchbench/run_task.sh index 99d14f6755..25b9c54e83 100644 --- a/responses_api_agents/pinchbench/run_task.sh +++ b/responses_api_agents/pinchbench/run_task.sh @@ -80,6 +80,17 @@ rc=$? kill "$GW_PID" 2>/dev/null || true kill -9 "$GW_PID" 2>/dev/null || true +if [ "${NEMO_GYM_OBSERVABILITY_ENABLED:-0}" = "1" ]; then + # Agent directories also contain credentials; retain only session metadata and transcripts. + for sessions_dir in "$HOME"/.openclaw/agents/*/sessions; do + [ -d "$sessions_dir" ] || continue + agent_id=$(basename "$(dirname "$sessions_dir")") + dest="$OUT/openclaw_sessions/agents/$agent_id" + mkdir -p "$dest" + cp -a "$sessions_dir" "$dest/" 2>/dev/null || true + done +fi + # Package $OUT so the host can download it (Sandbox API pulls one file). Tar to a temp # path then move in, so the archive never tries to include itself. tar czf "$TMPDIR/out.tgz" -C "$OUT" . 2>/dev/null || true diff --git a/responses_api_agents/pinchbench/tests/test_app.py b/responses_api_agents/pinchbench/tests/test_app.py index 88906675d9..0900ae09b8 100644 --- a/responses_api_agents/pinchbench/tests/test_app.py +++ b/responses_api_agents/pinchbench/tests/test_app.py @@ -23,6 +23,9 @@ import pytest +from nemo_gym.config_types import ModelServerRef +from nemo_gym.rollout_observability import AgentInvocation, ToolCallObservation +from nemo_gym.sandbox import SandboxExecResult from nemo_gym.server_utils import ServerClient from responses_api_agents.pinchbench.app import ( NG_FAILURE_CLASS_KEY, @@ -30,6 +33,7 @@ NG_TERMINAL_KEY, PinchBenchAgent, PinchBenchAgentConfig, + PinchBenchRunRequest, SandboxKilledError, _classify_task_failure, ) @@ -57,6 +61,10 @@ def make_agent(**over) -> PinchBenchAgent: return PinchBenchAgent(config=make_config(**over), server_client=MagicMock(spec=ServerClient)) +def _records(bundle, record_type): + return [record for record in bundle.records if isinstance(record, record_type)] + + def test_sanity_construct(): agent = make_agent() assert agent.config.task_timeout_s == 1800 @@ -76,6 +84,25 @@ def test_task_env_gateway_mode(): assert env["MODEL_NAME"] == "vendor/model" assert env["JUDGE_BASE_URL"] == "http://endpoint/v1" assert env["BRAVE_API_KEY"] == "brave-key" + assert "NEMO_GYM_OBSERVABILITY_ENABLED" not in env + assert make_agent()._task_env("task_x", "1-2")["NEMO_GYM_OBSERVABILITY_ENABLED"] == "1" + + +def test_task_env_prefixes_configured_gym_model_servers(): + agent = make_agent( + model_server=ModelServerRef(type="responses_api_models", name="policy"), + judge_model_server=ModelServerRef(type="responses_api_models", name="judge"), + ) + with patch.object( + PinchBenchAgent, + "resolve_model_base_url", + side_effect=lambda _self, name, rollout_id: f"http://{name}/ng-rollout/{rollout_id}/v1", + autospec=True, + ): + env = agent._task_env("task_x", "7-2") + + assert env["MODEL_BASE_URL"] == "http://policy/ng-rollout/7-2/v1" + assert env["JUDGE_BASE_URL"] == "http://judge/ng-rollout/7-2/v1" def test_direct_exec_wrapper_sets_provider_and_agent_timeout_ceiling(tmp_path): @@ -165,6 +192,7 @@ def test_response_from_transcript(tmp_path): (tdir / "task_x.jsonl").write_text("\n".join(json.dumps(e) for e in events)) resp = make_agent()._response_from_transcript("task_x", tmp_path) assert resp.output[0].content[0].text == "Done." + assert make_agent()._content_text(resp.output[0].content) == "Done." def test_response_from_transcript_common_output_items_and_usage(tmp_path): @@ -244,7 +272,7 @@ async def test_run_returns_zero_on_failure_never_raises(tmp_path, monkeypatch): otherwise ng_collect_rollouts (fail-fast) aborts the whole collection.""" agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "arch")) - async def boom(task_id, out_dir): + async def boom(task_id, out_dir, rollout_id=None): raise RuntimeError("sandbox exploded") monkeypatch.setattr(agent, "_run_in_sandbox", boom) @@ -270,6 +298,34 @@ def _run_body(task_id="task_x"): return body +def _observed_run_body(task_id="task_x"): + return PinchBenchRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "verifier_metadata": {"task_id": task_id}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + +def test_hierarchy_discovery_reports_specific_root_gap(tmp_path): + sessions_dir = tmp_path / "openclaw_sessions" / "agents" / "main" / "sessions" + sessions_dir.mkdir(parents=True) + (sessions_dir / "sessions.json").write_text(json.dumps({"other-root": {"sessionId": "other-session"}})) + + observations = make_agent()._build_observations( + _observed_run_body(), + make_agent()._empty_response("task_x"), + [], + tmp_path, + "expected-session", + ) + + hierarchy_gaps = [gap for gap in observations.gaps if gap.code == "subagent_hierarchy_unavailable"] + assert [gap.detail for gap in hierarchy_gaps] == ["root_session_not_found"] + + @pytest.mark.asyncio @pytest.mark.parametrize( "exc,expected_class,no_persist,terminal", @@ -282,7 +338,7 @@ def _run_body(task_id="task_x"): async def test_failure_routing_sentinels(exc, expected_class, no_persist, terminal, tmp_path, monkeypatch): agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "arch")) - async def fail(task_id, out_dir): + async def fail(task_id, out_dir, rollout_id=None): raise exc monkeypatch.setattr(agent, "_run_in_sandbox", fail) @@ -297,7 +353,7 @@ async def test_successful_task_carries_no_routing_sentinels(tmp_path, monkeypatc """Scored rollouts must keep landing in the main jsonl (no sentinel keys).""" agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "arch")) - async def ok(task_id, out_dir): + async def ok(task_id, out_dir, rollout_id=None): return None monkeypatch.setattr(agent, "_run_in_sandbox", ok) @@ -312,15 +368,133 @@ async def ok(task_id, out_dir): "status": "success", }, ) - monkeypatch.setattr(agent, "_response_from_transcript", lambda task_id, out_dir: agent._empty_response(task_id)) monkeypatch.setattr(agent, "_collect_transcript", lambda task_id, out_dir, run_id: ([], "")) 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_correlated_openclaw_observations(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} + rollout_ids = [] + + async def run_in_sandbox(task_id, out_dir, rollout_id=None): + rollout_ids.append(rollout_id) + transcript_dir = out_dir / "0001_transcripts" + transcript_dir.mkdir(parents=True) + events = [ + {"type": "session", "id": "session-1"}, + { + "type": "message", + "message": { + "role": "assistant", + "content": [ + {"type": "reasoning", "text": "need to search"}, + {"type": "toolCall", "id": "call-1", "name": "search"}, + ], + }, + }, + { + "type": "message", + "message": { + "role": "toolResult", + "toolCallId": "call-1", + "content": [{"type": "text", "text": "result"}], + "timestamp": 1_750_000_002_000, + "details": {"durationMs": 500, "status": "completed"}, + }, + }, + ] + (transcript_dir / f"{task_id}.jsonl").write_text("\n".join(json.dumps(event) for event in events)) + + monkeypatch.setattr(agent, "_run_in_sandbox", run_in_sandbox) + monkeypatch.setattr( + agent, + "_parse_result", + lambda *_: { + "reward": 1.0, + "grading_type": "automated", + "breakdown": {}, + "notes": "ok", + "status": "success", + }, + ) + + result = await agent.run(body=_observed_run_body()) + + assert rollout_ids == ["1-2"] + assert result.ng_agent_observations.source == "openclaw" + assert _records(result.ng_agent_observations, AgentInvocation)[0].invocation_id == "session-1" + assert _records(result.ng_agent_observations, ToolCallObservation)[0].duration_ms == 500 + + +@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, rollout_id=None): + return None + + monkeypatch.setattr(agent, "_run_in_sandbox", run_in_sandbox) + monkeypatch.setattr( + agent, + "_parse_result", + lambda *_: { + "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")), + ) + + result = await agent.run(body=_observed_run_body()) + + assert result.reward == 1.0 + assert [gap.code for gap in result.ng_agent_observations.gaps] == ["observation_capture_failed"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error_type,exception_type", + [("timeout", TimeoutError), ("sandbox", RuntimeError)], +) +async def test_managed_sandbox_exec_errors_skip_download(tmp_path, monkeypatch, error_type, exception_type): + agent = make_agent(sandbox_provider={"opensandbox": {}}) + + class FailedSandbox: + download_called = False + + async def start(self, spec): + return None + + async def exec(self, command, timeout_s): + return SandboxExecResult(stdout=None, stderr="provider failed", return_code=125, error_type=error_type) + + async def download(self, source, target): + self.download_called = True + + async def stop(self): + return None + + sandbox = FailedSandbox() + monkeypatch.setattr("responses_api_agents.pinchbench.app.AsyncSandbox", lambda _: sandbox) + + with pytest.raises(exception_type): + await agent._run_in_sandbox("task_x", tmp_path) + assert sandbox.download_called is False + + @pytest.mark.parametrize( "exc,expected", [ @@ -378,7 +552,7 @@ async def test_run_raises_on_missing_task_id(tmp_path): async def test_non_clean_exit_rc_present_in_raw_rollout(tmp_path, monkeypatch): agent = make_agent(work_root=str(tmp_path / "work"), transcripts_dir=str(tmp_path / "arch")) - async def non_clean(task_id, out_dir): + async def non_clean(task_id, out_dir, rollout_id=None): return 1 monkeypatch.setattr(agent, "_run_in_sandbox", non_clean) @@ -451,16 +625,27 @@ def test_response_from_transcript_uses_details_when_content_empty(tmp_path): def test_read_transcript_events_tolerates_malformed_json(tmp_path): tdir = tmp_path / "0001_transcripts" tdir.mkdir() - (tdir / "task_x.jsonl").write_text('{"valid": true}\nNOT JSON\n{"also": "valid"}') + (tdir / "task_x.jsonl").write_text('{"valid": true}\nNOT JSON\nnull\n[]\n{"also": "valid"}') events = make_agent()._read_transcript_events("task_x", tmp_path) - assert len(events) == 3 - assert "raw" in events[1] + assert len(events) == 5 + assert all("raw" in events[index] for index in (1, 2, 3)) + + +def test_transcript_parsing_ignores_non_object_message_and_usage(): + events = [ + {"type": "message", "message": "invalid"}, + {"type": "message", "message": {"role": "assistant", "content": "Done.", "usage": "invalid"}}, + ] + + response = make_agent()._response_from_transcript_events("task_x", events) + + assert response.output[0].content[0].text == "Done." + assert response.usage.total_tokens == 0 def test_tool_call_arguments_with_dict_is_json_serialized(): - assert json.loads(make_agent()._tool_call_arguments({"name": "search", "arguments": {"q": "AAPL"}})) == { - "q": "AAPL" - } + block = {"name": "search", "arguments": {"q": "AAPL"}, "partialArgs": '{"q":"stale"}'} + assert json.loads(make_agent()._tool_call_arguments(block)) == {"q": "AAPL"} def test_tool_call_arguments_absent_returns_empty_object():