[rollout-observability][2/7] Add Claude Code rollout observations

fern/versions/latest/pages/model-server/model-call-capture.mdx

@@ -126,12 +126,15 @@ invocation's `conversation` contains the ordered, normalized items exposed by th
 
 Agent observations and model-call capture are separate evidence. Join an invocation's model-call
 references by `model_call_id`, or by the exact `(model_ref, response_id)` pair when the harness sees
-the protocol response ID. Do not infer ownership from timestamps, text, or list position. The full
-model request and response remain in `CaptureStore`; rollout attachments intentionally omit them.
+the protocol response ID. An integration may resolve an otherwise hidden call only through a
+producer-specific, unique exact match against its retained artifact and the raw capture. Ambiguous
+matches remain unowned; timestamps or list position alone are never sufficient. The full model
+request and response remain in `CaptureStore`; rollout attachments intentionally omit them.
 
 Compaction records distinguish the calls immediately before and after the context change from
 `model_calls` used to perform the compaction. Compaction calls are exact references to calls owned by
-the enclosing invocation; opaque integrations leave them empty rather than infer them.
+the enclosing invocation. Integrations without an explicit identifier or a unique exact match leave
+them empty and report the gap.
 
 Tool timestamps are UTC Unix seconds when the source provides wall-clock time. `duration_ms` is the
 measured interval, and `timing_source` identifies executor, harness, or artifact-derived timing.

nemo_gym/base_responses_api_model.py

@@ -1287,7 +1287,22 @@ def merge_model_call_capture_into_record(
     if observations is not None:
         try:
             bundle = AgentObservationBundle.model_validate(observations)
-            record["ng_agent_observations"] = join_model_call_observations(bundle, calls).model_dump(mode="json")
+            if bundle.source == "claude_code":
+                try:
+                    from responses_api_agents.claude_code_agent.observability import (
+                        associate_claude_code_compaction_calls,
+                    )
+
+                    bundle = associate_claude_code_compaction_calls(bundle, calls)
+                except Exception:
+                    logger.warning(
+                        "Could not associate Claude Code compaction calls for rollout %s.",
+                        rollout_id,
+                        exc_info=True,
+                    )
+                    bundle.gaps.append(ObservationGap(code="compaction_model_call_join_failed"))
+            bundle = join_model_call_observations(bundle, calls)
+            record["ng_agent_observations"] = bundle.model_dump(mode="json")
         except Exception:
             logger.warning("Could not join agent observations for rollout %s.", rollout_id, exc_info=True)
             gaps.append(ObservationGap(code="agent_observation_join_failed"))

nemo_gym/rollout_observability.py

@@ -131,7 +131,10 @@ class ContextCompactionObservation(ObservationModel):
     )
     model_calls: list[ModelCallRef] = Field(
         default_factory=list,
-        description=("Invocation-owned model calls used for compaction, in producer-observed order; never inferred."),
+        description=(
+            "Invocation-owned model calls used for compaction, joined by explicit identifiers or a unique "
+            "producer-specific exact match."
+        ),
     )
     after_model_call: Optional[ModelCallRef] = Field(
         default=None,

responses_api_agents/claude_code_agent/README.md

@@ -196,5 +196,6 @@ The skills path is resolved like `input_jsonl_fpath` (relative paths check the w
 ## Limitations
 
 - Eval only for now. Token IDs and logprobs are not wired up yet.
-- Does not go through Gym's model server. Token counts come from Claude Code's own usage reporting.
-- `turns_used` counts assistant messages right now, not tool calls.
+- With `model_server`, model calls go through Gym and can be captured. Direct Anthropic or
+  `anthropic_base_url` runs bypass Gym capture.
+- `turns_used` counts assistant messages, not tool calls.

responses_api_agents/claude_code_agent/app.py

@@ -22,13 +22,14 @@
 import subprocess
 import tempfile
 from asyncio import Semaphore
+from contextlib import suppress
 from pathlib import Path
-from time import time
-from typing import Any, Optional
+from time import monotonic, time
+from typing import Any, Callable, Optional
 from uuid import uuid4
 
 from fastapi import Request
-from pydantic import ConfigDict, PrivateAttr
+from pydantic import ConfigDict, Field, PrivateAttr
 
 from nemo_gym.base_resources_server import NEMO_GYM_MCP_METADATA_KEY, BaseRunRequest, BaseVerifyResponse
 from nemo_gym.base_responses_api_agent import BaseResponsesAPIAgentConfig, Body, SimpleResponsesAPIAgent
@@ -46,8 +47,10 @@
     NeMoGymResponseOutputTokensDetails,
     NeMoGymResponseUsage,
 )
+from nemo_gym.rollout_observability import AgentEpisode, AgentObservationBundle, ObservationGap
 from nemo_gym.server_utils import apply_rollout_prefix, get_response_json, raise_for_status
 from nemo_gym.skills import stage_skills
+from responses_api_agents.claude_code_agent.observability import extract_claude_code_observations
 from responses_api_agents.claude_code_agent.setup_claude_code import ensure_claude_code
 
 
@@ -86,6 +89,9 @@ def parse_stream_json(stdout: str) -> tuple[list[Any], dict]:
     total_input = 0
     total_output = 0
     num_turns: Optional[int] = None
+    result_metadata: dict[str, Any] = {}
+    compacting_sessions: set[str] = set()
+    compaction_attempts: list[dict[str, str]] = []
 
     for event in raw_events:
         etype = event.get("type")
@@ -97,6 +103,13 @@ def parse_stream_json(stdout: str) -> tuple[list[Any], dict]:
             # Claude Code's authoritative turn counter (what --max-turns bounds).
             if event.get("num_turns") is not None:
                 num_turns = int(event["num_turns"])
+            if isinstance(event.get("subtype"), str):
+                result_metadata["subtype"] = event["subtype"]
+            if isinstance(event.get("is_error"), bool):
+                result_metadata["is_error"] = event["is_error"]
+            duration_ms = event.get("duration_ms")
+            if isinstance(duration_ms, (int, float)) and not isinstance(duration_ms, bool) and duration_ms >= 0:
+                result_metadata["duration_ms"] = float(duration_ms)
 
         elif etype == "assistant":
             message = event.get("message", {})
@@ -171,12 +184,44 @@ def parse_stream_json(stdout: str) -> tuple[list[Any], dict]:
                     )
                 )
 
+        elif etype == "system" and event.get("subtype") == "status":
+            session_id = event.get("session_id")
+            if not isinstance(session_id, str) or not session_id:
+                continue
+            if event.get("status") == "compacting":
+                compacting_sessions.add(session_id)
+                continue
+            compact_result = event.get("compact_result")
+            if compact_result in {"failed", "success"}:
+                if compact_result == "failed":
+                    compaction_attempts.append({"invocation_id": session_id, "outcome": "failed"})
+                compacting_sessions.discard(session_id)
+
+    compaction_attempts.extend(
+        {"invocation_id": session_id, "outcome": "unknown"} for session_id in compacting_sessions
+    )
     metadata: dict = {"input_tokens": total_input, "output_tokens": total_output}
     if num_turns is not None:
         metadata["num_turns"] = num_turns
+    if compaction_attempts:
+        metadata["compaction_attempts"] = compaction_attempts
+    metadata.update(result_metadata)
     return output_items, metadata
 
 
+def _invocation_outcome(metadata: dict[str, Any], returncode: int | None) -> tuple[str, str | None]:
+    subtype = metadata.get("subtype")
+    if subtype == "error_max_turns":
+        return "incomplete", subtype
+    if metadata.get("is_error") is True or (isinstance(subtype, str) and subtype.startswith("error_")):
+        return "failed", subtype if isinstance(subtype, str) else "agent_error"
+    if returncode not in (0, None):
+        return "failed", f"process_exit_{returncode}"
+    if subtype == "success":
+        return "completed", None
+    return "incomplete", "result_missing"
+
+
 def _extract_instruction(body_input) -> tuple[str, Optional[str]]:
     """Return (user_message, system_message) from a responses body input list."""
     items = list(body_input)
@@ -243,6 +288,9 @@ class ClaudeCodeAgentVerifyResponse(BaseVerifyResponse):
     model_config = ConfigDict(extra="allow")
     turns_used: int = 0
     finished_naturally: bool = False
+    ng_agent_observations: Optional[AgentObservationBundle] = Field(
+        default=None, exclude_if=lambda value: value is None
+    )
 
 
 class ClaudeCodeAgent(SimpleResponsesAPIAgent):
@@ -381,8 +429,9 @@ async def _run_claude_code(
         mcp_config: Optional[str] = None,
         skills_path: Optional[str] = None,
         rollout_id: Optional[str] = None,
-    ) -> tuple[str, str]:
-        """Run claude -p --output-format=stream-json and return (stdout, model_name).
+        observation_collector: Optional[Callable[[Path, dict[str, Any]], None]] = None,
+    ) -> tuple[list[Any], str, dict[str, Any]]:
+        """Run Claude Code and return parsed output, model name, and run metadata.
 
         When ``rollout_id`` is set and a model server is configured, the per-rollout capture prefix is
         applied to ANTHROPIC_BASE_URL so the CLI's streaming /v1/messages calls correlate to this rollout.
@@ -393,6 +442,7 @@ async def _run_claude_code(
         api_key = self.config.anthropic_api_key
 
         claude_config_dir = None
+        run_metadata: dict[str, Any] = {"status": "unknown"}
         try:
             # Inside the try so a bad skills.path (raising in stage_skills) still cleans up the
             # partially-created config dir in the finally rather than leaking it per failing request.
@@ -420,28 +470,60 @@ async def _run_claude_code(
                 skills_active=bool(skills_path),
             )
 
+            process_started_at = monotonic()
             proc = await asyncio.create_subprocess_exec(
                 *cmd,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
                 env=env,
             )
+            communication = asyncio.create_task(proc.communicate())
             try:
-                stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.config.timeout)
+                stdout, stderr = await asyncio.wait_for(
+                    asyncio.shield(communication),
+                    timeout=self.config.timeout,
+                )
             except asyncio.TimeoutError:
-                proc.kill()
-                await proc.communicate()
+                if proc.returncode is None:
+                    with suppress(ProcessLookupError):
+                        proc.kill()
+                stdout, _ = await communication
                 LOG.warning("claude-code timed out after %ds", self.config.timeout)
-                return "", model
+                _, run_metadata = parse_stream_json(stdout.decode(errors="replace"))
+                run_metadata.update(
+                    status="incomplete",
+                    error_type="timeout",
+                    duration_ms=(monotonic() - process_started_at) * 1000,
+                )
+                return [], model, run_metadata
+            except asyncio.CancelledError:
+                if proc.returncode is None:
+                    with suppress(ProcessLookupError):
+                        proc.kill()
+                await asyncio.gather(communication, return_exceptions=True)
+                raise
 
             if proc.returncode not in (0, None):
                 LOG.warning("claude-code exited %d: %s", proc.returncode, stderr.decode(errors="replace")[:500])
 
-            LOG.debug("claude-code stdout (%d chars): %s", len(stdout), stdout[:2000].decode(errors="replace"))
-            return stdout.decode(errors="replace"), model
+            stdout_text = stdout.decode(errors="replace")
+            LOG.debug("claude-code stdout (%d chars): %s", len(stdout), stdout_text[:2000])
+            output_items, run_metadata = parse_stream_json(stdout_text)
+            run_metadata.setdefault("duration_ms", (monotonic() - process_started_at) * 1000)
+            status, error_type = _invocation_outcome(run_metadata, proc.returncode)
+            run_metadata["status"] = status
+            if error_type is not None:
+                run_metadata["error_type"] = error_type
+            return output_items, model, run_metadata
         finally:
             if claude_config_dir is not None:
-                shutil.rmtree(claude_config_dir, ignore_errors=True)
+                try:
+                    if observation_collector is not None:
+                        await asyncio.to_thread(observation_collector, claude_config_dir, run_metadata)
+                except Exception:
+                    LOG.exception("failed to collect Claude Code observations")
+                finally:
+                    shutil.rmtree(claude_config_dir, ignore_errors=True)
 
     def _resources_server_base_url(self) -> str:
         cfg = get_first_server_config_dict(
@@ -509,6 +591,7 @@ async def _create_response(
         mcp_config: Optional[str] = None,
         skills_path: Optional[str] = None,
         rollout_id: Optional[str] = None,
+        observation_collector: Optional[Callable[[Path, dict[str, Any]], None]] = None,
     ) -> NeMoGymResponse:
         body = body.model_copy(deep=True)
         if isinstance(body.input, str):
@@ -518,14 +601,14 @@ async def _create_response(
         system_parts = [p for p in [self.config.system_prompt, input_system] if p]
         system_prompt = "\n\n".join(system_parts) if system_parts else None
 
-        stdout, model_name = await self._run_claude_code(
+        output_items, model_name, run_metadata = await self._run_claude_code(
             user_message,
             system_prompt=system_prompt,
             mcp_config=mcp_config,
             skills_path=skills_path,
             rollout_id=rollout_id,
+            observation_collector=observation_collector,
         )
-        output_items, usage = parse_stream_json(stdout)
 
         if not any(
             getattr(item, "type", None) == "message" and getattr(item, "role", None) == "assistant"
@@ -542,8 +625,8 @@ async def _create_response(
                 )
             )
 
-        input_tokens = usage.get("input_tokens", 0)
-        output_tokens = usage.get("output_tokens", 0)
+        input_tokens = run_metadata.get("input_tokens", 0)
+        output_tokens = run_metadata.get("output_tokens", 0)
 
         return NeMoGymResponse(
             id=f"resp_{uuid4().hex}",
@@ -568,7 +651,52 @@ async def responses(
         request: Request,
         body: NeMoGymResponseCreateParamsNonStreaming = Body(),
     ) -> NeMoGymResponse:
-        return await self._create_response(body)
+        return await self._create_response(body, rollout_id=request.path_params.get("rollout_id"))
+
+    async def _create_episode(
+        self,
+        body: NeMoGymResponseCreateParamsNonStreaming,
+        *,
+        mcp_config: Optional[str] = None,
+        skills_path: Optional[str] = None,
+        rollout_id: Optional[str] = None,
+    ) -> AgentEpisode:
+        observations: Optional[AgentObservationBundle] = None
+
+        def collect(config_dir: Path, run_metadata: dict[str, Any]) -> None:
+            nonlocal observations
+            try:
+                observations = extract_claude_code_observations(
+                    config_dir,
+                    model_ref=self.config.model_server,
+                    root_status=run_metadata["status"],
+                    root_duration_ms=run_metadata.get("duration_ms"),
+                    root_error_type=run_metadata.get("error_type"),
+                    compaction_attempts=run_metadata.get("compaction_attempts"),
+                )
+                if self.config.model_server is None:
+                    observations.gaps.append(ObservationGap(code="model_call_ownership_unavailable"))
+            except Exception:
+                LOG.exception("failed to extract Claude Code observations")
+                observations = AgentObservationBundle(
+                    source="claude_code",
+                    gaps=[ObservationGap(code="observation_parse_failed")],
+                )
+
+        response = await self._create_response(
+            body,
+            mcp_config=mcp_config,
+            skills_path=skills_path,
+            rollout_id=rollout_id,
+            observation_collector=collect,
+        )
+        if observations is None:
+            observations = AgentObservationBundle(
+                source="claude_code",
+                gaps=[ObservationGap(code="agent_transcript_unavailable")],
+            )
+        observations.gaps.append(ObservationGap(code="no_sandbox_runtime"))
+        return AgentEpisode(response=response, observations=observations)
 
     async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> ClaudeCodeAgentVerifyResponse:
         async with self.sem:
@@ -593,12 +721,21 @@ async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> Claude
 
             with tempfile.TemporaryDirectory(prefix="nemo_gym_claude_mcp_") as mcp_config_dir:
                 mcp_config = self._write_rollout_mcp_config(seed_resp_json, Path(mcp_config_dir))
-                agent_resp = await self._create_response(
-                    body.responses_create_params,
-                    mcp_config=mcp_config,
-                    skills_path=skills_path,
-                    rollout_id=rollout_id,
-                )
+                if rollout_id is not None:
+                    episode = await self._create_episode(
+                        body.responses_create_params,
+                        mcp_config=mcp_config,
+                        skills_path=skills_path,
+                        rollout_id=rollout_id,
+                    )
+                    agent_resp, observations = episode.response, episode.observations
+                else:
+                    agent_resp = await self._create_response(
+                        body.responses_create_params,
+                        mcp_config=mcp_config,
+                        skills_path=skills_path,
+                    )
+                    observations = None
                 agent_resp_json = agent_resp.model_dump(mode="json")
 
             verify_resp = await self.server_client.post(
@@ -619,9 +756,10 @@ async def run(self, request: Request, body: ClaudeCodeAgentRunRequest) -> Claude
             last = gym_resp.output[-1] if gym_resp.output else None
             naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant"
 
-            return ClaudeCodeAgentVerifyResponse.model_validate(
-                verify_json | {"turns_used": turns, "finished_naturally": naturally}
-            )
+            result = verify_json | {"turns_used": turns, "finished_naturally": naturally}
+            if observations is not None:
+                result["ng_agent_observations"] = observations.model_dump(mode="json")
+            return ClaudeCodeAgentVerifyResponse.model_validate(result)
 
 
 if __name__ == "__main__":