diff --git a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx index 039c0293a3..52c4da57c9 100644 --- a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx +++ b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx @@ -62,6 +62,7 @@ For C1, C2, and C4, `V` evaluates the correlated Gym Model Server path; direct-p | `non_executing_simple_agent` | V | V | X | V | X | X | X | | `openclaw_agent` | V | V | X | O | V | V | V | | `opencode_agent` | V | V | X | V | V | V | V | +| `opencode_sandboxed_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 | @@ -70,7 +71,8 @@ For C1, C2, and C4, `V` evaluates the correlated Gym Model Server path; direct-p | `simple_agent` | V | V | O | V | O | X | V | | `speed_bench_agent` | V | V | X | V | X | X | X | | `stirrup_agent` | O | O | X | O | X | X | X | -| `swe_agents` | X | X | X | X | X | X | X | +| `swe_agents` / OpenCode (legacy) | V | V | X | V | X | X | V | +| `swe_agents` / OpenHands (legacy) | X | X | X | O | X | X | V | | `tau2` | V | V | X | V | X | X | X | | `tool_simulation_agent` | V | V | X | V | X | X | X | | `toolsandbox_agent` | V | V | X | V | X | X | X | @@ -81,4 +83,8 @@ CVDP Simple path, Finance and Remote aggregate usage, LabBench image redaction, 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. +OpenCode coverage includes both the standalone producer and the decoupled `opencode_sandboxed_agent` path. The decoupled +path composes agent observations with verifier-sandbox evidence returned by its resources server. Legacy OpenHands retains +one cumulative root conversation, so its model-visible history remains partial; its pinned fork does not route model calls +through a rollout-prefixed Gym Model Server endpoint. C7 requires standardized agent-side trajectory evidence; model HTTP capture alone does not satisfy it. diff --git a/resources_servers/swebench/app.py b/resources_servers/swebench/app.py index 7f51deabd4..84d4a68c0a 100644 --- a/resources_servers/swebench/app.py +++ b/resources_servers/swebench/app.py @@ -17,12 +17,12 @@ from glob import glob from pathlib import Path from shutil import rmtree -from time import time +from time import monotonic, time from traceback import format_exc from typing import Any, Dict, Optional, Tuple from fastapi import Request -from pydantic import BaseModel +from pydantic import BaseModel, Field from swebench.harness.run_evaluation import make_test_spec from swebench.harness.test_spec.test_spec import LATEST, TestSpec @@ -36,6 +36,7 @@ SimpleResourcesServer, ) from nemo_gym.global_config import get_global_config_dict +from nemo_gym.rollout_observability import SandboxObservation from nemo_gym.sandbox import AsyncSandbox, SandboxResources, SandboxSpec from nemo_gym.sandbox.config import resolve_provider_config, resolve_provider_metadata from nemo_gym.server_utils import SESSION_ID_KEY @@ -105,6 +106,10 @@ class SWEBenchVerifyResponse(BaseVerifyResponse): log_dir: str + verifier_sandbox_observation: Optional[SandboxObservation] = Field( + default=None, exclude_if=lambda value: value is None + ) + # @bxyu-nvidia: This is a wrapper that can be passed directly to a very lightly modified version of `run_instance` # The method is almost identical to the original, just with async awaits rather than sync. @@ -114,6 +119,8 @@ class DockerContainer(BaseModel): instance_id: str _inner_container: AsyncSandbox + _eval_return_code: Optional[int] = None + _sandbox_error_type: Optional[str] = None async def exec_run( self, @@ -121,11 +128,17 @@ async def exec_run( workdir: Optional[str] = None, user: Optional[str] = None, ) -> ExecResult: - res = await self._inner_container.exec( - command=command, - cwd=workdir, - user=user, - ) + try: + res = await self._inner_container.exec( + command=command, + cwd=workdir, + user=user, + ) + except Exception as exc: + self._sandbox_error_type = self._sandbox_error_type or type(exc).__name__ + raise + if res.error_type is not None: + self._sandbox_error_type = self._sandbox_error_type or res.error_type return ExecResult( exit_code=res.return_code, @@ -143,6 +156,9 @@ async def exec_run_with_timeout(self, command: str, timeout: int) -> Tuple[str, # AsyncSandbox.exec takes timeout_s, not docker-py's timeout. timeout_s=timeout, ) + self._eval_return_code = res.return_code if res.error_type is None else None + if res.error_type is not None: + self._sandbox_error_type = res.error_type timed_out = False stdout = res.stdout or "" @@ -153,7 +169,11 @@ async def exec_run_with_timeout(self, command: str, timeout: int) -> Tuple[str, except TimeoutError: # Gym Sandbox API will throw a timeout error on actual timeout. timed_out = True + self._sandbox_error_type = "TimeoutError" test_output = "" + except Exception as exc: + self._sandbox_error_type = type(exc).__name__ + raise return (test_output, timed_out, time() - start_time) @@ -162,14 +182,41 @@ async def copy(self, src: Path, dest: Path) -> None: data = src.read_text() src.write_text(patch_swebench_multilingual_golden_patch_pass(data, self.instance_id)) - await self._inner_container.upload(local_path=src, remote_path=str(dest)) + try: + await self._inner_container.upload(local_path=src, remote_path=str(dest)) + except Exception as exc: + self._sandbox_error_type = self._sandbox_error_type or type(exc).__name__ + raise async def cleanup(self) -> None: try: await self._inner_container.stop() - except: + except Exception as exc: + self._sandbox_error_type = self._sandbox_error_type or type(exc).__name__ print("Failed to stop verification sandbox", format_exc(), file=sys.stderr) + def observation(self, *, wall_time_s: float, evaluation_completed: bool) -> SandboxObservation: + handle = self._inner_container._handle + normalized_error = self._sandbox_error_type.lower() if isinstance(self._sandbox_error_type, str) else "" + if "timeout" in normalized_error: + outcome = "timeout" + elif self._sandbox_error_type is not None: + outcome = "sandbox_error" + elif evaluation_completed: + outcome = "completed" + else: + outcome = "failed" + + return SandboxObservation( + role="verifier", + provider=handle.provider_name if handle is not None else None, + sandbox_id=handle.sandbox_id if handle is not None else None, + outcome=outcome, + exit_code=self._eval_return_code, + wall_time_s=wall_time_s, + error_type=self._sandbox_error_type, + ) + # TODO @bxyu-nvidia: Eventually once the sandbox server infra is ready, these seed_session types need to upgrade to pass a sandbox spec. # They can possibly even omitted once this graduates to core infra. @@ -283,9 +330,9 @@ async def verify(self, request: Request, body: SWEBenchVerifyRequest) -> SWEBenc test_spec = self._make_test_spec(body) - start_time = time() + verifier_sandbox_lifecycle_started_at = monotonic() eval_sandbox = await self._create_sandbox(test_spec) - eval_sandbox_start_time_taken = time() - start_time + eval_sandbox_start_time_taken = monotonic() - verifier_sandbox_lifecycle_started_at model_patch = "" if self.config.is_verifying_golden_patch: @@ -323,6 +370,16 @@ async def verify(self, request: Request, body: SWEBenchVerifyRequest) -> SWEBenc rewrite_reports=False, ) patch_verification_time_taken = time() - start_time + verifier_sandbox_wall_time_s = monotonic() - verifier_sandbox_lifecycle_started_at + + try: + verifier_sandbox_observation = mock_container.observation( + wall_time_s=verifier_sandbox_wall_time_s, + evaluation_completed=res["completed"], + ) + except Exception: + verifier_sandbox_observation = None + print("Failed to build verification sandbox observation", format_exc(), file=sys.stderr) log_dir = Path(__file__).parent / "logs/run_evaluation" / run_id @@ -347,6 +404,7 @@ async def verify(self, request: Request, body: SWEBenchVerifyRequest) -> SWEBenc model_patch=model_patch or None, test_output=test_output, log_dir=str(log_dir), + verifier_sandbox_observation=verifier_sandbox_observation, ) diff --git a/resources_servers/swebench/tests/test_app.py b/resources_servers/swebench/tests/test_app.py index d410ca440e..3e4739070c 100644 --- a/resources_servers/swebench/tests/test_app.py +++ b/resources_servers/swebench/tests/test_app.py @@ -12,13 +12,36 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from pathlib import Path from unittest.mock import AsyncMock, MagicMock +import pytest from fastapi.testclient import TestClient from pytest import MonkeyPatch +from nemo_gym.sandbox import SandboxExecResult, SandboxHandle from nemo_gym.server_utils import ServerClient -from resources_servers.swebench.app import SwebenchResourcesServer, SwebenchResourcesServerConfig +from resources_servers.swebench.app import ( + DockerContainer, + SwebenchResourcesServer, + SwebenchResourcesServerConfig, + SWEBenchVerifyResponse, +) + + +def make_sandbox( + *, + exec_result: SandboxExecResult | None = None, + exec_error: Exception | None = None, + upload_error: Exception | None = None, + stop_error: Exception | None = None, +) -> MagicMock: + sandbox = MagicMock() + sandbox._handle = SandboxHandle(sandbox_id="sandbox-123", provider_name="test-provider", raw=None) + sandbox.exec = AsyncMock(return_value=exec_result, side_effect=exec_error) + sandbox.upload = AsyncMock(side_effect=upload_error) + sandbox.stop = AsyncMock(side_effect=stop_error) + return sandbox class TestApp: @@ -37,8 +60,10 @@ def test_sanity(self, monkeypatch: MonkeyPatch) -> None: client = TestClient(app) + eval_sandbox = make_sandbox() monkeypatch.setattr( - "resources_servers.swebench.app.SwebenchResourcesServer._create_sandbox", AsyncMock(start=AsyncMock()) + "resources_servers.swebench.app.SwebenchResourcesServer._create_sandbox", + AsyncMock(return_value=eval_sandbox), ) monkeypatch.setattr( "resources_servers.swebench.app.run_instance", @@ -77,3 +102,130 @@ def test_sanity(self, monkeypatch: MonkeyPatch) -> None: }, ) assert res.status_code == 200 + observation = res.json()["verifier_sandbox_observation"] + assert observation.pop("wall_time_s") >= 0 + assert observation == { + "kind": "sandbox", + "role": "verifier", + "provider": "test-provider", + "sandbox_id": "sandbox-123", + "outcome": "completed", + "exit_code": None, + "cpu_time_s": None, + "peak_memory_mib": None, + "resource_usage_source": None, + "error_type": None, + } + + def test_unobserved_response_omits_optional_field(self) -> None: + response = SWEBenchVerifyResponse.model_construct(verifier_sandbox_observation=None) + + assert "verifier_sandbox_observation" not in response.model_dump() + + async def test_eval_exit_code_is_observed_without_treating_failed_tests_as_sandbox_failure(self) -> None: + sandbox = make_sandbox(exec_result=SandboxExecResult(stdout="test output", stderr=None, return_code=7)) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + test_output, timed_out, _ = await container.exec_run_with_timeout("/bin/bash /eval.sh", timeout=60) + observation = container.observation(wall_time_s=3.5, evaluation_completed=True) + + assert test_output == "test output" + assert timed_out is False + assert observation.outcome == "completed" + assert observation.exit_code == 7 + assert observation.wall_time_s == 3.5 + + async def test_timeout_is_observed_without_changing_harness_timeout_behavior(self) -> None: + sandbox = make_sandbox( + exec_result=SandboxExecResult( + stdout=None, + stderr="backend failed", + return_code=125, + error_type="sandbox", + ) + ) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + await container.exec_run("git apply patch.diff") + sandbox.exec.side_effect = TimeoutError("timed out") + test_output, timed_out, _ = await container.exec_run_with_timeout("/bin/bash /eval.sh", timeout=60) + observation = container.observation(wall_time_s=60.0, evaluation_completed=False) + + assert test_output == "" + assert timed_out is True + assert observation.outcome == "timeout" + assert observation.exit_code is None + assert observation.error_type == "TimeoutError" + + async def test_runtime_error_is_observed_and_still_propagates(self) -> None: + sandbox = make_sandbox(exec_error=RuntimeError("Sandbox was OOM-killed")) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + with pytest.raises(RuntimeError, match="OOM-killed"): + await container.exec_run_with_timeout("/bin/bash /eval.sh", timeout=60) + + observation = container.observation(wall_time_s=1.0, evaluation_completed=False) + assert observation.outcome == "sandbox_error" + assert observation.error_type == "RuntimeError" + assert observation.exit_code is None + + @pytest.mark.parametrize( + ("error_type", "expected_outcome"), + [("sandbox", "sandbox_error"), ("TimeoutError", "timeout")], + ) + async def test_provider_error_does_not_report_sentinel_as_process_exit_code( + self, error_type: str, expected_outcome: str + ) -> None: + sandbox = make_sandbox( + exec_result=SandboxExecResult(stdout=None, stderr="backend failed", return_code=125, error_type=error_type) + ) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + _, timed_out, _ = await container.exec_run_with_timeout("/bin/bash /eval.sh", timeout=60) + observation = container.observation(wall_time_s=1.0, evaluation_completed=False) + + assert timed_out is False + assert observation.outcome == expected_outcome + assert observation.error_type == error_type + assert observation.exit_code is None + + async def test_pre_eval_provider_error_is_observed(self) -> None: + sandbox = make_sandbox( + exec_result=SandboxExecResult(stdout=None, stderr="backend failed", return_code=125, error_type="sandbox") + ) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + await container.exec_run("git apply patch.diff") + observation = container.observation(wall_time_s=1.0, evaluation_completed=False) + + assert observation.outcome == "sandbox_error" + assert observation.error_type == "sandbox" + assert observation.exit_code is None + + async def test_upload_error_is_observed(self, tmp_path: Path) -> None: + sandbox = make_sandbox(upload_error=RuntimeError("upload failed")) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + with pytest.raises(RuntimeError, match="upload failed"): + await container.copy(tmp_path / "patch.diff", Path("/tmp/patch.diff")) + + observation = container.observation(wall_time_s=1.0, evaluation_completed=False) + assert observation.outcome == "sandbox_error" + assert observation.error_type == "RuntimeError" + + async def test_cleanup_error_is_fail_open_and_observed(self) -> None: + sandbox = make_sandbox(stop_error=RuntimeError("stop failed")) + container = DockerContainer(id="run-id", instance_id="instance-id") + container._inner_container = sandbox + + await container.cleanup() + observation = container.observation(wall_time_s=2.0, evaluation_completed=True) + + assert observation.outcome == "sandbox_error" + assert observation.error_type == "RuntimeError" diff --git a/responses_api_agents/opencode_sandboxed_agent/app.py b/responses_api_agents/opencode_sandboxed_agent/app.py index d7ece3efe1..de94ffab5c 100644 --- a/responses_api_agents/opencode_sandboxed_agent/app.py +++ b/responses_api_agents/opencode_sandboxed_agent/app.py @@ -14,6 +14,7 @@ # limitations under the License. import json +import sqlite3 import sys from pathlib import Path from shlex import quote @@ -42,10 +43,22 @@ NeMoGymResponseFunctionToolCall, NeMoGymResponseInputTokensDetails, NeMoGymResponseOutputItem, + NeMoGymResponseOutputMessage, + NeMoGymResponseOutputText, NeMoGymResponseOutputTokensDetails, + NeMoGymResponseReasoningItem, NeMoGymResponseUsage, + NeMoGymSummary, ) from nemo_gym.responses_converter import ResponsesConverter +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ContextCompactionObservation, + ObservationGap, + SandboxObservation, + ToolCallObservation, +) from nemo_gym.sandbox import AsyncSandbox, SandboxResources, SandboxSpec, create_provider from nemo_gym.sandbox.config import resolve_provider_config, resolve_provider_metadata from nemo_gym.server_utils import ( @@ -57,6 +70,316 @@ ) +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_observations(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, + ) + + class OpenCodeSandboxedAgentConfig(BaseResponsesAPIAgentConfig): resources_server: ResourcesServerRef model_server: ModelServerRef @@ -94,6 +417,10 @@ class OpenCodeSandboxedAgentVerifyResponse(BaseVerifyResponse): opencode_run_stderr: str opencode_finished: bool opencode_export_found: bool + ng_agent_observations: Optional[AgentObservationBundle] = Field( + default=None, + exclude_if=lambda value: value is None, + ) class OpenCodeSandboxedAgent(SimpleResponsesAPIAgent): @@ -142,6 +469,37 @@ async def _start_sandbox(self, sandbox_id: Optional[str] = None) -> AsyncSandbox return sandbox + def _agent_sandbox_observation( + self, + *, + sandbox: AsyncSandbox, + return_code: Any, + error_type: Any, + finished: bool, + ) -> SandboxObservation: + handle = getattr(sandbox, "_handle", None) + handle_provider = getattr(handle, "provider_name", None) + handle_sandbox_id = getattr(handle, "sandbox_id", None) + normalized_error = error_type.lower() if isinstance(error_type, str) else "" + if "timeout" in normalized_error: + outcome = "timeout" + elif normalized_error: + outcome = "sandbox_error" + elif return_code == 0 and finished: + outcome = "completed" + elif isinstance(return_code, int): + outcome = "failed" if return_code != 0 else "unknown" + else: + outcome = "unknown" + return SandboxObservation( + role="agent", + provider=handle_provider if isinstance(handle_provider, str) else None, + sandbox_id=handle_sandbox_id if isinstance(handle_sandbox_id, str) else None, + outcome=outcome, + exit_code=return_code if not normalized_error and isinstance(return_code, int) else None, + error_type=error_type if isinstance(error_type, str) else None, + ) + async def _create_opencode_config(self, request: Request) -> Dict[str, Any]: base_url = ( self.base_url_for_run( @@ -311,26 +669,36 @@ async def responses( """ opencode_config_content = json.dumps(await self._create_opencode_config(request)) + observation_invocation_id = getattr(request.state, "_ng_observation_invocation_id", None) + observation_invocation_id = observation_invocation_id if isinstance(observation_invocation_id, str) else None + collect_observations = observation_invocation_id is not None + opencode_env = { + "OPENCODE_CONFIG_CONTENT": opencode_config_content, + # @bxyu-nvidia: OpenCode defaults to 32k here https://github.com/anomalyco/opencode/blob/58a99916bb96edf5cf605dc03e1be1e4bacf9ff7/packages/opencode/src/provider/transform.ts#L21 + # and there is no way to set it to null. + # Here, we set an exorbitantly high number that cannot ever be reached. + # In future versions of OpenCode, this can be directly passed via maxOutputTokens in the limit config above https://github.com/anomalyco/opencode/blob/1b18a50418f730aca32630ccfcde850f2b5fc360/packages/opencode/src/provider/transform.ts#L1418 + "OPENCODE_EXPERIMENTAL_OUTPUT_TOKEN_MAX": str(1_000_000_000), + } + remote_data_home = None + if collect_observations: + remote_data_home = f"/tmp/nemo-gym-opencode-{uuid4().hex}" + opencode_env["XDG_DATA_HOME"] = remote_data_home if self.config.debug: print(f"Running command:\n```bash\n{command}\n```\n", file=sys.stderr) print(f"OpenCode config JSON str: {opencode_config_content}", file=sys.stderr) + run_error_type = None try: result = await sandbox.exec( command=command, timeout_s=self.config.sandbox_timeout, - env={ - "OPENCODE_CONFIG_CONTENT": opencode_config_content, - # @bxyu-nvidia: OpenCode defaults to 32k here https://github.com/anomalyco/opencode/blob/58a99916bb96edf5cf605dc03e1be1e4bacf9ff7/packages/opencode/src/provider/transform.ts#L21 - # and there is no way to set it to null. - # Here, we set an exorbitantly high number that cannot ever be reached. - # In future versions of OpenCode, this can be directly passed via maxOutputTokens in the limit config above https://github.com/anomalyco/opencode/blob/1b18a50418f730aca32630ccfcde850f2b5fc360/packages/opencode/src/provider/transform.ts#L1418 - "OPENCODE_EXPERIMENTAL_OUTPUT_TOKEN_MAX": str(1_000_000_000), - }, + env=opencode_env, ) - except: + except Exception as exc: result = None + run_error_type = type(exc).__name__ print("OpenCode exec hit error.", format_exc(), file=sys.stderr) if self.config.debug and result: @@ -343,7 +711,8 @@ async def responses( command=f"""export PATH=$HOME/.opencode/bin:$PATH \ && (command -v jq >/dev/null 2>&1 || (apt-get update && apt-get install -y --no-install-recommends jq)) \ && session_id=$(opencode session list --format json | jq -r '.[0].id') \ - && opencode export $session_id > {export_fname}""" + && opencode export $session_id > {export_fname}""", + env={"XDG_DATA_HOME": remote_data_home} if remote_data_home is not None else None, ) except: export_result = None @@ -373,6 +742,45 @@ async def responses( print("Export stdout:\n", export_result.stdout, file=sys.stderr) print("Export stderr:\n", export_result.stderr, file=sys.stderr) + observations = None + if collect_observations: + assert remote_data_home is not None + observations_remote_fpath = f"{remote_data_home}/opencode/opencode.db" + snapshot_remote_fpath = f"{remote_data_home}/opencode/nemo-gym-observations.db" + observations_local_fpath = results_dir / "opencode.db" + observations_local_fpath.unlink(missing_ok=True) + try: + snapshot_script = ( + "import sqlite3,sys;" + "source=sqlite3.connect(f'file:{sys.argv[1]}?mode=ro',uri=True);" + "destination=sqlite3.connect(sys.argv[2]);" + "source.backup(destination);destination.close();source.close()" + ) + snapshot_result = await sandbox.exec( + command=( + f"python3 -c {quote(snapshot_script)} " + f"{quote(observations_remote_fpath)} {quote(snapshot_remote_fpath)}" + ) + ) + if snapshot_result.return_code != 0 or snapshot_result.error_type is not None: + raise RuntimeError("OpenCode database snapshot failed") + await sandbox.download(snapshot_remote_fpath, observations_local_fpath) + observations = parse_opencode_observations(observations_local_fpath, observation_invocation_id) + except Exception: + print("Failed to capture OpenCode observations", format_exc(), file=sys.stderr) + observations = AgentObservationBundle( + source="opencode", + records=[AgentInvocation(invocation_id=observation_invocation_id)], + gaps=[ + ObservationGap(code="agent_artifact_unavailable"), + ObservationGap(code="agent_transcript_unavailable"), + ObservationGap(code="model_call_ownership_unavailable"), + ObservationGap(code="observation_capture_failed"), + ], + ) + finally: + observations_local_fpath.unlink(missing_ok=True) + opencode_export = dict() if results_local_fpath.exists(): opencode_export = json.loads(results_local_fpath.read_text().strip() or "{}") @@ -386,13 +794,43 @@ async def responses( output = self._opencode_export_to_output_items(opencode_export)[1:] usage = NeMoGymResponseUsage.sum_from_list(self._opencode_export_to_usages(opencode_export)) - self._sandbox_id_to_run_result[request.cookies["sandbox_id"]] = { + result_stdout = (result.stdout if result else "") or "" + result_stderr = (result.stderr if result else "") or "" + opencode_finished = "OpenCode run finished" in result_stdout + + if collect_observations and observations is not None: + agent_sandbox_observation = self._agent_sandbox_observation( + sandbox=sandbox, + return_code=getattr(result, "return_code", None), + error_type=getattr(result, "error_type", None) or run_error_type, + finished=opencode_finished, + ) + for record in observations.records: + if isinstance(record, ToolCallObservation): + record.sandbox_id = agent_sandbox_observation.sandbox_id + elif isinstance(record, AgentInvocation) and record.parent_invocation_id is None: + status = { + "completed": "completed", + "failed": "failed", + "sandbox_error": "failed", + "timeout": "incomplete", + "cancelled": "incomplete", + }.get(agent_sandbox_observation.outcome) + if status is not None: + record.status = status + observations.records.append(agent_sandbox_observation) + observations.gaps.append(ObservationGap(code="sandbox_lifecycle_timing_unavailable")) + + run_result = { "opencode_results_fpath": str(results_local_fpath) if opencode_export_found else "", - "opencode_run_stdout": (result.stdout if result else "") or "", - "opencode_run_stderr": (result.stderr if result else "") or "", + "opencode_run_stdout": result_stdout, + "opencode_run_stderr": result_stderr, "opencode_export_found": opencode_export_found, - "opencode_finished": ("OpenCode run finished" in (result.stdout or "") if result else False), + "opencode_finished": opencode_finished, } + if collect_observations: + run_result["_ng_agent_observations"] = observations + self._sandbox_id_to_run_result[request.cookies["sandbox_id"]] = run_result return NeMoGymResponse( id=f"resp_{uuid4().hex}", @@ -410,6 +848,8 @@ async def run( self, request: Request, body: OpenCodeSandboxedAgentRunRequest ) -> OpenCodeSandboxedAgentVerifyResponse: cookies = request.cookies + session_key = request.session[SESSION_ID_KEY] + rollout_id = self.rollout_id_from_run(body) seed_session_response = await self.server_client.post( server_name=self.config.resources_server.name, @@ -423,14 +863,23 @@ async def run( # @bxyu-nvidia: "sandbox_handle" comes from resources_servers/swebench/app.py # Once we graduate to use the sandbox server, this will be in a generic seed_session type that can be model validated. seed_session_result = await seed_session_response.json() - sandbox = await self._start_sandbox(sandbox_id=seed_session_result.get("sandbox_handle")) - self._sandbox_id_to_sandbox[request.session[SESSION_ID_KEY]] = sandbox + provider_sandbox_id = seed_session_result.get("sandbox_handle") + provider_sandbox_id = provider_sandbox_id if isinstance(provider_sandbox_id, str) else None + sandbox = await self._start_sandbox(sandbox_id=provider_sandbox_id) + self._sandbox_id_to_sandbox[session_key] = sandbox # Propagating the sandbox handle - cookies["sandbox_id"] = request.session[SESSION_ID_KEY] + cookies["sandbox_id"] = session_key request._cookies = cookies - response = await self.responses(request, body.responses_create_params) + request.state._ng_observation_invocation_id = rollout_id + observations = None + try: + response = await self.responses(request, body.responses_create_params) + finally: + del request.state._ng_observation_invocation_id + run_result = self._sandbox_id_to_run_result.get(session_key, {}) + observations = run_result.pop("_ng_agent_observations", None) verify_request = OpenCodeSandboxedAgentVerifyRequest.model_validate(body.model_dump() | {"response": response}) @@ -444,19 +893,40 @@ async def run( try: await sandbox.stop() - except: + except Exception: print("Failed to stop sandbox", format_exc(), file=sys.stderr) - self._sandbox_id_to_sandbox.pop(request.session[SESSION_ID_KEY]) + self._sandbox_id_to_sandbox.pop(session_key, None) # @bxyu-nvidia: This is scraped from the raw create params. Later on we can dynamically set this if OpenCode exports this :rofl: opencode_system_prompt = "You are opencode, an interactive CLI tool that helps users with software engineering tasks. Use the instructions below and the tools available to you to assist the user.\n\nIMPORTANT: You must NEVER generate or guess URLs for the user unless you are confident that the URLs are for helping the user with programming. You may use URLs provided by the user in their messages or local files.\n\nIf the user asks for help or wants to give feedback inform them of the following:\n- /help: Get help with using opencode\n- To give feedback, users should report the issue at https://github.com/anomalyco/opencode/issues\n\nWhen the user directly asks about opencode (eg 'can opencode do...', 'does opencode have...') or asks in second person (eg 'are you able...', 'can you do...'), first use the WebFetch tool to gather information to answer the question from opencode docs at https://opencode.ai\n\n# Tone and style\nYou should be concise, direct, and to the point. When you run a non-trivial bash command, you should explain what the command does and why you are running it, to make sure the user understands what you are doing (this is especially important when you are running a command that will make changes to the user's system).\nRemember that your output will be displayed on a command line interface. Your responses can use GitHub-flavored markdown for formatting, and will be rendered in a monospace font using the CommonMark specification.\nOutput text to communicate with the user; all text you output outside of tool use is displayed to the user. Only use tools to complete tasks. Never use tools like Bash or code comments as means to communicate with the user during the session.\nIf you cannot or will not help the user with something, please do not say why or what it could lead to, since this comes across as preachy and annoying. Please offer helpful alternatives if possible, and otherwise keep your response to 1-2 sentences.\nOnly use emojis if the user explicitly requests it. Avoid using emojis in all communication unless asked.\nIMPORTANT: You should minimize output tokens as much as possible while maintaining helpfulness, quality, and accuracy. Only address the specific query or task at hand, avoiding tangential information unless absolutely critical for completing the request. If you can answer in 1-3 sentences or a short paragraph, please do.\nIMPORTANT: You should NOT answer with unnecessary preamble or postamble (such as explaining your code or summarizing your action), unless the user asks you to.\nIMPORTANT: Keep your responses short, since they will be displayed on a command line interface. You MUST answer concisely with fewer than 4 lines (not including tool use or code generation), unless user asks for detail. Answer the user's question directly, without elaboration, explanation, or details. One word answers are best. Avoid introductions, conclusions, and explanations. You MUST avoid text before/after your response, such as \"The answer is .\", \"Here is the content of the file...\" or \"Based on the information provided, the answer is...\" or \"Here is what I will do next...\". Here are some examples to demonstrate appropriate verbosity:\n\nuser: what is 2+2?\nassistant: 4\n\n\n\nuser: is 11 a prime number?\nassistant: Yes\n\n\n\nuser: what command should I run to list files in the current directory?\nassistant: ls\n\n\n\nuser: what command should I run to watch files in the current directory?\nassistant: [use the ls tool to list the files in the current directory, then read docs/commands in the relevant file to find out how to watch files]\nnpm run dev\n\n\n\nuser: what files are in the directory src/?\nassistant: [runs ls and sees foo.c, bar.c, baz.c]\nuser: which file contains the implementation of foo?\nassistant: src/foo.c\n\n\n\nuser: write tests for new feature\nassistant: [uses grep and glob search tools to find where similar tests are defined, uses concurrent read file tool use blocks in one tool call to read relevant files at the same time, uses edit file tool to write new tests]\n\n\n# Proactiveness\nYou are allowed to be proactive, but only when the user asks you to do something. You should strive to strike a balance between:\n1. Doing the right thing when asked, including taking actions and follow-up actions\n2. Not surprising the user with actions you take without asking\nFor example, if the user asks you how to approach something, you should do your best to answer their question first, and not immediately jump into taking actions.\n3. Do not add additional code explanation summary unless requested by the user. After working on a file, just stop, rather than providing an explanation of what you did.\n\n# Following conventions\nWhen making changes to files, first understand the file's code conventions. Mimic code style, use existing libraries and utilities, and follow existing patterns.\n- NEVER assume that a given library is available, even if it is well known. Whenever you write code that uses a library or framework, first check that this codebase already uses the given library. For example, you might look at neighboring files, or check the package.json (or cargo.toml, and so on depending on the language).\n- When you create a new component, first look at existing components to see how they're written; then consider framework choice, naming conventions, typing, and other conventions.\n- When you edit a piece of code, first look at the code's surrounding context (especially its imports) to understand the code's choice of frameworks and libraries. Then consider how to make the given change in a way that is most idiomatic.\n- Always follow security best practices. Never introduce code that exposes or logs secrets and keys. Never commit secrets or keys to the repository.\n\n# Code style\n- IMPORTANT: DO NOT ADD ***ANY*** COMMENTS unless asked\n\n# Doing tasks\nThe user will primarily request you perform software engineering tasks. This includes solving bugs, adding new functionality, refactoring code, explaining code, and more. For these tasks the following steps are recommended:\n- Use the available search tools to understand the codebase and the user's query. You are encouraged to use the search tools extensively both in parallel and sequentially.\n- Implement the solution using all tools available to you\n- Verify the solution if possible with tests. NEVER assume specific test framework or test script. Check the README or search codebase to determine the testing approach.\n- VERY IMPORTANT: When you have completed a task, you MUST run the lint and typecheck commands (e.g. npm run lint, npm run typecheck, ruff, etc.) with Bash if they were provided to you to ensure your code is correct. If you are unable to find the correct command, ask the user for the command to run and if they supply it, proactively suggest writing it to AGENTS.md so that you will know to run it next time.\nNEVER commit changes unless the user explicitly asks you to. It is VERY IMPORTANT to only commit when explicitly asked, otherwise the user will feel that you are being too proactive.\n\n- Tool results and user messages may include tags. tags contain useful information and reminders. They are NOT part of the user's provided input or the tool result.\n\n# Tool usage policy\n- When doing file search, prefer to use the Task tool in order to reduce context usage.\n- You have the capability to call multiple tools in a single response. When multiple independent pieces of information are requested, batch your tool calls together for optimal performance. When making multiple bash tool calls, you MUST send a single message with multiple tools calls to run the calls in parallel. For example, if you need to run \"git status\" and \"git diff\", send a single message with two tool calls to run the calls in parallel.\n\nYou MUST answer concisely with fewer than 4 lines of text (not including tool use or code generation), unless user asks for detail.\n\nIMPORTANT: Before you begin work, think about what the code you're editing is supposed to do based on the filenames directory structure.\n\n# Code References\n\nWhen referencing specific functions or pieces of code include the pattern `file_path:line_number` to allow the user to easily navigate to the source code location.\n\n\nuser: Where are errors from the client handled?\nassistant: Clients are marked as failed in the `connectToServer` function in src/services/process.ts:712.\n\n\nYou are powered by the model named dummy_model. The exact model ID is nemo_gym/dummy_model\nHere is some useful information about the environment you are running in:\n\n Working directory: /testbed\n Workspace root folder: /testbed\n Is directory a git repo: yes\n Platform: linux\n Today's date: Tue Aug 04 2026\n\nSkills provide specialized instructions and workflows for specific tasks.\nUse the skill tool to load a skill when a task matches its description.\n\n \n customize-opencode\n Use ONLY when the user is editing or creating opencode's own configuration: opencode.json, opencode.jsonc, files under .opencode/, or files under ~/.config/opencode/. Also use when creating or fixing opencode agents, subagents, skills, plugins, MCP servers, or permission rules. Do not use for the user's own application code, or for any project that is not configuring opencode itself.\n file:///testbed/%3Cbuilt-in%3E\n \n" response_dict = await get_response_json(verify_response) - response_dict |= self._sandbox_id_to_run_result[cookies["sandbox_id"]] + run_result = self._sandbox_id_to_run_result.pop(session_key) + response_dict |= run_result + raw_verifier_sandbox_observation = response_dict.pop("verifier_sandbox_observation", None) response_dict["responses_create_params"]["input"].insert( 0, {"content": opencode_system_prompt, "role": "system"} ) + + if rollout_id is not None: + if observations is None: + observations = AgentObservationBundle( + source="opencode", + records=[AgentInvocation(invocation_id=rollout_id)], + gaps=[ObservationGap(code="observation_capture_failed")], + ) + if raw_verifier_sandbox_observation is not None: + try: + verifier_observation = SandboxObservation.model_validate(raw_verifier_sandbox_observation) + if verifier_observation.role != "verifier": + raise ValueError("resources server returned a non-verifier sandbox observation") + observations.records.append(verifier_observation) + except Exception: + observations.gaps.append(ObservationGap(code="verifier_sandbox_observation_invalid")) + else: + observations.gaps.append(ObservationGap(code="verifier_sandbox_observation_unavailable")) + response_dict["ng_agent_observations"] = observations.model_dump(mode="json") return OpenCodeSandboxedAgentVerifyResponse.model_validate(response_dict) diff --git a/responses_api_agents/opencode_sandboxed_agent/configs/opencode_agent.yaml b/responses_api_agents/opencode_sandboxed_agent/configs/opencode_agent.yaml index 502916c985..19f2243f66 100644 --- a/responses_api_agents/opencode_sandboxed_agent/configs/opencode_agent.yaml +++ b/responses_api_agents/opencode_sandboxed_agent/configs/opencode_agent.yaml @@ -9,6 +9,7 @@ opencode_sandboxed_agent: model_server: type: responses_api_models name: policy_model + token_id_capture: true debug: false diff --git a/responses_api_agents/opencode_sandboxed_agent/tests/test_app.py b/responses_api_agents/opencode_sandboxed_agent/tests/test_app.py index 7dbd7761d9..0dbfc7c852 100644 --- a/responses_api_agents/opencode_sandboxed_agent/tests/test_app.py +++ b/responses_api_agents/opencode_sandboxed_agent/tests/test_app.py @@ -13,11 +13,16 @@ # See the License for the specific language governing permissions and # limitations under the License. import json +import shlex +import sqlite3 +import subprocess +import sys from pathlib import Path +from types import SimpleNamespace from typing import Any, Dict from unittest.mock import AsyncMock, MagicMock -from pytest import MonkeyPatch, fixture +from pytest import MonkeyPatch, fixture, mark from nemo_gym.config_types import ModelServerRef, ResourcesServerRef from nemo_gym.openai_utils import ( @@ -34,11 +39,29 @@ NeMoGymResponseUsage, NeMoGymSummary, ) +from nemo_gym.rollout_observability import ( + AgentInvocation, + SandboxObservation, + ToolCallObservation, +) +from nemo_gym.sandbox import SandboxHandle from nemo_gym.server_utils import SESSION_ID_KEY, ServerClient -from responses_api_agents.opencode_sandboxed_agent.app import OpenCodeSandboxedAgent, OpenCodeSandboxedAgentConfig +from responses_api_agents.opencode_sandboxed_agent.app import ( + OpenCodeSandboxedAgent, + OpenCodeSandboxedAgentConfig, + OpenCodeSandboxedAgentRunRequest, +) class TestOpenCodeSandboxedAgent: + def test_import_does_not_load_standalone_opencode_agent(self) -> None: + code = ( + "import sys; import responses_api_agents.opencode_sandboxed_agent.app; " + "assert not any(name == 'responses_api_agents.opencode_agent' " + "or name.startswith('responses_api_agents.opencode_agent.') for name in sys.modules)" + ) + subprocess.run([sys.executable, "-c", code], check=True, timeout=30) + def _create_config(self) -> OpenCodeSandboxedAgentConfig: return OpenCodeSandboxedAgentConfig( host="0.0.0.0", @@ -52,6 +75,7 @@ def _create_config(self) -> OpenCodeSandboxedAgentConfig: sandbox_config=dict(), sandbox_timeout=0, opencode_max_context_window=0, + token_id_capture=True, ) @fixture @@ -133,13 +157,18 @@ async def test_responses_sanity(self, opencode_export_test_data: Dict[str, Any], config = self._create_config() server = OpenCodeSandboxedAgent(config=config, server_client=MagicMock(spec=ServerClient)) - sandbox_mock = AsyncMock() + sandbox_mock = MagicMock() + sandbox_mock.exec = AsyncMock( + side_effect=[ + SimpleNamespace(stdout="OpenCode run finished", stderr="", return_code=0, error_type=None), + SimpleNamespace(stdout="", stderr="", return_code=0, error_type=None), + SimpleNamespace(stdout="my dir"), + ] + ) + sandbox_mock.download = AsyncMock() monkeypatch.setattr(server, "_sandbox_id_to_sandbox", {"": sandbox_mock}) monkeypatch.setattr(server, "_create_opencode_config", AsyncMock(return_value=dict())) - sandbox_mock.return_value.exec.return_value = MagicMock() - sandbox_mock.return_value.exec.return_value.stdout = "my dir" - monkeypatch.setattr( "responses_api_agents.opencode_sandboxed_agent.app.Path.exists", lambda self: True, @@ -155,7 +184,11 @@ async def test_responses_sanity(self, opencode_export_test_data: Dict[str, Any], monkeypatch.setattr("responses_api_agents.opencode_sandboxed_agent.app.time", MagicMock(return_value=0.0)) actual_response = await server.responses( - request=MagicMock(session={SESSION_ID_KEY: "my session"}, cookies={"sandbox_id": ""}), + request=MagicMock( + session={SESSION_ID_KEY: "my session"}, + cookies={"sandbox_id": ""}, + path_params={"rollout_id": "direct-call"}, + ), body=NeMoGymResponseCreateParamsNonStreaming( input=[{"role": "user", "content": "hello"}], ), @@ -238,3 +271,258 @@ async def test_responses_sanity(self, opencode_export_test_data: Dict[str, Any], ) assert expected_response == actual_response + assert not any(key.startswith("_ng_") for key in server._sandbox_id_to_run_result[""]) + assert "XDG_DATA_HOME" not in sandbox_mock.exec.await_args_list[0].kwargs["env"] + assert sandbox_mock.exec.await_args_list[1].kwargs["env"] is None + + def test_agent_sandbox_observation_classifies_timeout_errors(self) -> None: + server = OpenCodeSandboxedAgent( + config=self._create_config(), + server_client=MagicMock(spec=ServerClient), + ) + sandbox = MagicMock() + sandbox._handle = SandboxHandle(sandbox_id="connected-sandbox", provider_name="opensandbox", raw=None) + + observation = server._agent_sandbox_observation( + sandbox=sandbox, + return_code=125, + error_type="TimeoutError", + finished=False, + ) + + assert observation.outcome == "timeout" + assert observation.exit_code is None + assert observation.sandbox_id == "connected-sandbox" + assert observation.provider == "opensandbox" + + observation = server._agent_sandbox_observation( + sandbox=sandbox, + return_code=137, + error_type="OutOfMemoryError", + finished=False, + ) + assert observation.outcome == "sandbox_error" + assert observation.exit_code is None + + @mark.parametrize( + ("observability_enabled", "token_capture_enabled", "expected_base_url"), + [ + (False, False, "http://model-server/v1"), + (True, False, "http://model-server/ng-rollout/7-2/v1"), + (False, True, "http://model-server/ng-rollout/7-2/training-token-capture/v1"), + (True, True, "http://model-server/ng-rollout/7-2/training-token-capture/v1"), + ], + ids=("disabled", "observability-only", "token-capture-only", "both"), + ) + async def test_create_opencode_config_routes_each_capture_state( + self, + monkeypatch: MonkeyPatch, + observability_enabled: bool, + token_capture_enabled: bool, + expected_base_url: str, + ) -> None: + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = { + "observability_enabled": observability_enabled, + "token_id_capture": {"enabled": token_capture_enabled, "all_agents": False}, + } + server = OpenCodeSandboxedAgent(config=self._create_config(), server_client=server_client) + monkeypatch.setattr( + "responses_api_agents.opencode_sandboxed_agent.app.get_server_url", + lambda _name: "http://model-server", + ) + request = MagicMock() + request.json = AsyncMock( + return_value={ + "responses_create_params": {"input": "solve"}, + "_ng_task_index": 7, + "_ng_rollout_index": 2, + } + ) + + config = await server._create_opencode_config(request) + + assert config["provider"]["nemo_gym"]["options"]["baseURL"] == expected_base_url + + async def test_run_builds_observations_from_live_wal_snapshot( + self, + tmp_path: Path, + opencode_export_test_data: Dict[str, Any], + monkeypatch: MonkeyPatch, + ) -> None: + class Response: + ok = True + + def __init__(self, payload: dict[str, Any], cookies: dict[str, str] | None = None): + self.payload = payload + self.cookies = cookies or {} + + async def json(self) -> dict[str, Any]: + return self.payload + + async def read(self) -> bytes: + return json.dumps(self.payload).encode() + + class RunRequest: + def __init__(self) -> None: + self._cookies: dict[str, str] = {} + self.session = {SESSION_ID_KEY: "session-1"} + self.state = SimpleNamespace() + + @property + def cookies(self) -> dict[str, str]: + return self._cookies + + db_path = tmp_path / "source.db" + connection = sqlite3.connect(db_path) + connection.execute("pragma journal_mode=wal") + connection.execute("create table session (id text, parent_id text, time_created integer)") + connection.execute("create table message (id text, session_id text, data text, time_created integer)") + connection.execute( + "create table part (id text, message_id text, session_id text, data text, time_created integer)" + ) + connection.commit() + connection.execute("pragma wal_checkpoint(truncate)") + connection.execute("insert into session values ('root', null, 0)") + connection.execute( + "insert into message values (?, ?, ?, ?)", + ("m1", "root", json.dumps({"role": "assistant", "time": {"created": 1, "completed": 3}}), 1), + ) + connection.execute( + "insert into part values (?, ?, ?, ?, ?)", + ( + "p1", + "m1", + "root", + json.dumps( + { + "type": "tool", + "tool": "bash", + "callID": "call-1", + "state": { + "status": "completed", + "input": {"command": "true"}, + "output": "", + "time": {"start": 1_000, "end": 2_000}, + }, + } + ), + 1, + ), + ) + connection.commit() + assert db_path.with_name(f"{db_path.name}-wal").stat().st_size > 0 + main_only_path = tmp_path / "main-only.db" + main_only_path.write_bytes(db_path.read_bytes()) + with sqlite3.connect(main_only_path) as main_only: + assert main_only.execute("select count(*) from session").fetchone() == (0,) + + server_client = MagicMock(spec=ServerClient) + server_client.global_config_dict = { + "observability_enabled": True, + "token_id_capture": {"enabled": False, "all_agents": False}, + } + server = OpenCodeSandboxedAgent(config=self._create_config(), server_client=server_client) + server._create_opencode_config = AsyncMock(return_value={}) + + sandbox = MagicMock() + sandbox._handle = SandboxHandle(sandbox_id="connected-sandbox", provider_name="opensandbox", raw=None) + sandbox.exec = AsyncMock( + side_effect=[ + SimpleNamespace(stdout="OpenCode run finished", stderr="", return_code=0, error_type=None), + SimpleNamespace(stdout="", stderr="", return_code=0, error_type=None), + SimpleNamespace(stdout="/workspace\n"), + SimpleNamespace(stdout="", stderr="", return_code=0, error_type=None), + ] + ) + snapshot_path = tmp_path / "snapshot.db" + + def local_quote(value: str) -> str: + if value.endswith("/opencode/opencode.db"): + value = str(db_path) + elif value.endswith("/opencode/nemo-gym-observations.db"): + value = str(snapshot_path) + return shlex.quote(value) + + monkeypatch.setattr("responses_api_agents.opencode_sandboxed_agent.app.quote", local_quote) + + async def download(remote_path: str, local_path: Path) -> None: + if remote_path == "/workspace/export.json": + local_path.write_text(json.dumps(opencode_export_test_data)) + else: + assert remote_path.endswith("/opencode/nemo-gym-observations.db") + subprocess.run(shlex.split(sandbox.exec.await_args_list[-1].kwargs["command"]), check=True) + local_path.write_bytes(snapshot_path.read_bytes()) + + sandbox.download = AsyncMock(side_effect=download) + sandbox.stop = AsyncMock(side_effect=RuntimeError("resource server already stopped the sandbox")) + server._start_sandbox = AsyncMock(return_value=sandbox) + monkeypatch.setattr( + "responses_api_agents.opencode_sandboxed_agent.app.__file__", + str(tmp_path / "app.py"), + ) + + verifier_sandbox = SandboxObservation( + role="verifier", + provider="opensandbox", + sandbox_id="verify-sandbox", + outcome="completed", + wall_time_s=2.0, + ) + + async def post(server_name, url_path, json=None, cookies=None): + if url_path == "/seed_session": + return Response({"sandbox_handle": "seed-sandbox"}) + assert url_path == "/verify" + return Response( + json + | { + "reward": 1.0, + "verifier_sandbox_observation": verifier_sandbox.model_dump(mode="json"), + } + ) + + server_client.post = AsyncMock(side_effect=post) + request = RunRequest() + body = OpenCodeSandboxedAgentRunRequest.model_validate( + { + "responses_create_params": {"input": [{"role": "user", "content": "solve"}]}, + "_ng_task_index": 7, + "_ng_rollout_index": 2, + } + ) + + try: + result = await server.run(request, body) + finally: + connection.close() + + assert result.ng_agent_observations is not None + [invocation] = [ + record for record in result.ng_agent_observations.records if isinstance(record, AgentInvocation) + ] + assert invocation.invocation_id == "root" + assert invocation.status == "completed" + [tool] = [record for record in result.ng_agent_observations.records if isinstance(record, ToolCallObservation)] + assert tool.tool_call_id == "call-1" + assert tool.sandbox_id == "connected-sandbox" + assert tool.duration_ms == 1_000 + sandbox_records = [ + record for record in result.ng_agent_observations.records if isinstance(record, SandboxObservation) + ] + assert [(record.role, record.sandbox_id) for record in sandbox_records] == [ + ("agent", "connected-sandbox"), + ("verifier", "verify-sandbox"), + ] + assert sandbox_records[0].provider == "opensandbox" + assert sandbox_records[0].outcome == "completed" + assert sandbox_records[0].wall_time_s is None + assert "sandbox_lifecycle_timing_unavailable" in {gap.code for gap in result.ng_agent_observations.gaps} + assert "sandbox_cleanup_failed" not in {gap.code for gap in result.ng_agent_observations.gaps} + run_env = sandbox.exec.await_args_list[0].kwargs["env"] + export_env = sandbox.exec.await_args_list[1].kwargs["env"] + assert run_env["XDG_DATA_HOME"].startswith("/tmp/nemo-gym-opencode-") + assert export_env["XDG_DATA_HOME"] == run_env["XDG_DATA_HOME"] + assert not hasattr(request.state, "_ng_observation_invocation_id") + assert server._sandbox_id_to_run_result == {} + assert not (tmp_path / "results" / "session-1" / "opencode.db").exists() diff --git a/responses_api_agents/swe_agents/app.py b/responses_api_agents/swe_agents/app.py index b90914c9b7..b6b49fc12a 100644 --- a/responses_api_agents/swe_agents/app.py +++ b/responses_api_agents/swe_agents/app.py @@ -69,7 +69,14 @@ NeMoGymResponseCreateParamsNonStreaming, ) from nemo_gym.profiling import Profiler -from nemo_gym.server_utils import get_first_server_config_dict +from nemo_gym.rollout_observability import AgentObservationBundle, ObservationGap +from nemo_gym.server_utils import apply_rollout_prefix, get_first_server_config_dict +from responses_api_agents.swe_agents.observability import ( + OBSERVATIONS_FILENAME, + build_swe_observations, + materialize_completion, + sandbox_observations_from_metrics, +) from responses_api_models.vllm_model.app import VLLMConverter, split_responses_input_output_items @@ -219,6 +226,8 @@ class ExecuteContainerCommandArgs(BaseModel): class SWEBenchWrapperInstanceConfig(SWEBenchWrapperServerConfig, SWEBenchWrapperConfig): + rollout_id: Optional[str] = Field(default=None, exclude_if=lambda value: value is None) + token_id_capture_enabled: bool = False metrics_fpath: Path problem_info: Dict[str, Any] body: NeMoGymResponseCreateParamsNonStreaming @@ -306,6 +315,13 @@ class SWEBenchMetrics(BaseModel): class SWEBenchVerifyResponse(SWEBenchMetrics, BaseVerifyResponse): instance_config: SWEBenchWrapperInstanceConfig subagent_trajectories: Optional[List[Dict[str, Any]]] = None + ng_agent_observations: Optional[AgentObservationBundle] = Field( + default=None, exclude_if=lambda value: value is None + ) + + +class SWEBenchRunRequest(BaseRunRequest): + model_config = ConfigDict(extra="allow") ######################################## @@ -2086,7 +2102,11 @@ def get_run_command(self) -> ExecuteContainerCommandArgs: # openai_model.yaml uses `openai_model`; vllm_model.yaml uses `model`. try: model_server_cfg = get_first_server_config_dict(get_global_config_dict(), self.config.model_server_name) - model_server_base_url = f"http://{model_server_cfg.host}:{model_server_cfg.port}" + model_server_base_url = apply_rollout_prefix( + f"http://{model_server_cfg.host}:{model_server_cfg.port}", + self.config.rollout_id, + token_capture=self.config.token_id_capture_enabled, + ) default_model_name = ( getattr(model_server_cfg, "openai_model", None) or getattr(model_server_cfg, "model", None) or "" ) @@ -2493,6 +2513,27 @@ def _openhands_dir_copy_from_host(self, output_file_path: Optional[str]) -> Opti str(eval_dir_on_host / "**" / "llm_completions" / "*" / "*.json"), recursive=True, ) + if self.config.rollout_id is not None: + try: + observations = build_swe_observations( + (Path(path) for path in completion_candidates), + framework=self.config.agent_framework, + model_ref=self.config.model_server, + ) + except Exception: + observations = AgentObservationBundle( + source=f"swe_{self.config.agent_framework}", + gaps=[ + ObservationGap( + code="observation_parse_failed", + ) + ], + ) + try: + (self.config.persistent_dir / OBSERVATIONS_FILENAME).write_text(observations.model_dump_json()) + except OSError: + pass + # When subagents are enabled (opencode) we get multiple sessions, each # writing its own per-turn JSONs. Group by session_id (from the file # payload) and copy each session's most recent turn — that file's @@ -2507,7 +2548,7 @@ def _openhands_dir_copy_from_host(self, output_file_path: Optional[str]) -> Opti payload = json.load(f) if isinstance(payload, dict) and payload.get("session_id"): sess_id = str(payload["session_id"]) - except (OSError, json.JSONDecodeError): + except (OSError, UnicodeError, json.JSONDecodeError): pass mtime = os.path.getmtime(path_str) if mtime > session_mtime.get(sess_id, -1): @@ -2935,31 +2976,6 @@ def model_post_init(self, context: Any) -> None: # START Results processing logic ######################################## - @staticmethod - def _materialize_trajectory(data: dict) -> tuple[list, list]: - """Inflate one completion-file payload into (messages, tools).""" - messages = list(data.get("messages") or []) - tools = data.get("kwargs", {}).get("tools", []) - provider_specific_fields = data.get("provider_specific_fields", {}) - try: - final_assistant_message = data["response"]["choices"][0]["message"] - except (KeyError, IndexError): - return messages, tools - - for key in [ - "prompt_token_ids", - "generation_token_ids", - "generation_log_probs", - "routed_experts", - ]: - if key in provider_specific_fields: - final_assistant_message[key] = provider_specific_fields[key] - - if final_assistant_message.get("content") or final_assistant_message.get("tool_calls"): - messages.append(final_assistant_message) - - return messages, tools - def get_openhands_trajectory_from_completions(self, trajectories_dir: Path, instance_id: str) -> tuple: """Extract the main session's trajectory for the API response. @@ -3011,7 +3027,7 @@ def get_openhands_trajectory_from_completions(self, trajectories_dir: Path, inst with open(completion_files[-1], "r") as f: main_data = json.load(f) - messages, tools = self._materialize_trajectory(main_data) + messages, tools = materialize_completion(main_data) return messages, tools, first_prefix_count def get_all_session_trajectories_from_completions(self, trajectories_dir: Path, instance_id: str) -> list[dict]: @@ -3037,7 +3053,7 @@ def get_all_session_trajectories_from_completions(self, trajectories_dir: Path, continue by_session[sess_id] = data for sess_id, data in by_session.items(): - messages, tools = self._materialize_trajectory(data) + messages, tools = materialize_completion(data) out.append( { "session_id": sess_id, @@ -3517,7 +3533,11 @@ def _item_type(item) -> Optional[str]: return json.dumps(chat_messages) def _setup_params( - self, body: NeMoGymResponseCreateParamsNonStreaming + self, + body: NeMoGymResponseCreateParamsNonStreaming, + rollout_id: Optional[str] = None, + *, + token_id_capture_enabled: bool = False, ) -> Tuple[SWEBenchWrapperInstanceConfig, BaseDatasetHarnessProcessor]: problem_info = body.metadata | {"container_formatter": self.config.container_formatter} instance_id = problem_info.get("instance_id", "unknown") @@ -3591,6 +3611,8 @@ def _setup_params( params: SWEBenchWrapperInstanceConfig = SWEBenchWrapperInstanceConfig( **self.config.model_dump(), **self._swe_bench_wrapper_server_config.model_dump(), + rollout_id=rollout_id, + token_id_capture_enabled=token_id_capture_enabled, problem_info=problem_info, body=body, persistent_dir=persistent_dir, @@ -3665,7 +3687,20 @@ def _setup_params( return params, dataset_processor async def responses(self, body: NeMoGymResponseCreateParamsNonStreaming = Body()) -> NeMoGymResponse: - params, dataset_processor = self._setup_params(body) + return await self._responses(body) + + async def _responses( + self, + body: NeMoGymResponseCreateParamsNonStreaming, + rollout_id: Optional[str] = None, + *, + token_id_capture_enabled: bool = False, + ) -> NeMoGymResponse: + params, dataset_processor = self._setup_params( + body, + rollout_id, + token_id_capture_enabled=token_id_capture_enabled, + ) with (params.eval_private_dir / "params.json").open("w") as f: f.write(params.model_dump_json(indent=4)) @@ -3783,6 +3818,62 @@ def _item_field(item, name: str): updated_metrics = update_and_read_metrics(params.metrics_fpath, metrics_to_update) + observations: Optional[AgentObservationBundle] = None + if params.rollout_id is not None: + observations_path = params.persistent_dir / OBSERVATIONS_FILENAME + try: + observations = AgentObservationBundle.model_validate_json(observations_path.read_text()) + except (OSError, ValueError): + observations = AgentObservationBundle( + source=f"swe_{params.agent_framework}", + gaps=[ + ObservationGap( + code="agent_artifact_unavailable", + ) + ], + ) + if params.agent_framework == "openhands": + observations.gaps.append( + ObservationGap( + code="model_call_capture_correlation_unavailable", + ) + ) + try: + sandbox_observations = sandbox_observations_from_metrics(updated_metrics) + except ValueError: + print(f"Error creating sandbox observations: {format_exc()}", flush=True) + observations.gaps.append(ObservationGap(code="sandbox_observation_unavailable")) + sandbox_observations = [] + observations.records.extend(sandbox_observations) + for sandbox in sandbox_observations: + observations.gaps.append( + ObservationGap( + code="sandbox_identity_unavailable", + detail=sandbox.role, + ) + ) + if sandbox.cpu_time_s is None: + observations.gaps.append( + ObservationGap( + code="sandbox_cpu_time_unavailable", + detail=sandbox.role, + ) + ) + if sandbox.peak_memory_mib is None: + observations.gaps.append( + ObservationGap( + code="sandbox_memory_usage_unavailable", + detail=sandbox.role, + ) + ) + else: + observations.gaps.append( + ObservationGap( + code="sandbox_memory_usage_sampled", + detail=sandbox.role, + ) + ) + # body.model can be None (replay JSONLs omit it; the openai_model proxy # picks the backend). NeMoGymResponse.model is a required non-None string, # so fall back to the agent's configured model server name. @@ -3798,6 +3889,8 @@ def _item_field(item, name: str): if entry.get("parent_session_id") ] metadata["subagent_trajectories"] = json.dumps(subagent_trajectories) + if observations is not None: + metadata["agent_observations"] = observations.model_dump_json() return NeMoGymResponse( id=f"swebench-{params.instance_id}", @@ -3811,12 +3904,18 @@ def _item_field(item, name: str): metadata=metadata, ) - async def run(self, body: BaseRunRequest) -> SWEBenchVerifyResponse: + async def run(self, body: SWEBenchRunRequest) -> SWEBenchVerifyResponse: async with self._sem: body.responses_create_params.parallel_tool_calls = True body.responses_create_params.tool_choice = "auto" - response = await self.responses(body.responses_create_params) + rollout_id = self.rollout_id_from_run(body) + token_id_capture_enabled = self._token_id_capture_enabled() + response = await self._responses( + body.responses_create_params, + rollout_id, + token_id_capture_enabled=token_id_capture_enabled, + ) metadata, response.metadata = response.metadata, None responses_create_params = body.responses_create_params.model_dump() | { @@ -3827,6 +3926,9 @@ async def run(self, body: BaseRunRequest) -> SWEBenchVerifyResponse: subagent_trajectories = None if "subagent_trajectories" in metadata: subagent_trajectories = json.loads(metadata["subagent_trajectories"]) + observations = None + if rollout_id is not None and "agent_observations" in metadata: + observations = AgentObservationBundle.model_validate_json(metadata["agent_observations"]) return SWEBenchVerifyResponse( responses_create_params=responses_create_params, @@ -3837,6 +3939,7 @@ async def run(self, body: BaseRunRequest) -> SWEBenchVerifyResponse: metadata["instance_config"] ).model_dump(), subagent_trajectories=subagent_trajectories, + ng_agent_observations=observations, ) diff --git a/responses_api_agents/swe_agents/observability.py b/responses_api_agents/swe_agents/observability.py new file mode 100644 index 0000000000..89fa49053a --- /dev/null +++ b/responses_api_agents/swe_agents/observability.py @@ -0,0 +1,264 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Normalize completion artifacts emitted by the SWE agent harnesses.""" + +from __future__ import annotations + +import json +from collections import defaultdict +from collections.abc import Iterable +from pathlib import Path +from typing import Any, Literal + +from nemo_gym.config_types import ModelServerRef +from nemo_gym.responses_converter import VLLMConverter +from nemo_gym.rollout_observability import ( + AgentInvocation, + AgentObservationBundle, + ModelCallRef, + ObservationGap, + SandboxObservation, +) + + +OBSERVATIONS_FILENAME = "agent_observations.json" + + +def _value(metrics: Any, name: str) -> Any: + return metrics.get(name) if isinstance(metrics, dict) else getattr(metrics, name, None) + + +def sandbox_observations_from_metrics(metrics: Any) -> list[SandboxObservation]: + """Map only resource and outcome facts reported by the SWE runner.""" + observations: list[SandboxObservation] = [] + phases = ( + ( + "agent", + _value(metrics, "openhands_run_time"), + _value(metrics, "agent_peak_rss_mb"), + _value(metrics, "agent_timed_out"), + _value(metrics, "oom_killed"), + ), + ( + "verifier", + _value(metrics, "final_eval_time"), + _value(metrics, "eval_peak_rss_mb"), + _value(metrics, "eval_timed_out"), + _value(metrics, "eval_oom_killed"), + ), + ) + for role, wall_time_s, peak_memory_mib, timed_out, oom_killed in phases: + if wall_time_s is None and peak_memory_mib is None and not timed_out and not oom_killed: + continue + outcome = "oom" if oom_killed else "timeout" if timed_out else "unknown" + observations.append( + SandboxObservation( + role=role, + provider="apptainer", + outcome=outcome, + wall_time_s=wall_time_s, + peak_memory_mib=peak_memory_mib, + resource_usage_source="proc_tree_watchdog" if peak_memory_mib is not None else None, + ) + ) + return observations + + +def materialize_completion(data: dict[str, Any]) -> tuple[list[dict[str, Any]], list[Any]]: + """Return the cumulative chat history and tools stored in one artifact.""" + messages = [dict(message) for message in data.get("messages") or [] if isinstance(message, dict)] + tools = list((data.get("kwargs") or {}).get("tools") or []) + try: + final_assistant_message = dict(data["response"]["choices"][0]["message"]) + except (KeyError, IndexError, TypeError): + return messages, tools + + provider_fields = data.get("provider_specific_fields") or {} + for key in ( + "prompt_token_ids", + "generation_token_ids", + "generation_log_probs", + "routed_experts", + ): + if key in provider_fields: + final_assistant_message[key] = provider_fields[key] + + if final_assistant_message.get("content") or final_assistant_message.get("tool_calls"): + messages.append(final_assistant_message) + return messages, tools + + +def _artifact_order(path: Path, data: dict[str, Any]) -> tuple[int, float, str]: + turn = data.get("turn") + if isinstance(turn, int) and not isinstance(turn, bool): + return (2, float(turn), path.name) + timestamp = data.get("timestamp") + if isinstance(timestamp, (int, float)) and not isinstance(timestamp, bool): + return (1, float(timestamp), path.name) + try: + mtime = path.stat().st_mtime + except OSError: + mtime = 0.0 + return (0, mtime, path.name) + + +def _gap(code: str, *, invocation_id: str | None = None, detail: str | None = None) -> ObservationGap: + return ObservationGap(code=code, invocation_id=invocation_id, detail=detail) + + +def _tool_call_ids(conversation: Iterable[Any]) -> Iterable[str]: + for item in conversation: + if getattr(item, "type", None) not in {"function_call", "custom_tool_call"}: + continue + call_id = getattr(item, "call_id", None) + if isinstance(call_id, str) and call_id: + yield call_id + + +def build_swe_observations( + completion_paths: Iterable[Path], + *, + framework: Literal["openhands", "opencode"], + model_ref: ModelServerRef, +) -> AgentObservationBundle: + """Build one compact bundle while retaining every exact model response ID. + + Completion files carry cumulative histories. Only the latest file per + invocation supplies the conversation; all files contribute their response ID. + """ + source = f"swe_{framework}" + gaps: list[ObservationGap] = [] + artifacts: dict[str, list[tuple[Path, dict[str, Any]]]] = defaultdict(list) + + paths = tuple(sorted(Path(path) for path in completion_paths)) + for path in paths: + try: + data = json.loads(path.read_text()) + except (OSError, UnicodeError, json.JSONDecodeError): + gaps.append(_gap("agent_artifact_parse_failed", detail=path.name)) + continue + if not isinstance(data, dict): + gaps.append(_gap("agent_artifact_parse_failed", detail=path.name)) + continue + + if framework == "openhands": + invocation_id = "root" + else: + invocation_id = data.get("session_id") + if not isinstance(invocation_id, str) or not invocation_id: + gaps.append(_gap("agent_session_id_missing", detail=path.name)) + continue + artifacts[invocation_id].append((path, data)) + + if framework == "openhands" and "root" not in artifacts: + artifacts["root"] = [] + if not paths: + gaps.append(_gap("agent_artifact_unavailable")) + + converter = VLLMConverter(return_token_id_information=True) + invocations: list[AgentInvocation] = [] + parents: dict[str, str | None] = {} + response_owners: dict[str, str] = {} + + for invocation_id, entries in artifacts.items(): + entries.sort(key=lambda item: _artifact_order(*item)) + response_ids: list[str] = [] + seen_response_ids: set[str] = set() + parent_values: set[str | None] = set() + + for path, data in entries: + response = data.get("response") + response_id = response.get("id") if isinstance(response, dict) else None + if isinstance(response_id, str) and response_id and response_id != "unknown": + owner = response_owners.setdefault(response_id, invocation_id) + if owner != invocation_id: + gaps.append( + _gap( + "model_response_owner_conflict", + invocation_id=invocation_id, + detail=response_id, + ) + ) + elif response_id not in seen_response_ids: + response_ids.append(response_id) + seen_response_ids.add(response_id) + else: + gaps.append(_gap("model_response_id_missing", invocation_id=invocation_id, detail=path.name)) + + if framework == "opencode": + parent = data.get("parent_session_id") + if parent in (None, ""): + parent_values.add(None) + elif isinstance(parent, str): + parent_values.add(parent) + else: + gaps.append(_gap("parent_invocation_id_invalid", invocation_id=invocation_id, detail=path.name)) + + parent: str | None = None + if framework == "opencode": + if len(parent_values) == 1: + parent = next(iter(parent_values)) + elif len(parent_values) > 1: + gaps.append(_gap("parent_invocation_conflict", invocation_id=invocation_id)) + parents[invocation_id] = parent + + conversation = [] + if entries: + _, latest = entries[-1] + try: + messages, _ = materialize_completion(latest) + conversation = converter.chat_completions_messages_to_responses_items(messages) + except Exception as exc: + gaps.append( + _gap( + "conversation_conversion_failed", + invocation_id=invocation_id, + detail=type(exc).__name__, + ) + ) + + for tool_call_id in _tool_call_ids(conversation): + gaps.append( + _gap( + "tool_timing_unavailable", + invocation_id=invocation_id, + detail=tool_call_id, + ) + ) + gaps.append( + _gap( + "tool_outcome_unavailable", + invocation_id=invocation_id, + detail=tool_call_id, + ) + ) + + invocations.append( + AgentInvocation( + invocation_id=invocation_id, + parent_invocation_id=parent, + model_calls=[ + ModelCallRef(model_ref=model_ref, response_id=response_id) for response_id in response_ids + ], + conversation=conversation, + ) + ) + + invocation_ids = set(parents) + for invocation_id, parent in parents.items(): + if parent is not None and parent not in invocation_ids: + gaps.append(_gap("parent_invocation_missing", invocation_id=invocation_id, detail=parent)) + if parent is not None: + gaps.append(_gap("subagent_spawn_tool_unavailable", invocation_id=invocation_id)) + + if framework == "openhands": + gaps.append(_gap("subagent_hierarchy_unavailable", invocation_id="root")) + gaps.append(_gap("context_compaction_unavailable")) + + invocations.sort(key=lambda invocation: (invocation.parent_invocation_id is not None, invocation.invocation_id)) + return AgentObservationBundle( + source=source, + records=invocations, + gaps=gaps, + ) diff --git a/responses_api_agents/swe_agents/tests/test_app.py b/responses_api_agents/swe_agents/tests/test_app.py index b1ea4025e6..9991aa395e 100644 --- a/responses_api_agents/swe_agents/tests/test_app.py +++ b/responses_api_agents/swe_agents/tests/test_app.py @@ -30,6 +30,7 @@ NeMoGymResponse, NeMoGymResponseCreateParamsNonStreaming, ) +from nemo_gym.rollout_observability import AgentInvocation, AgentObservationBundle, ModelCallRef, SandboxObservation from nemo_gym.server_utils import ServerClient from responses_api_agents.swe_agents.app import ( ActiveContainerCommand, @@ -44,6 +45,7 @@ SweBenchDatasetProcessor, SWEBenchMetrics, SweBenchMultilingualDatasetProcessor, + SWEBenchRunRequest, SWEBenchVerifyResponse, SWEBenchWrapper, SWEBenchWrapperConfig, @@ -57,6 +59,7 @@ runner_ray_remote, update_and_read_metrics, ) +from responses_api_agents.swe_agents.observability import OBSERVATIONS_FILENAME, sandbox_observations_from_metrics SWE_AGENTS_DIR = Path(__file__).resolve().parent.parent @@ -343,6 +346,12 @@ def test_resolved_defaults(self) -> None: assert config.resolved_agent_cls == "CodeActAgent" assert config.resolved_diversify_tool_names is False assert config.resolved_camel_case_tool_names is False + serialized = config.model_dump() + assert "rollout_id" not in serialized + assert serialized["token_id_capture_enabled"] is False + + restored = SWEBenchWrapperInstanceConfig.model_validate(serialized) + assert restored.token_id_capture_enabled is False class TestSWEBenchMetrics: @@ -359,6 +368,34 @@ def test_with_values(self) -> None: assert metrics.resolved is True assert metrics.ray_queue_time == 1.5 + def test_maps_explicit_sandbox_metrics_without_inferring_cpu_or_oom(self) -> None: + observations = sandbox_observations_from_metrics( + SWEBenchMetrics( + openhands_run_time=12.5, + agent_peak_rss_mb=2048, + final_eval_time=3.0, + eval_timed_out=True, + ) + ) + + assert [(item.role, item.outcome) for item in observations] == [ + ("agent", "unknown"), + ("verifier", "timeout"), + ] + assert observations[0].wall_time_s == 12.5 + assert observations[0].peak_memory_mib == 2048 + assert observations[0].sandbox_id is None + assert observations[0].cpu_time_s is None + assert observations[0].resource_usage_source == "proc_tree_watchdog" + assert observations[1].wall_time_s == 3.0 + assert observations[1].peak_memory_mib is None + + def test_explicit_oom_takes_precedence_over_timeout(self) -> None: + [observation] = sandbox_observations_from_metrics(SWEBenchMetrics(agent_timed_out=True, oom_killed=True)) + + assert observation.role == "agent" + assert observation.outcome == "oom" + class TestSWEBenchVerifyResponse: def test_fields_exist(self) -> None: @@ -368,6 +405,11 @@ def test_fields_exist(self) -> None: assert "instance_config" in fields assert "subagent_trajectories" in fields + def test_optional_observations_are_excluded_when_disabled(self) -> None: + response = SWEBenchVerifyResponse.model_construct(ng_agent_observations=None) + + assert "ng_agent_observations" not in response.model_dump() + ######################################## # update_metrics tests @@ -1218,6 +1260,33 @@ def test_get_run_command_writes_script(self, _stub_model_server_lookup) -> None: assert "--max-turns" not in script # max_turns is positional assert str(config.agent_max_turns) in script + @pytest.mark.parametrize( + ("rollout_id", "token_id_capture_enabled", "expected_base_url"), + [ + (None, False, "http://test-host:12345"), + ("7-2", False, "http://test-host:12345/ng-rollout/7-2"), + ("7-2", True, "http://test-host:12345/ng-rollout/7-2/training-token-capture"), + ], + ids=("disabled", "observability", "token-capture"), + ) + def test_get_run_command_routes_each_capture_state( + self, + _stub_model_server_lookup, + rollout_id: str | None, + token_id_capture_enabled: bool, + expected_base_url: str, + ) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + config = self._opencode_config( + tmpdir, + rollout_id=rollout_id, + token_id_capture_enabled=token_id_capture_enabled, + ) + OpenCodeHarnessProcessor(config=config).get_run_command() + + script = self._read_agent_script(config) + assert f"NEMO_GYM_MODEL_SERVER_BASE_URL={expected_base_url}" in script + def test_get_run_command_subagents_disabled_by_default(self, _stub_model_server_lookup) -> None: with tempfile.TemporaryDirectory() as tmpdir: config = self._opencode_config(tmpdir) @@ -1341,13 +1410,24 @@ def test_returns_empty_for_non_string_non_dict(self): ######################################## -def _write_completion(path: Path, *, session_id, parent_session_id, turn, content_text="hello"): +def _write_completion( + path: Path, + *, + session_id, + parent_session_id, + turn, + content_text="hello", + response_id=None, +): path.parent.mkdir(parents=True, exist_ok=True) + response = {"choices": [{"message": {"role": "assistant", "content": content_text}}]} + if response_id is not None: + response["id"] = response_id path.write_text( json.dumps( { "messages": [{"role": "user", "content": "Fix bug"}], - "response": {"choices": [{"message": {"role": "assistant", "content": content_text}}]}, + "response": response, "provider_specific_fields": {"prompt_token_ids": [1, 2, 3]}, "kwargs": {"tools": [{"name": "edit"}]}, "session_id": session_id, @@ -1363,7 +1443,7 @@ class TestOpencodeMultiSessionCopy: """`_openhands_dir_copy_from_host` must keep latest-per-session, not just one global latest, when the opencode bench writes session-tagged JSONs.""" - def _agent(self, tmpdir) -> RunOpenHandsAgent: + def _agent(self, tmpdir, **overrides) -> RunOpenHandsAgent: opencode_setup_dir = Path(tmpdir) / "opencode_setup" opencode_setup_dir.mkdir(parents=True, exist_ok=True) cfg = _make_instance_config( @@ -1372,6 +1452,7 @@ def _agent(self, tmpdir) -> RunOpenHandsAgent: opencode_setup_dir=opencode_setup_dir, agent_framework_repo="https://example.invalid/opencode.git", agent_framework_commit="deadbeef", + **overrides, ) return RunOpenHandsAgent(config=cfg) @@ -1447,6 +1528,86 @@ def test_falls_back_to_single_latest_when_files_untagged(self) -> None: copied = sorted((traj_root / "llm_completions" / inst).glob("*.json")) assert [p.name for p in copied] == ["new.json"] + assert not (agent.config.persistent_dir / "agent_observations.json").exists() + + def test_observation_failure_does_not_break_existing_copy_path(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + agent = self._agent(tmpdir, rollout_id="7-2") + eval_dir = self._eval_dir(agent) + inst = agent.config.instance_id + comp_root = eval_dir / inst / "bench_run" / "llm_completions" / inst + artifact = comp_root / "turn.json" + _write_completion( + artifact, + session_id="main", + parent_session_id=None, + turn=0, + response_id="resp-0", + ) + + with patch.object(swe_app, "build_swe_observations", side_effect=ValueError("invalid artifact")): + agent._openhands_dir_copy_from_host(output_file_path=None) + + copied = agent.config.trajectories_root / "llm_completions" / inst / artifact.name + assert copied.is_file() + bundle = AgentObservationBundle.model_validate_json( + (agent.config.persistent_dir / "agent_observations.json").read_text() + ) + assert [gap.code for gap in bundle.gaps] == ["observation_parse_failed"] + + def test_persists_all_response_ids_before_source_cleanup(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + agent = self._agent(tmpdir, rollout_id="7-2") + eval_dir = self._eval_dir(agent) + inst = agent.config.instance_id + comp_root = eval_dir / inst / "bench_run" / "llm_completions" / inst + _write_completion( + comp_root / "turn-0.json", + session_id="main", + parent_session_id=None, + turn=0, + response_id="resp-0", + ) + _write_completion( + comp_root / "turn-1.json", + session_id="main", + parent_session_id=None, + turn=1, + response_id="resp-1", + ) + + agent._openhands_dir_copy_from_host(output_file_path=None) + + assert not eval_dir.exists() + bundle = AgentObservationBundle.model_validate_json( + (agent.config.persistent_dir / "agent_observations.json").read_text() + ) + [invocation] = [record for record in bundle.records if isinstance(record, AgentInvocation)] + assert [ref.response_id for ref in invocation.model_calls] == ["resp-0", "resp-1"] + + def test_token_capture_only_builds_observation_artifacts(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + agent = self._agent(tmpdir, rollout_id="7-2", token_id_capture_enabled=True) + eval_dir = self._eval_dir(agent) + inst = agent.config.instance_id + comp_root = eval_dir / inst / "bench_run" / "llm_completions" / inst + artifact = comp_root / "turn.json" + _write_completion( + artifact, + session_id="main", + parent_session_id=None, + turn=0, + response_id="resp-0", + ) + + agent._openhands_dir_copy_from_host(output_file_path=None) + + assert (agent.config.trajectories_root / "llm_completions" / inst / artifact.name).is_file() + bundle = AgentObservationBundle.model_validate_json( + (agent.config.persistent_dir / "agent_observations.json").read_text() + ) + [invocation] = [record for record in bundle.records if isinstance(record, AgentInvocation)] + assert [ref.response_id for ref in invocation.model_calls] == ["resp-0"] class TestGetOpenhandsTrajectoryFromCompletions: @@ -2516,11 +2677,123 @@ async def test_responses_exception_writes_traceback(self, monkeypatch) -> None: with pytest.raises(RuntimeError, match="test error"): await wrapper.responses(body) + @pytest.mark.parametrize( + ("rollout_id", "token_id_capture_enabled", "expects_observations"), + [ + (None, False, False), + ("7-2", True, True), + ], + ids=("direct", "token-capture-only"), + ) + @pytest.mark.asyncio + async def test_inner_responses_gates_observations_on_rollout_id( + self, + monkeypatch, + rollout_id: str | None, + token_id_capture_enabled: bool, + expects_observations: bool, + ) -> None: + wrapper = _create_wrapper(monkeypatch) + runner = MagicMock() + runner.remote = AsyncMock(return_value=None) + monkeypatch.setattr(swe_app, "runner_ray_remote", runner) + + with tempfile.TemporaryDirectory() as tmpdir: + params = _make_instance_config( + tmpdir, + rollout_id=rollout_id, + token_id_capture_enabled=token_id_capture_enabled, + ) + params.metrics_fpath.write_text(json.dumps({"openhands_run_time": 1.0})) + observations = AgentObservationBundle( + source="swe_openhands", + records=[ + AgentInvocation( + invocation_id="root", + model_calls=[ + ModelCallRef( + model_ref=params.model_server, + response_id="resp-openhands-0", + ) + ], + ) + ], + ) + (params.persistent_dir / OBSERVATIONS_FILENAME).write_text(observations.model_dump_json()) + monkeypatch.setattr( + SWEBenchWrapper, + "get_openhands_trajectory_from_completions", + MagicMock(return_value=([], [], 0)), + ) + + response = await wrapper._inner_responses(params, MagicMock()) + serialized = response.metadata or {} + + assert ("agent_observations" in serialized) is expects_observations + if expects_observations: + bundle = AgentObservationBundle.model_validate_json(serialized["agent_observations"]) + [invocation] = [record for record in bundle.records if isinstance(record, AgentInvocation)] + assert [ref.response_id for ref in invocation.model_calls] == ["resp-openhands-0"] + assert "model_call_capture_correlation_unavailable" in {gap.code for gap in bundle.gaps} + [sandbox] = [record for record in bundle.records if isinstance(record, SandboxObservation)] + assert sandbox.role == "agent" + assert sandbox.sandbox_id is None + assert ("sandbox_identity_unavailable", "agent") in {(gap.code, gap.detail) for gap in bundle.gaps} + + @pytest.mark.asyncio + async def test_inner_responses_invalid_sandbox_metrics_fail_open(self, monkeypatch) -> None: + wrapper = _create_wrapper(monkeypatch) + monkeypatch.setattr(swe_app, "runner_ray_remote", MagicMock(remote=AsyncMock(return_value=None))) + + with tempfile.TemporaryDirectory() as tmpdir: + params = _make_instance_config(tmpdir, rollout_id="7-2") + params.metrics_fpath.write_text(json.dumps({"openhands_run_time": -1})) + monkeypatch.setattr( + SWEBenchWrapper, + "get_openhands_trajectory_from_completions", + MagicMock(return_value=([], [], 0)), + ) + + response = await wrapper._inner_responses(params, MagicMock()) + + bundle = AgentObservationBundle.model_validate_json(response.metadata["agent_observations"]) + assert not any(isinstance(record, SandboxObservation) for record in bundle.records) + assert [gap.code for gap in bundle.gaps if gap.code.startswith("sandbox_")] == [ + "sandbox_observation_unavailable" + ] + class TestSWEBenchWrapperRun: + @pytest.mark.parametrize( + ("model_call_capture_enabled", "token_id_capture_enabled", "expected_rollout_id"), + [ + (False, False, None), + (True, False, "7-2-a1"), + (False, True, "7-2-a1"), + (True, True, "7-2-a1"), + ], + ids=("disabled", "observability-only", "token-capture-only", "both"), + ) @pytest.mark.asyncio - async def test_run_resolved(self, monkeypatch) -> None: + async def test_run_resolved_routes_each_capture_state( + self, + monkeypatch, + model_call_capture_enabled: bool, + token_id_capture_enabled: bool, + expected_rollout_id: str | None, + ) -> None: wrapper = _create_wrapper(monkeypatch) + wrapper.server_client.global_config_dict = { + "observability_enabled": model_call_capture_enabled, + "token_id_capture": { + "enabled": token_id_capture_enabled, + "all_agents": token_id_capture_enabled, + }, + } + observations = AgentObservationBundle( + source="swe_opencode", + records=[AgentInvocation(invocation_id="main")], + ) mock_response = NeMoGymResponse( id="swebench-test", @@ -2535,13 +2808,14 @@ async def test_run_resolved(self, monkeypatch) -> None: "input": "[]", "metrics": json.dumps({"resolved": True, "patch_exists": True}), "instance_config": _make_instance_config(tempfile.mkdtemp()).model_dump_json(), + "agent_observations": observations.model_dump_json(), }, ) - with patch.object(SWEBenchWrapper, "responses", new_callable=AsyncMock, return_value=mock_response): - from nemo_gym.base_resources_server import BaseRunRequest - - body = BaseRunRequest( + with patch.object( + SWEBenchWrapper, "_responses", new_callable=AsyncMock, return_value=mock_response + ) as responses_mock: + body = SWEBenchRunRequest( responses_create_params=NeMoGymResponseCreateParamsNonStreaming( model="test-model", input=[], @@ -2553,12 +2827,20 @@ async def test_run_resolved(self, monkeypatch) -> None: "split": "test", "instance_dict": "{}", }, - ) + ), + _ng_task_index=7, + _ng_rollout_index=2, + _ng_attempt_index=1, ) result = await wrapper.run(body) assert isinstance(result, SWEBenchVerifyResponse) assert result.reward == 1.0 + assert result.ng_agent_observations == (observations if expected_rollout_id is not None else None) + assert responses_mock.await_args.args[1] == expected_rollout_id + assert responses_mock.await_args.kwargs == { + "token_id_capture_enabled": token_id_capture_enabled, + } @pytest.mark.asyncio async def test_run_not_resolved(self, monkeypatch) -> None: @@ -2580,10 +2862,8 @@ async def test_run_not_resolved(self, monkeypatch) -> None: }, ) - with patch.object(SWEBenchWrapper, "responses", new_callable=AsyncMock, return_value=mock_response): - from nemo_gym.base_resources_server import BaseRunRequest - - body = BaseRunRequest( + with patch.object(SWEBenchWrapper, "_responses", new_callable=AsyncMock, return_value=mock_response): + body = SWEBenchRunRequest( responses_create_params=NeMoGymResponseCreateParamsNonStreaming( model="test-model", input=[], diff --git a/responses_api_agents/swe_agents/tests/test_observability.py b/responses_api_agents/swe_agents/tests/test_observability.py new file mode 100644 index 0000000000..573df0784f --- /dev/null +++ b/responses_api_agents/swe_agents/tests/test_observability.py @@ -0,0 +1,184 @@ +# 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.rollout_observability import AgentInvocation +from responses_api_agents.swe_agents.observability import build_swe_observations + + +MODEL_REF = ModelServerRef(type="responses_api_models", name="policy_model") + + +def _invocations(bundle) -> list[AgentInvocation]: + return [record for record in bundle.records if isinstance(record, AgentInvocation)] + + +def _completion( + path: Path, + *, + response_id: str | None, + content: str, + turn: int, + session_id: str | None = None, + parent_session_id: str | None = None, + tool_call_id: str | None = None, +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + response = { + "choices": [{"message": {"role": "assistant", "content": content}}], + } + if response_id is not None: + response["id"] = response_id + messages = [{"role": "user", "content": "Fix the bug"}] + if tool_call_id is not None: + response["choices"][0]["message"]["tool_calls"] = [ + { + "id": tool_call_id, + "type": "function", + "function": {"name": "shell", "arguments": '{"cmd":"pytest"}'}, + } + ] + messages.append( + { + "role": "tool", + "tool_call_id": tool_call_id, + "content": "passed", + } + ) + path.write_text( + json.dumps( + { + "messages": messages, + "response": response, + "kwargs": {"tools": []}, + "session_id": session_id, + "parent_session_id": parent_session_id, + "turn": turn, + } + ) + ) + + +def test_opencode_preserves_tree_and_all_response_ids(tmp_path: Path) -> None: + _completion( + tmp_path / "main-0.json", + response_id="resp-main-0", + content="first", + turn=0, + session_id="main", + ) + _completion( + tmp_path / "main-1.json", + response_id="resp-main-1", + content="latest", + turn=1, + session_id="main", + ) + _completion( + tmp_path / "child.json", + response_id="resp-child", + content="child", + turn=0, + session_id="child", + parent_session_id="main", + tool_call_id="call-child", + ) + + bundle = build_swe_observations(tmp_path.glob("*.json"), framework="opencode", model_ref=MODEL_REF) + + records = _invocations(bundle) + invocations = {item.invocation_id: item for item in records} + assert invocations["main"].parent_invocation_id is None + assert invocations["child"].parent_invocation_id == "main" + assert [ref.response_id for ref in invocations["main"].model_calls] == [ + "resp-main-0", + "resp-main-1", + ] + assert [ref.response_id for ref in invocations["child"].model_calls] == ["resp-child"] + assert all(ref.model_ref == MODEL_REF for invocation in records for ref in invocation.model_calls) + assert "latest" in json.dumps([item.model_dump() for item in invocations["main"].conversation]) + assert "subagent_hierarchy_unavailable" not in {gap.code for gap in bundle.gaps} + assert "subagent_spawn_tool_unavailable" in {gap.code for gap in bundle.gaps} + assert {(gap.code, gap.invocation_id) for gap in bundle.gaps} >= { + ("tool_timing_unavailable", "child"), + ("tool_outcome_unavailable", "child"), + } + + +def test_openhands_uses_one_root_and_latest_cumulative_conversation(tmp_path: Path) -> None: + old = tmp_path / "old.json" + latest = tmp_path / "latest.json" + _completion(old, response_id="resp-0", content="old", turn=0) + _completion(latest, response_id="resp-1", content="latest", turn=1) + bundle = build_swe_observations((old, latest), framework="openhands", model_ref=MODEL_REF) + + [root] = _invocations(bundle) + assert root.invocation_id == "root" + assert root.parent_invocation_id is None + assert [ref.response_id for ref in root.model_calls] == ["resp-0", "resp-1"] + assert "latest" in json.dumps([item.model_dump() for item in root.conversation]) + assert {gap.code for gap in bundle.gaps} == { + "subagent_hierarchy_unavailable", + "context_compaction_unavailable", + } + + +def test_missing_response_id_is_reported_without_a_synthetic_ref(tmp_path: Path) -> None: + _completion( + tmp_path / "missing.json", + response_id=None, + content="answer", + turn=0, + session_id="main", + ) + + bundle = build_swe_observations(tmp_path.glob("*.json"), framework="opencode", model_ref=MODEL_REF) + + [invocation] = _invocations(bundle) + assert invocation.model_calls == [] + assert "model_response_id_missing" in {gap.code for gap in bundle.gaps} + + +def test_empty_openhands_capture_still_has_one_root_and_an_exact_gap() -> None: + bundle = build_swe_observations((), framework="openhands", model_ref=MODEL_REF) + + assert [invocation.invocation_id for invocation in _invocations(bundle)] == ["root"] + assert "agent_artifact_unavailable" in {gap.code for gap in bundle.gaps} + + +def test_untagged_opencode_artifact_does_not_fabricate_an_invocation( + tmp_path: Path, +) -> None: + artifact = tmp_path / "legacy.json" + _completion( + artifact, + response_id="resp-legacy", + content="legacy", + turn=0, + ) + + bundle = build_swe_observations((artifact,), framework="opencode", model_ref=MODEL_REF) + + assert _invocations(bundle) == [] + assert "agent_session_id_missing" in {gap.code for gap in bundle.gaps} + + +def test_duplicate_response_id_is_not_assigned_to_two_invocations(tmp_path: Path) -> None: + _completion(tmp_path / "root.json", response_id="resp-1", content="root", turn=0, session_id="root") + _completion( + tmp_path / "child.json", + response_id="resp-1", + content="child", + turn=0, + session_id="child", + parent_session_id="root", + ) + + bundle = build_swe_observations(tmp_path.glob("*.json"), framework="opencode", model_ref=MODEL_REF) + + refs = [ref for invocation in _invocations(bundle) for ref in invocation.model_calls] + assert len(refs) == 1 + assert "model_response_owner_conflict" in {gap.code for gap in bundle.gaps}