feat: enable Relay streaming for Pi

adapters/README.md

@@ -142,7 +142,7 @@ and additive extension maps because their support does not vary by adapter:
 | `skills.paths` | Yes | Yes | Yes | Yes | No | InteractiveAgent: Yes; BenchAgent: No | No | Yes | No |
 | `mcp.servers.<name>.transport`, `.url` with `harness_native` exposure | Yes | Yes | Yes | Yes | No | InteractiveAgent: Yes; BenchAgent: No | No | No | No |
 | `mcp.servers.<name>.exposure = "fabric_managed"` | No; not implemented | No; not implemented | No; not implemented | No; not implemented | No | No; not implemented | No | No | No |
-| `telemetry.providers.relay` | Yes | Yes | Yes | Yes | Yes | Yes | No | Yes | Yes, supports collector-backed ATOF streaming |
+| `telemetry.providers.relay` | Yes | Yes | Yes | Yes | Yes | Yes | No | Yes, supports embedded collector-backed ATOF streaming | Yes, supports collector-backed ATOF streaming |
 | `telemetry.providers.native` | No | Yes; OpenTelemetry | Yes; OpenTelemetry and OpenInference | No | No | No | No | No | No |
 | `telemetry.providers.<provider>.config` | Declared-provider pass-through | Declared-provider pass-through | Declared-provider pass-through | Declared-provider pass-through | Declared-provider pass-through | Declared-provider pass-through | No | Declared-provider pass-through | No |
 | `relay.project`, `.output_dir`, `.observability` | Yes | Yes | Yes | Yes | Yes | Yes | No | Yes | Uses the named external collector sink when selected; config is not sent to the remote service |
@@ -184,7 +184,7 @@ and produces normalized trajectories in Agent Trajectory Interchange Format
 | [mini-SWE-agent](python/mini-swe-agent/README.md) | Conversation history | Adapter-owned subclass with NeMo Relay Python SDK scopes | Creates a fresh Relay plugin and request scope, emits step, model, and bash-action telemetry, and collects artifacts | Clears the agent and Relay state | Not implemented |
 | [NOOA](python/nooa/README.md) | InteractiveAgent queue dispatcher or BenchAgent task state | Adapter-owned Relay middleware and generated Relay configuration | InteractiveAgent dispatches queued requests; BenchAgent evaluates one task | Closes agent resources and Relay state | Not implemented |
 | [OpenCode](typescript/opencode/README.md) | Embedded OpenCode host and session | Not supported | Reuses the session and calls `prompt()`, `wait()`, and `context()` for ordered text input | Removes the session and closes the host | Not implemented |
-| [Pi](typescript/pi/README.md) | In-memory Pi `AgentSession` | Runtime-owned Relay 0.9 CLI gateway and explicit Pi extension | Reuses the session, calls `prompt()` for ordered text input, and collects ATOF; `relay_artifacts` does not include local ATIF | Aborts work, emits extension shutdown so local ATIF finalizes on disk, disposes the session, and then stops the gateway | Not implemented |
+| [Pi](typescript/pi/README.md) | In-memory Pi `AgentSession` | Runtime-owned Relay 0.9 CLI gateway and explicit Pi extension | Reuses the session and calls `prompt()` for ordered text input; with `streaming=True`, routes every model turn's ATOF through the embedded collector; `relay_artifacts` does not include local ATIF | Aborts work, emits extension shutdown so local ATIF finalizes on disk, disposes the session, and then stops the gateway | Not implemented |
 | [Remote Agent](python/remote-agent/README.md) | `httpx.AsyncClient` and user/assistant transcript | Remote Relay publishes to a shared ATOF collector | Registers the request ID, maps it into body metadata, sends one HTTP request, and retains the completed transcript | Closes the HTTP client | Implemented over HTTP(S) |
 
 Telemetry output names use the descriptor contract values. Claude, Codex,

adapters/typescript/pi/README.md

@@ -24,6 +24,8 @@ The adapter supports:
 - Slash commands registered by those explicit extensions
 - NeMo Relay 0.9 telemetry through a runtime-owned gateway and an explicitly
   configured Relay Pi extension
+- Live ATOF records from every Pi model turn through the default embedded
+  NeMo Fabric collector
 - Ordered plain-text invocations with a `{ "response": "..." }` terminal
   output, Relay runtime details, and collected ATOF artifacts
 
@@ -154,8 +156,43 @@ results do not prevent subsequent turns.
 Session, turn, and tool telemetry does not depend on model redirection. Model
 telemetry is available only when Relay supports the selected model API and the
 gateway upstream matches the model endpoint. A skipped redirect is recorded as
-a `model_redirect` mark with the reason. Relay-backed
-`Runtime.invoke_stream()` correlation is not yet supported for Pi.
+a `model_redirect` mark with the reason.
+
+Install the matching collector for the embedded streaming path:
+
+```bash
+pip install "nemo-fabric[streaming]"
+```
+
+Start the runtime with streaming enabled to consume live ATOF records from all
+model turns in one Pi invocation:
+
+```python
+from nemo_fabric import Fabric
+
+async with await Fabric().start_runtime(config, streaming=True) as runtime:
+    stream = runtime.invoke_stream(input="Review the latest patch")
+    async for record in stream:
+        print(record)
+    result = await stream.result()
+```
+
+The terminal `RunResult` remains separate from the ATOF records. Fully consume
+each stream, or call `await stream.aclose()` if iteration stops early, before
+starting another invocation; the same runtime can then alternate
+`invoke_stream()` and `invoke()` calls. The embedded collector serializes both
+methods behind one Pi invocation lease. Streaming capture begins at the first
+Pi `turn_start` and closes at `agent_settled`. If Relay output is interrupted or
+late, the collector discards the remaining records through that same terminal
+marker before allowing another invocation to start. Use the default embedded
+collector for Pi streaming. The Pi extension does not attach NeMo Fabric
+request IDs, so
+`start_runtime(..., streaming=True, launch_collector=False)` cannot correlate
+its records through an externally managed collector.
+
+This Relay-backed path runs the adapter's ordinary `invoke` operation. It is
+independent of native OpenAI streaming, so the adapter descriptor's
+`capabilities.streaming` value remains `false`.
 
 ## Custom Tool Modules
 
@@ -204,11 +241,14 @@ path explicitly:
 .venv/bin/python -m examples.code_review_agent \
   --variant pi \
   --relay \
+  --stream \
   --pi-relay-extension-path /path/to/NeMo-Relay/crates/cli/assets/pi-extension \
   --input "Review calculator.py"
 ```
 
-MCP is not currently supported. Do not combine the Pi variant with `--stream`.
+The command collects Relay ATOF records from every model turn, then prints one
+JSON document containing `atof_records` and the separate terminal `result`. MCP
+is not currently supported.
 
 ## Dependency Rationale
 

docs/getting-started/install.mdx

@@ -256,8 +256,9 @@ directory; relative paths resolve from `environment.workspace`. Relay-enabled
 Pi configurations must also set `runtime.artifacts` so NeMo Fabric can provide
 the adapter-owned Relay configuration. Refer to the
 [Pi adapter guide](../integrations/harness/pi.mdx) for the complete
-configuration. Relay-backed `Runtime.invoke_stream()` correlation is not yet
-supported for Pi.
+configuration and the embedded collector streaming workflow. Do not set
+`launch_collector=False` for Pi streaming because the Pi extension does not
+attach the NeMo Fabric request IDs required by an externally managed collector.
 
 ## Source Installation Method
 

docs/integrations/harness/pi.mdx

@@ -173,6 +173,42 @@ telemetry is available only when Relay supports the selected model API and the
 gateway upstream matches the model endpoint. A skipped redirect is recorded as
 a `model_redirect` mark with the reason.
 
+Install the matching collector for the embedded streaming path:
+
+```bash
+pip install "nemo-fabric[streaming]"
+```
+
+Start the runtime with streaming enabled to consume live ATOF records from all
+model turns in one Pi invocation:
+
+```python
+from nemo_fabric import Fabric
+
+async with await Fabric().start_runtime(config, streaming=True) as runtime:
+    stream = runtime.invoke_stream(input="Review the latest patch")
+    async for record in stream:
+        print(record)
+    result = await stream.result()
+```
+
+The terminal `RunResult` remains separate from the ATOF records. Fully consume
+each stream, or call `await stream.aclose()` if iteration stops early, before
+starting another invocation; the same runtime can then alternate
+`invoke_stream()` and `invoke()` calls. The embedded collector serializes both
+methods behind one Pi invocation lease. Streaming capture begins at the first
+Pi `turn_start` and closes at `agent_settled`. If Relay output is interrupted or
+late, the collector discards the remaining records through that same terminal
+marker before allowing another invocation to start. Use the default embedded
+collector for Pi streaming. The Pi extension does not attach NeMo Fabric
+request IDs, so
+`start_runtime(..., streaming=True, launch_collector=False)` cannot correlate
+its records through an externally managed collector.
+
+This Relay-backed path runs the adapter's ordinary `invoke` operation. It is
+independent of native OpenAI streaming, so the adapter descriptor's
+`capabilities.streaming` value remains `false`.
+
 ## Configure a NeMo Fabric Tool Definition
 
 Pi accepts trusted local JavaScript and TypeScript tool factories. The
@@ -226,11 +262,9 @@ skills, extensions, custom tools, or workspace.
 ## Current Limitations
 
 The current adapter does not expose MCP, native OpenAI streaming, caller-driven
-cancellation, or remote-service execution. Relay-backed
-`Runtime.invoke_stream()` correlation is not yet supported for Pi. Enabling a
-third-party extension can add extension-defined skills, prompts, or themes even
-though ambient Pi resources remain disabled. The Relay extension does not add
-those resources.
+cancellation, or remote-service execution. Enabling a third-party extension
+can add extension-defined skills, prompts, or themes even though ambient Pi
+resources remain disabled. The Relay extension does not add those resources.
 
 The Relay Pi hook route is available only on the loopback gateway and does not
 authenticate hook posts. Treat the loopback port as a local trust boundary.

docs/reference/api/python-library-reference/nemo_fabric.client.md

@@ -170,7 +170,7 @@ async def start_runtime(
 
 Start a stateful runtime for one or more ordered invocations.
 
-Each call starts a new logical runtime. Runtime-scoped overrides are recursively merged below invocation-scoped overrides. With NVIDIA NeMo Relay enabled, ``streaming=True`` uses collector-backed streaming. By default, streaming starts an embedded collector. Set ``launch_collector=False`` to use an externally managed collector.
+Each call starts a new logical runtime. Runtime-scoped overrides are recursively merged below invocation-scoped overrides. With NVIDIA NeMo Relay enabled, ``streaming=True`` uses collector-backed streaming. By default, streaming starts an embedded collector. Set ``launch_collector=False`` to use an externally managed collector. Pi requires the embedded collector because its ATOF records do not carry Fabric request IDs.
 
 
 
@@ -180,7 +180,7 @@ Each call starts a new logical runtime. Runtime-scoped overrides are recursively
  - <b>`base_dir`</b>:  Base directory for resolving relative paths.
  - <b>`overrides`</b>:  JSON-compatible overrides applied to every invocation  in the runtime unless superseded by invocation overrides.
  - <b>`streaming`</b>:  Whether to enable collector-backed NeMo Relay ATOF  streaming for ``Runtime.invoke_stream()``.
- - <b>`launch_collector`</b>:  Whether to launch an embedded collector. ``None``  defaults to ``True`` when streaming is enabled. ``False`` uses  an externally managed collector. This argument cannot be set  unless ``streaming=True``.
+ - <b>`launch_collector`</b>:  Whether to launch an embedded collector. ``None``  defaults to ``True`` when streaming is enabled. ``False`` uses  an externally managed collector. Pi does not support ``False``.  This argument cannot be set unless ``streaming=True``.
 
 
 
@@ -191,7 +191,7 @@ Each call starts a new logical runtime. Runtime-scoped overrides are recursively
 
 **Raises:**
 
- - <b>`FabricConfigError`</b>:  If inputs or overrides are invalid, streaming is  requested without NeMo Relay enabled, ``launch_collector`` is  set without streaming, or an external collector has no sink.
+ - <b>`FabricConfigError`</b>:  If inputs or overrides are invalid, streaming is  requested without NeMo Relay enabled, ``launch_collector`` is  set without streaming, Pi is configured with an external  collector, or an external collector has no sink.
  - <b>`FabricNativeUnavailableError`</b>:  If the native extension is not  installed.
  - <b>`FabricRuntimeError`</b>:  If runtime startup fails.
 

docs/sdk/python.mdx

@@ -641,7 +641,10 @@ Streaming has the following v0.1 constraints:
   `launch_collector=False` to use an externally managed collector. In that
   case, configure exactly one `nemo-fabric-stream` sink with an `http` or
   `https` collector base URL and `transport="ndjson"`; Fabric sends Relay to
-  `<base-url>/v1/atof`.
+  `<base-url>/v1/atof`. Pi is the exception: Pi streaming supports only the
+  default embedded collector. Do not set `launch_collector=False` for Pi
+  because its extension does not attach the NeMo Fabric request IDs required
+  for external correlation.
 - The default `streaming=False` leaves the existing NeMo Relay configuration
   unchanged; it neither enables nor disables ATOF. It also avoids collector
   startup and streaming control traffic for runtimes that use only ATIF,
@@ -666,17 +669,22 @@ Streaming has the following v0.1 constraints:
   terminal scope. In-process harnesses expose scope-level progress.
 - Reconstruct nested and parallel work with `uuid` and `parent_uuid`. Stream
   order alone does not define the Deep Agents scope tree.
-- Before invoking the agent, the runtime registers its request ID with the
-  collector. The collector routes the ATOF root scope carrying that request ID
-  and its descendants, based on UUID ancestry, to that invocation’s private
-  queue. With `relay_streaming: true`, Remote Agent services receive the ID as
-  `metadata.nemo_fabric_request_id` in the mapped request body and must carry it
-  into Relay metadata. Records that cannot be correlated to a registered
-  request are discarded.
-- An external collector can be shared by multiple Fabric runtimes. It must be
-  reachable by both Relay publishers and Fabric clients. Use HTTPS and its
-  configured authentication when it is deployed beyond a trusted local
-  boundary.
+- For request-correlated adapters, the runtime registers its request ID with
+  the collector before invoking the agent. The collector routes the ATOF root
+  scope carrying that request ID and its descendants, based on UUID ancestry,
+  to that invocation's private queue. With `relay_streaming: true`, Remote
+  Agent services receive the ID as `metadata.nemo_fabric_request_id` in the
+  mapped request body and must carry it into Relay metadata. Records that
+  cannot be correlated to a registered request are discarded. Pi instead
+  serializes both streaming and plain invocations behind one embedded-collector
+  lease. Streaming capture begins at the first `turn_start` and closes at
+  `agent_settled`. If output is interrupted or late, the collector discards the
+  remainder through that same terminal marker before allowing the next
+  invocation to start.
+- An external collector can be shared by multiple Fabric runtimes whose
+  adapters propagate request IDs. It must be reachable by both Relay
+  publishers and Fabric clients. Use HTTPS and its configured authentication
+  when it is deployed beyond a trusted local boundary.
 
 Claude and Codex streaming use the NeMo Relay `nemo-relay` gateway CLI and
 require a release in the `>=0.7.2,<0.8` range. Their adapter `harness` and

examples/README.md

@@ -40,8 +40,9 @@ just build-all
 
 The Pi variant supports Relay telemetry with `nemo-relay>=0.9.0,<0.10.0` and an
 explicit extension path. Pass
-`--variant pi --relay --pi-relay-extension-path <PATH>` but do not combine the Pi
-variant with `--stream`.
+`--variant pi --relay --stream --pi-relay-extension-path <PATH>` to collect ATOF
+records from every model turn, then print one JSON document containing
+`atof_records` and the separate terminal `result`.
 
 ## LangGraph Custom Agent
 

examples/code_review_agent/README.md

@@ -195,9 +195,12 @@ and pass the Relay Pi extension path explicitly:
 .venv/bin/python -m examples.code_review_agent \
   --variant pi \
   --relay \
+  --stream \
   --pi-relay-extension-path /path/to/NeMo-Relay/crates/cli/assets/pi-extension \
   --input "Review calculator.py"
 ```
 
-The Pi variant does not yet support Relay-backed streaming, so do not add
-`--stream`.
+The Pi variant uses the default embedded collector to collect ATOF records from
+every model turn, then prints one JSON document containing `atof_records` and
+the separate terminal `result`. Omit `--stream` to retain Relay artifacts
+without collecting records for that JSON output.

examples/code_review_agent/__main__.py

@@ -80,8 +80,6 @@ async def main() -> None:
         parser.error("--stream requires --relay")
     if args.stream and args.plan:
         parser.error("--stream cannot be combined with --plan")
-    if args.variant == "pi" and args.stream:
-        parser.error("the Pi adapter does not support Relay-backed streaming yet")
     if args.pi_relay_extension_path is not None and args.variant != "pi":
         parser.error("--pi-relay-extension-path requires --variant pi")
     if (
@@ -98,9 +96,7 @@ async def main() -> None:
     elif args.no_skills:
         config = with_skill_paths(config)
     if args.pi_relay_extension_path is not None:
-        config.harness.settings["relay_extension_path"] = (
-            args.pi_relay_extension_path
-        )
+        config.harness.settings["relay_extension_path"] = args.pi_relay_extension_path
     if args.relay:
         config = with_relay(config)
 

examples/harbor/README.md

@@ -84,8 +84,9 @@ The `nemo-fabric` package installs the runtime. The `relay` extra installs the
 NeMo Relay Python package, not the CLI required by Claude.
 Pi Relay configurations must set `runtime.artifacts` and
 `harness.settings.relay_extension_path`; relative extension paths resolve from
-the task's `environment.workspace`. The Pi adapter does not currently support
-Relay-backed `Runtime.invoke_stream()`.
+the task's `environment.workspace`. Applications that use Pi with Relay-backed
+`Runtime.invoke_stream()` must also install `nemo-fabric[streaming]` and use the
+default embedded collector.
 
 Hermes Agent 0.20 and later is no longer installable from PyPI. Prepare Hermes
 Agent task images by following the