diff --git a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx index ffc94ab1f8..f49eac6bf3 100644 --- a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx +++ b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx @@ -60,7 +60,7 @@ For C1, C2, and C4, `V` evaluates the correlated Gym Model Server path; direct-p | `mini_swe_agent_2` | V | V | X | V | X | X | X | | `non_executing_simple_agent` | V | V | X | V | X | X | X | | `openclaw_agent` | V | V | X | O | V | V | V | -| `opencode_agent` | X | X | X | X | X | X | X | +| `opencode_agent` | V | V | X | V | V | V | V | | `osworld_agent` | O | O | X | O | X | X | X | | `pi_agent` | V | V | X | V | V | V | V | | `proof_refinement_agent` | V | V | X | V | X | X | X | diff --git a/responses_api_agents/opencode_agent/app.py b/responses_api_agents/opencode_agent/app.py index e628ac32bc..e03b48b67b 100644 --- a/responses_api_agents/opencode_agent/app.py +++ b/responses_api_agents/opencode_agent/app.py @@ -22,6 +22,7 @@ import shutil import sqlite3 from asyncio import Semaphore +from collections.abc import Mapping from pathlib import Path from time import time from typing import Any, Optional @@ -37,7 +38,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, @@ -48,17 +48,338 @@ NeMoGymResponseOutputMessage, NeMoGymResponseOutputText, NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, NeMoGymResponseUsage, + NeMoGymSummary, +) +from nemo_gym.rollout_observability import ( + AgentEpisode, + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ObservationGap, + ToolCallObservation, ) from nemo_gym.server_utils import get_response_json, raise_for_status from responses_api_agents.opencode_agent.setup_opencode import ensure_opencode LOG = logging.getLogger(__name__) +_INTERNAL_OBSERVATIONS_KEY = "_ng_agent_observations" + + +def _load_json(value: Any) -> dict[str, Any]: + try: + parsed = json.loads(value) + except (json.JSONDecodeError, TypeError): + return {} + return parsed if isinstance(parsed, dict) else {} + + +def _milliseconds(value: Any) -> Optional[float]: + """Convert OpenCode's Date.now()-based epoch milliseconds to seconds.""" + if not isinstance(value, (int, float)) or isinstance(value, bool) or value < 0: + return None + return float(value) / 1000 + + +def _parse_opencode_session(db_path: Path, fallback_invocation_id: str) -> AgentObservationBundle: + """Read OpenCode's persisted session tree before its workspace is removed.""" + if not db_path.is_file(): + return AgentObservationBundle( + source="opencode", + records=[AgentInvocation(invocation_id=fallback_invocation_id)], + gaps=[ + ObservationGap(code="agent_artifact_unavailable"), + ObservationGap(code="agent_transcript_unavailable"), + ObservationGap(code="model_call_ownership_unavailable"), + ], + ) + + con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + con.row_factory = sqlite3.Row + try: + session_rows = con.execute( + "select id, parent_id, time_created from session order by time_created, id" + ).fetchall() + message_rows = con.execute( + "select id, session_id, data, time_created from message order by time_created, id" + ).fetchall() + part_rows = con.execute( + "select id, message_id, session_id, data, time_created from part order by time_created, id" + ).fetchall() + finally: + con.close() + + messages = {row["id"]: _load_json(row["data"]) for row in message_rows} + message_sessions = {row["id"]: row["session_id"] for row in message_rows} + conversations: dict[str, list[Any]] = {row["id"]: [] for row in session_rows} + invocation_status: dict[str, str] = {row["id"]: "unknown" for row in session_rows} + tools: list[ToolCallObservation] = [] + child_tools: dict[str, set[str]] = {} + child_status: dict[str, str] = {} + compaction_parts: list[tuple[str, str, float | None, dict[str, Any]]] = [] + gaps: list[ObservationGap] = [] + summary_text: dict[str, list[str]] = {} + summaries_by_parent: dict[str, list[str]] = {} + first_item_id_by_message: dict[tuple[str, str], str] = {} + + for row in message_rows: + message = messages[row["id"]] + session_id = row["session_id"] + if not isinstance(session_id, str) or session_id not in invocation_status: + gaps.append(ObservationGap(code="agent_artifact_record_unowned", detail=row["id"])) + elif message.get("role") == "assistant": + if isinstance(message.get("error"), dict): + invocation_status[session_id] = "failed" + message_time = message.get("time") if isinstance(message.get("time"), dict) else {} + if invocation_status[session_id] != "failed" and _milliseconds(message_time.get("completed")) is not None: + invocation_status[session_id] = "completed" + if message.get("summary") is True: + summary_text[row["id"]] = [] + parent_id = message.get("parentID") + if isinstance(parent_id, str): + summaries_by_parent.setdefault(parent_id, []).append(row["id"]) + + for row in part_rows: + part = _load_json(row["data"]) + if not part: + gaps.append(ObservationGap(code="agent_artifact_record_unparseable")) + continue + ptype = part.get("type") + message_id = row["message_id"] + message = messages.get(message_id, {}) + session_id = row["session_id"] or message_sessions.get(message_id) + if not isinstance(session_id, str): + gaps.append(ObservationGap(code="agent_artifact_record_unowned")) + continue + conversation = conversations.setdefault(session_id, []) + role = message.get("role") + + if ptype == "step-finish": + continue + + text = part.get("text") + if ptype == "text" and isinstance(text, str) and text.strip(): + if message_id in summary_text: + summary_text[message_id].append(text) + if role == "user" and part.get("ignored") is not True: + conversation.append(NeMoGymEasyInputMessage(role="user", content=text)) + elif role == "assistant": + item = NeMoGymResponseOutputMessage( + id=row["id"], + content=[NeMoGymResponseOutputText(type="output_text", text=text, annotations=[])], + role="assistant", + status="completed", + type="message", + ) + conversation.append(item) + first_item_id_by_message.setdefault((session_id, message_id), row["id"]) + continue + if ptype == "reasoning" and role == "assistant" and isinstance(text, str) and text.strip(): + conversation.append( + NeMoGymResponseReasoningItem( + id=row["id"], + summary=[NeMoGymSummary(type="summary_text", text=text)], + ) + ) + first_item_id_by_message.setdefault((session_id, message_id), row["id"]) + continue + if ptype == "tool" and role == "assistant": + state = part.get("state") if isinstance(part.get("state"), dict) else {} + native_call_id = part.get("callID") + observed_call_id = native_call_id if isinstance(native_call_id, str) and native_call_id else None + call_id = observed_call_id or f"call-{uuid4().hex[:8]}" + tool_input = state.get("input") or {} + arguments = json.dumps(tool_input) if isinstance(tool_input, (dict, list)) else str(tool_input) + native_status = state.get("status") + response_status = "completed" if native_status == "completed" else "incomplete" + call = NeMoGymResponseFunctionToolCall( + arguments=arguments, + call_id=call_id, + name=part.get("tool", ""), + type="function_call", + id=call_id, + status=response_status, + ) + conversation.append(call) + first_item_id_by_message.setdefault((session_id, message_id), call_id) + native_time = state.get("time") if isinstance(state.get("time"), dict) else {} + # OpenCode retains raw output in SQLite but substitutes this literal in later model inputs after pruning. + if native_status == "completed" and native_time.get("compacted") is not None: + observed_tool_output = "[Old tool result content cleared]" + else: + observed_tool_output = state.get("output") if state.get("output") is not None else state.get("error") + if observed_tool_output is not None: + result = NeMoGymFunctionCallOutput( + type="function_call_output", + call_id=call_id, + output=str(observed_tool_output), + status=response_status, + ) + conversation.append(result) + + native_start = native_time.get("start") + native_end = native_time.get("end") + valid_interval = ( + isinstance(native_start, (int, float)) + and not isinstance(native_start, bool) + and isinstance(native_end, (int, float)) + and not isinstance(native_end, bool) + and native_end >= native_start + ) + started_at = _milliseconds(native_start) if valid_interval else None + completed_at = _milliseconds(native_end) if valid_interval else None + duration_ms = float(native_end - native_start) if valid_interval else None + status = { + "completed": "completed", + "error": "failed", + "running": "incomplete", + "pending": "incomplete", + }.get(native_status, "unknown") + if observed_call_id is not None: + tools.append( + ToolCallObservation( + invocation_id=session_id, + tool_call_id=observed_call_id, + tool_name=part.get("tool") if isinstance(part.get("tool"), str) else None, + started_at=started_at, + completed_at=completed_at, + duration_ms=duration_ms, + timing_source="artifact" if started_at is not None else None, + status=status, + error_type="tool_error" if native_status == "error" else None, + ) + ) + else: + gaps.append( + ObservationGap( + code="tool_call_identity_unavailable", + invocation_id=session_id, + detail=row["id"], + ) + ) + if observed_call_id is not None and ( + started_at is None or (native_status in {"completed", "error"} and completed_at is None) + ): + gaps.append( + ObservationGap( + code="tool_timing_unavailable", + invocation_id=session_id, + detail=observed_call_id, + ) + ) + metadata = state.get("metadata") if isinstance(state.get("metadata"), dict) else {} + if not metadata and isinstance(part.get("metadata"), dict): + metadata = part["metadata"] + child_id = metadata.get("sessionId") + if isinstance(child_id, str) and observed_call_id is not None: + child_tools.setdefault(child_id, set()).add(observed_call_id) + child_status[child_id] = status + continue + if ptype == "compaction": + compaction_parts.append((session_id, message_id, _milliseconds(row["time_created"]), part)) + continue + + compactions: list[ContextCompactionObservation] = [] + for session_id, message_id, observed_at, part in compaction_parts: + summary_ids = summaries_by_parent.get(message_id, []) + summary = "\n".join(summary_text.get(summary_ids[0], [])) if len(summary_ids) == 1 else None + if len(summary_ids) > 1: + gaps.append( + ObservationGap( + code="compaction_summary_ambiguous", + invocation_id=session_id, + ) + ) + trigger = "overflow" if part.get("overflow") is True else "automatic" if part.get("auto") is True else "manual" + tail_start_id = part.get("tail_start_id") if isinstance(part.get("tail_start_id"), str) else None + first_kept_item_id = ( + first_item_id_by_message.get((session_id, tail_start_id)) if tail_start_id is not None else None + ) + compactions.append( + ContextCompactionObservation( + invocation_id=session_id, + observed_at=observed_at, + trigger=trigger, + outcome="completed" if summary else "unknown", + summary=summary, + first_kept_item_id=first_kept_item_id, + ) + ) + if tail_start_id is not None and first_kept_item_id is None: + gaps.append( + ObservationGap( + code="compaction_first_kept_item_unavailable", + invocation_id=session_id, + detail=tail_start_id, + ) + ) + if not summary: + gaps.append(ObservationGap(code="compaction_summary_unavailable", invocation_id=session_id)) + gaps.append(ObservationGap(code="compaction_token_counts_unavailable", invocation_id=session_id)) + gaps.append( + ObservationGap( + code="compaction_model_call_boundary_unavailable", + invocation_id=session_id, + ) + ) + + session_ids = {row["id"] for row in session_rows} + invocations = [] + for row in session_rows: + invocation_id = row["id"] + parent_id = row["parent_id"] + spawn_candidates = child_tools.get(invocation_id, set()) + if parent_id is not None and parent_id not in session_ids: + gaps.append( + ObservationGap( + code="subagent_parent_unavailable", + invocation_id=invocation_id, + detail=parent_id, + ) + ) + if len(spawn_candidates) > 1: + gaps.append( + ObservationGap( + code="subagent_spawn_ambiguous", + invocation_id=invocation_id, + ) + ) + elif parent_id is not None and not spawn_candidates: + gaps.append( + ObservationGap( + code="subagent_spawn_tool_unavailable", + invocation_id=invocation_id, + ) + ) + invocations.append( + AgentInvocation( + invocation_id=invocation_id, + parent_invocation_id=parent_id, + spawned_by_tool_call_id=next(iter(spawn_candidates)) if len(spawn_candidates) == 1 else None, + status=( + invocation_status.get(invocation_id, "unknown") + if invocation_status.get(invocation_id, "unknown") != "unknown" + else child_status.get(invocation_id, "unknown") + ), + conversation=conversations.get(invocation_id, []), + ) + ) + if not invocations: + invocations = [AgentInvocation(invocation_id=fallback_invocation_id)] + gaps.append(ObservationGap(code="agent_transcript_unavailable")) + gaps.append(ObservationGap(code="model_call_ownership_unavailable")) + + return AgentObservationBundle( + source="opencode", + records=[*invocations, *tools, *compactions], + gaps=gaps, + ) def parse_opencode_session(db_path: Path) -> tuple[list[Any], dict[str, int]]: - """convert the sqlite session db into Gym format including tool calls""" + """Convert an OpenCode session database into the existing Gym response shape.""" output_items: list[Any] = [] input_tokens = 0 output_tokens = 0 @@ -68,29 +389,20 @@ def parse_opencode_session(db_path: Path) -> tuple[list[Any], dict[str, int]]: con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) con.row_factory = sqlite3.Row try: - roles = {r["id"]: json.loads(r["data"]).get("role") for r in con.execute("select id, data from message")} + roles = {row["id"]: _load_json(row["data"]).get("role") for row in con.execute("select id, data from message")} rows = con.execute("select message_id, data from part order by time_created").fetchall() finally: con.close() - for r in rows: - try: - part = json.loads(r["data"]) - except (json.JSONDecodeError, TypeError): - continue + for row in rows: + part = _load_json(row["data"]) ptype = part.get("type") - if ptype == "step-finish": tokens = part.get("tokens") or {} cache = tokens.get("cache") or {} input_tokens += int(tokens.get("input") or 0) + int(cache.get("read") or 0) output_tokens += int(tokens.get("output") or 0) - continue - - if roles.get(r["message_id"]) != "assistant": - continue - - if ptype == "text" and (part.get("text") or "").strip(): + elif roles.get(row["message_id"]) == "assistant" and ptype == "text" and (part.get("text") or "").strip(): output_items.append( NeMoGymResponseOutputMessage( id=f"msg-{len(output_items)}", @@ -100,7 +412,7 @@ def parse_opencode_session(db_path: Path) -> tuple[list[Any], dict[str, int]]: type="message", ) ) - elif ptype == "tool": + elif roles.get(row["message_id"]) == "assistant" and ptype == "tool": state = part.get("state") or {} call_id = part.get("callID") or f"call-{uuid4().hex[:8]}" tool_input = state.get("input") or {} @@ -120,7 +432,7 @@ def parse_opencode_session(db_path: Path) -> tuple[list[Any], dict[str, int]]: NeMoGymFunctionCallOutput( type="function_call_output", call_id=call_id, - output=str(state.get("output")), + output=str(state["output"]), status="completed", ) ) @@ -195,6 +507,9 @@ class OpenCodeAgentVerifyResponse(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 OpenCodeAgent(SimpleResponsesAPIAgent): @@ -237,26 +552,21 @@ def _repo_dir(self, fallback: Path) -> Path: root.mkdir(parents=True, exist_ok=True) return root - 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 - def _build_opencode_config(self) -> dict[str, Any]: + def _build_opencode_config(self, rollout_id: Optional[str] = None) -> dict[str, Any]: config = self._deep_merge({}, copy.deepcopy(self.config.opencode_config)) if self.config.model_server: providers = config.setdefault("provider", {}) nemo = providers.setdefault("nemo", {"npm": "@ai-sdk/openai-compatible"}) nemo.setdefault("options", {}).update( - {"baseURL": self._resolve_model_base_url(), "apiKey": "EMPTY"} # pragma: allowlist secret + {"baseURL": self._resolve_model_base_url(rollout_id), "apiKey": "EMPTY"} # pragma: allowlist secret ) model = nemo.setdefault("models", {}).get(self.config.model, {}) self._deep_merge( @@ -270,34 +580,44 @@ def _build_opencode_config(self) -> dict[str, Any]: nemo["models"] = {self.config.model: model} return config - def _write_opencode_config(self, work_dir: Path) -> None: - config = self._build_opencode_config() + def _write_opencode_config(self, work_dir: Path, rollout_id: Optional[str] = None) -> None: + config = self._build_opencode_config(rollout_id) if not config: return (work_dir / "opencode.json").write_text(json.dumps(config, indent=2)) - def _env(self, data_home: str) -> dict[str, str]: + def _env(self, data_home: str, rollout_id: Optional[str] = None) -> dict[str, str]: env = {**os.environ, "XDG_DATA_HOME": data_home} - base_url = self._resolve_model_base_url() if self.config.model_server else self.config.openai_base_url + base_url = ( + self._resolve_model_base_url(rollout_id) if self.config.model_server else self.config.openai_base_url + ) api_key = "EMPTY" if self.config.model_server else self.config.openai_api_key # pragma: allowlist secret if base_url: env["OPENAI_BASE_URL"] = base_url if api_key: env["OPENAI_API_KEY"] = api_key env.update({k: v for k, v in self.config.env.items() if v}) + if self.config.model_server is not None: + env["OPENAI_BASE_URL"] = base_url + env["OPENAI_API_KEY"] = "EMPTY" # pragma: allowlist secret return env async def _run_opencode( - self, instruction: str, system_prompt: Optional[str] - ) -> tuple[list[Any], dict[str, int], str]: - """Run one headless opencode run. Returns (output_items, usage, model_name).""" + self, + instruction: str, + system_prompt: Optional[str], + *, + rollout_id: Optional[str] = None, + collect_observations: bool = True, + ) -> tuple[list[Any], dict[str, int], str, AgentObservationBundle]: + """Run one headless OpenCode session and read its persisted artifact.""" prompt = instruction if not system_prompt else f"{system_prompt}\n\n{instruction}" work_dir = self._workspace_root() project_dir = self._repo_dir(work_dir) data_home = work_dir / ".opencode-data" data_home.mkdir(parents=True, exist_ok=True) - self._write_opencode_config(project_dir) - env = self._env(str(data_home)) + self._write_opencode_config(project_dir, rollout_id) + env = self._env(str(data_home), rollout_id) cmd = [*self.config.command_parts, "run", "-m", self._effective_model(), "--dir", str(project_dir)] if self.config.thinking: @@ -306,6 +626,7 @@ async def _run_opencode( cmd.append(prompt) try: + timed_out = False proc = await asyncio.create_subprocess_exec( *cmd, cwd=str(project_dir), @@ -317,23 +638,52 @@ async def _run_opencode( _, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.config.timeout) except asyncio.TimeoutError: proc.kill() - await proc.communicate() + _, stderr = await proc.communicate() + timed_out = True LOG.warning("opencode timed out after %ds", self.config.timeout) - return [], {"input_tokens": 0, "output_tokens": 0}, self.config.model if proc.returncode not in (0, None): LOG.warning("opencode exited %d: %s", proc.returncode, stderr.decode(errors="replace")[:500]) - output_items, usage = parse_opencode_session(data_home / "opencode" / "opencode.db") - return output_items, usage, self.config.model + db_path = data_home / "opencode" / "opencode.db" + invocation_id = rollout_id or f"opencode-{uuid4().hex}" + output_items, usage = ( + ([], {"input_tokens": 0, "output_tokens": 0}) if timed_out else parse_opencode_session(db_path) + ) + observations = AgentObservationBundle(source="opencode") + if collect_observations: + try: + observations = _parse_opencode_session(db_path, invocation_id) + except Exception: + LOG.exception("failed to read OpenCode session artifact") + observations = AgentObservationBundle( + source="opencode", + records=[AgentInvocation(invocation_id=invocation_id)], + gaps=[ + ObservationGap(code="agent_artifact_unavailable"), + ObservationGap(code="agent_transcript_unavailable"), + ObservationGap(code="model_call_ownership_unavailable"), + ], + ) + run_status = "incomplete" if timed_out else "completed" if proc.returncode == 0 else "failed" + for invocation in observations.records: + if not isinstance(invocation, AgentInvocation): + continue + if invocation.parent_invocation_id is None: + invocation.status = run_status + if timed_out: + observations.gaps.append(ObservationGap(code="agent_run_timeout")) + return output_items, usage, self.config.model, observations finally: shutil.rmtree(work_dir, ignore_errors=True) - async def responses( + async def _create_episode( self, - request: Request, - body: NeMoGymResponseCreateParamsNonStreaming = Body(), - ) -> NeMoGymResponse: + body: NeMoGymResponseCreateParamsNonStreaming, + *, + rollout_id: Optional[str] = None, + collect_observations: bool = True, + ) -> AgentEpisode: body = body.model_copy(deep=True) if isinstance(body.input, str): body.input = [NeMoGymEasyInputMessage(role="user", content=body.input)] @@ -341,8 +691,30 @@ async def responses( user_message, input_system = _extract_instruction(body.input) system_parts = [p for p in [self.config.system_prompt, input_system] if p] system_prompt = "\n\n".join(system_parts) if system_parts else None + prompt = user_message if system_prompt is None else f"{system_prompt}\n\n{user_message}" - output_items, usage, model_name = await self._run_opencode(user_message, system_prompt) + output_items, usage, model_name, observations = await self._run_opencode( + user_message, + system_prompt, + rollout_id=rollout_id, + collect_observations=collect_observations, + ) + if collect_observations: + observations.gaps.append(ObservationGap(code="no_sandbox_runtime")) + + if collect_observations: + root = next( + ( + record + for record in observations.records + if isinstance(record, AgentInvocation) and record.parent_invocation_id is None + ), + None, + ) + if root is not None and not any( + getattr(item, "role", None) in {"user", "system", "developer"} for item in root.conversation + ): + root.conversation = [NeMoGymEasyInputMessage(role="user", content=prompt), *root.conversation] if not any( getattr(item, "type", None) == "message" and getattr(item, "role", None) == "assistant" @@ -362,7 +734,7 @@ async def responses( input_tokens = usage.get("input_tokens", 0) output_tokens = usage.get("output_tokens", 0) - return NeMoGymResponse( + response = NeMoGymResponse( id=f"resp_{uuid4().hex}", created_at=int(time()), model=model_name, @@ -379,6 +751,25 @@ async def responses( total_tokens=input_tokens + output_tokens, ), ) + return AgentEpisode(response=response, observations=observations) + + 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 + episode = await self._create_episode( + body, + rollout_id=rollout_id, + collect_observations=isinstance(rollout_id, str), + ) + if not isinstance(rollout_id, str): + return episode.response + return episode.response.model_copy( + update={_INTERNAL_OBSERVATIONS_KEY: episode.observations.model_dump(mode="json")} + ) async def run(self, request: Request, body: OpenCodeAgentRunRequest) -> OpenCodeAgentVerifyResponse: async with self.sem: @@ -393,15 +784,22 @@ async def run(self, request: Request, body: OpenCodeAgentRunRequest) -> OpenCode 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, @@ -422,7 +820,9 @@ async def run(self, request: Request, body: OpenCodeAgentRunRequest) -> OpenCode naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant" return OpenCodeAgentVerifyResponse.model_validate( - verify_json | {"turns_used": turns, "finished_naturally": naturally} + verify_json + | {"turns_used": turns, "finished_naturally": naturally} + | ({"ng_agent_observations": observations} if observations is not None else {}) ) diff --git a/responses_api_agents/opencode_agent/tests/test_app.py b/responses_api_agents/opencode_agent/tests/test_app.py index 1d61ad726a..56f102f6b3 100644 --- a/responses_api_agents/opencode_agent/tests/test_app.py +++ b/responses_api_agents/opencode_agent/tests/test_app.py @@ -24,14 +24,23 @@ from nemo_gym.openai_utils import ( NeMoGymEasyInputMessage, NeMoGymFunctionCallOutput, + NeMoGymResponseCreateParamsNonStreaming, NeMoGymResponseFunctionToolCall, NeMoGymResponseOutputMessage, ) +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ToolCallObservation, +) from nemo_gym.server_utils import ServerClient from responses_api_agents.opencode_agent.app import ( OpenCodeAgent, OpenCodeAgentConfig, + OpenCodeAgentRunRequest, _extract_instruction, + _parse_opencode_session, parse_opencode_session, ) @@ -47,6 +56,18 @@ def _config(**kwargs) -> OpenCodeAgentConfig: ) +def _invocations(bundle: AgentObservationBundle) -> list[AgentInvocation]: + return [record for record in bundle.records if isinstance(record, AgentInvocation)] + + +def _tool_calls(bundle: AgentObservationBundle) -> list[ToolCallObservation]: + return [record for record in bundle.records if isinstance(record, ToolCallObservation)] + + +def _compactions(bundle: AgentObservationBundle) -> list[ContextCompactionObservation]: + return [record for record in bundle.records if isinstance(record, ContextCompactionObservation)] + + def _make_agent(**kwargs) -> OpenCodeAgent: with patch("responses_api_agents.opencode_agent.app.OpenCodeAgent.model_post_init"): agent = OpenCodeAgent(config=_config(**kwargs), server_client=MagicMock(spec=ServerClient)) @@ -54,20 +75,30 @@ def _make_agent(**kwargs) -> OpenCodeAgent: return agent -def _session_db(tmp_path, messages) -> Path: - """Build a minimal opencode sqlite db. messages is a list of (role, [part_dicts]).""" +def _session_db(tmp_path, messages, sessions=None) -> Path: + """Build the subset of OpenCode's v1.17.11 artifact used by the adapter.""" import sqlite3 db = tmp_path / "opencode.db" con = sqlite3.connect(db) - con.execute("create table message (id text, data text, time_created integer)") - con.execute("create table part (id text, message_id text, data text, time_created integer)") + sessions = sessions or [("root", None)] + con.execute("create table session (id text, parent_id text, time_created integer)") + con.execute("create table message (id text, session_id text, data text, time_created integer)") + con.execute("create table part (id text, message_id text, session_id text, data text, time_created integer)") + for index, (session_id, parent_id) in enumerate(sessions): + con.execute("insert into session values (?,?,?)", (session_id, parent_id, index)) t = 0 - for mi, (role, parts) in enumerate(messages): + for mi, entry in enumerate(messages): + session_id, role_or_message, parts = ("root", *entry) if len(entry) == 2 else entry mid = f"m{mi}" - con.execute("insert into message values (?,?,?)", (mid, json.dumps({"role": role}), mi)) + message = ( + {"role": role_or_message, "time": {"created": mi, "completed": mi + 1}} + if isinstance(role_or_message, str) + else role_or_message + ) + con.execute("insert into message values (?,?,?,?)", (mid, session_id, json.dumps(message), mi)) for p in parts: - con.execute("insert into part values (?,?,?,?)", (f"p{t}", mid, json.dumps(p), t)) + con.execute("insert into part values (?,?,?,?,?)", (f"p{t}", mid, session_id, json.dumps(p), t)) t += 1 con.commit() con.close() @@ -137,7 +168,12 @@ def test_tool_call_and_output(self, tmp_path) -> None: "type": "tool", "callID": "c1", "tool": "bash", - "state": {"input": {"command": "echo 6"}, "output": "6\n"}, + "state": { + "status": "completed", + "input": {"command": "echo 6"}, + "output": "6\n", + "time": {"start": 1000, "end": 1200}, + }, }, {"type": "text", "text": "answer is 6"}, ], @@ -162,6 +198,164 @@ def test_step_finish_usage(self, tmp_path) -> None: assert usage["input_tokens"] == 105 assert usage["output_tokens"] == 20 + def test_preserves_tree_parallel_tools_compaction_and_reasoning(self, tmp_path) -> None: + db = _session_db( + tmp_path, + [ + ( + "root", + "user", + [ + {"type": "text", "text": "solve"}, + {"type": "text", "text": "not replayed", "ignored": True}, + ], + ), + ( + "root", + "assistant", + [ + { + "type": "tool", + "callID": "task-1", + "tool": "task", + "state": { + "status": "completed", + "input": {"prompt": "inspect"}, + "output": "done", + "metadata": {"sessionId": "child"}, + "time": {"start": 1785813051824, "end": 1785813053824}, + }, + }, + { + "type": "tool", + "callID": "bash-1", + "tool": "bash", + "state": { + "status": "completed", + "input": {"command": "pwd"}, + "output": "/workspace", + "time": { + "start": 1785813052000, + "end": 1785813052400, + "compacted": 1785813052500, + }, + }, + }, + ], + ), + ("child", "user", [{"type": "text", "text": "inspect"}]), + ("child", "assistant", [{"type": "reasoning", "text": "checking files"}]), + ( + "child", + "user", + [{"type": "compaction", "auto": True, "overflow": True, "tail_start_id": "m3"}], + ), + ( + "child", + { + "role": "assistant", + "summary": True, + "parentID": "m4", + "time": {"created": 5, "completed": 6}, + }, + [{"type": "text", "text": "condensed context"}], + ), + ], + sessions=[("root", None), ("child", "root")], + ) + + bundle = _parse_opencode_session(db, "fallback") + + root, child = _invocations(bundle) + assert child.parent_invocation_id == root.invocation_id + assert child.spawned_by_tool_call_id == "task-1" + assert any(item.type == "reasoning" for item in child.conversation) + assert all(getattr(item, "content", None) != "not replayed" for item in root.conversation) + assert ( + next( + item.output + for item in root.conversation + if isinstance(item, NeMoGymFunctionCallOutput) and item.call_id == "bash-1" + ) + == "[Old tool result content cleared]" + ) + tools = {tool.tool_call_id: tool for tool in _tool_calls(bundle)} + assert {tool_id: tool.duration_ms for tool_id, tool in tools.items()} == { + "task-1": 2000, + "bash-1": 400, + } + assert tools["task-1"].started_at == 1785813051.824 + assert tools["task-1"].completed_at == 1785813053.824 + compaction = _compactions(bundle)[0] + assert compaction.trigger == "overflow" + assert compaction.summary == "condensed context" + assert compaction.first_kept_item_id == "p5" + assert "compaction_model_call_boundary_unavailable" in {gap.code for gap in bundle.gaps} + + def test_reports_unaddressable_compaction_boundary(self, tmp_path) -> None: + db = _session_db( + tmp_path, + [ + ("user", [{"type": "text", "text": "keep this"}]), + ("user", [{"type": "compaction", "tail_start_id": "m0"}]), + ], + ) + + bundle = _parse_opencode_session(db, "fallback") + + assert _compactions(bundle)[0].first_kept_item_id is None + assert any(gap.code == "compaction_first_kept_item_unavailable" and gap.detail == "m0" for gap in bundle.gaps) + + def test_preserves_parts_owned_by_session_when_message_is_unowned(self, tmp_path) -> None: + import sqlite3 + + db = _session_db(tmp_path, [("assistant", [{"type": "text", "text": "kept"}])]) + con = sqlite3.connect(db) + con.execute("update message set session_id = NULL where id = 'm0'") + con.commit() + con.close() + + bundle = _parse_opencode_session(db, "fallback") + + root = _invocations(bundle)[0] + assert root.status == "unknown" + assert root.conversation[0].content[0].text == "kept" + assert any(gap.code == "agent_artifact_record_unowned" and gap.detail == "m0" for gap in bundle.gaps) + + def test_reports_invalid_timing_and_unresolved_tree_edges(self, tmp_path) -> None: + db = _session_db( + tmp_path, + [ + ( + "root", + "assistant", + [ + { + "type": "tool", + "callID": call_id, + "tool": "task", + "state": { + "status": "completed", + "metadata": {"sessionId": "child"}, + "time": {"start": 2000, "end": 1000}, + }, + } + for call_id in ("task-1", "task-2") + ], + ) + ], + sessions=[("root", None), ("child", "missing-parent")], + ) + + bundle = _parse_opencode_session(db, "fallback") + + assert all(tool.started_at is None and tool.completed_at is None for tool in _tool_calls(bundle)) + assert {gap.code for gap in bundle.gaps} >= { + "tool_timing_unavailable", + "subagent_parent_unavailable", + "subagent_spawn_ambiguous", + } + class TestDeepMerge: def test_nested_merge(self) -> None: @@ -196,6 +390,93 @@ def test_model_server_builds_local_provider(self) -> None: assert provider["models"]["Qwen3.6-35B-A3B"]["limit"]["output"] == 131072 +class TestRolloutObservability: + def test_routes_model_server_without_mutating_config(self, tmp_path: Path) -> None: + opencode_config = {"provider": {"openai": {"options": {"baseURL": "https://api.openai.com/v1"}}}} + agent = _make_agent( + model_server=ModelServerRef(type="responses_api_models", name="policy"), + opencode_config=opencode_config, + env={"OPENAI_BASE_URL": "https://wrong.invalid/v1"}, + ) + + with patch.object(OpenCodeAgent, "resolve_model_base_url", return_value="http://policy/ng-rollout/1-2/v1"): + agent._write_opencode_config(tmp_path, "1-2") + env = agent._env(str(tmp_path), "1-2") + + written = json.loads((tmp_path / "opencode.json").read_text()) + assert written["provider"]["nemo"]["options"]["baseURL"] == "http://policy/ng-rollout/1-2/v1" + assert written["provider"]["openai"]["options"]["baseURL"] == "https://api.openai.com/v1" + assert env["OPENAI_BASE_URL"] == "http://policy/ng-rollout/1-2/v1" + assert env["OPENAI_API_KEY"] == "EMPTY" # pragma: allowlist secret + assert agent.config.opencode_config == opencode_config + + def test_padding_is_not_reported_as_artifact_evidence(self, tmp_path: Path) -> None: + _, usage = parse_opencode_session(tmp_path / "missing.db") + observations = _parse_opencode_session(tmp_path / "missing.db", "1-2") + agent = _make_agent(system_prompt="configured system") + agent._run_opencode = AsyncMock(return_value=([], usage, "model", observations)) + body = NeMoGymResponseCreateParamsNonStreaming( + input=[ + NeMoGymEasyInputMessage(role="system", content="request system"), + NeMoGymEasyInputMessage(role="user", content="old question"), + NeMoGymEasyInputMessage(role="assistant", content="old answer"), + NeMoGymEasyInputMessage(role="user", content="solve"), + ] + ) + + episode = asyncio.run(agent._create_episode(body, rollout_id="1-2")) + + assert episode.response.output + assert agent._run_opencode.await_args.args == ("solve", "configured system\n\nrequest system") + assert _invocations(episode.observations)[0].conversation == [ + NeMoGymEasyInputMessage(role="user", content="configured system\n\nrequest system\n\nsolve") + ] + assert "agent_transcript_unavailable" in {gap.code for gap in episode.observations.gaps} + + def test_run_attaches_artifact_observations_when_enabled(self, tmp_path: Path) -> None: + db = _session_db(tmp_path, [("assistant", [{"type": "text", "text": "done"}])]) + items, usage = parse_opencode_session(db) + observations = _parse_opencode_session(db, "1-2") + agent = _make_agent() + agent.server_client.global_config_dict = {"observability_enabled": True} + agent._run_opencode = AsyncMock(return_value=(items, usage, "model", observations)) + + class Response: + ok = True + cookies = {} + + def __init__(self, payload): + self.payload = payload + + async def read(self): + return json.dumps(self.payload).encode() + + async def post(server_name, url_path, json=None, cookies=None, **kwargs): + if url_path.endswith("/v1/responses"): + response = await agent.responses(MagicMock(path_params={"rollout_id": "1-2"}), json) + return Response(response.model_dump(mode="json")) + return Response(json | {"reward": 1.0}) if url_path == "/verify" else Response({}) + + agent.server_client.post = AsyncMock(side_effect=post) + request = MagicMock(cookies={}) + body = OpenCodeAgentRunRequest.model_validate( + { + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 1, + "_ng_rollout_index": 2, + } + ) + + result = asyncio.run(agent.run(request, body)) + + assert result.ng_agent_observations is not None + assert _invocations(result.ng_agent_observations)[0].conversation + assert agent._run_opencode.await_args.kwargs["rollout_id"] == "1-2" + assert agent.server_client.post.await_args_list[1].kwargs["url_path"] == "/ng-rollout/1-2/v1/responses" + verify_json = agent.server_client.post.await_args_list[2].kwargs["json"] + assert "_ng_agent_observations" not in verify_json["response"] + + class TestRepoDir: def test_creates_configured_repo_dir(self, tmp_path: Path) -> None: repo_dir = tmp_path / "nested" / "repo" @@ -210,6 +491,7 @@ async def test_preserves_configured_repo_and_cleans_workspace(self, tmp_path: Pa process = MagicMock(returncode=0) process.communicate = AsyncMock(return_value=(b"", b"")) agent = _make_agent(repo_dir=str(repo_dir)) + scored = [NeMoGymResponseOutputMessage(id="scored", content=[])] with ( patch.object(agent, "_workspace_root", return_value=workspace), @@ -217,9 +499,22 @@ async def test_preserves_configured_repo_and_cleans_workspace(self, tmp_path: Pa "responses_api_agents.opencode_agent.app.asyncio.create_subprocess_exec", AsyncMock(return_value=process), ), + patch( + "responses_api_agents.opencode_agent.app.parse_opencode_session", + return_value=(scored, {"input_tokens": 1, "output_tokens": 2}), + ), + patch( + "responses_api_agents.opencode_agent.app._parse_opencode_session", + side_effect=ValueError("invalid observation artifact"), + ), ): - await agent._run_opencode("fix the issue", None) + output, usage, _, observations = await agent._run_opencode( + "fix the issue", None, collect_observations=True + ) + assert output == scored + assert usage == {"input_tokens": 1, "output_tokens": 2} + assert "agent_artifact_unavailable" in {gap.code for gap in observations.gaps} assert repo_dir.is_dir() assert not workspace.exists()