From e6897b41cb1a8ad013f7a561805a23a42bacabca Mon Sep 17 00:00:00 2001
From: Michal Bien <mbien@nvidia.com>
Date: Thu, 23 Jul 2026 11:42:46 +0200
Subject: [PATCH 1/4] Add Pi rollout observations

Signed-off-by: Michal Bien <mbien@nvidia.com>
---
 responses_api_agents/pi_agent/app.py          | 363 ++++++++++++++++--
 .../pi_agent/tests/test_app.py                | 264 ++++++++++++-
 2 files changed, 592 insertions(+), 35 deletions(-)

diff --git a/responses_api_agents/pi_agent/app.py b/responses_api_agents/pi_agent/app.py
index a85f784f3b..33c3280a70 100644
--- a/responses_api_agents/pi_agent/app.py
+++ b/responses_api_agents/pi_agent/app.py
@@ -21,6 +21,7 @@
 import shlex
 import shutil
 from asyncio import Semaphore
+from collections.abc import Mapping
 from pathlib import Path
 from time import time
 from typing import Any, Optional
@@ -36,7 +37,6 @@
     SimpleResponsesAPIAgent,
 )
 from nemo_gym.config_types import ModelServerRef, ResourcesServerRef
-from nemo_gym.global_config import get_first_server_config_dict
 from nemo_gym.openai_utils import (
     NeMoGymEasyInputMessage,
     NeMoGymFunctionCallOutput,
@@ -49,11 +49,21 @@
     NeMoGymResponseOutputTokensDetails,
     NeMoGymResponseUsage,
 )
+from nemo_gym.rollout_observability import (
+    AgentEpisode,
+    AgentInvocation,
+    AgentObservationBundle,
+    ContextCompactionObservation,
+    ModelCallRef,
+    ObservationGap,
+    ToolCallObservation,
+)
 from nemo_gym.server_utils import get_response_json, raise_for_status
 from responses_api_agents.pi_agent.setup_pi import ensure_pi
 
 
 LOG = logging.getLogger(__name__)
+_INTERNAL_OBSERVATIONS_KEY = "_ng_agent_observations"
 
 
 def parse_pi_events(stdout: str) -> tuple[list[Any], dict[str, int]]:
@@ -67,11 +77,15 @@ def parse_pi_events(stdout: str) -> tuple[list[Any], dict[str, int]]:
             continue
         try:
             event = json.loads(line)
-        except json.JSONDecodeError:
+        except (json.JSONDecodeError, RecursionError):
+            continue
+        if not isinstance(event, dict):
             continue
         if event.get("type") != "message_end":
             continue
         message = event.get("message") or {}
+        if not isinstance(message, dict):
+            continue
         role = message.get("role")
         content = message.get("content")
         if not isinstance(content, list):
@@ -79,6 +93,8 @@ def parse_pi_events(stdout: str) -> tuple[list[Any], dict[str, int]]:
 
         if role == "assistant":
             usage = message.get("usage") or {}
+            if not isinstance(usage, dict):
+                usage = {}
             input_tokens += int(usage.get("input") or 0) + int(usage.get("cacheRead") or 0)
             output_tokens += int(usage.get("output") or 0)
             texts = [b["text"] for b in content if isinstance(b, dict) and (b.get("text") or "").strip()]
@@ -126,6 +142,217 @@ def parse_pi_events(stdout: str) -> tuple[list[Any], dict[str, int]]:
     return output_items, {"input_tokens": input_tokens, "output_tokens": output_tokens}
 
 
+async def _read_pi_stdout(stream: asyncio.StreamReader) -> tuple[str, list[tuple[float, dict[str, Any]]]]:
+    lines: list[str] = []
+    events: list[tuple[float, dict[str, Any]]] = []
+
+    def consume(line: bytes) -> None:
+        observed_at = time()
+        text = line.decode(errors="replace")
+        lines.append(text)
+        try:
+            event = json.loads(text)
+        except (json.JSONDecodeError, RecursionError):
+            return
+        if isinstance(event, dict):
+            events.append((observed_at, event))
+
+    pending = bytearray()
+    while chunk := await stream.read(64 * 1024):
+        pending.extend(chunk)
+        while (newline := pending.find(b"\n")) >= 0:
+            consume(bytes(pending[: newline + 1]))
+            del pending[: newline + 1]
+    if pending:
+        consume(bytes(pending))
+    return "".join(lines), events
+
+
+def _build_pi_observations(
+    events: list[tuple[float, dict[str, Any]]],
+    invocation_id: str,
+    model_ref: Optional[ModelServerRef],
+    conversation: list[Any],
+    *,
+    transcript_available: bool = True,
+) -> AgentObservationBundle:
+    def gap(code: str, detail: Optional[str] = None) -> ObservationGap:
+        return ObservationGap(code=code, invocation_id=invocation_id, detail=detail)
+
+    gaps = [gap("subagent_hierarchy_unavailable")]
+    if not transcript_available:
+        gaps.append(gap("agent_transcript_unavailable"))
+    model_calls: list[ModelCallRef] = []
+    model_call_join_missing = False
+    starts: dict[str, tuple[float, Optional[str]]] = {}
+    tools: dict[str, ToolCallObservation] = {}
+    compaction_start: Optional[tuple[float, Optional[str], Optional[ModelCallRef]]] = None
+    compactions: list[ContextCompactionObservation] = []
+    compactions_waiting_for_call: list[ContextCompactionObservation] = []
+    last_model_call: Optional[ModelCallRef] = None
+
+    for observed_at, event in events:
+        event_type = event.get("type")
+        message = event.get("message")
+        call_id = event.get("toolCallId")
+        tool_name = event.get("toolName")
+        tool_name = tool_name if isinstance(tool_name, str) and tool_name else None
+
+        if event_type == "message_end" and isinstance(message, dict) and message.get("role") == "assistant":
+            response_id = message.get("responseId")
+            if model_ref is not None and isinstance(response_id, str) and response_id:
+                last_model_call = ModelCallRef(model_ref=model_ref, response_id=response_id)
+                model_calls.append(last_model_call)
+            else:
+                last_model_call = None
+                model_call_join_missing = True
+            for compaction in compactions_waiting_for_call:
+                compaction.after_model_call = last_model_call
+                if last_model_call is None:
+                    gaps.append(gap("compaction_after_model_call_unavailable"))
+            compactions_waiting_for_call.clear()
+        elif event_type == "tool_execution_start" and isinstance(call_id, str):
+            starts[call_id] = (observed_at, tool_name)
+        elif event_type == "tool_execution_end" and isinstance(call_id, str):
+            completed_at = observed_at
+            started_at, started_name = starts.pop(call_id, (None, None))
+            valid_interval = started_at is not None and completed_at >= started_at
+            duration_ms = (completed_at - started_at) * 1000 if started_at is not None and valid_interval else None
+            tools[call_id] = ToolCallObservation(
+                invocation_id=invocation_id,
+                tool_call_id=call_id,
+                tool_name=tool_name or started_name,
+                started_at=started_at,
+                completed_at=completed_at,
+                duration_ms=duration_ms,
+                timing_source="harness",
+                status=(
+                    "failed"
+                    if event.get("isError") is True
+                    else "completed"
+                    if event.get("isError") is False
+                    else "unknown"
+                ),
+            )
+            if not valid_interval:
+                gaps.append(gap("tool_timing_unavailable", call_id))
+            if not isinstance(event.get("isError"), bool):
+                gaps.append(gap("tool_outcome_unavailable", call_id))
+        elif event_type == "compaction_start":
+            reason = event.get("reason")
+            compaction_start = (observed_at, reason if isinstance(reason, str) else None, last_model_call)
+        elif event_type == "compaction_end":
+            reason = event.get("reason")
+            started_at, started_reason, before_model_call = compaction_start or (observed_at, None, None)
+            raw_result = event.get("result")
+            result: dict[str, Any] = raw_result if isinstance(raw_result, dict) else {}
+            before = result.get("tokensBefore")
+            after = result.get("estimatedTokensAfter")
+            summary = result.get("summary")
+            first_kept_item_id = result.get("firstKeptEntryId")
+            outcome = (
+                "aborted"
+                if event.get("aborted") is True
+                else "completed"
+                if result
+                else "failed"
+                if isinstance(event.get("errorMessage"), str)
+                else "unknown"
+            )
+            compaction = ContextCompactionObservation(
+                invocation_id=invocation_id,
+                observed_at=started_at,
+                trigger=reason if isinstance(reason, str) else started_reason,
+                tokens_before=before if type(before) is int and before >= 0 else None,
+                tokens_after=after if type(after) is int and after >= 0 else None,
+                outcome=outcome,
+                summary=summary if isinstance(summary, str) else None,
+                first_kept_item_id=first_kept_item_id if isinstance(first_kept_item_id, str) else None,
+                before_model_call=before_model_call,
+            )
+            compactions.append(compaction)
+            compactions_waiting_for_call.append(compaction)
+            if compaction_start is None:
+                gaps.append(gap("compaction_start_unavailable"))
+            if not result:
+                gaps.append(gap("compaction_result_unavailable"))
+            else:
+                if type(before) is not int or before < 0:
+                    gaps.append(gap("compaction_tokens_before_unavailable"))
+                if not isinstance(summary, str):
+                    gaps.append(gap("compaction_summary_unavailable"))
+                if not isinstance(first_kept_item_id, str):
+                    gaps.append(gap("compaction_boundary_unavailable"))
+                if type(after) is not int or after < 0:
+                    gaps.append(gap("compaction_tokens_after_unavailable"))
+            if outcome == "unknown":
+                gaps.append(gap("compaction_outcome_unavailable"))
+            compaction_start = None
+    if not model_calls or model_call_join_missing:
+        gaps.append(gap("model_call_ownership_unavailable"))
+
+    for call_id, (started_at, tool_name) in starts.items():
+        tools[call_id] = ToolCallObservation(
+            invocation_id=invocation_id,
+            tool_call_id=call_id,
+            tool_name=tool_name,
+            started_at=started_at,
+            timing_source="harness",
+            status="incomplete",
+        )
+        gaps.append(gap("tool_timing_unavailable", call_id))
+    if compaction_start is not None:
+        started_at, reason, before_model_call = compaction_start
+        compactions.append(
+            ContextCompactionObservation(
+                invocation_id=invocation_id,
+                observed_at=started_at,
+                trigger=reason,
+                before_model_call=before_model_call,
+            )
+        )
+        gaps.append(gap("compaction_result_unavailable"))
+        gaps.append(gap("compaction_outcome_unavailable"))
+    for _ in compactions_waiting_for_call:
+        gaps.append(gap("compaction_after_model_call_unavailable"))
+
+    def field(item: Any, name: str) -> Any:
+        return item.get(name) if isinstance(item, dict) else getattr(item, name, None)
+
+    result_ids = {
+        field(item, "call_id")
+        for item in conversation
+        if field(item, "type") == "function_call_output" and isinstance(field(item, "call_id"), str)
+    }
+    for item in conversation:
+        if field(item, "type") != "function_call":
+            continue
+        call_id = field(item, "call_id")
+        if not isinstance(call_id, str) or not call_id or call_id in tools:
+            continue
+        tools[call_id] = ToolCallObservation(
+            invocation_id=invocation_id,
+            tool_call_id=call_id,
+            tool_name=field(item, "name"),
+            status="unknown" if call_id in result_ids else "incomplete",
+        )
+        gaps.append(gap("tool_timing_unavailable", call_id))
+
+    return AgentObservationBundle(
+        source="pi",
+        records=[
+            AgentInvocation(
+                invocation_id=invocation_id,
+                model_calls=model_calls,
+                conversation=conversation,
+            ),
+            *tools.values(),
+            *compactions,
+        ],
+        gaps=gaps,
+    )
+
+
 def _extract_instruction(body_input) -> tuple[str, Optional[str]]:
     """Return (user_message, system_message) from a responses body input list."""
     items = list(body_input)
@@ -188,6 +415,9 @@ class PiAgentVerifyResponse(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 PiAgent(SimpleResponsesAPIAgent):
@@ -216,26 +446,21 @@ def _env(self, home: Path) -> dict[str, str]:
         env.update({k: v for k, v in self.config.env.items() if v})
         return env
 
-    def _resolve_model_base_url(self) -> str:
+    def _resolve_model_base_url(self, rollout_id: Optional[str] = None) -> str:
         if self.config.model_server is None:
             return ""
-        config = get_first_server_config_dict(
-            self.server_client.global_config_dict,
-            self.config.model_server.name,
-        )
-        base_url = self.server_client._build_server_base_url(config).rstrip("/")
-        return base_url if base_url.endswith("/v1") else f"{base_url}/v1"
+        return self.resolve_model_base_url(self.config.model_server.name, rollout_id)
 
     def _effective_model(self) -> str:
         return f"nemo/{self.config.model}" if self.config.model_server else self.config.model
 
-    def _build_models_config(self) -> dict[str, Any]:
+    def _build_models_config(self, rollout_id: Optional[str] = None) -> dict[str, Any]:
         config = copy.deepcopy(self.config.models_config)
         if self.config.model_server is None:
             return config
         providers = config.setdefault("providers", {})
         providers["nemo"] = {
-            "baseUrl": self._resolve_model_base_url(),
+            "baseUrl": self._resolve_model_base_url(rollout_id),
             "api": "openai-completions",
             "apiKey": "EMPTY",  # pragma: allowlist secret
             "compat": {"supportsDeveloperRole": False, "supportsReasoningEffort": False},
@@ -251,13 +476,20 @@ def _build_models_config(self) -> dict[str, Any]:
         }
         return config
 
-    async def _run_pi(self, instruction: str, system_prompt: Optional[str]) -> tuple[list[Any], dict[str, int], str]:
+    async def _run_pi(
+        self,
+        instruction: str,
+        system_prompt: Optional[str],
+        *,
+        rollout_id: Optional[str] = None,
+        collect_observations: bool = True,
+    ) -> tuple[list[Any], dict[str, int], str, list[tuple[float, dict[str, Any]]]]:
         effective_model = self._effective_model()
         provider, _, model_id = effective_model.partition("/")
         work_dir = self._workspace_root()
         home = work_dir / ".pi-home"
         (home / ".pi" / "agent").mkdir(parents=True, exist_ok=True)
-        models_config = self._build_models_config()
+        models_config = self._build_models_config(rollout_id)
         if models_config:
             (home / ".pi" / "agent" / "models.json").write_text(json.dumps(models_config, indent=2))
         env = self._env(home)
@@ -283,26 +515,45 @@ async def _run_pi(self, instruction: str, system_prompt: Optional[str]) -> tuple
                 stderr=asyncio.subprocess.PIPE,
                 env=env,
             )
-            try:
-                stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.config.timeout)
-            except asyncio.TimeoutError:
-                proc.kill()
-                await proc.communicate()
-                LOG.warning("pi timed out after %ds", self.config.timeout)
-                return [], {"input_tokens": 0, "output_tokens": 0}, self.config.model
+            assert proc.stdout is not None and proc.stderr is not None
+            events: list[tuple[float, dict[str, Any]]] = []
+            if collect_observations:
+                stdout_task = asyncio.create_task(_read_pi_stdout(proc.stdout))
+                stderr_task = asyncio.create_task(proc.stderr.read())
+                output_task = asyncio.gather(stdout_task, stderr_task, proc.wait())
+                try:
+                    (stdout, events), stderr, _ = await asyncio.wait_for(
+                        asyncio.shield(output_task), timeout=self.config.timeout
+                    )
+                except asyncio.TimeoutError:
+                    if proc.returncode is None:
+                        proc.kill()
+                    (_, events), _, _ = await output_task
+                    LOG.warning("pi timed out after %ds", self.config.timeout)
+                    return [], {"input_tokens": 0, "output_tokens": 0}, self.config.model, events
+            else:
+                try:
+                    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.config.timeout)
+                except asyncio.TimeoutError:
+                    proc.kill()
+                    await proc.communicate()
+                    LOG.warning("pi timed out after %ds", self.config.timeout)
+                    return [], {"input_tokens": 0, "output_tokens": 0}, self.config.model, events
 
             if proc.returncode not in (0, None):
                 LOG.warning("pi exited %d: %s", proc.returncode, stderr.decode(errors="replace")[:500])
-            output_items, usage = parse_pi_events(stdout.decode(errors="replace"))
-            return output_items, usage, self.config.model
+            output_items, usage = parse_pi_events(stdout)
+            return output_items, usage, self.config.model, events
         finally:
             shutil.rmtree(work_dir, ignore_errors=True)
 
-    async def responses(
+    async def _create_episode(
         self,
-        request: Request,
-        body: NeMoGymResponseCreateParamsNonStreaming = Body(),
-    ) -> NeMoGymResponse:
+        body: NeMoGymResponseCreateParamsNonStreaming,
+        *,
+        rollout_id: Optional[str] = None,
+        collect_observations: bool = True,
+    ) -> AgentEpisode:
         body = body.model_copy(deep=True)
         if isinstance(body.input, str):
             body.input = [NeMoGymEasyInputMessage(role="user", content=body.input)]
@@ -311,7 +562,15 @@ async def responses(
         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
 
-        output_items, usage, model_name = await self._run_pi(user_message, system_prompt)
+        output_items, usage, model_name, events = await self._run_pi(
+            user_message,
+            system_prompt,
+            rollout_id=rollout_id,
+            collect_observations=collect_observations,
+        )
+        observed_output_items = list(output_items)
+        if not observed_output_items and events:
+            observed_output_items, _ = parse_pi_events("\n".join(json.dumps(event) for _, event in events))
 
         if not any(
             getattr(item, "type", None) == "message" and getattr(item, "role", None) == "assistant"
@@ -331,7 +590,7 @@ async def responses(
         input_tokens = usage.get("input_tokens", 0)
         output_tokens = usage.get("output_tokens", 0)
 
-        return NeMoGymResponse(
+        response = NeMoGymResponse(
             id=f"resp_{uuid4().hex}",
             created_at=int(time()),
             model=model_name,
@@ -348,6 +607,41 @@ async def responses(
                 total_tokens=input_tokens + output_tokens,
             ),
         )
+        observations = AgentObservationBundle(source="pi")
+        if collect_observations:
+            invocation_id = rollout_id or response.id
+            try:
+                observations = _build_pi_observations(
+                    events,
+                    invocation_id,
+                    self.config.model_server,
+                    [*body.input, *observed_output_items],
+                    transcript_available=bool(observed_output_items),
+                )
+            except Exception:
+                LOG.exception("failed to build Pi observations")
+                observations = AgentObservationBundle(
+                    source="pi", gaps=[ObservationGap(code="observation_parse_failed")]
+                )
+        return AgentEpisode(response=response, observations=observations)
+
+    async def responses(
+        self,
+        request: Request,
+        body: NeMoGymResponseCreateParamsNonStreaming = Body(),
+    ) -> NeMoGymResponse:
+        path_params = getattr(request, "path_params", None)
+        rollout_id = path_params.get("rollout_id") if isinstance(path_params, Mapping) else None
+        episode = await self._create_episode(
+            body,
+            rollout_id=rollout_id,
+            collect_observations=isinstance(rollout_id, str),
+        )
+        if not isinstance(rollout_id, str):
+            return episode.response
+        return episode.response.model_copy(
+            update={_INTERNAL_OBSERVATIONS_KEY: episode.observations.model_dump(mode="json")}
+        )
 
     async def run(self, request: Request, body: PiAgentRunRequest) -> PiAgentVerifyResponse:
         async with self.sem:
@@ -362,15 +656,22 @@ async def run(self, request: Request, body: PiAgentRunRequest) -> PiAgentVerifyR
             await raise_for_status(seed_resp)
             cookies = seed_resp.cookies
 
+            rollout_id = self.rollout_id_from_run(body)
             agent_resp = await self.server_client.post(
                 server_name=self.config.name,
-                url_path="/v1/responses",
+                url_path=self.url_path_for_run("/v1/responses", body),
                 json=body.responses_create_params,
                 cookies=cookies,
             )
             await raise_for_status(agent_resp)
             cookies = agent_resp.cookies
             agent_resp_json = await get_response_json(agent_resp)
+            raw_observations = (
+                agent_resp_json.pop(_INTERNAL_OBSERVATIONS_KEY, None) if rollout_id is not None else None
+            )
+            observations = (
+                AgentObservationBundle.model_validate(raw_observations) if isinstance(raw_observations, dict) else None
+            )
 
             verify_resp = await self.server_client.post(
                 server_name=self.config.resources_server.name,
@@ -391,7 +692,9 @@ async def run(self, request: Request, body: PiAgentRunRequest) -> PiAgentVerifyR
             naturally = getattr(last, "type", None) == "message" and getattr(last, "role", None) == "assistant"
 
             return PiAgentVerifyResponse.model_validate(
-                verify_json | {"turns_used": turns, "finished_naturally": naturally}
+                verify_json
+                | {"turns_used": turns, "finished_naturally": naturally}
+                | ({"ng_agent_observations": observations} if observations is not None else {})
             )
 
 
diff --git a/responses_api_agents/pi_agent/tests/test_app.py b/responses_api_agents/pi_agent/tests/test_app.py
index 86c32b46cf..deef3fe3d9 100644
--- a/responses_api_agents/pi_agent/tests/test_app.py
+++ b/responses_api_agents/pi_agent/tests/test_app.py
@@ -16,7 +16,7 @@
 import asyncio
 import json
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import yaml
 
@@ -24,15 +24,20 @@
 from nemo_gym.openai_utils import (
     NeMoGymEasyInputMessage,
     NeMoGymFunctionCallOutput,
+    NeMoGymResponseCreateParamsNonStreaming,
     NeMoGymResponseFunctionToolCall,
     NeMoGymResponseOutputMessage,
 )
+from nemo_gym.rollout_observability import AgentInvocation, ContextCompactionObservation, ToolCallObservation
 from nemo_gym.server_utils import ServerClient
 from responses_api_agents.pi_agent.app import (
     PiAgent,
     PiAgentConfig,
+    PiAgentRunRequest,
     ResourcesServerRef,
+    _build_pi_observations,
     _extract_instruction,
+    _read_pi_stdout,
     parse_pi_events,
 )
 
@@ -59,6 +64,10 @@ def _msg_end(role, content, **extra) -> str:
     return json.dumps({"type": "message_end", "message": {"role": role, "content": content, **extra}})
 
 
+def _records(bundle, record_type):
+    return [record for record in bundle.records if isinstance(record, record_type)]
+
+
 class TestSanity:
     def test_config_defaults(self) -> None:
         cfg = _config()
@@ -139,7 +148,7 @@ def test_tool_call_and_result(self) -> None:
         assert isinstance(items[2], NeMoGymResponseOutputMessage)
 
     def test_malformed_lines_skipped(self) -> None:
-        line = "not-json\n" + _msg_end("assistant", [{"type": "text", "text": "ok"}])
+        line = "not-json\nnull\n[]\n" + _msg_end("assistant", [{"type": "text", "text": "ok"}])
         items, _ = parse_pi_events(line)
         assert len(items) == 1
 
@@ -155,18 +164,26 @@ def test_env_passthrough(self) -> None:
 
 class TestModelServer:
     def test_builds_pi_provider_config(self) -> None:
+        models_config = {"providers": {"custom": {"baseUrl": "https://example.test"}}}
         agent = _make_agent(
             model="Qwen3.6-35B-A3B",
             model_server=ModelServerRef(type="responses_api_models", name="policy_model"),
+            models_config=models_config,
         )
-        with patch.object(agent, "_resolve_model_base_url", return_value="http://model/v1"):
-            config = agent._build_models_config()
+        with patch.object(
+            PiAgent,
+            "resolve_model_base_url",
+            return_value="http://model/ng-rollout/1-2/v1",
+        ) as resolve:
+            config = agent._build_models_config("1-2")
 
         provider = config["providers"]["nemo"]
         assert agent._effective_model() == "nemo/Qwen3.6-35B-A3B"
-        assert provider["baseUrl"] == "http://model/v1"
+        assert provider["baseUrl"] == "http://model/ng-rollout/1-2/v1"
         assert provider["models"][0]["id"] == "Qwen3.6-35B-A3B"
         assert provider["models"][0]["maxTokens"] == 131072
+        assert agent.config.models_config == models_config
+        resolve.assert_called_once_with("policy_model", "1-2")
 
     def test_preserves_explicit_provider_without_model_server(self) -> None:
         config = {"providers": {"custom": {"baseUrl": "https://example.test"}}}
@@ -175,6 +192,243 @@ def test_preserves_explicit_provider_without_model_server(self) -> None:
         assert agent._build_models_config() == config
 
 
+class TestRolloutObservability:
+    async def test_reads_and_timestamps_json_events(self) -> None:
+        stream = asyncio.StreamReader()
+        stream.feed_data(b'{"type":"tool_execution_start","toolCallId":"a"}\nnot-json\n')
+        stream.feed_eof()
+
+        with patch("responses_api_agents.pi_agent.app.time", side_effect=[10.0, 11.0]):
+            stdout, events = await _read_pi_stdout(stream)
+
+        assert stdout.endswith("not-json\n")
+        assert events == [(10.0, {"type": "tool_execution_start", "toolCallId": "a"})]
+
+    async def test_reads_json_events_larger_than_streamreader_line_limit(self) -> None:
+        event = {"type": "message_end", "message": {"role": "assistant", "content": "x" * 70_000}}
+        payload = (json.dumps(event) + "\n").encode()
+        stream = asyncio.StreamReader()
+        stream.feed_data(payload)
+        stream.feed_eof()
+
+        stdout, events = await _read_pi_stdout(stream)
+
+        assert stdout.encode() == payload
+        assert events[0][1] == event
+
+    def test_exact_model_calls_parallel_tools_and_compaction(self) -> None:
+        assistant = {
+            "type": "message_end",
+            "message": {
+                "role": "assistant",
+                "responseId": "resp-upstream-1",
+                "content": [
+                    {"type": "toolCall", "id": "a", "name": "read", "arguments": {}},
+                    {"type": "toolCall", "id": "b", "name": "bash", "arguments": {}},
+                ],
+            },
+        }
+        final_assistant = {
+            "type": "message_end",
+            "message": {
+                "role": "assistant",
+                "responseId": "resp-upstream-2",
+                "content": [{"type": "text", "text": "done"}],
+            },
+        }
+        events = [
+            (1.0, assistant),
+            (2.0, {"type": "tool_execution_start", "toolCallId": "a", "toolName": "read"}),
+            (3.0, {"type": "tool_execution_start", "toolCallId": "b", "toolName": "bash"}),
+            (4.0, {"type": "tool_execution_end", "toolCallId": "b", "toolName": "bash", "isError": True}),
+            (5.0, {"type": "tool_execution_end", "toolCallId": "a", "toolName": "read", "isError": False}),
+            (6.0, {"type": "compaction_start", "reason": "threshold"}),
+            (
+                7.0,
+                {
+                    "type": "compaction_end",
+                    "reason": "threshold",
+                    "result": {
+                        "summary": "condensed history",
+                        "firstKeptEntryId": "entry-7",
+                        "tokensBefore": 150_000,
+                        "estimatedTokensAfter": 32_000,
+                    },
+                    "aborted": False,
+                },
+            ),
+            (8.0, final_assistant),
+        ]
+        stdout = "\n".join(json.dumps(event) for _, event in events)
+        items, _ = parse_pi_events(stdout)
+        model_ref = ModelServerRef(type="responses_api_models", name="policy")
+
+        bundle = _build_pi_observations(events, "rollout-1", model_ref, items)
+
+        [invocation] = _records(bundle, AgentInvocation)
+        assert invocation.model_calls[0].response_id == "resp-upstream-1"
+        assert invocation.model_calls[0].model_ref == model_ref
+        timings = {tool.tool_call_id: tool for tool in _records(bundle, ToolCallObservation)}
+        assert (
+            timings["a"].started_at < timings["b"].started_at < timings["b"].completed_at < timings["a"].completed_at
+        )
+        assert timings["a"].duration_ms == 3000
+        assert timings["b"].duration_ms == 1000
+        assert all(tool.timing_source == "harness" for tool in timings.values())
+        assert timings["a"].status == "completed"
+        assert timings["b"].status == "failed"
+        assert timings["b"].error_type is None
+        [compaction] = _records(bundle, ContextCompactionObservation)
+        assert compaction.trigger == "threshold"
+        assert compaction.tokens_before == 150_000
+        assert compaction.tokens_after == 32_000
+        assert compaction.outcome == "completed"
+        assert compaction.summary == "condensed history"
+        assert compaction.first_kept_item_id == "entry-7"
+        assert compaction.before_model_call is not None
+        assert compaction.after_model_call is not None
+        assert compaction.before_model_call.response_id == "resp-upstream-1"
+        assert compaction.after_model_call.response_id == "resp-upstream-2"
+        assert {gap.code for gap in bundle.gaps} == {"subagent_hierarchy_unavailable"}
+
+    def test_compaction_outcome_uses_native_status(self) -> None:
+        events = [
+            (1.0, {"type": "compaction_start", "reason": "manual"}),
+            (2.0, {"type": "compaction_end", "reason": "manual", "result": None, "aborted": True}),
+            (3.0, {"type": "compaction_start", "reason": "overflow"}),
+            (
+                4.0,
+                {
+                    "type": "compaction_end",
+                    "reason": "overflow",
+                    "result": None,
+                    "aborted": False,
+                    "errorMessage": "quota",
+                },
+            ),
+        ]
+
+        compactions = _records(
+            _build_pi_observations(events, "rollout-1", None, []),
+            ContextCompactionObservation,
+        )
+
+        assert [item.outcome for item in compactions] == ["aborted", "failed"]
+
+    def test_compaction_join_does_not_skip_unjoinable_model_call(self) -> None:
+        model_ref = ModelServerRef(type="responses_api_models", name="policy")
+        events = [
+            (1.0, {"type": "message_end", "message": {"role": "assistant", "responseId": "resp-1"}}),
+            (2.0, {"type": "compaction_start", "reason": "threshold"}),
+            (3.0, {"type": "compaction_end", "reason": "threshold", "result": None, "aborted": True}),
+            (4.0, {"type": "message_end", "message": {"role": "assistant"}}),
+            (5.0, {"type": "message_end", "message": {"role": "assistant", "responseId": "resp-3"}}),
+        ]
+
+        [compaction] = _records(
+            _build_pi_observations(events, "rollout-1", model_ref, []),
+            ContextCompactionObservation,
+        )
+
+        assert compaction.before_model_call is not None
+        assert compaction.before_model_call.response_id == "resp-1"
+        assert compaction.after_model_call is None
+
+    def test_reports_only_missing_evidence(self) -> None:
+        items, _ = parse_pi_events(_msg_end("assistant", [{"type": "toolCall", "id": "c1", "name": "bash"}]))
+        bundle = _build_pi_observations([], "rollout-1", None, items)
+
+        assert {gap.code for gap in bundle.gaps} == {
+            "model_call_ownership_unavailable",
+            "subagent_hierarchy_unavailable",
+            "tool_timing_unavailable",
+        }
+        assert _records(bundle, ContextCompactionObservation) == []
+
+    def test_episode_preserves_response_and_observations(self) -> None:
+        event = {
+            "type": "message_end",
+            "message": {
+                "role": "assistant",
+                "responseId": "resp-upstream-1",
+                "content": [{"type": "text", "text": "done"}],
+            },
+        }
+        items, usage = parse_pi_events(json.dumps(event))
+        agent = _make_agent(model_server=ModelServerRef(type="responses_api_models", name="policy"))
+        agent._run_pi = AsyncMock(return_value=(items, usage, "model", [(1.0, event)]))
+
+        episode = asyncio.run(
+            agent._create_episode(NeMoGymResponseCreateParamsNonStreaming(input="solve"), rollout_id="1-2")
+        )
+
+        assert agent._run_pi.await_args.kwargs["rollout_id"] == "1-2"
+        assert episode.response.output == items
+        [invocation] = _records(episode.observations, AgentInvocation)
+        assert invocation.conversation == [
+            NeMoGymEasyInputMessage(role="user", content="solve"),
+            *items,
+        ]
+        assert invocation.model_calls[0].response_id == "resp-upstream-1"
+
+    def test_padding_is_not_reported_as_agent_evidence(self) -> None:
+        agent = _make_agent()
+        agent._run_pi = AsyncMock(return_value=([], {"input_tokens": 0, "output_tokens": 0}, "model", []))
+
+        episode = asyncio.run(agent._create_episode(NeMoGymResponseCreateParamsNonStreaming(input="solve")))
+
+        assert episode.response.output
+        [invocation] = _records(episode.observations, AgentInvocation)
+        assert invocation.conversation == [NeMoGymEasyInputMessage(role="user", content="solve")]
+        assert "agent_transcript_unavailable" in {gap.code for gap in episode.observations.gaps}
+
+    def test_partial_events_survive_empty_scoring_output(self) -> None:
+        event = {"type": "tool_execution_start", "toolCallId": "call-1", "toolName": "bash"}
+        agent = _make_agent()
+        agent._run_pi = AsyncMock(return_value=([], {"input_tokens": 0, "output_tokens": 0}, "model", [(1.0, event)]))
+
+        episode = asyncio.run(agent._create_episode(NeMoGymResponseCreateParamsNonStreaming(input="solve")))
+
+        [tool] = _records(episode.observations, ToolCallObservation)
+        assert tool.tool_call_id == "call-1"
+        assert tool.status == "incomplete"
+
+    def test_run_uses_prefixed_response_boundary(self) -> None:
+        agent = _make_agent()
+        agent.server_client.global_config_dict = {"observability_enabled": True}
+        agent._run_pi = AsyncMock(return_value=([], {"input_tokens": 0, "output_tokens": 0}, "model", []))
+
+        def response(payload):
+            result = MagicMock(ok=True, cookies={})
+            result.read = AsyncMock(return_value=json.dumps(payload).encode())
+            return result
+
+        async def post(server_name, url_path, json=None, cookies=None, **kwargs):
+            if url_path.endswith("/v1/responses"):
+                agent_response = await agent.responses(MagicMock(path_params={"rollout_id": "1-2"}), json)
+                return response(agent_response.model_dump(mode="json"))
+            if url_path == "/verify":
+                return response(json | {"reward": 1.0})
+            return response({})
+
+        agent.server_client.post = AsyncMock(side_effect=post)
+        body = PiAgentRunRequest.model_validate(
+            {
+                "responses_create_params": {"input": "solve"},
+                "_ng_task_index": 1,
+                "_ng_rollout_index": 2,
+            }
+        )
+
+        result = asyncio.run(agent.run(MagicMock(cookies={}), body))
+
+        assert result.ng_agent_observations is not None
+        assert agent.server_client.post.await_args_list[1].kwargs["url_path"] == "/ng-rollout/1-2/v1/responses"
+        assert agent._run_pi.await_args.kwargs["rollout_id"] == "1-2"
+        verify_json = agent.server_client.post.await_args_list[2].kwargs["json"]
+        assert "_ng_agent_observations" not in verify_json["response"]
+
+
 class TestConfigYaml:
     def test_module_parses(self) -> None:
         app_path = Path(__file__).resolve().parent.parent / "app.py"

From bfd0acde818915588b855c8cbdea34a4bfe2c02a Mon Sep 17 00:00:00 2001
From: Michal Bien <mbien@nvidia.com>
Date: Fri, 24 Jul 2026 18:34:38 +0200
Subject: [PATCH 2/4] Complete Pi observation status

Signed-off-by: Michal Bien <mbien@nvidia.com>
---
 .../configs/math_with_judge_pi_agent.yaml     |  3 ++
 responses_api_agents/pi_agent/app.py          | 22 ++++++++++++
 .../pi_agent/tests/test_app.py                | 34 ++++++++++++++++++-
 3 files changed, 58 insertions(+), 1 deletion(-)

diff --git a/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml b/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
index c431e8a10e..62b65775d5 100644
--- a/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
+++ b/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
@@ -19,6 +19,9 @@ math_with_judge_pi_agent:
       resources_server:
         type: resources_servers
         name: math_with_judge
+      model_server:
+        type: responses_api_models
+        name: policy_model
       concurrency: 8
       command: pi
       model: nvinf/nvidia/qwen/qwen3-next-80b-a3b-instruct
diff --git a/responses_api_agents/pi_agent/app.py b/responses_api_agents/pi_agent/app.py
index 33c3280a70..606b2d358b 100644
--- a/responses_api_agents/pi_agent/app.py
+++ b/responses_api_agents/pi_agent/app.py
@@ -190,6 +190,7 @@ def gap(code: str, detail: Optional[str] = None) -> ObservationGap:
     compactions: list[ContextCompactionObservation] = []
     compactions_waiting_for_call: list[ContextCompactionObservation] = []
     last_model_call: Optional[ModelCallRef] = None
+    invocation_status = "unknown"
 
     for observed_at, event in events:
         event_type = event.get("type")
@@ -211,6 +212,23 @@ def gap(code: str, detail: Optional[str] = None) -> ObservationGap:
                 if last_model_call is None:
                     gaps.append(gap("compaction_after_model_call_unavailable"))
             compactions_waiting_for_call.clear()
+        elif event_type == "agent_end":
+            terminal_messages = event.get("messages")
+            if isinstance(terminal_messages, list):
+                stop_reason = next(
+                    (
+                        item.get("stopReason")
+                        for item in reversed(terminal_messages)
+                        if isinstance(item, dict) and item.get("role") == "assistant"
+                    ),
+                    None,
+                )
+                invocation_status = {
+                    "stop": "completed",
+                    "error": "failed",
+                    "aborted": "incomplete",
+                    "length": "incomplete",
+                }.get(stop_reason, "unknown")
         elif event_type == "tool_execution_start" and isinstance(call_id, str):
             starts[call_id] = (observed_at, tool_name)
         elif event_type == "tool_execution_end" and isinstance(call_id, str):
@@ -290,6 +308,8 @@ def gap(code: str, detail: Optional[str] = None) -> ObservationGap:
             compaction_start = None
     if not model_calls or model_call_join_missing:
         gaps.append(gap("model_call_ownership_unavailable"))
+    if invocation_status == "unknown":
+        gaps.append(gap("invocation_outcome_unavailable"))
 
     for call_id, (started_at, tool_name) in starts.items():
         tools[call_id] = ToolCallObservation(
@@ -343,6 +363,7 @@ def field(item: Any, name: str) -> Any:
         records=[
             AgentInvocation(
                 invocation_id=invocation_id,
+                status=invocation_status,
                 model_calls=model_calls,
                 conversation=conversation,
             ),
@@ -623,6 +644,7 @@ async def _create_episode(
                 observations = AgentObservationBundle(
                     source="pi", gaps=[ObservationGap(code="observation_parse_failed")]
                 )
+            observations.gaps.append(ObservationGap(code="no_sandbox_runtime"))
         return AgentEpisode(response=response, observations=observations)
 
     async def responses(
diff --git a/responses_api_agents/pi_agent/tests/test_app.py b/responses_api_agents/pi_agent/tests/test_app.py
index deef3fe3d9..b7e281036c 100644
--- a/responses_api_agents/pi_agent/tests/test_app.py
+++ b/responses_api_agents/pi_agent/tests/test_app.py
@@ -18,6 +18,7 @@
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
 import yaml
 
 from nemo_gym.config_types import ModelServerRef
@@ -289,7 +290,10 @@ def test_exact_model_calls_parallel_tools_and_compaction(self) -> None:
         assert compaction.after_model_call is not None
         assert compaction.before_model_call.response_id == "resp-upstream-1"
         assert compaction.after_model_call.response_id == "resp-upstream-2"
-        assert {gap.code for gap in bundle.gaps} == {"subagent_hierarchy_unavailable"}
+        assert {gap.code for gap in bundle.gaps} == {
+            "invocation_outcome_unavailable",
+            "subagent_hierarchy_unavailable",
+        }
 
     def test_compaction_outcome_uses_native_status(self) -> None:
         events = [
@@ -315,6 +319,32 @@ def test_compaction_outcome_uses_native_status(self) -> None:
 
         assert [item.outcome for item in compactions] == ["aborted", "failed"]
 
+    @pytest.mark.parametrize(
+        ("stop_reason", "expected"),
+        [
+            ("stop", "completed"),
+            ("error", "failed"),
+            ("aborted", "incomplete"),
+            ("length", "incomplete"),
+            (None, "unknown"),
+            ("toolUse", "unknown"),
+        ],
+    )
+    def test_agent_end_sets_invocation_status(self, stop_reason, expected) -> None:
+        message = {"role": "assistant"}
+        if stop_reason is not None:
+            message["stopReason"] = stop_reason
+        bundle = _build_pi_observations(
+            [(1.0, {"type": "agent_end", "messages": [message]})],
+            "rollout-1",
+            None,
+            [],
+        )
+
+        [invocation] = _records(bundle, AgentInvocation)
+        assert invocation.status == expected
+        assert any(gap.code == "invocation_outcome_unavailable" for gap in bundle.gaps) is (expected == "unknown")
+
     def test_compaction_join_does_not_skip_unjoinable_model_call(self) -> None:
         model_ref = ModelServerRef(type="responses_api_models", name="policy")
         events = [
@@ -339,6 +369,7 @@ def test_reports_only_missing_evidence(self) -> None:
         bundle = _build_pi_observations([], "rollout-1", None, items)
 
         assert {gap.code for gap in bundle.gaps} == {
+            "invocation_outcome_unavailable",
             "model_call_ownership_unavailable",
             "subagent_hierarchy_unavailable",
             "tool_timing_unavailable",
@@ -370,6 +401,7 @@ def test_episode_preserves_response_and_observations(self) -> None:
             *items,
         ]
         assert invocation.model_calls[0].response_id == "resp-upstream-1"
+        assert "no_sandbox_runtime" in {gap.code for gap in episode.observations.gaps}
 
     def test_padding_is_not_reported_as_agent_evidence(self) -> None:
         agent = _make_agent()

From 3950e444b4b62b18eaa0780ae50eb557ede28129 Mon Sep 17 00:00:00 2001
From: Michal Bien <mbien@nvidia.com>
Date: Mon, 10 Aug 2026 12:37:51 +0200
Subject: [PATCH 3/4] Fix Pi observation fidelity

Signed-off-by: Michal Bien <mbien@nvidia.com>
---
 .../reference/trajectory-capabilities.mdx     |  2 +-
 responses_api_agents/pi_agent/app.py          | 10 +++++++--
 .../pi_agent/tests/test_app.py                | 21 ++++++++++++++++---
 3 files changed, 27 insertions(+), 6 deletions(-)

diff --git a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx
index 0ab4efad09..ffc94ab1f8 100644
--- a/fern/versions/latest/pages/reference/trajectory-capabilities.mdx
+++ b/fern/versions/latest/pages/reference/trajectory-capabilities.mdx
@@ -62,7 +62,7 @@ For C1, C2, and C4, `V` evaluates the correlated Gym Model Server path; direct-p
 | `openclaw_agent` | V | V | X | O | V | V | V |
 | `opencode_agent` | X | X | X | X | X | X | X |
 | `osworld_agent` | O | O | X | O | X | X | X |
-| `pi_agent` | X | X | X | X | X | X | X |
+| `pi_agent` | V | V | X | V | V | V | V |
 | `proof_refinement_agent` | V | V | X | V | X | X | X |
 | `remote_agent` | X | O | X | X | X | X | X |
 | `scicode_agent` | X | X | X | X | X | X | X |
diff --git a/responses_api_agents/pi_agent/app.py b/responses_api_agents/pi_agent/app.py
index 606b2d358b..d1dede9f4a 100644
--- a/responses_api_agents/pi_agent/app.py
+++ b/responses_api_agents/pi_agent/app.py
@@ -66,7 +66,9 @@
 _INTERNAL_OBSERVATIONS_KEY = "_ng_agent_observations"
 
 
-def parse_pi_events(stdout: str) -> tuple[list[Any], dict[str, int]]:
+def parse_pi_events(stdout: str | bytes) -> tuple[list[Any], dict[str, int]]:
+    if isinstance(stdout, bytes):
+        stdout = stdout.decode(errors="replace")
     output_items: list[Any] = []
     input_tokens = 0
     output_tokens = 0
@@ -582,6 +584,10 @@ async def _create_episode(
         user_message, input_system = _extract_instruction(body.input)
         system_parts = [p for p in [self.config.system_prompt, input_system] if p]
         system_prompt = "\n\n".join(system_parts) if system_parts else None
+        conversation_input = (
+            [NeMoGymEasyInputMessage(role="system", content=system_prompt)] if system_prompt is not None else []
+        )
+        conversation_input.append(NeMoGymEasyInputMessage(role="user", content=user_message))
 
         output_items, usage, model_name, events = await self._run_pi(
             user_message,
@@ -636,7 +642,7 @@ async def _create_episode(
                     events,
                     invocation_id,
                     self.config.model_server,
-                    [*body.input, *observed_output_items],
+                    [*conversation_input, *observed_output_items],
                     transcript_available=bool(observed_output_items),
                 )
             except Exception:
diff --git a/responses_api_agents/pi_agent/tests/test_app.py b/responses_api_agents/pi_agent/tests/test_app.py
index b7e281036c..f20385b87d 100644
--- a/responses_api_agents/pi_agent/tests/test_app.py
+++ b/responses_api_agents/pi_agent/tests/test_app.py
@@ -149,7 +149,7 @@ def test_tool_call_and_result(self) -> None:
         assert isinstance(items[2], NeMoGymResponseOutputMessage)
 
     def test_malformed_lines_skipped(self) -> None:
-        line = "not-json\nnull\n[]\n" + _msg_end("assistant", [{"type": "text", "text": "ok"}])
+        line = b"\xff\nnot-json\nnull\n[]\n" + _msg_end("assistant", [{"type": "text", "text": "ok"}]).encode()
         items, _ = parse_pi_events(line)
         assert len(items) == 1
 
@@ -386,17 +386,32 @@ def test_episode_preserves_response_and_observations(self) -> None:
             },
         }
         items, usage = parse_pi_events(json.dumps(event))
-        agent = _make_agent(model_server=ModelServerRef(type="responses_api_models", name="policy"))
+        agent = _make_agent(
+            model_server=ModelServerRef(type="responses_api_models", name="policy"),
+            system_prompt="configured system",
+        )
         agent._run_pi = AsyncMock(return_value=(items, usage, "model", [(1.0, event)]))
 
         episode = asyncio.run(
-            agent._create_episode(NeMoGymResponseCreateParamsNonStreaming(input="solve"), rollout_id="1-2")
+            agent._create_episode(
+                NeMoGymResponseCreateParamsNonStreaming(
+                    input=[
+                        NeMoGymEasyInputMessage(role="system", content="request system"),
+                        NeMoGymEasyInputMessage(role="user", content="old question"),
+                        NeMoGymEasyInputMessage(role="assistant", content="old answer"),
+                        NeMoGymEasyInputMessage(role="user", content="solve"),
+                    ]
+                ),
+                rollout_id="1-2",
+            )
         )
 
         assert agent._run_pi.await_args.kwargs["rollout_id"] == "1-2"
+        assert agent._run_pi.await_args.args == ("solve", "configured system\n\nrequest system")
         assert episode.response.output == items
         [invocation] = _records(episode.observations, AgentInvocation)
         assert invocation.conversation == [
+            NeMoGymEasyInputMessage(role="system", content="configured system\n\nrequest system"),
             NeMoGymEasyInputMessage(role="user", content="solve"),
             *items,
         ]

From c429e65eb79ca48d49e6f3f1a8805a4a3ba5a05c Mon Sep 17 00:00:00 2001
From: Michal Bien <mbien@nvidia.com>
Date: Tue, 11 Aug 2026 15:03:57 +0200
Subject: [PATCH 4/4] Fix Pi observation timing without changing benchmark
 config

Signed-off-by: Michal Bien <mbien@nvidia.com>
---
 .../configs/math_with_judge_pi_agent.yaml     |  3 --
 responses_api_agents/pi_agent/app.py          |  2 +-
 .../pi_agent/tests/test_app.py                | 30 +++++++++++++++++++
 3 files changed, 31 insertions(+), 4 deletions(-)

diff --git a/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml b/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
index 62b65775d5..c431e8a10e 100644
--- a/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
+++ b/resources_servers/math_with_judge/configs/math_with_judge_pi_agent.yaml
@@ -19,9 +19,6 @@ math_with_judge_pi_agent:
       resources_server:
         type: resources_servers
         name: math_with_judge
-      model_server:
-        type: responses_api_models
-        name: policy_model
       concurrency: 8
       command: pi
       model: nvinf/nvidia/qwen/qwen3-next-80b-a3b-instruct
diff --git a/responses_api_agents/pi_agent/app.py b/responses_api_agents/pi_agent/app.py
index d1dede9f4a..7b5159bfb5 100644
--- a/responses_api_agents/pi_agent/app.py
+++ b/responses_api_agents/pi_agent/app.py
@@ -242,7 +242,7 @@ def gap(code: str, detail: Optional[str] = None) -> ObservationGap:
                 invocation_id=invocation_id,
                 tool_call_id=call_id,
                 tool_name=tool_name or started_name,
-                started_at=started_at,
+                started_at=started_at if valid_interval else None,
                 completed_at=completed_at,
                 duration_ms=duration_ms,
                 timing_source="harness",
diff --git a/responses_api_agents/pi_agent/tests/test_app.py b/responses_api_agents/pi_agent/tests/test_app.py
index f20385b87d..1961f69e67 100644
--- a/responses_api_agents/pi_agent/tests/test_app.py
+++ b/responses_api_agents/pi_agent/tests/test_app.py
@@ -295,6 +295,20 @@ def test_exact_model_calls_parallel_tools_and_compaction(self) -> None:
             "subagent_hierarchy_unavailable",
         }
 
+    def test_invalid_tool_interval_keeps_only_valid_timing(self) -> None:
+        events = [
+            (2.0, {"type": "tool_execution_start", "toolCallId": "call-1"}),
+            (1.0, {"type": "tool_execution_end", "toolCallId": "call-1", "isError": False}),
+        ]
+
+        bundle = _build_pi_observations(events, "rollout-1", None, [])
+
+        [tool] = _records(bundle, ToolCallObservation)
+        assert tool.started_at is None
+        assert tool.completed_at == 1.0
+        assert tool.duration_ms is None
+        assert any(gap.code == "tool_timing_unavailable" and gap.detail == "call-1" for gap in bundle.gaps)
+
     def test_compaction_outcome_uses_native_status(self) -> None:
         events = [
             (1.0, {"type": "compaction_start", "reason": "manual"}),
@@ -475,6 +489,22 @@ async def post(server_name, url_path, json=None, cookies=None, **kwargs):
         verify_json = agent.server_client.post.await_args_list[2].kwargs["json"]
         assert "_ng_agent_observations" not in verify_json["response"]
 
+    def test_prefixed_responses_preserves_observations_through_fastapi(self) -> None:
+        from fastapi.testclient import TestClient
+
+        agent = _make_agent()
+        agent._run_pi = AsyncMock(return_value=([], {"input_tokens": 0, "output_tokens": 0}, "model", []))
+
+        response = TestClient(agent.setup_webserver()).post(
+            "/ng-rollout/1-2/v1/responses",
+            json={"input": "solve"},
+        )
+
+        assert response.status_code == 200
+        observations = response.json()["_ng_agent_observations"]
+        assert observations["source"] == "pi"
+        assert observations["records"][0]["invocation_id"] == "1-2"
+
 
 class TestConfigYaml:
     def test_module_parses(self) -> None: