Add agent-side rollout observability

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

@@ -77,9 +77,10 @@ totals = aggregate_model_call_metrics(store, rollout_id)
 ```
 
 `ModelCallRecord` is an observability serialization model derived from captured HTTP exchanges. It
-contains a unique server-generated `model_call_id`, typed `model_ref`, wall-clock `started_at` and
-`completed_at`, a `call_index`, API dialect, token and cache usage, latency, error details, tool
-calls, reasoning content, and the captured request and response. `started_at` is recorded immediately
+contains a unique server-generated `model_call_id`, the protocol `response_id` when present, typed
+`model_ref`, wall-clock `started_at` and `completed_at`, a `call_index`, API dialect, token and cache
+usage, latency, error details, tool calls, reasoning content, and the captured request and response.
+`started_at` is recorded immediately
 before invoking the downstream ASGI application; `completed_at` is recorded when that invocation
 returns or raises, before capture parsing and persistence. Both are UTC Unix seconds for external
 trace correlation; durations use the monotonic latency fields. `call_index` reflects durable append
@@ -113,6 +114,35 @@ for that exact rollout-attempt id, including a kill-shaped attempt being redispa
 The attachment is additive: it does not replace or rewrite the existing response, reward,
 `NeMoGymResponse`, token-id, or log-prob fields. Downstream consumers can choose whether to read it.
 
+## Agent observations
+
+Supported Agent Servers may also attach `ng_agent_observations` when observability is enabled. It
+contains agent invocations, tool-call intervals, explicit context-compaction events, and gaps for
+facts the integration could not observe. An invocation's `conversation` is the ordered, normalized
+set of conversation items supported by that producer. Unsupported or unavailable evidence is
+reported in `gaps` instead of being guessed; `agent_transcript_unavailable` means any available
+items came from a fallback output rather than a harness transcript.
+
+Agent observations and model-call capture are separate evidence. Join an invocation's model-call
+references by `model_call_id`, or by the exact `(model_ref, response_id)` pair when the harness sees
+the protocol response ID. Do not infer ownership from timestamps, text, or list position. The full
+model request and response remain in `CaptureStore`; rollout attachments intentionally omit them.
+
+Tool timestamps are UTC Unix seconds when the source provides wall-clock time. `duration_ms` is the
+measured interval, and `timing_source` distinguishes an executor measurement from an artifact-derived
+interval. A missing capability is represented in `gaps`, not approximated.
+
+Model-visible tool calls and results remain in `AgentInvocation.conversation` as `function_call` and
+`function_call_output` items. Execution timing and outcome, when observable, live in
+`ToolCallObservation` and join through `(invocation_id, tool_call_id)`.
+
+Producer coverage is integration-specific. Claude Code exports transcript-derived hierarchy,
+model-call references, tool intervals, and compaction markers. Hermes exports hierarchy, executor
+tool timing, and compaction markers, but not exact model-call ownership. OpenClaw and PinchBench
+expose normalized response items and report gaps for hierarchy, timing, compaction, and model-call
+ownership. Those items duplicate response output and can make rollout records substantially larger.
+Observation payloads are retained in rollout records and excluded from aggregate-metrics requests.
+
 ## Limitations
 
 The model-server boundary observes model HTTP requests and responses. It can record tool calls,

nemo_gym/base_responses_api_model.py

@@ -391,6 +391,7 @@ class ModelCallRecord(BaseModel):
 
     # Unique server-generated identity for each persisted call.
     model_call_id: Optional[str] = None
+    response_id: Optional[str] = None
 
     # Durable append order, not a causal or semantic order for concurrent calls.
     call_index: int
@@ -441,6 +442,7 @@ def build_model_call_record(exchange: dict[str, Any], *, call_index: int) -> Mod
     tool_calls, reasoning_content = _tool_calls_and_reasoning(response)
     return ModelCallRecord(
         model_call_id=exchange.get("model_call_id"),
+        response_id=response.get("id") if isinstance(response.get("id"), str) else None,
         call_index=call_index,
         model_ref=exchange.get("model_ref"),
         dialect=exchange.get("dialect"),
@@ -664,11 +666,14 @@ def _reconstruct_chat_sse(events: list[dict[str, Any]]) -> Optional[dict[str, An
     tool_calls: dict[int, dict[str, Any]] = {}
     usage: Optional[dict[str, Any]] = None
     model: Optional[str] = None
+    response_id: Optional[str] = None
     role = "assistant"
     finish_reason: Optional[str] = None
     saw_choice = False
     for chunk in events:
         model = chunk.get("model") or model
+        if isinstance(chunk.get("id"), str):
+            response_id = chunk["id"]
         if chunk.get("usage"):
             usage = chunk["usage"]
         for choice in chunk.get("choices") or []:
@@ -707,6 +712,8 @@ def _reconstruct_chat_sse(events: list[dict[str, Any]]) -> Optional[dict[str, An
         "model": model,
         "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}],
     }
+    if response_id is not None:
+        result["id"] = response_id
     if usage:
         result["usage"] = usage
     return result

nemo_gym/rollout_collection.py

@@ -643,7 +643,11 @@ async def _fetch_agent_metrics(agent_name: str, agent_result_list: List[Dict]) -
             # Strip heavyweight fields before sending, but preserve response.usage
             stripped = []
             for r in agent_result_list:
-                entry = {k: v for k, v in r.items() if k not in ("response", "responses_create_params")}
+                entry = {
+                    k: v
+                    for k, v in r.items()
+                    if k not in ("response", "responses_create_params", "ng_agent_observations")
+                }
                 usage = (r.get("response") or {}).get("usage")
                 if usage:
                     entry["response"] = {"usage": usage}

nemo_gym/rollout_observability.py

@@ -0,0 +1,93 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Small shared contract for observations exposed by Agent integrations."""
+
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field, model_validator
+
+from nemo_gym.config_types import ModelServerRef
+from nemo_gym.openai_utils import NeMoGymResponse, NeMoGymResponseInputItem
+
+
+class ModelCallRef(BaseModel):
+    """Stable identifiers an Agent integration can observe for one model call."""
+
+    model_call_id: Optional[str] = None
+    model_ref: Optional[ModelServerRef] = None
+    response_id: Optional[str] = None
+
+    @model_validator(mode="after")
+    def validate_join_key(self) -> "ModelCallRef":
+        if not self.model_call_id and not (self.model_ref is not None and self.response_id):
+            raise ValueError("model_call_id or both model_ref and response_id are required")
+        return self
+
+
+class AgentInvocation(BaseModel):
+    """One root Agent or subagent conversation observed by a harness."""
+
+    invocation_id: str
+    parent_invocation_id: Optional[str] = None
+    spawned_by_tool_call_id: Optional[str] = None
+    status: Literal["completed", "failed", "incomplete", "unknown"] = Field(
+        default="unknown", description="Harness-reported invocation outcome; unknown when not explicit."
+    )
+    model_calls: list[ModelCallRef] = Field(default_factory=list)
+    conversation: list[NeMoGymResponseInputItem] = Field(
+        default_factory=list,
+        description="Normalized conversation items supported by this producer; gaps describe unavailable evidence.",
+    )
+
+
+class ToolCallObservation(BaseModel):
+    """Timing observed for one tool call at an Agent-owned boundary."""
+
+    invocation_id: str
+    tool_call_id: str
+    tool_name: Optional[str] = None
+    started_at: Optional[float] = None
+    completed_at: Optional[float] = None
+    duration_ms: Optional[float] = None
+    clock_id: Optional[str] = None
+    timing_source: Optional[Literal["executor", "artifact"]] = None
+    status: Literal["completed", "failed", "timeout", "incomplete", "unknown"] = "unknown"
+
+
+class ContextCompactionObservation(BaseModel):
+    """An explicit context-compaction event reported by the Agent harness."""
+
+    invocation_id: str
+    observed_at: Optional[float] = None
+    trigger: Optional[str] = None
+    tokens_before: Optional[int] = None
+    tokens_after: Optional[int] = None
+
+
+class ObservationGap(BaseModel):
+    """A fact that the selected integration could not observe or join exactly."""
+
+    code: str
+    source: str
+    invocation_id: Optional[str] = None
+    detail: Optional[str] = None
+
+
+class AgentObservationBundle(BaseModel):
+    """Normalized observations returned by one Agent Server for one rollout."""
+
+    source: str
+    invocations: list[AgentInvocation] = Field(default_factory=list)
+    tool_calls: list[ToolCallObservation] = Field(default_factory=list)
+    compactions: list[ContextCompactionObservation] = Field(default_factory=list)
+    gaps: list[ObservationGap] = Field(default_factory=list)
+
+
+class AgentEpisode(BaseModel):
+    """An Agent response and the observations available at its execution boundary."""
+
+    response: NeMoGymResponse
+    observations: AgentObservationBundle

resources_servers/gdpval/app.py

@@ -41,7 +41,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Literal, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from nemo_gym.base_resources_server import (
     BaseResourcesServerConfig,
@@ -50,7 +50,7 @@
     SimpleResourcesServer,
 )
 from nemo_gym.config_types import AggregateMetrics, AggregateMetricsRequest, ModelServerRef
-from nemo_gym.server_utils import get_server_url
+from nemo_gym.server_utils import apply_rollout_prefix, get_server_url
 from resources_servers.gdpval.judge_panel import (
     ResolvedJudge,
     dir_contains_audio_video,
@@ -234,6 +234,7 @@ class GDPValResourcesServerConfig(BaseResourcesServerConfig):
 
 
 class GDPValVerifyRequest(BaseVerifyRequest):
+    rollout_id: Optional[str] = Field(default=None, exclude_if=lambda value: value is None)
     task_id: str
     sector: Optional[str] = None
     occupation: Optional[str] = None
@@ -324,7 +325,7 @@ def _effective_panel(self) -> List[JudgePanelMember]:
             JudgePanelMember(create_params_overrides=dict(self.config.judge_responses_create_params_overrides or {}))
         ]
 
-    def _resolve_judges(self) -> List[ResolvedJudge]:
+    def _resolve_judges(self, rollout_id: Optional[str] = None) -> List[ResolvedJudge]:
         """Resolve the (always non-empty) panel to concrete upstream coordinates.
 
         Every judge — including the single-judge special case (see
@@ -337,7 +338,7 @@ def _resolve_judges(self) -> List[ResolvedJudge]:
         legacy_overrides = dict(self.config.judge_responses_create_params_overrides or {})
 
         def _url(server: ModelServerRef) -> str:
-            return get_server_url(server.name) + "/v1"
+            return apply_rollout_prefix(get_server_url(server.name), rollout_id) + "/v1"
 
         judges: List[ResolvedJudge] = []
         for i, member in enumerate(self._effective_panel()):
@@ -373,7 +374,7 @@ async def _verify_rubric(self, body: GDPValVerifyRequest) -> GDPValVerifyRespons
                 invalid_judge_response=True,
             )
 
-        judges = self._resolve_judges()
+        judges = self._resolve_judges(body.rollout_id)
         # Route tasks with audio/video deliverables to the AV-capable judge(s) —
         # most judges can't read those modalities natively.
         if dir_contains_audio_video(body.deliverables_dir):
@@ -534,7 +535,7 @@ async def _verify_comparison(self, body: GDPValVerifyRequest) -> GDPValVerifyRes
         # Build the judge panel. Members may share a single proxy server (so we
         # reuse one OpenAI client per distinct upstream) and differ only by model
         # + reasoning settings. run_trials samples one member per trial.
-        resolved_judges = self._resolve_judges()
+        resolved_judges = self._resolve_judges(body.rollout_id)
         client_cache: Dict[tuple, Any] = {}
 
         def _client_for(judge: ResolvedJudge) -> Any:

resources_servers/gdpval/tests/test_app.py

@@ -103,6 +103,9 @@ def test_missing_dir_returns_empty(self, tmp_path) -> None:
 
 
 class TestApp:
+    def test_rollout_id_is_absent_when_correlation_is_disabled(self) -> None:
+        assert "rollout_id" not in _verify_request().model_dump()
+
     def test_sanity_rubric(self) -> None:
         _server(reward_mode="rubric")
 
@@ -241,7 +244,7 @@ async def fake_score_with_rubric(**kwargs):
             captured.update(kwargs)
             return 0.5, {"overall_score": 0.5}
 
-        body = _verify_request(rubric_json=[{"criterion": "clarity", "score": 1}])
+        body = _verify_request(rubric_json=[{"criterion": "clarity", "score": 1}], rollout_id="7-3")
 
         with (
             patch("resources_servers.gdpval.scoring.score_with_rubric", side_effect=fake_score_with_rubric),
@@ -259,7 +262,7 @@ async def fake_score_with_rubric(**kwargs):
         assert judges[0].create_overrides == {"reasoning_effort": "medium"}
         assert judges[2].weight == 2.0
         # All share the single proxy base_url.
-        assert {j.base_url for j in judges} == {"http://localhost:9999/v1"}
+        assert {j.base_url for j in judges} == {"http://localhost:9999/ng-rollout/7-3/v1"}
         # A seeded rng is threaded through for reproducible sampling.
         assert captured["rng"] is not None