From f27cbb1b9b2d0124488f1af6bb64b18e97d10231 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 08:32:24 -0700 Subject: [PATCH 01/10] Add async session boundary to the Python SDK (start/invoke/stream/cancel/stop) Finish the async SDK boundary beyond the stable oneshot `run`: a client-side multi-turn Session over the existing stateless adapter call. No new wire schemas -- turns reuse RunRequest / RunResult. - FabricClient.start / start_config open a Session (FabricSessionUnsupportedError when the resolved adapter is not session-capable). - Session.invoke runs one turn, replaying the accumulated transcript as request.context.history and merging per-turn overrides. - Session.stream yields normalized events then the final RunResult (buffered; the async-iterator shape is forward-compatible with token streaming). - Session.cancel cooperatively aborts the in-flight turn (idle or running) and marks the session cancelled; Session.stop finalizes (idempotent, async cm). - hermes-sdk adapter: resolve_history threads request.context.history into run_conversation (request context wins over static settings; oneshot unchanged). Session state is client-side (transcript replayed as history); a persistent, harness-stateful session is a later phase, bounded by Hermes support. Tests: python/tests/smoke_sdk_sessions.py (dependency-free; fakes the inline adapter to assert multi-turn threading, buffered stream, cooperative cancel, idempotent stop, gating). tests/smoke_hermes_session.py (opt-in real-Hermes multi-turn, gated by RUN_FABRIC_HERMES_INTEGRATION). examples/session_quickstart.py. Signed-off-by: Ajay Thorve --- README.md | 24 ++ .../hermes_sdk/adapter.py | 14 +- examples/session_quickstart.py | 50 +++++ python/src/nemo_fabric/__init__.py | 18 +- python/src/nemo_fabric/client.py | 212 +++++++++++++++++- python/tests/smoke_sdk_sessions.py | 175 +++++++++++++++ tests/smoke_hermes_session.py | 76 +++++++ 7 files changed, 565 insertions(+), 4 deletions(-) create mode 100644 examples/session_quickstart.py create mode 100644 python/tests/smoke_sdk_sessions.py create mode 100644 tests/smoke_hermes_session.py diff --git a/README.md b/README.md index 1b074a903..c34969e7a 100644 --- a/README.md +++ b/README.md @@ -174,6 +174,30 @@ plan = client.plan_config( ) ``` +For multi-turn sessions, open a `Session` and invoke it repeatedly. The session +replays the accumulated transcript as conversation history so the harness sees +prior turns: + +```python +from nemo_fabric import FabricClient + +async def chat(): + async with await FabricClient().start( + "examples/code-review-agent", profile="hermes_sdk" + ) as session: + await session.invoke("My name is Robin.") + reply = await session.invoke("What's my name?") # recalls "Robin" + print(session.id, session.status, len(session.messages)) + print(reply["output"]["response"]) + +asyncio.run(chat()) +``` + +Sessions require the native binding and a session-capable (inline Python) +adapter; `start_config(...)` is the typed-config equivalent. `stream(...)` yields +events then the final result (buffered today); `cancel()` cooperatively aborts an +in-flight turn. See `examples/session_quickstart.py`. + When installed from the repository root, `FabricClient()` uses the native Rust binding. If the selected Python adapter descriptor provides a `runner.module` and `runner.callable`, the SDK imports and invokes that adapter inline. The diff --git a/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py b/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py index 95ce45be8..f3241aaf6 100644 --- a/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py +++ b/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py @@ -67,6 +67,18 @@ def settings_payload(payload: dict[str, Any]) -> dict[str, Any]: return harness.get("settings") or payload.get("settings") or {} +def resolve_history(payload: dict[str, Any]) -> Any: + """Conversation history for this turn. + + A per-invocation request context wins over static harness settings, so the + SDK can drive multi-turn sessions by passing accumulated messages in + ``request.context.history`` without mutating the agent config. + """ + + context = request_payload(payload).get("context") or {} + return context.get("history") or settings_payload(payload).get("history") + + def models_payload(payload: dict[str, Any]) -> dict[str, Any]: return fabric_config(payload).get("models") or payload.get("models") or {} @@ -393,7 +405,7 @@ def run_hermes_sdk(payload: dict[str, Any]) -> dict[str, Any]: conversation_kwargs = filter_supported_call_kwargs( agent.run_conversation, system_message=settings.get("system_prompt"), - conversation_history=settings.get("history"), + conversation_history=resolve_history(payload), sync_honcho=False, dont_review=True, ) diff --git a/examples/session_quickstart.py b/examples/session_quickstart.py new file mode 100644 index 000000000..2dc04191a --- /dev/null +++ b/examples/session_quickstart.py @@ -0,0 +1,50 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Quickstart: a multi-turn Fabric session (start -> invoke -> stream -> stop). + +The session replays the accumulated transcript as conversation history on each +turn, so the harness remembers prior turns. + +Run it with an interpreter that has the ``nemo_fabric`` native binding and Hermes +installed, with an API key available: + + set -a; . ./.env; set +a # provides NVIDIA_API_KEY + /bin/python examples/session_quickstart.py + +For a zero-setup local check of the session mechanics (no native binding, no +Hermes, no API key), run the unit smoke instead: + + python3 python/tests/smoke_sdk_sessions.py +""" + +from __future__ import annotations + +import asyncio + +from nemo_fabric import FabricClient + + +async def main() -> None: + async with await FabricClient().start( + "examples/code-review-agent", profile="hermes_sdk" + ) as session: + print(f"session {session.id} [{session.status.value}]") + + result = await session.invoke("My name is Robin. Please remember it for later.") + print(f"\n> remember my name\n {(result.get('output') or {}).get('response')}") + + # stream() yields events as they arrive, then the final RunResult (last item). + print("\n> what is my name? (streamed)") + async for item in session.stream("What is my name? Reply with just the name."): + if "status" in item: # terminal RunResult + print(f" = {(item.get('output') or {}).get('response')}") + else: # incremental event + print(f" . {item.get('kind')}: {item.get('message')}") + + print(f"\ntranscript turns accumulated: {len(session.messages)}") + print(f"\nsession [{session.status.value}] after context exit") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/src/nemo_fabric/__init__.py b/python/src/nemo_fabric/__init__.py index bcf6cf3c5..db38618a1 100644 --- a/python/src/nemo_fabric/__init__.py +++ b/python/src/nemo_fabric/__init__.py @@ -3,6 +3,20 @@ """Python SDK surface for NeMo Fabric.""" -from nemo_fabric.client import FabricCliError, FabricClient, FabricNativeUnavailableError +from nemo_fabric.client import ( + FabricCliError, + FabricClient, + FabricNativeUnavailableError, + FabricSessionUnsupportedError, + Session, + SessionStatus, +) -__all__ = ["FabricCliError", "FabricClient", "FabricNativeUnavailableError"] +__all__ = [ + "FabricCliError", + "FabricClient", + "FabricNativeUnavailableError", + "FabricSessionUnsupportedError", + "Session", + "SessionStatus", +] diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index 147df9a1f..32ce5b6f3 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -21,9 +21,10 @@ import sys import time import uuid -from collections.abc import Mapping +from collections.abc import AsyncIterator, Mapping from contextlib import contextmanager from dataclasses import dataclass +from enum import Enum from pathlib import Path from typing import Any, Iterable, Sequence @@ -48,6 +49,10 @@ class FabricNativeUnavailableError(RuntimeError): """Raised when a typed-config SDK method needs the native extension.""" +class FabricSessionUnsupportedError(RuntimeError): + """Raised when start()/start_config() resolve a non-session-capable adapter.""" + + @dataclass(frozen=True) class FabricClient: """Python entrypoint for Fabric config, planning, diagnostics, and runs.""" @@ -241,6 +246,35 @@ async def run_config( ) ) + async def start( + self, + path: str | Path, + *, + profile: str | Sequence[str] | None = None, + overrides: dict[str, Any] | None = None, + ) -> "Session": + """Open a multi-turn session over a session-capable agent/profile.""" + + self._require_native_module("start") + plan = self.plan(path, profile=profile) + return _make_session(self, plan, overrides) + + async def start_config( + self, + config: Mapping[str, Any] | Any, + *, + profile_configs: Sequence[Mapping[str, Any] | Any] | None = None, + base_dir: str | Path | None = None, + overrides: dict[str, Any] | None = None, + ) -> "Session": + """Open a multi-turn session over an in-memory typed config.""" + + self._require_native_module("start_config") + plan = self.plan_config( + config, profile_configs=profile_configs, base_dir=base_dir + ) + return _make_session(self, plan, overrides) + def _command(self) -> tuple[str, ...]: if self.command is not None: return self.command @@ -306,6 +340,182 @@ def _require_native_module(self, method: str) -> Any: return native +class SessionStatus(str, Enum): + """Lifecycle state of a :class:`Session`.""" + + ACTIVE = "active" + STOPPED = "stopped" + CANCELLED = "cancelled" + + +class Session: + """A multi-turn session over a session-capable Fabric adapter. + + Created by :meth:`FabricClient.start` / :meth:`FabricClient.start_config`. + Each :meth:`invoke` runs one turn through the resolved plan, replaying the + accumulated transcript as conversation history so the harness sees prior + turns. The session is stateless on the Fabric side: the running transcript + lives in Python and is threaded back in via ``request.context.history``. A + persistent, harness-stateful session is a later phase. + """ + + def __init__( + self, + *, + client: "FabricClient", + plan: dict[str, Any], + entrypoint: tuple[str, str], + overrides: dict[str, Any] | None = None, + ) -> None: + self._client = client + self._plan = plan + self._entrypoint = entrypoint + self._overrides = overrides + self._messages: list[Any] = [] + self._status = SessionStatus.ACTIVE + self._current_task: asyncio.Task[Any] | None = None + self.id = _new_id("session") + + @property + def status(self) -> SessionStatus: + return self._status + + @property + def messages(self) -> list[Any]: + """Read-only copy of the accumulated transcript.""" + + return list(self._messages) + + @property + def info(self) -> dict[str, Any]: + return { + "session_id": self.id, + "agent_name": self._plan.get("agent_name"), + "profile": self._plan.get("profile"), + "harness_type": _harness_type(self._plan), + "adapter_kind": _adapter_kind(self._plan), + } + + async def invoke( + self, + input_text: str | None = None, + *, + request: dict[str, Any] | None = None, + overrides: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Run one turn, replaying the accumulated transcript as history.""" + + if self._status is not SessionStatus.ACTIVE: + raise RuntimeError(f"cannot invoke a {self._status.value} session") + request_payload = _run_request_payload( + input_text=input_text or "", + input_file=None, + request=request, + request_file=None, + ) + if self._messages: + request_payload["context"].setdefault("history", self._messages) + merged_overrides = _merge_overrides(self._overrides, overrides) + if merged_overrides is not None: + request_payload.setdefault("overrides", merged_overrides) + self._current_task = asyncio.current_task() + try: + result = await _run_inline_adapter( + self._plan, request_payload, self._entrypoint + ) + finally: + self._current_task = None + self._absorb(result) + return result + + async def stream( + self, + input_text: str | None = None, + *, + request: dict[str, Any] | None = None, + overrides: dict[str, Any] | None = None, + ) -> AsyncIterator[dict[str, Any]]: + """Run one turn and yield its events, then the final ``RunResult``. + + Buffered: the turn runs to completion via :meth:`invoke`, then the + normalized ``events`` are yielded in order, followed by the terminal + ``RunResult`` (the last item). The async-iterator shape is + forward-compatible with live token streaming if a harness exposes one. + """ + + result = await self.invoke(input_text, request=request, overrides=overrides) + for event in result.get("events") or []: + yield event + yield result + + async def cancel(self) -> None: + """Cancel the in-flight turn and close the session. Idempotent. + + Cooperative: cancels the awaiting :meth:`invoke` / :meth:`stream` + coroutine and marks the session ``CANCELLED``. The inline adapter runs + in a worker thread that Python cannot hard-kill, so an already-dispatched + harness call may run to completion and its result is discarded; the + process-backed path can terminate the subprocess. + """ + + if self._status is not SessionStatus.ACTIVE: + return + self._status = SessionStatus.CANCELLED + task = self._current_task + if task is not None and not task.done() and task is not asyncio.current_task(): + task.cancel() + + async def stop(self) -> None: + """Finalize the session. Idempotent.""" + + if self._status is SessionStatus.ACTIVE: + self._status = SessionStatus.STOPPED + + def _absorb(self, result: Any) -> None: + """Advance the transcript and session id from a turn's ``RunResult``.""" + + output = result.get("output") if isinstance(result, dict) else None + if not isinstance(output, dict): + return + messages = output.get("messages") + if isinstance(messages, list) and messages: + self._messages = messages + session_id = output.get("session_id") + if session_id: + self.id = str(session_id) + + async def __aenter__(self) -> "Session": + return self + + async def __aexit__(self, exc_type: object, exc: object, traceback: object) -> None: + await self.stop() + + +def _make_session( + client: "FabricClient", + plan: dict[str, Any], + overrides: dict[str, Any] | None, +) -> "Session": + entrypoint = _inline_adapter_entrypoint(plan) + if entrypoint is None: + raise FabricSessionUnsupportedError( + "sessions require an inline Python adapter (adapter_kind='python'); " + f"resolved adapter_kind={_adapter_kind(plan)!r}" + ) + return Session(client=client, plan=plan, entrypoint=entrypoint, overrides=overrides) + + +def _merge_overrides( + base: dict[str, Any] | None, extra: dict[str, Any] | None +) -> dict[str, Any] | None: + merged: dict[str, Any] = {} + if isinstance(base, dict): + merged.update(base) + if isinstance(extra, dict): + merged.update(extra) + return merged or None + + def _profile_args(profile: str | Sequence[str] | None) -> list[str]: if profile is None: return [] diff --git a/python/tests/smoke_sdk_sessions.py b/python/tests/smoke_sdk_sessions.py new file mode 100644 index 000000000..487e5928c --- /dev/null +++ b/python/tests/smoke_sdk_sessions.py @@ -0,0 +1,175 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Smoke: the SDK Session boundary -- start / invoke / stream / cancel / stop. + +Dependency-free -- no native extension and no Hermes. A fake inline adapter +stands in for ``_run_inline_adapter`` and echoes the conversation history it +received, so we can assert that turn N sees the transcript produced by turn +N-1 (the core of stateless multi-turn), plus the full lifecycle: buffered +stream, cooperative cancel (idle and in-flight), idempotent stop, and +session-support gating. +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from nemo_fabric import ( + FabricClient, + FabricSessionUnsupportedError, + Session, + SessionStatus, +) +from nemo_fabric import client as client_mod +from nemo_fabric.client import _make_session + +seen_history: list[list] = [] + + +async def _fake_inline(plan, request, entrypoint): + """Stand-in for _run_inline_adapter: echo history, return a full transcript.""" + + history = (request.get("context") or {}).get("history") or [] + seen_history.append(list(history)) + turn = len(history) // 2 + 1 + transcript = list(history) + [ + {"role": "user", "content": request.get("input")}, + {"role": "assistant", "content": f"reply-{turn}"}, + ] + return { + "status": "succeeded", + "events": [{"event_id": f"evt-{turn}", "kind": "log", "message": f"turn {turn}"}], + "output": { + "messages": transcript, + "response": f"reply-{turn}", + "session_id": "sess-fake", + }, + } + + +def _session() -> Session: + return Session( + client=FabricClient(), + plan={"agent_name": "demo", "profile": "hermes_sdk"}, + entrypoint=("fake.module", "run"), + ) + + +async def multi_turn_threads_history() -> None: + seen_history.clear() + client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] + session = _session() + assert session.status is SessionStatus.ACTIVE + assert session.messages == [] + + await session.invoke("My name is Robin.") + assert seen_history[0] == [], seen_history[0] + after_turn1 = session.messages + assert len(after_turn1) == 2, after_turn1 + + await session.invoke("What's my name?") + assert seen_history[1] == after_turn1, (seen_history[1], after_turn1) + assert len(session.messages) == 4, session.messages + assert session.id == "sess-fake", session.id + + +async def stream_yields_events_then_result() -> None: + client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] + session = _session() + items = [item async for item in session.stream("hello")] + assert items[-1]["status"] == "succeeded", items[-1] # RunResult is the last item + events = items[:-1] + assert events and all(e.get("kind") == "log" for e in events), events + assert len(session.messages) == 2, session.messages # the streamed turn advanced it + + +async def stop_is_idempotent_and_blocks_invoke() -> None: + client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] + session = _session() + await session.stop() + assert session.status is SessionStatus.STOPPED + await session.stop() # idempotent + try: + await session.invoke("too late") + except RuntimeError: + pass + else: + raise AssertionError("invoke after stop should raise") + + async with _session() as ctx: + await ctx.invoke("hi") + assert ctx.status is SessionStatus.ACTIVE + assert ctx.status is SessionStatus.STOPPED # context manager auto-stops + + +async def cancel_when_idle_marks_cancelled() -> None: + client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] + session = _session() + await session.cancel() + assert session.status is SessionStatus.CANCELLED + await session.cancel() # idempotent + try: + await session.invoke("after cancel") + except RuntimeError: + pass + else: + raise AssertionError("invoke after cancel should raise") + + +async def cancel_aborts_in_flight_turn() -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def _blocking_inline(plan, request, entrypoint): + started.set() + await release.wait() # never released; cancellation is the only way out + return {"status": "succeeded", "output": {}} + + client_mod._run_inline_adapter = _blocking_inline # type: ignore[assignment] + session = _session() + turn = asyncio.create_task(session.invoke("long running")) + await started.wait() + await session.cancel() + assert session.status is SessionStatus.CANCELLED + try: + await turn + except asyncio.CancelledError: + pass + else: + raise AssertionError("in-flight invoke should be cancelled") + + +async def gating_rejects_non_session_adapter() -> None: + try: + _make_session( + FabricClient(), + {"adapter_descriptor": {"descriptor": {"adapter_kind": "process"}}}, + None, + ) + except FabricSessionUnsupportedError: + pass + else: + raise AssertionError("process adapter should not be session-capable") + + +async def main() -> None: + original = client_mod._run_inline_adapter + try: + await multi_turn_threads_history() + await stream_yields_events_then_result() + await stop_is_idempotent_and_blocks_invoke() + await cancel_when_idle_marks_cancelled() + await cancel_aborts_in_flight_turn() + await gating_rejects_non_session_adapter() + finally: + client_mod._run_inline_adapter = original # type: ignore[assignment] + print("smoke_sdk_sessions ok") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/smoke_hermes_session.py b/tests/smoke_hermes_session.py new file mode 100644 index 000000000..5f64bf44e --- /dev/null +++ b/tests/smoke_hermes_session.py @@ -0,0 +1,76 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Opt-in integration smoke for the SDK multi-turn Session path (real Hermes). + +Drives ``FabricClient.start -> invoke -> invoke -> stop`` against the Hermes SDK +adapter and asserts the session carries conversation memory across turns +(stateless multi-turn via history replay). + +Unlike ``smoke_hermes_sdk.py`` (which shells out to the CLI), the session path is +SDK-only and runs the inline adapter in-process, so this must be executed by an +interpreter that has BOTH the nemo_fabric native extension and Hermes importable: + + RUN_FABRIC_HERMES_INTEGRATION=1 NVIDIA_API_KEY=... \\ + /bin/python tests/smoke_hermes_session.py +""" + +from __future__ import annotations + +import asyncio +import importlib.util +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "python" / "src")) + + +def main() -> None: + if os.environ.get("RUN_FABRIC_HERMES_INTEGRATION") != "1": + print("skipping: set RUN_FABRIC_HERMES_INTEGRATION=1 to run") + return + if not os.environ.get("NVIDIA_API_KEY"): + raise SystemExit("NVIDIA_API_KEY is required") + if importlib.util.find_spec("nemo_fabric._native") is None: + print( + "skipping: the SDK session path needs the nemo_fabric native extension " + "(pip install -e . into this interpreter)" + ) + return + if importlib.util.find_spec("run_agent") is None: + print( + "skipping: Hermes (run_agent) is not importable; run with the Hermes " + "venv python (set HERMES_PYTHON or invoke it directly)" + ) + return + asyncio.run(_run()) + + +async def _run() -> None: + from nemo_fabric import FabricClient, SessionStatus + + agent = str(ROOT / "examples" / "code-review-agent") + async with await FabricClient().start(agent, profile="hermes_sdk") as session: + assert session.status is SessionStatus.ACTIVE, session.status + + r1 = await session.invoke("My name is Robin. Please remember it for later.") + assert r1["status"] == "succeeded", r1 + after_turn1 = session.messages + assert len(after_turn1) >= 2, after_turn1 + + r2 = await session.invoke("What is my name? Reply with just the name.") + assert r2["status"] == "succeeded", r2 + # Transcript must grow (history accumulated across turns). + assert len(session.messages) > len(after_turn1), session.messages + # And the model must recall the name supplied in turn 1. + response = (r2["output"].get("response") or "").lower() + assert "robin" in response, response + + assert session.status is SessionStatus.STOPPED, session.status + print("smoke_hermes_session ok") + + +if __name__ == "__main__": + main() From 9e7a2016847d5770973f6e53c59d24240a594c67 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 09:23:58 -0700 Subject: [PATCH 02/10] ci(python): run the session smoke Add the dependency-free session smoke (smoke_sdk_sessions) to the explicit CI smoke list so the Session boundary is enforced on every PR. The real-Hermes session smoke stays gated (RUN_FABRIC_HERMES_INTEGRATION) and out of CI. Signed-off-by: Ajay Thorve --- .github/workflows/ci_python.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci_python.yml b/.github/workflows/ci_python.yml index f292509ba..ffe51ac73 100644 --- a/.github/workflows/ci_python.yml +++ b/.github/workflows/ci_python.yml @@ -69,6 +69,7 @@ jobs: python/tests/smoke_typed_config.py python/tests/smoke_consumer_neutral.py python/tests/smoke_readme_examples.py + python/tests/smoke_sdk_sessions.py tests/smoke_cli.py tests/smoke_hermes_cli.py tests/smoke_hermes_config_mapping.py From 37755fb81eb42b7cc41f6ba89534da19f8df8955 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 13:16:39 -0700 Subject: [PATCH 03/10] Surface per-turn invocation handles on Session Track {request_id, runtime_id, invocation_id} per turn and expose them via Session.invocations, so a session is correlatable to its runtimes, telemetry, and artifacts even when it spans multiple runtimes (one per turn) -- the case when the harness has no resumable runtime (e.g. Hermes). Aligns the session surface toward the runtime-mode contract without faking a persistent runtime. Signed-off-by: Ajay Thorve --- python/src/nemo_fabric/client.py | 28 ++++++++++++++++++++++++++-- python/tests/smoke_sdk_sessions.py | 12 ++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index 32ce5b6f3..42882e138 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -372,6 +372,7 @@ def __init__( self._entrypoint = entrypoint self._overrides = overrides self._messages: list[Any] = [] + self._invocations: list[dict[str, Any]] = [] self._status = SessionStatus.ACTIVE self._current_task: asyncio.Task[Any] | None = None self.id = _new_id("session") @@ -386,6 +387,18 @@ def messages(self) -> list[Any]: return list(self._messages) + @property + def invocations(self) -> list[dict[str, Any]]: + """Per-turn ``{request_id, runtime_id, invocation_id}`` for correlating the + session to its runtimes, telemetry, and artifacts. + + A Fabric session may span multiple runtimes -- one per turn -- where the + harness exposes no resumable runtime (e.g. Hermes), so identity is tracked + per invocation rather than via a single stable ``runtime_id``. + """ + + return list(self._invocations) + @property def info(self) -> dict[str, Any]: return { @@ -472,9 +485,20 @@ async def stop(self) -> None: self._status = SessionStatus.STOPPED def _absorb(self, result: Any) -> None: - """Advance the transcript and session id from a turn's ``RunResult``.""" + """Record the turn's handles and advance the transcript from its ``RunResult``.""" - output = result.get("output") if isinstance(result, dict) else None + if not isinstance(result, dict): + return + # Per-turn identity for correlation; runtime_id may differ each turn when + # the harness has no resumable runtime. + self._invocations.append( + { + "request_id": result.get("request_id"), + "runtime_id": result.get("runtime_id"), + "invocation_id": result.get("invocation_id"), + } + ) + output = result.get("output") if not isinstance(output, dict): return messages = output.get("messages") diff --git a/python/tests/smoke_sdk_sessions.py b/python/tests/smoke_sdk_sessions.py index 487e5928c..fb5c0ecd9 100644 --- a/python/tests/smoke_sdk_sessions.py +++ b/python/tests/smoke_sdk_sessions.py @@ -43,6 +43,9 @@ async def _fake_inline(plan, request, entrypoint): ] return { "status": "succeeded", + "request_id": request.get("request_id"), + "runtime_id": f"runtime-{turn}", + "invocation_id": f"invocation-{turn}", "events": [{"event_id": f"evt-{turn}", "kind": "log", "message": f"turn {turn}"}], "output": { "messages": transcript, @@ -77,6 +80,15 @@ async def multi_turn_threads_history() -> None: assert len(session.messages) == 4, session.messages assert session.id == "sess-fake", session.id + # Per-turn handles are tracked for correlation, and a session can span + # multiple runtimes (one per turn) where the harness has no resumable runtime. + assert len(session.invocations) == 2, session.invocations + assert session.invocations[0]["runtime_id"] == "runtime-1", session.invocations + assert session.invocations[1]["runtime_id"] == "runtime-2", session.invocations + assert ( + session.invocations[0]["runtime_id"] != session.invocations[1]["runtime_id"] + ), session.invocations + async def stream_yields_events_then_result() -> None: client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] From bca7285bea681940f1e5dd14f7eb423359e77b70 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 14:14:33 -0700 Subject: [PATCH 04/10] Document session SDK methods (params, returns, errors) Add Args/Returns/Raises (and Yields for stream) to the session surface so the generated API reference documents the full contract: start, start_config, Session.invoke, Session.stream, and the previously-undocumented Session.info. Records the raised errors per method (FabricNativeUnavailableError, FabricSessionUnsupportedError, RuntimeError on a non-active session). The full-SDK docstring pass + Fern regeneration is FABRIC-2 / PR #7. Signed-off-by: Ajay Thorve --- python/src/nemo_fabric/client.py | 72 ++++++++++++++++++++++++++++++-- 1 file changed, 69 insertions(+), 3 deletions(-) diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index 42882e138..3eb02ebbc 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -253,7 +253,24 @@ async def start( profile: str | Sequence[str] | None = None, overrides: dict[str, Any] | None = None, ) -> "Session": - """Open a multi-turn session over a session-capable agent/profile.""" + """Open a multi-turn session over a session-capable agent/profile. + + Args: + path: Agent package directory or config file to resolve. + profile: Profile name, or several applied in order, layered onto the + base config. + overrides: Config overrides applied to every turn in the session; a + turn's own ``overrides`` merge over these. + + Returns: + An active :class:`Session` bound to the resolved plan. + + Raises: + FabricNativeUnavailableError: The native extension is unavailable + (sessions are not supported over the CLI fallback). + FabricSessionUnsupportedError: The resolved adapter is not + session-capable (no inline Python entrypoint). + """ self._require_native_module("start") plan = self.plan(path, profile=profile) @@ -267,7 +284,25 @@ async def start_config( base_dir: str | Path | None = None, overrides: dict[str, Any] | None = None, ) -> "Session": - """Open a multi-turn session over an in-memory typed config.""" + """Open a multi-turn session over an in-memory typed config. + + Args: + config: Typed Fabric config as a mapping or a Pydantic-like object + (``model_dump()``/``dict()``); no agent directory required. + profile_configs: Profile configs layered onto the base config, in order. + base_dir: Resolution root for relative paths and package-local + adapters. ``None`` resolves against the process working directory. + overrides: Config overrides applied to every turn; a turn's own + ``overrides`` merge over these. + + Returns: + An active :class:`Session` bound to the resolved plan. + + Raises: + FabricNativeUnavailableError: The native extension is unavailable. + FabricSessionUnsupportedError: The resolved adapter is not + session-capable. + """ self._require_native_module("start_config") plan = self.plan_config( @@ -401,6 +436,9 @@ def invocations(self) -> list[dict[str, Any]]: @property def info(self) -> dict[str, Any]: + """Summary handle: ``session_id``, ``agent_name``, ``profile``, + ``harness_type``, and ``adapter_kind``.""" + return { "session_id": self.id, "agent_name": self._plan.get("agent_name"), @@ -416,7 +454,22 @@ async def invoke( request: dict[str, Any] | None = None, overrides: dict[str, Any] | None = None, ) -> dict[str, Any]: - """Run one turn, replaying the accumulated transcript as history.""" + """Run one turn, replaying the accumulated transcript as history. + + Args: + input_text: Text input for the turn. Ignored when ``request`` is given. + request: A full ``RunRequest`` mapping for the turn, as an alternative + to ``input_text``. + overrides: Per-turn config overrides, merged over the session-level + overrides passed to :meth:`FabricClient.start`. + + Returns: + The turn's normalized ``RunResult`` mapping. The transcript + (:attr:`messages`) and :attr:`invocations` advance as a side effect. + + Raises: + RuntimeError: The session is not active (already stopped or cancelled). + """ if self._status is not SessionStatus.ACTIVE: raise RuntimeError(f"cannot invoke a {self._status.value} session") @@ -454,6 +507,19 @@ async def stream( normalized ``events`` are yielded in order, followed by the terminal ``RunResult`` (the last item). The async-iterator shape is forward-compatible with live token streaming if a harness exposes one. + + Args: + input_text: Text input for the turn. Ignored when ``request`` is given. + request: A full ``RunRequest`` mapping for the turn. + overrides: Per-turn config overrides, merged over the session-level + overrides. + + Yields: + Each ``fabric-event`` mapping for the turn, in order, then the final + ``RunResult`` mapping as the terminal item. + + Raises: + RuntimeError: The session is not active (already stopped or cancelled). """ result = await self.invoke(input_text, request=request, overrides=overrides) From 3a4ae87368bb388fbed82d3195c6720a7b4f8159 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 14:18:38 -0700 Subject: [PATCH 05/10] Add session unit tests pytest unit tests for the Session boundary (start/invoke/stream/cancel/stop), matching the suite added in #13: dependency-free via a monkeypatched inline adapter. Covers history replay, per-turn override merge, invocation-handle correlation, output.session_id adoption, buffered stream, idempotent stop + context manager, cooperative cancel (idle and in-flight), the info summary, defensive-copy semantics of messages/invocations, an empty-output turn, adapter gating, and the native-required error. Auto-discovered by the CI `uv run pytest`. Signed-off-by: Ajay Thorve --- tests/test_session.py | 236 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 236 insertions(+) create mode 100644 tests/test_session.py diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 000000000..97ae45734 --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,236 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the SDK Session boundary: start / invoke / stream / cancel / stop. + +Dependency-free: the inline adapter is monkeypatched, so these exercise the +Session orchestration (history replay, per-turn overrides, handle correlation, +lifecycle, gating, errors) without the native extension or a real harness. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from nemo_fabric import ( + FabricClient, + FabricNativeUnavailableError, + FabricSessionUnsupportedError, + Session, + SessionStatus, +) +from nemo_fabric import client as client_mod + + +def _plan(adapter_kind: str = "python") -> dict: + return { + "agent_name": "demo", + "profile": "hermes_sdk", + "adapter_descriptor": { + "descriptor": {"adapter_kind": adapter_kind, "adapter_id": "test.fabric.shim"} + }, + } + + +def _session(overrides: dict | None = None) -> Session: + return Session( + client=FabricClient(), + plan=_plan(), + entrypoint=("fake.module", "run"), + overrides=overrides, + ) + + +@pytest.fixture(name="seen_history") +def echo_adapter_fixture(monkeypatch: pytest.MonkeyPatch) -> list[list]: + """Patch _run_inline_adapter to echo history and emit per-turn handles. + + Returns the list of histories the adapter saw, one entry per turn. + """ + + seen: list[list] = [] + + async def _fake(plan, request, entrypoint): + history = (request.get("context") or {}).get("history") or [] + seen.append(list(history)) + turn = len(history) // 2 + 1 + transcript = list(history) + [ + {"role": "user", "content": request.get("input")}, + {"role": "assistant", "content": f"reply-{turn}"}, + ] + return { + "status": "succeeded", + "request_id": request.get("request_id"), + "runtime_id": f"runtime-{turn}", + "invocation_id": f"invocation-{turn}", + "events": [{"event_id": f"evt-{turn}", "kind": "log", "message": f"turn {turn}"}], + "output": { + "messages": transcript, + "response": f"reply-{turn}", + "session_id": "sess-1", + }, + } + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _fake) + return seen + + +async def test_new_session_is_active_and_empty(seen_history: list[list]) -> None: + session = _session() + assert session.status is SessionStatus.ACTIVE + assert session.messages == [] + assert session.invocations == [] + + +async def test_invoke_threads_accumulated_history(seen_history: list[list]) -> None: + session = _session() + + await session.invoke("My name is Robin.") + assert seen_history[0] == [] # turn 1 sees no prior history + after_turn1 = session.messages + assert len(after_turn1) == 2 + + await session.invoke("What's my name?") + assert seen_history[1] == after_turn1 # turn 2 sees turn 1's transcript + assert len(session.messages) == 4 + + +async def test_invoke_records_per_turn_handles(seen_history: list[list]) -> None: + session = _session() + await session.invoke("a") + await session.invoke("b") + + assert [inv["runtime_id"] for inv in session.invocations] == ["runtime-1", "runtime-2"] + assert session.invocations[0]["invocation_id"] == "invocation-1" + # A session may span multiple runtimes when the harness has no resumable one. + assert session.invocations[0]["runtime_id"] != session.invocations[1]["runtime_id"] + + +async def test_invoke_adopts_session_id_from_output(seen_history: list[list]) -> None: + session = _session() + await session.invoke("hi") + assert session.id == "sess-1" + + +async def test_invoke_merges_session_and_turn_overrides( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict = {} + + async def _fake(plan, request, entrypoint): + captured["overrides"] = request.get("overrides") + return {"status": "succeeded", "output": {}} + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _fake) + session = _session(overrides={"model": "a"}) + await session.invoke("x", overrides={"temperature": 0.0}) + + assert captured["overrides"] == {"model": "a", "temperature": 0.0} + + +async def test_stream_yields_events_then_result(seen_history: list[list]) -> None: + session = _session() + items = [item async for item in session.stream("hi")] + + assert items[-1]["status"] == "succeeded" # RunResult is the terminal item + events = items[:-1] + assert events and all(event.get("kind") == "log" for event in events) + assert len(session.messages) == 2 # the streamed turn advanced the transcript + + +async def test_stop_is_idempotent_and_blocks_invoke(seen_history: list[list]) -> None: + session = _session() + await session.stop() + assert session.status is SessionStatus.STOPPED + await session.stop() # idempotent + with pytest.raises(RuntimeError): + await session.invoke("too late") + + +async def test_context_manager_auto_stops(seen_history: list[list]) -> None: + async with _session() as session: + await session.invoke("hi") + assert session.status is SessionStatus.ACTIVE + assert session.status is SessionStatus.STOPPED + + +async def test_cancel_when_idle_marks_cancelled(seen_history: list[list]) -> None: + session = _session() + await session.cancel() + assert session.status is SessionStatus.CANCELLED + await session.cancel() # idempotent + with pytest.raises(RuntimeError): + await session.invoke("after cancel") + + +async def test_cancel_aborts_in_flight_turn(monkeypatch: pytest.MonkeyPatch) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def _blocking(plan, request, entrypoint): + started.set() + await release.wait() # only cancellation unblocks this + return {"status": "succeeded", "output": {}} + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _blocking) + session = _session() + turn = asyncio.create_task(session.invoke("long running")) + await started.wait() + + await session.cancel() + assert session.status is SessionStatus.CANCELLED + with pytest.raises(asyncio.CancelledError): + await turn + + +async def test_info_summarizes_the_session(seen_history: list[list]) -> None: + session = _session() + info = session.info + assert info["session_id"] == session.id + assert info["agent_name"] == "demo" + assert info["profile"] == "hermes_sdk" + assert info["adapter_kind"] == "python" + assert info["harness_type"] == "test.fabric.shim" + + +async def test_messages_and_invocations_return_copies(seen_history: list[list]) -> None: + session = _session() + await session.invoke("hi") + + snapshot = session.messages + snapshot.append({"role": "user", "content": "tampered"}) + invocations = session.invocations + invocations.clear() + + # Mutating the returned lists must not affect internal session state. + assert len(session.messages) == 2 + assert len(session.invocations) == 1 + + +async def test_invoke_without_output_messages_keeps_transcript( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def _fake(plan, request, entrypoint): + return {"status": "succeeded", "runtime_id": "r1", "invocation_id": "i1", "output": {}} + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _fake) + session = _session() + await session.invoke("hi") + + assert session.messages == [] # no echoed messages -> transcript unchanged + assert len(session.invocations) == 1 # the turn is still recorded for correlation + + +async def test_make_session_rejects_non_session_adapter() -> None: + with pytest.raises(FabricSessionUnsupportedError): + client_mod._make_session(FabricClient(), _plan(adapter_kind="process"), None) + + +async def test_start_requires_native_extension() -> None: + # A CLI-pinned client has no native module, so the typed session path must + # fail loudly rather than silently degrade. + client = FabricClient(command=("fabric",)) + with pytest.raises(FabricNativeUnavailableError): + await client.start("any/agent") From c59f0be92154e0b5557fdab69ffc7ad81dd23c84 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 15:05:59 -0700 Subject: [PATCH 06/10] Add hermes_session example profile; note sessions are SDK-only The example's hermes_sdk/cli/relay profiles override the base agent to runtime.mode: oneshot, so the session quickstart/smoke previously ran over a oneshot-declared config. Add a hermes_session profile (hermes-sdk adapter, runtime.mode: session) and point the quickstart, the gated real-Hermes smoke, and the README session snippet at it, so the example demonstrates session mode honestly. Document that sessions are SDK-only (no fabric CLI equivalent). Signed-off-by: Ajay Thorve --- README.md | 5 ++- .../profiles/hermes-session.yaml | 38 +++++++++++++++++++ examples/session_quickstart.py | 2 +- tests/smoke_hermes_session.py | 2 +- 4 files changed, 43 insertions(+), 4 deletions(-) create mode 100644 examples/code-review-agent/profiles/hermes-session.yaml diff --git a/README.md b/README.md index c34969e7a..5b446c263 100644 --- a/README.md +++ b/README.md @@ -183,7 +183,7 @@ from nemo_fabric import FabricClient async def chat(): async with await FabricClient().start( - "examples/code-review-agent", profile="hermes_sdk" + "examples/code-review-agent", profile="hermes_session" ) as session: await session.invoke("My name is Robin.") reply = await session.invoke("What's my name?") # recalls "Robin" @@ -196,7 +196,8 @@ asyncio.run(chat()) Sessions require the native binding and a session-capable (inline Python) adapter; `start_config(...)` is the typed-config equivalent. `stream(...)` yields events then the final result (buffered today); `cancel()` cooperatively aborts an -in-flight turn. See `examples/session_quickstart.py`. +in-flight turn. Sessions are SDK-only — there is no `fabric` CLI equivalent (the +CLI runs one invocation per process). See `examples/session_quickstart.py`. When installed from the repository root, `FabricClient()` uses the native Rust binding. If the selected Python adapter descriptor provides a `runner.module` diff --git a/examples/code-review-agent/profiles/hermes-session.yaml b/examples/code-review-agent/profiles/hermes-session.yaml new file mode 100644 index 000000000..7d549c661 --- /dev/null +++ b/examples/code-review-agent/profiles/hermes-session.yaml @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +schema_version: fabric.profile/v1alpha1 +name: hermes_session +description: Drive the Hermes Python adapter as a multi-turn session (runtime mode session). + +harness: + adapter_id: nvidia.fabric.hermes.sdk + resolution: preinstalled + settings: + python_env: HERMES_PYTHON + agent_profile: code_reviewer + workspace: ./repos/my-service + hermes_home: ./artifacts/hermes-home + base_url: https://integrate.api.nvidia.com/v1 + max_turns: 1 + max_tokens: 512 + temperature: 0.0 + reasoning_config: + effort: none + enabled_toolsets: [] + system_prompt: You are a concise smoke test assistant. + +runtime: + mode: session + transport: library + input_schema: chat + output_schema: message + artifacts: ./artifacts/hermes-session + +environment: + provider: local + workspace: ./repos/my-service + artifacts: ./artifacts/hermes-session + +telemetry: + enabled: false diff --git a/examples/session_quickstart.py b/examples/session_quickstart.py index 2dc04191a..27daf2e8d 100644 --- a/examples/session_quickstart.py +++ b/examples/session_quickstart.py @@ -27,7 +27,7 @@ async def main() -> None: async with await FabricClient().start( - "examples/code-review-agent", profile="hermes_sdk" + "examples/code-review-agent", profile="hermes_session" ) as session: print(f"session {session.id} [{session.status.value}]") diff --git a/tests/smoke_hermes_session.py b/tests/smoke_hermes_session.py index 5f64bf44e..0eaacfc9f 100644 --- a/tests/smoke_hermes_session.py +++ b/tests/smoke_hermes_session.py @@ -52,7 +52,7 @@ async def _run() -> None: from nemo_fabric import FabricClient, SessionStatus agent = str(ROOT / "examples" / "code-review-agent") - async with await FabricClient().start(agent, profile="hermes_sdk") as session: + async with await FabricClient().start(agent, profile="hermes_session") as session: assert session.status is SessionStatus.ACTIVE, session.status r1 = await session.invoke("My name is Robin. Please remember it for later.") From 3bf05c696066b8df7d2dad8a49a31e5ec2c18d21 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 15:18:29 -0700 Subject: [PATCH 07/10] Address CodeRabbit review on the session boundary - adapter resolve_history: check `history` key presence so an explicit empty history ([]) clears the conversation instead of falling back to static settings. - Session: the accumulated transcript is authoritative (thread it over any caller-supplied request.context.history); request-level overrides are merged (session < request < per-turn) rather than bypassing the merge; messages are deep-copied on store, expose, and thread so caller/adapter mutation cannot corrupt the session. - README: use session.status.value for consistency with the quickstart. - tests: cover authoritative history, request-override merge, and deep-copy isolation. (Skipped the @pytest.mark.asyncio nitpick: asyncio_mode=auto is the repo convention and the #13 suite omits the decorator.) Signed-off-by: Ajay Thorve --- README.md | 2 +- .../hermes_sdk/adapter.py | 6 ++- python/src/nemo_fabric/client.py | 20 ++++++---- tests/test_session.py | 37 ++++++++++++++++++- 4 files changed, 55 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 5b446c263..9b67eda19 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ async def chat(): ) as session: await session.invoke("My name is Robin.") reply = await session.invoke("What's my name?") # recalls "Robin" - print(session.id, session.status, len(session.messages)) + print(session.id, session.status.value, len(session.messages)) print(reply["output"]["response"]) asyncio.run(chat()) diff --git a/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py b/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py index f3241aaf6..6cff75f01 100644 --- a/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py +++ b/adapters/hermes-sdk/src/nemo_fabric_adapters/hermes_sdk/adapter.py @@ -76,7 +76,11 @@ def resolve_history(payload: dict[str, Any]) -> Any: """ context = request_payload(payload).get("context") or {} - return context.get("history") or settings_payload(payload).get("history") + # Check key presence so an explicit empty history ([]) clears the conversation + # rather than falling back to static harness settings. + if isinstance(context, dict) and "history" in context: + return context["history"] + return settings_payload(payload).get("history") def models_payload(payload: dict[str, Any]) -> dict[str, Any]: diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index 3eb02ebbc..2e5abfd1e 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -23,6 +23,7 @@ import uuid from collections.abc import AsyncIterator, Mapping from contextlib import contextmanager +from copy import deepcopy from dataclasses import dataclass from enum import Enum from pathlib import Path @@ -418,9 +419,9 @@ def status(self) -> SessionStatus: @property def messages(self) -> list[Any]: - """Read-only copy of the accumulated transcript.""" + """Read-only deep copy of the accumulated transcript.""" - return list(self._messages) + return deepcopy(self._messages) @property def invocations(self) -> list[dict[str, Any]]: @@ -479,11 +480,16 @@ async def invoke( request=request, request_file=None, ) - if self._messages: - request_payload["context"].setdefault("history", self._messages) - merged_overrides = _merge_overrides(self._overrides, overrides) + # The session transcript is authoritative: thread it (a deep copy, so the + # adapter cannot mutate our state) and override any history a caller passed + # via ``request``. + request_payload["context"]["history"] = deepcopy(self._messages) + # Merge overrides as session < request < per-turn; request-level overrides + # must not bypass the documented session/turn merge. + merged_overrides = _merge_overrides(self._overrides, request_payload.get("overrides")) + merged_overrides = _merge_overrides(merged_overrides, overrides) if merged_overrides is not None: - request_payload.setdefault("overrides", merged_overrides) + request_payload["overrides"] = merged_overrides self._current_task = asyncio.current_task() try: result = await _run_inline_adapter( @@ -569,7 +575,7 @@ def _absorb(self, result: Any) -> None: return messages = output.get("messages") if isinstance(messages, list) and messages: - self._messages = messages + self._messages = deepcopy(messages) session_id = output.get("session_id") if session_id: self.id = str(session_id) diff --git a/tests/test_session.py b/tests/test_session.py index 97ae45734..198b4f63c 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -201,11 +201,13 @@ async def test_messages_and_invocations_return_copies(seen_history: list[list]) snapshot = session.messages snapshot.append({"role": "user", "content": "tampered"}) + snapshot[0]["content"] = "mutated" # deep mutation of a returned message object invocations = session.invocations invocations.clear() - # Mutating the returned lists must not affect internal session state. + # Mutating the returned lists or their items must not affect session state. assert len(session.messages) == 2 + assert session.messages[0]["content"] != "mutated" assert len(session.invocations) == 1 @@ -223,6 +225,39 @@ async def _fake(plan, request, entrypoint): assert len(session.invocations) == 1 # the turn is still recorded for correlation +async def test_session_history_is_authoritative_over_request( + seen_history: list[list], +) -> None: + session = _session() + await session.invoke("turn one") + transcript = session.messages + + # A caller-supplied request carrying stale history must not override the + # session's accumulated transcript. + stale = {"input": "turn two", "context": {"history": [{"role": "user", "content": "stale"}]}} + await session.invoke(request=stale) + + assert seen_history[1] == transcript + + +async def test_request_level_overrides_are_merged(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + async def _fake(plan, request, entrypoint): + captured["overrides"] = request.get("overrides") + return {"status": "succeeded", "output": {}} + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _fake) + session = _session(overrides={"a": "session"}) + await session.invoke( + request={"input": "x", "overrides": {"b": "request"}}, + overrides={"c": "turn"}, + ) + + # session < request < per-turn, all merged (none bypassed). + assert captured["overrides"] == {"a": "session", "b": "request", "c": "turn"} + + async def test_make_session_rejects_non_session_adapter() -> None: with pytest.raises(FabricSessionUnsupportedError): client_mod._make_session(FabricClient(), _plan(adapter_kind="process"), None) From 83aef9172e460f833f0d2335e0aad0a74781b790 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 15:18:29 -0700 Subject: [PATCH 08/10] Remove unused agent_profile from the example config agent_profile was set in agent.yaml and every hermes profile but is read by neither the adapters nor the core, so it was inert. Drop it. Signed-off-by: Ajay Thorve --- examples/code-review-agent/agent.yaml | 1 - examples/code-review-agent/profiles/hermes-cli.yaml | 1 - examples/code-review-agent/profiles/hermes-relay.yaml | 1 - examples/code-review-agent/profiles/hermes-sdk.yaml | 1 - examples/code-review-agent/profiles/hermes-session.yaml | 1 - 5 files changed, 5 deletions(-) diff --git a/examples/code-review-agent/agent.yaml b/examples/code-review-agent/agent.yaml index 6ddda2867..bf27b2f72 100644 --- a/examples/code-review-agent/agent.yaml +++ b/examples/code-review-agent/agent.yaml @@ -11,7 +11,6 @@ harness: adapter_id: nvidia.fabric.hermes.sdk resolution: preinstalled settings: - agent_profile: code_reviewer workspace: ./repos/my-service models: diff --git a/examples/code-review-agent/profiles/hermes-cli.yaml b/examples/code-review-agent/profiles/hermes-cli.yaml index b598f1b1d..4609bfdfe 100644 --- a/examples/code-review-agent/profiles/hermes-cli.yaml +++ b/examples/code-review-agent/profiles/hermes-cli.yaml @@ -9,7 +9,6 @@ harness: adapter_id: nvidia.fabric.hermes.cli resolution: preinstalled settings: - agent_profile: code_reviewer workspace: ./repos/my-service hermes_home: ./artifacts/hermes-cli/home base_url: https://integrate.api.nvidia.com/v1 diff --git a/examples/code-review-agent/profiles/hermes-relay.yaml b/examples/code-review-agent/profiles/hermes-relay.yaml index c2413fa08..39b6bed27 100644 --- a/examples/code-review-agent/profiles/hermes-relay.yaml +++ b/examples/code-review-agent/profiles/hermes-relay.yaml @@ -10,7 +10,6 @@ harness: resolution: preinstalled settings: python_env: HERMES_PYTHON - agent_profile: code_reviewer workspace: ./repos/my-service hermes_home: ./artifacts/hermes-relay/home base_url: https://integrate.api.nvidia.com/v1 diff --git a/examples/code-review-agent/profiles/hermes-sdk.yaml b/examples/code-review-agent/profiles/hermes-sdk.yaml index 0d3a32d95..7abe481e7 100644 --- a/examples/code-review-agent/profiles/hermes-sdk.yaml +++ b/examples/code-review-agent/profiles/hermes-sdk.yaml @@ -10,7 +10,6 @@ harness: resolution: preinstalled settings: python_env: HERMES_PYTHON - agent_profile: code_reviewer workspace: ./repos/my-service hermes_home: ./artifacts/hermes-home base_url: https://integrate.api.nvidia.com/v1 diff --git a/examples/code-review-agent/profiles/hermes-session.yaml b/examples/code-review-agent/profiles/hermes-session.yaml index 7d549c661..8eb4c94e0 100644 --- a/examples/code-review-agent/profiles/hermes-session.yaml +++ b/examples/code-review-agent/profiles/hermes-session.yaml @@ -10,7 +10,6 @@ harness: resolution: preinstalled settings: python_env: HERMES_PYTHON - agent_profile: code_reviewer workspace: ./repos/my-service hermes_home: ./artifacts/hermes-home base_url: https://integrate.api.nvidia.com/v1 From c77a46f3be4e73382699df7a37349c7b57085ee0 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 15:21:52 -0700 Subject: [PATCH 09/10] Address CodeRabbit: single-flight session turns; README example import - Session.invoke claims an in-flight guard before any await, so concurrent invoke()/stream() calls on one session are rejected (turns are ordered) rather than racing the transcript and _absorb(). - README session example: add the missing `import asyncio`. - test: a second concurrent invoke is rejected while a turn is in flight. Signed-off-by: Ajay Thorve --- README.md | 2 ++ python/src/nemo_fabric/client.py | 42 ++++++++++++++++++-------------- tests/test_session.py | 22 +++++++++++++++++ 3 files changed, 48 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 9b67eda19..33aa03fcb 100644 --- a/README.md +++ b/README.md @@ -179,6 +179,8 @@ replays the accumulated transcript as conversation history so the harness sees prior turns: ```python +import asyncio + from nemo_fabric import FabricClient async def chat(): diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index 2e5abfd1e..b3913cd09 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -474,31 +474,37 @@ async def invoke( if self._status is not SessionStatus.ACTIVE: raise RuntimeError(f"cannot invoke a {self._status.value} session") - request_payload = _run_request_payload( - input_text=input_text or "", - input_file=None, - request=request, - request_file=None, - ) - # The session transcript is authoritative: thread it (a deep copy, so the - # adapter cannot mutate our state) and override any history a caller passed - # via ``request``. - request_payload["context"]["history"] = deepcopy(self._messages) - # Merge overrides as session < request < per-turn; request-level overrides - # must not bypass the documented session/turn merge. - merged_overrides = _merge_overrides(self._overrides, request_payload.get("overrides")) - merged_overrides = _merge_overrides(merged_overrides, overrides) - if merged_overrides is not None: - request_payload["overrides"] = merged_overrides + if self._current_task is not None: + raise RuntimeError( + "session is already running a turn; turns are ordered (one at a time)" + ) + # Claim the turn before any await so concurrent invoke()/stream() calls + # cannot both replay the transcript and race _absorb(). self._current_task = asyncio.current_task() try: + request_payload = _run_request_payload( + input_text=input_text or "", + input_file=None, + request=request, + request_file=None, + ) + # The session transcript is authoritative: thread it (a deep copy, so + # the adapter cannot mutate our state) and override any history a caller + # passed via ``request``. + request_payload["context"]["history"] = deepcopy(self._messages) + # Merge overrides as session < request < per-turn; request-level + # overrides must not bypass the documented session/turn merge. + merged_overrides = _merge_overrides(self._overrides, request_payload.get("overrides")) + merged_overrides = _merge_overrides(merged_overrides, overrides) + if merged_overrides is not None: + request_payload["overrides"] = merged_overrides result = await _run_inline_adapter( self._plan, request_payload, self._entrypoint ) + self._absorb(result) + return result finally: self._current_task = None - self._absorb(result) - return result async def stream( self, diff --git a/tests/test_session.py b/tests/test_session.py index 198b4f63c..24196976a 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -258,6 +258,28 @@ async def _fake(plan, request, entrypoint): assert captured["overrides"] == {"a": "session", "b": "request", "c": "turn"} +async def test_concurrent_invokes_are_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def _blocking(plan, request, entrypoint): + started.set() + await release.wait() + return {"status": "succeeded", "output": {}} + + monkeypatch.setattr(client_mod, "_run_inline_adapter", _blocking) + session = _session() + first = asyncio.create_task(session.invoke("turn one")) + await started.wait() # first turn is in flight + + # A session is ordered/single-flight: a second concurrent turn is rejected. + with pytest.raises(RuntimeError): + await session.invoke("turn two") + + release.set() + await first # the in-flight turn still completes cleanly + + async def test_make_session_rejects_non_session_adapter() -> None: with pytest.raises(FabricSessionUnsupportedError): client_mod._make_session(FabricClient(), _plan(adapter_kind="process"), None) From 0b026127fa1a1d5918fbb50dc674de1a0e673911 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 23 Jun 2026 16:10:04 -0700 Subject: [PATCH 10/10] Apply suggestion from @dagardner-nv Signed-off-by: Ajay Thorve --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 33aa03fcb..37a9daa5b 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,7 @@ plan = client.plan_config( ``` For multi-turn sessions, open a `Session` and invoke it repeatedly. The session -replays the accumulated transcript as conversation history so the harness sees +replays the accumulated transcript as conversation history so the harness has access to prior turns: ```python