[DO NOT MERGE]feat: add streaming API POC

streaming-poc/.gitignore

@@ -0,0 +1,5 @@
+# Runtime artifacts from run_harness.py
+.work/
+*.out.jsonl
+out.atof.jsonl
+__pycache__/

streaming-poc/README.md

@@ -0,0 +1,60 @@
+# NeMo Fabric streaming POC
+
+Proof-of-concept for a Fabric streaming API built on **NeMo Relay-generated ATOF**.
+Hermes and Deep Agents were run for real — their raw native events (teed before
+Relay) and ATOF captured, and the same `invoke_stream` prototype exercised.
+Codex and Claude are stubs pending a usable API key. Conclusion + production plan:
+[`synthesis/`](synthesis/README.md).
+
+## The v0.1 contract (recommended)
+```python
+runtime = await fabric.start_runtime(config)   # Relay enabled → loopback ATOF endpoint injected
+stream  = runtime.invoke_stream(input="...")
+async for atof_record in stream:   # RAW canonical ATOF record (dict), one per Relay event
+    ...
+result = await stream.result()     # RunResult, out of band
+```
+Relay-only; available only when Relay is enabled; raw ATOF pass-through (no
+normalization in v0.1); `RunResult` out of band. Why normalization is deferred:
+[`synthesis/README.md`](synthesis/README.md).
+
+## Layout
+```
+streaming-poc/
+├── implementation-spec.md   architecture + end-to-end flow (mermaid diagram)
+├── common/          the experimental prototype: loopback listener, invoke_stream,
+│                    run_harness, native_recorder
+├── hermes/          in-process · native-events.jsonl · events.atof.jsonl · findings.md
+├── deepagents/      in-process · native-events.jsonl · events.atof.jsonl · findings.md
+├── claude/          gateway · findings.md (stub: pending ANTHROPIC_API_KEY)
+├── codex/           gateway · findings.md (stub: pending a funded OpenAI key)
+└── synthesis/       cross-harness conclusion + production work breakdown
+```
+Each completed harness folder carries `native-events.jsonl` (the SDK stream teed
+*before* Relay via `common/native_recorder.py`), `events.atof.jsonl` (the Relay
+ATOF that crossed the Fabric boundary), and `findings.md` (native→ATOF mapping,
+loss analysis, deltas-vs-terminal, duplicate-rendering risks, recommendation).
+
+| Harness | mode | status |
+|---|---|---|
+| Hermes | in-process | ✅ complete — real run, native + ATOF captured |
+| Deep Agents | in-process | ✅ complete — real run w/ delegated subagents |
+| Codex / Claude | gateway | ⏸ stub — pending a usable API key (native seam identified) |
+
+## Reproduce (Hermes / Deep Agents)
+Prereqs: a native extension matching the Python SDK (`just build-python`, or
+`PYO3_PYTHON=$PWD/.venv/bin/python cargo build -p fabric-python --release` then copy
+`target/release/lib_native.dylib` → `python/src/nemo_fabric/_native.abi3.so`), and
+`NVIDIA_API_KEY`.
+
+```bash
+python streaming-poc/common/run_harness.py nvidia.fabric.hermes out.atof.jsonl "your prompt"
+```
+To also tee native events, set `POC_NATIVE_RECORD=<path>` and
+`POC_RECORDER_DIR=streaming-poc/common` and apply the seam patch documented in
+`common/native_recorder.py` (POC-only; revert after capture). Gateway harnesses
+(Claude/Codex) additionally need `nemo-relay` ≥0.6.0 and the respective key.
+
+## Fixture note
+Oversized full-request snapshot records (>20 KB/line) have their `data` truncated
+in the committed fixtures; the streaming deltas and IDs are preserved.

streaming-poc/claude/findings.md

@@ -0,0 +1,20 @@
+# Claude streaming POC — stub (pending a usable API key)
+
+**Harness:** `nvidia.fabric.claude` · gateway (Relay CLI) mode · part of the
+Codex + Claude child POC.
+
+**Status: pending a usable `ANTHROPIC_API_KEY`.** None is available in this
+environment, so a real Claude run cannot be captured. Claude runs against the
+Anthropic Messages API through the Relay gateway; the relay-only `invoke_stream`
+prototype ([../common/](../common/README.md)) is harness-agnostic and validated on
+the same gateway path, so only a real run is pending.
+
+**To complete when a key is available:** run the harness through
+`common/run_harness.py`, and capture native evidence by teeing
+`ClaudeSDKClient.receive_response()` with `include_partial_messages=True` — record
+every `Message` / `StreamEvent.event` before normalization — at
+[adapters/claude/.../adapter.py:889](../../adapters/claude/src/nemo_fabric_adapters/claude/adapter.py#L889),
+alongside the Relay listener. Then add `native-events.jsonl`, `events.atof.jsonl`,
+and the native→ATOF mapping here.
+
+Cross-harness recommendation: [../synthesis/README.md](../synthesis/README.md).

streaming-poc/codex/findings.md

@@ -0,0 +1,22 @@
+# Codex streaming POC — stub (pending a usable API key)
+
+**Harness:** `nvidia.fabric.codex` · gateway (Relay CLI) mode · part of the
+Codex + Claude child POC.
+
+**Status: pending a usable OpenAI API key.** Codex runs against the OpenAI
+Responses API through the Relay gateway. A clean end-to-end capture is blocked —
+the available key is out of quota, and NVIDIA's inference endpoint rejects Codex's
+OpenAI-proprietary tool types. The relay-only `invoke_stream` prototype
+([../common/](../common/README.md)) is harness-agnostic and already validated on
+the other gateway path, so no Codex-specific prototype work is pending — only a
+real run.
+
+**To complete when a key is available:** run the harness through
+`common/run_harness.py`, and capture native evidence by teeing
+`AsyncTurnHandle.stream()` — record every notification's `method` + `payload`
+before the terminal-result collector — at
+[adapters/codex/.../adapter.py:927](../../adapters/codex/src/nemo_fabric_adapters/codex/adapter.py#L927),
+alongside the Relay listener. Then add `native-events.jsonl`, `events.atof.jsonl`,
+and the native→ATOF mapping here.
+
+Cross-harness recommendation: [../synthesis/README.md](../synthesis/README.md).

streaming-poc/common/README.md

@@ -0,0 +1,13 @@
+# common/ — shared prototype
+
+Harness-agnostic prototype used by all four child POCs (one mechanism, no
+per-harness streaming code).
+
+| file | purpose |
+|---|---|
+| `atof_stream.py` | `AtofStreamListener` — loopback ndjson sink; async queue of raw ATOF; bounded-queue backpressure; handles >512 KB gateway records |
+| `fabric_stream.py` | `Runtime.invoke_stream()` prototype — `StreamingRuntime` / `InvokeStream`; endpoint injection at `start_runtime`; raw ATOF + out-of-band `result()`; honest early-exit |
+| `run_harness.py` | run one real harness through `invoke_stream` and save its raw ATOF stream |
+| `native_recorder.py` | **POC-only** recorder that tees a harness's native SDK stream *before* Relay (produces `native-events.jsonl`); documents the per-harness capture seams. Not a Fabric API. |
+
+Requires a built native extension and provider creds — see [`../README.md`](../README.md).

streaming-poc/common/atof_stream.py

@@ -0,0 +1,79 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""SDK-side loopback ATOF stream listener (Fabric streaming POC).
+
+Stands up a loopback ndjson sink that NeMo Relay pushes ATOF records into, and
+exposes them as an async queue. Handles the large records gateway harnesses
+(Claude/Codex) emit — they embed the full model request/response, so a single
+line can exceed aiohttp's default 512 KB readline limit; we read raw chunks and
+split on newlines ourselves.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from aiohttp import web
+
+
+class AtofStreamListener:
+    """Receive live ndjson ATOF push; expose raw records via an async queue."""
+
+    def __init__(
+        self, host: str = "127.0.0.1", port: int = 0, maxsize: int = 0
+    ) -> None:
+        self._host = host
+        self._port = port
+        # maxsize>0 bounds memory: a full queue blocks the handler's put(), which
+        # stops reading the socket -> TCP backpressure -> Relay's sender backs off.
+        self._queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=maxsize)
+        self._runner: web.AppRunner | None = None
+        self._bound_port: int | None = None
+
+    @property
+    def url(self) -> str:
+        if self._bound_port is None:
+            raise RuntimeError("listener not started")
+        return f"http://{self._host}:{self._bound_port}/atof"
+
+    @property
+    def records(self) -> "asyncio.Queue[Any]":
+        return self._queue
+
+    async def start(self) -> "AtofStreamListener":
+        app = web.Application(client_max_size=256 * 1024 * 1024)
+        app.router.add_route("*", "/{tail:.*}", self._handle)
+        self._runner = web.AppRunner(app, access_log=None)
+        await self._runner.setup()
+        site = web.TCPSite(self._runner, self._host, self._port)
+        await site.start()
+        self._bound_port = self._runner.addresses[0][1]  # discover ephemeral port
+        return self
+
+    async def _handle(self, request: web.Request) -> web.Response:
+        buf = b""
+        async for chunk in request.content.iter_chunked(65536):
+            buf += chunk
+            while b"\n" in buf:
+                line, buf = buf.split(b"\n", 1)
+                await self._emit(line)
+        await self._emit(buf)
+        return web.Response(text="ok")
+
+    async def _emit(self, line: bytes) -> None:
+        line = line.strip()
+        if not line:
+            return
+        try:
+            rec = json.loads(line)
+        except json.JSONDecodeError:
+            return
+        await self._queue.put(rec)
+
+    async def close(self) -> None:
+        if self._runner is not None:
+            await self._runner.cleanup()
+            self._runner = None

streaming-poc/common/fabric_stream.py

@@ -0,0 +1,158 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""`Runtime.invoke_stream()` prototype — sugar over `Runtime.invoke()`.
+
+v0.1 contract (per the streaming decision):
+    runtime = await start_streaming_runtime(fabric, config)   # relay enabled
+    stream  = runtime.invoke_stream(input="...")
+    async for record in stream:      # RAW canonical ATOF record (dict)
+        ...
+    result = await stream.result()   # RunResult, out of band
+
+Relay's config (incl. the ATOF endpoint) is fixed when the adapter subprocess is
+spawned at start_runtime, so the loopback endpoint is injected there; `invoke_stream`
+is then pure sugar over `invoke`. No Fabric-specific event normalization in v0.1 —
+consumers read raw Relay-generated ATOF.
+
+In the shipped SDK this injection would live inside `Fabric.start_runtime`; here a
+factory models it without modifying the SDK.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from nemo_fabric.errors import FabricCapabilityError
+from nemo_fabric.models import (
+    FabricConfig,
+    RelayAtofConfig,
+    RelayAtofEndpointConfig,
+    RelayObservabilityConfig,
+    TelemetryConfig,
+)
+
+from atof_stream import AtofStreamListener
+
+_DRAIN = 0.25  # seconds to drain trailing socket data after a turn completes
+
+
+def _relay_enabled(config: FabricConfig) -> bool:
+    """Streaming availability signal = Relay in telemetry providers.
+
+    NOTE: distinct from ``RuntimeCapabilities.streaming`` (that flag is native
+    adapter progressive output, always False for Relay-based streaming).
+    """
+    tel = config.telemetry
+    if tel is None:
+        return False
+    if not isinstance(tel, TelemetryConfig):
+        tel = TelemetryConfig.model_validate(tel)
+    return "relay" in (tel.providers or {})
+
+
+def _coerce(value: Any, model: type) -> Any:
+    if value is None:
+        return model()
+    return value if isinstance(value, model) else model.model_validate(value)
+
+
+def with_atof_endpoint(config: FabricConfig, url: str) -> FabricConfig:
+    """Deep copy with a loopback ndjson ATOF endpoint appended (preserves sinks)."""
+    cfg = config.model_copy(deep=True)
+    obs = _coerce(
+        cfg.relay.observability if cfg.relay else None, RelayObservabilityConfig
+    )
+    atof = _coerce(obs.atof, RelayAtofConfig)
+    atof.enabled = True
+    atof.endpoints = list(atof.endpoints or []) + [
+        RelayAtofEndpointConfig(url=url, transport="ndjson")
+    ]
+    obs.atof = atof
+    cfg.enable_relay(observability=obs)
+    return cfg
+
+
+class InvokeStream:
+    """Async-iterable of raw ATOF records for one invocation; ``result()`` = RunResult.
+
+    Early-exit contract: ``aclose()`` detaches the consumer but does NOT interrupt
+    the turn — ``invoke`` is a blocking native call on a worker thread and runs to
+    completion. Only ``runtime.stop()`` aborts a turn.
+    """
+
+    def __init__(
+        self, runtime: Any, listener: AtofStreamListener, *, input: Any, request: Any
+    ):
+        self._listener = listener
+        self._task = asyncio.ensure_future(runtime.invoke(input=input, request=request))
+
+    def __aiter__(self) -> "InvokeStream":
+        return self
+
+    async def __anext__(self) -> dict[str, Any]:
+        q = self._listener.records
+        while True:
+            if not q.empty():
+                return q.get_nowait()
+            if self._task.done():
+                try:  # let trailing records land after the turn finishes
+                    return await asyncio.wait_for(q.get(), _DRAIN)
+                except asyncio.TimeoutError:
+                    raise StopAsyncIteration
+            getter = asyncio.ensure_future(q.get())
+            await asyncio.wait(
+                {getter, self._task}, return_when=asyncio.FIRST_COMPLETED
+            )
+            if getter.done() and not getter.cancelled():
+                return getter.result()
+            getter.cancel()
+
+    async def result(self) -> Any:
+        return await self._task
+
+    async def aclose(self) -> None:
+        if not self._task.done():
+            self._task.cancel()
+            try:
+                await self._task
+            except BaseException:
+                pass
+
+
+class StreamingRuntime:
+    """A relay-enabled Runtime that also exposes ``invoke_stream()``."""
+
+    def __init__(self, runtime: Any, listener: AtofStreamListener):
+        self._runtime = runtime
+        self._listener = listener
+
+    def invoke_stream(self, *, input: Any = None, request: Any = None) -> InvokeStream:
+        return InvokeStream(self._runtime, self._listener, input=input, request=request)
+
+    async def aclose(self) -> None:
+        try:
+            await self._runtime.stop()
+        finally:
+            await self._listener.close()
+
+    async def __aenter__(self) -> "StreamingRuntime":
+        return self
+
+    async def __aexit__(self, *exc: Any) -> None:
+        await self.aclose()
+
+
+async def start_streaming_runtime(
+    fabric: Any, config: FabricConfig, *, base_dir: str | None = None
+) -> StreamingRuntime:
+    """Sugar over ``Fabric.start_runtime``: inject the loopback endpoint, start, wrap."""
+    if not _relay_enabled(config):
+        raise FabricCapabilityError(
+            "streaming requires relay telemetry enabled", capability="streaming"
+        )
+    listener = await AtofStreamListener().start()
+    cfg = with_atof_endpoint(config, listener.url)
+    runtime = await fabric.start_runtime(cfg, base_dir=base_dir)
+    return StreamingRuntime(runtime, listener)