fix(agents): prevent unbounded fabric runtime for stateless invocations

plugins/nemo-agents/README.md

@@ -781,6 +781,11 @@ http://127.0.0.1:8080/apis/agents/v2/workspaces/default/agents/react-agent/-/v1/
 
 You can call it directly with any OpenAI-compatible client using the same path.
 
+Requests without ``X-Nemo-Session-Id`` use a one-shot Fabric runtime that is
+stopped when the response or response stream completes. To retain runtime
+context across turns, send a stable session ID in that header; the registered
+runtime then follows the Platform session lifecycle.
+
 The agent is still running — continue to the [Evaluation](#evaluation) section
 below, or see [Cleanup](#cleanup-optional) to tear everything down.
 

plugins/nemo-agents/src/nemo_agents_plugin/fabric/runtime.py

@@ -15,12 +15,13 @@
 
 import asyncio
 from collections.abc import AsyncIterator, Mapping, Sequence
+from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
 # CI type-checks this plugin via ty extra-paths without installing nemo-agents deps.
-from nemo_fabric import (  # ty: ignore[unresolved-import]
+from nemo_fabric import (
     Fabric,
     FabricConfig,
     FabricError,
@@ -131,6 +132,10 @@ class FabricRuntimeExecutionError(RuntimeError):
     """Raised when Fabric cannot return a normalized runtime result."""
 
 
+class FabricRuntimeStartError(FabricRuntimeExecutionError):
+    """Raised when Fabric cannot start a runtime."""
+
+
 class FabricRuntimeTimeoutError(FabricRuntimeExecutionError):
     """Raised when a Fabric runtime invocation times out."""
 
@@ -184,35 +189,80 @@ async def run_fabric_agent_once(
 ) -> FabricRuntimeResult:
     """Start an ephemeral Fabric runtime, invoke it once, and stop it."""
     fabric_client = fabric or Fabric()
+    runtime = await _start_one_shot_runtime(request, fabric=fabric_client)
 
+    runtime_entered = False
     try:
-        result = await asyncio.wait_for(
-            _invoke_fabric_agent_once(request, fabric=fabric_client),
-            timeout=request.timeout_seconds,
-        )
-    except TimeoutError as error:
-        raise FabricRuntimeTimeoutError(
-            _timeout_error_message(request.timeout_seconds),
-        ) from error
+        async with runtime:
+            runtime_entered = True
+            try:
+                result = await asyncio.wait_for(
+                    runtime.invoke(request=_with_platform_invocation_context(request)),
+                    timeout=request.timeout_seconds,
+                )
+            except TimeoutError as error:
+                raise FabricRuntimeTimeoutError(
+                    _timeout_error_message(request.timeout_seconds),
+                ) from error
+            except FabricError as error:
+                raise FabricRuntimeExecutionError(
+                    f"Fabric runtime invocation failed: {error}",
+                ) from error
     except FabricError as error:
-        raise FabricRuntimeExecutionError(
-            f"Fabric runtime invocation failed: {error}",
-        ) from error
+        raise _runtime_context_error(error, entered=runtime_entered) from error
 
     return _normalize_fabric_run_result(result)
 
 
-async def _invoke_fabric_agent_once(
+@asynccontextmanager
+async def stream_fabric_agent_once(
+    request: FabricOneShotRequest,
+    *,
+    fabric: Any | None = None,
+) -> AsyncIterator[FabricRuntimeStream]:
+    """Start an ephemeral Fabric runtime and keep it alive for one stream."""
+    fabric_client = fabric or Fabric()
+
+    runtime = await _start_one_shot_runtime(request, fabric=fabric_client, streaming=True)
+
+    runtime_entered = False
+    try:
+        async with runtime:
+            runtime_entered = True
+            yield stream_fabric_runtime(
+                runtime,
+                FabricInvocationRequest(
+                    input=request.input,
+                    request_id=request.request_id,
+                    caller_context=request.caller_context,
+                    timeout_seconds=request.timeout_seconds,
+                ),
+            )
+    except FabricError as error:
+        raise _runtime_context_error(error, entered=runtime_entered) from error
+
+
+def _runtime_context_error(error: FabricError, *, entered: bool) -> FabricRuntimeExecutionError:
+    if entered:
+        return FabricRuntimeExecutionError(f"Fabric runtime cleanup failed: {error}")
+    return FabricRuntimeStartError(f"Fabric runtime startup failed: {error}")
+
+
+async def _start_one_shot_runtime(
     request: FabricOneShotRequest,
     *,
     fabric: Any,
-) -> RunResult:
-    async with await fabric.start_runtime(
-        request.fabric_config,
-        base_dir=request.base_dir,
-        overrides=request.overrides,
-    ) as runtime:
-        return await runtime.invoke(request=_with_platform_invocation_context(request))
+    streaming: bool = False,
+) -> Runtime:
+    try:
+        return await fabric.start_runtime(
+            request.fabric_config,
+            base_dir=request.base_dir,
+            overrides=request.overrides,
+            streaming=streaming,
+        )
+    except FabricError as error:
+        raise FabricRuntimeStartError(f"Fabric runtime startup failed: {error}") from error
 
 
 def _with_platform_invocation_context(request: FabricInvocationRequest | FabricOneShotRequest) -> RunRequest: