diff --git a/README.md b/README.md index 37a9daa5b..0f1090372 100644 --- a/README.md +++ b/README.md @@ -175,8 +175,8 @@ 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 has access to -prior turns: +keeps one Fabric runtime handle active across turns; harness/adapter state is +authoritative rather than reconstructed from a Python-side transcript: ```python import asyncio @@ -189,23 +189,23 @@ 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.value, len(session.messages)) + print(session.runtime_id, session.status.value, 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. Sessions are SDK-only — there is no `fabric` CLI equivalent (the -CLI runs one invocation per process). See `examples/session_quickstart.py`. +Sessions require the native binding; `start_config(...)` is the typed-config +equivalent. `stream(...)` yields events then the final result (buffered today); +`cancel()` cooperatively aborts an in-flight turn. Sessions are SDK-only — there +is no `fabric` CLI equivalent (the CLI runs one invocation per process). The +real-Hermes integration check is `tests/smoke_hermes_session.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 -`fabric` CLI continues to use the adapter script/process path, which is useful -for local debugging and environment-backed consumers. +binding. SDK `run(...)`, `start(...)`, and their typed-config equivalents all +drive the core Fabric runtime lifecycle (`start_runtime` / `invoke_runtime` / +`stop_runtime`) so one-shot and session paths use the same adapter execution +contract. For source-tree debugging, pass an explicit CLI command: diff --git a/adapters/hermes-sdk/README.md b/adapters/hermes-sdk/README.md index 72e34b3fb..fff9fbed8 100644 --- a/adapters/hermes-sdk/README.md +++ b/adapters/hermes-sdk/README.md @@ -9,12 +9,11 @@ This adapter runs Hermes through its Python SDK. It is the preferred Hermes path for Python consumers such as NeMo Platform, Gym-style agent servers, and direct Fabric SDK use. -The adapter exposes two entrypoints: - -- `runner.module` + `runner.callable` in `fabric-adapter.json` point to the - inline SDK entrypoint used by `FabricClient`. -- `runner.script` points to a thin executable wrapper used by `fabric run` and - process-style fallback paths. +The adapter descriptor records both callable and script metadata, but Fabric's +current SDK and CLI paths invoke the adapter through the core runtime lifecycle +and its `runner.script` wrapper. Keep the callable and script pointing at the +same `run(payload: dict) -> dict` implementation so future in-process adapter +execution remains equivalent. ## What It Maps @@ -33,10 +32,8 @@ Keep `fabric-adapter.json` aligned with the Python implementation: - `adapter_id` is the stable id selected by `harness.adapter_id`. - `adapter_kind` is `python` because Fabric can invoke it through Python. -- `runner.module` and `runner.callable` define the inline SDK entrypoint with - the shape `run(payload: dict) -> dict`. -- `runner.script` is the process fallback and must remain a thin wrapper around - the same callable. +- `runner.module`, `runner.callable`, and `runner.script` must remain thin + routes to the same `run(payload: dict) -> dict` implementation. - `requirements` powers `fabric doctor`; keep required env vars, binaries, or packages current. - `config.accepts` must match the Fabric sections this adapter maps into Hermes. 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 c894ca5cf..8176c370b 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 @@ -37,7 +37,7 @@ def main() -> None: def run(payload: dict[str, Any]) -> dict[str, Any]: - """Inline Fabric adapter entrypoint used by the Python SDK.""" + """Fabric adapter entrypoint used by script and native SDK runtime calls.""" return run_hermes_sdk(payload) @@ -51,20 +51,28 @@ def resolve_hermes_toolsets(settings: dict[str, Any], config: dict[str, Any]) -> platform = settings.get("toolset_platform", "cli") return sorted(_get_platform_tools(config, platform)) -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. - """ +def runtime_session_id(payload: dict[str, Any]) -> str | None: + runtime_id = hermes_common.runtime_context(payload).get("runtime_id") + if runtime_id: + return str(runtime_id) + return None - context = hermes_common.request_payload(payload).get("context") or {} - # 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 hermes_common.settings_payload(payload).get("history") + +def load_runtime_history(session_db: Any, session_id: str | None) -> list[dict[str, Any]] | None: + if not session_id: + return None + + resolved_id = session_id + resolve_session = getattr(session_db, "resolve_resume_session_id", None) + if resolve_session is not None: + resolved_id = resolve_session(session_id) or session_id + if session_db.get_session(resolved_id) is None: + return None + + messages = session_db.get_messages_as_conversation(resolved_id) + messages = [message for message in messages if message.get("role") != "session_meta"] + return messages or None def run_hermes_sdk(payload: dict[str, Any]) -> dict[str, Any]: @@ -79,6 +87,7 @@ def run_hermes_sdk(payload: dict[str, Any]) -> dict[str, Any]: os.environ["HERMES_HOME"] = str(hermes_home) os.environ.setdefault("HERMES_YOLO_MODE", "1") os.environ.setdefault("HERMES_ACCEPT_HOOKS", "1") + os.environ["HERMES_SESSION_SOURCE"] = "fabric" os.environ.setdefault("TERMINAL_ENV", settings.get("terminal_backend", "local")) os.environ.setdefault("TERMINAL_TIMEOUT", str(settings.get("terminal_timeout", 60))) relay_plugin_config = hermes_common.configure_hermes_relay(payload) @@ -105,11 +114,15 @@ def run_hermes_sdk(payload: dict[str, Any]) -> dict[str, Any]: with redirect_stdout(hermes_stdout): from hermes_cli.config import load_config from hermes_cli.plugins import discover_plugins, invoke_hook + from hermes_state import SessionDB from run_agent import AIAgent discover_plugins(force=True) loaded_hermes_config = load_config() enabled_toolsets = resolve_hermes_toolsets(settings, loaded_hermes_config) + session_id = runtime_session_id(payload) + session_db = SessionDB() + conversation_history = load_runtime_history(session_db, session_id) agent = None agent = AIAgent(**filter_supported_kwargs( AIAgent, @@ -128,12 +141,15 @@ def run_hermes_sdk(payload: dict[str, Any]) -> dict[str, Any]: temperature=settings.get("temperature", model_config.get("temperature", 0.0)), reasoning_config=settings.get("reasoning_config", {"effort": "none"}), insert_reasoning=bool(settings.get("insert_reasoning", False)), + platform="fabric", + session_id=session_id, + session_db=session_db, )) try: conversation_kwargs = filter_supported_call_kwargs( agent.run_conversation, system_message=settings.get("system_prompt"), - conversation_history=resolve_history(payload), + conversation_history=conversation_history, sync_honcho=False, dont_review=True, ) diff --git a/crates/fabric-python/src/lib.rs b/crates/fabric-python/src/lib.rs index daa55cbe3..cad9a362b 100644 --- a/crates/fabric-python/src/lib.rs +++ b/crates/fabric-python/src/lib.rs @@ -6,8 +6,8 @@ use std::path::PathBuf; use fabric_core::{ - FabricConfig, ProfileConfig, ResolveContext, RunRequest, doctor_plan, load_fabric_document, - resolve_effective_config_with_profiles, resolve_run_plan_from_config, + FabricConfig, ProfileConfig, ResolveContext, RunPlan, RunRequest, RuntimeHandle, doctor_plan, + load_fabric_document, resolve_effective_config_with_profiles, resolve_run_plan_from_config, resolve_run_plan_with_profiles, run_plan, }; use pyo3::exceptions::PyRuntimeError; @@ -158,6 +158,44 @@ fn run_config( to_json(&result) } +/// Start a runtime for a resolved run plan and return its RuntimeHandle JSON. +#[pyfunction] +fn start_runtime(py: Python<'_>, plan_json: String) -> PyResult { + let plan = parse_run_plan(plan_json)?; + let runtime = py + .detach(|| fabric_core::start_runtime(&plan)) + .map_err(to_py_error)?; + to_json(&runtime) +} + +/// Invoke a previously started runtime and return RunResult JSON. +#[pyfunction] +fn invoke_runtime( + py: Python<'_>, + plan_json: String, + runtime_json: String, + request_json: String, +) -> PyResult { + let plan = parse_run_plan(plan_json)?; + let runtime = parse_runtime_handle(runtime_json)?; + let request = parse_run_request(request_json)?; + let result = py + .detach(|| fabric_core::invoke_runtime(&plan, &runtime, request)) + .map_err(to_py_error)?; + to_json(&result) +} + +/// Stop a previously started runtime and return FabricEvent list JSON. +#[pyfunction] +fn stop_runtime(py: Python<'_>, plan_json: String, runtime_json: String) -> PyResult { + let plan = parse_run_plan(plan_json)?; + let runtime = parse_runtime_handle(runtime_json)?; + let events = py + .detach(|| fabric_core::stop_runtime(&plan, &runtime)) + .map_err(to_py_error)?; + to_json(&events) +} + #[pymodule] fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(version, m)?)?; @@ -169,6 +207,9 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(doctor_config, m)?)?; m.add_function(wrap_pyfunction!(run, m)?)?; m.add_function(wrap_pyfunction!(run_config, m)?)?; + m.add_function(wrap_pyfunction!(start_runtime, m)?)?; + m.add_function(wrap_pyfunction!(invoke_runtime, m)?)?; + m.add_function(wrap_pyfunction!(stop_runtime, m)?)?; Ok(()) } @@ -220,3 +261,11 @@ fn parse_profiles(contents: Option) -> PyResult> { fn parse_run_request(contents: String) -> PyResult { serde_json::from_str(&contents).map_err(|error| PyRuntimeError::new_err(error.to_string())) } + +fn parse_run_plan(contents: String) -> PyResult { + serde_json::from_str(&contents).map_err(|error| PyRuntimeError::new_err(error.to_string())) +} + +fn parse_runtime_handle(contents: String) -> PyResult { + serde_json::from_str(&contents).map_err(|error| PyRuntimeError::new_err(error.to_string())) +} diff --git a/examples/session_quickstart.py b/examples/session_quickstart.py deleted file mode 100644 index 27daf2e8d..000000000 --- a/examples/session_quickstart.py +++ /dev/null @@ -1,50 +0,0 @@ -# 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_session" - ) 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 db38618a1..8fcbd95e9 100644 --- a/python/src/nemo_fabric/__init__.py +++ b/python/src/nemo_fabric/__init__.py @@ -7,7 +7,6 @@ FabricCliError, FabricClient, FabricNativeUnavailableError, - FabricSessionUnsupportedError, Session, SessionStatus, ) @@ -16,7 +15,6 @@ "FabricCliError", "FabricClient", "FabricNativeUnavailableError", - "FabricSessionUnsupportedError", "Session", "SessionStatus", ] diff --git a/python/src/nemo_fabric/_native.pyi b/python/src/nemo_fabric/_native.pyi index dccf0607e..bd0252af8 100644 --- a/python/src/nemo_fabric/_native.pyi +++ b/python/src/nemo_fabric/_native.pyi @@ -33,3 +33,10 @@ def run_config( request_json: str | None = None, request_file: str | None = None, ) -> str: ... +def start_runtime(plan_json: str) -> str: ... +def invoke_runtime( + plan_json: str, + runtime_json: str, + request_json: str, +) -> str: ... +def stop_runtime(plan_json: str, runtime_json: str) -> str: ... diff --git a/python/src/nemo_fabric/client.py b/python/src/nemo_fabric/client.py index b3913cd09..37b2728f6 100644 --- a/python/src/nemo_fabric/client.py +++ b/python/src/nemo_fabric/client.py @@ -12,17 +12,13 @@ import asyncio import concurrent.futures -import inspect import importlib import json import os import shlex import subprocess -import sys -import time import uuid from collections.abc import AsyncIterator, Mapping -from contextlib import contextmanager from copy import deepcopy from dataclasses import dataclass from enum import Enum @@ -47,11 +43,7 @@ def __init__(self, command: Sequence[str], returncode: int, stdout: str, stderr: 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.""" + """Raised when an SDK method requires the native extension.""" @dataclass(frozen=True) @@ -176,21 +168,7 @@ async def run( request_file=request_file, ) plan = json.loads(native.plan(str(path), native_profile)) - inline_entrypoint = _inline_adapter_entrypoint(plan) - if inline_entrypoint is not None: - return await _run_inline_adapter(plan, request_payload, inline_entrypoint) - return await _call_blocking( - lambda: json.loads( - native.run( - str(path), - native_profile, - input_text, - None if input_file is None else str(input_file), - None if request is None else json.dumps(request), - None if request_file is None else str(request_file), - ) - ) - ) + return await _run_native_lifecycle(native, plan, request_payload) args = ["run", str(path)] args.extend(_profile_args(profile)) if request_file is not None: @@ -230,22 +208,7 @@ async def run_config( None if base_dir is None else str(base_dir), ) ) - inline_entrypoint = _inline_adapter_entrypoint(plan) - if inline_entrypoint is not None: - return await _run_inline_adapter(plan, request_payload, inline_entrypoint) - return await _call_blocking( - lambda: json.loads( - native.run_config( - _config_json(config), - _profiles_json(profile_configs), - None if base_dir is None else str(base_dir), - input_text, - None if input_file is None else str(input_file), - None if request is None else json.dumps(request), - None if request_file is None else str(request_file), - ) - ) - ) + return await _run_native_lifecycle(native, plan, request_payload) async def start( self, @@ -254,7 +217,7 @@ 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 an agent/profile runtime. Args: path: Agent package directory or config file to resolve. @@ -269,13 +232,14 @@ async def start( 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") + native = self._require_native_module("start") plan = self.plan(path, profile=profile) - return _make_session(self, plan, overrides) + runtime = await _call_blocking( + lambda: json.loads(native.start_runtime(json.dumps(plan))) + ) + return Session(client=self, plan=plan, runtime=runtime, overrides=overrides) async def start_config( self, @@ -301,15 +265,16 @@ async def start_config( Raises: FabricNativeUnavailableError: The native extension is unavailable. - FabricSessionUnsupportedError: The resolved adapter is not - session-capable. """ - self._require_native_module("start_config") + native = 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) + runtime = await _call_blocking( + lambda: json.loads(native.start_runtime(json.dumps(plan))) + ) + return Session(client=self, plan=plan, runtime=runtime, overrides=overrides) def _command(self) -> tuple[str, ...]: if self.command is not None: @@ -385,14 +350,12 @@ class SessionStatus(str, Enum): class Session: - """A multi-turn session over a session-capable Fabric adapter. + """A multi-turn session over a Fabric runtime. 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. + Each :meth:`invoke` runs one turn through the same core ``RuntimeHandle``. + Harness state is owned by the selected adapter/runtime, not replayed from a + Python-side transcript. """ def __init__( @@ -400,18 +363,18 @@ def __init__( *, client: "FabricClient", plan: dict[str, Any], - entrypoint: tuple[str, str], + runtime: dict[str, Any], overrides: dict[str, Any] | None = None, ) -> None: self._client = client self._plan = plan - self._entrypoint = entrypoint + self._runtime = runtime 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") + self._closing = False @property def status(self) -> SessionStatus: @@ -425,23 +388,28 @@ def messages(self) -> list[Any]: @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``. - """ + """Per-turn ``{request_id, runtime_id, invocation_id}`` correlation data.""" return list(self._invocations) + @property + def runtime(self) -> dict[str, Any]: + """Read-only deep copy of the active ``RuntimeHandle``.""" + + return deepcopy(self._runtime) + + @property + def runtime_id(self) -> str: + """Canonical Fabric runtime id for this session.""" + + return str(self._runtime["runtime_id"]) + @property def info(self) -> dict[str, Any]: - """Summary handle: ``session_id``, ``agent_name``, ``profile``, - ``harness_type``, and ``adapter_kind``.""" + """Summary handle for the active Fabric runtime and selected adapter.""" return { - "session_id": self.id, + "runtime_id": self._runtime.get("runtime_id"), "agent_name": self._plan.get("agent_name"), "profile": self._plan.get("profile"), "harness_type": _harness_type(self._plan), @@ -455,7 +423,7 @@ 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 on the session runtime. Args: input_text: Text input for the turn. Ignored when ``request`` is given. @@ -465,8 +433,8 @@ async def invoke( 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. + The turn's normalized ``RunResult`` mapping. ``messages`` is updated + only when the adapter returns a ``messages`` list in its output. Raises: RuntimeError: The session is not active (already stopped or cancelled). @@ -474,12 +442,14 @@ async def invoke( if self._status is not SessionStatus.ACTIVE: raise RuntimeError(f"cannot invoke a {self._status.value} session") + if self._closing: + raise RuntimeError("cannot invoke while session shutdown is in progress") 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(). + # Claim the turn before any await so callers cannot concurrently invoke + # the same runtime handle. self._current_task = asyncio.current_task() try: request_payload = _run_request_payload( @@ -488,18 +458,21 @@ async def invoke( 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 + native = self._client._require_native_module("invoke") + result = await _call_blocking( + lambda: json.loads( + native.invoke_runtime( + json.dumps(self._plan), + json.dumps(self._runtime), + json.dumps(request_payload), + ) + ) ) self._absorb(result) return result @@ -543,32 +516,59 @@ 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. + coroutine and marks the session ``CANCELLED``. Already-dispatched + blocking native calls may run to completion and their result is discarded. """ if self._status is not SessionStatus.ACTIVE: return - self._status = SessionStatus.CANCELLED + if self._closing: + raise RuntimeError("session shutdown is already in progress") + self._closing = True task = self._current_task if task is not None and not task.done() and task is not asyncio.current_task(): task.cancel() + try: + await self._stop_runtime() + except Exception: + self._closing = False + raise + else: + self._status = SessionStatus.CANCELLED + self._closing = False async def stop(self) -> None: """Finalize the session. Idempotent.""" if self._status is SessionStatus.ACTIVE: - self._status = SessionStatus.STOPPED + task = self._current_task + if task is not None and not task.done() and task is not asyncio.current_task(): + raise RuntimeError("cannot stop while a turn is in flight; use cancel()") + if self._closing: + raise RuntimeError("session shutdown is already in progress") + self._closing = True + try: + await self._stop_runtime() + except Exception: + self._closing = False + raise + else: + self._status = SessionStatus.STOPPED + self._closing = False + + async def _stop_runtime(self) -> None: + native = self._client._require_native_module("stop") + await _call_blocking( + lambda: json.loads( + native.stop_runtime(json.dumps(self._plan), json.dumps(self._runtime)) + ) + ) def _absorb(self, result: Any) -> None: """Record the turn's handles and advance the transcript from its ``RunResult``.""" 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"), @@ -580,11 +580,8 @@ def _absorb(self, result: Any) -> None: if not isinstance(output, dict): return messages = output.get("messages") - if isinstance(messages, list) and messages: + if isinstance(messages, list): self._messages = deepcopy(messages) - session_id = output.get("session_id") - if session_id: - self.id = str(session_id) async def __aenter__(self) -> "Session": return self @@ -593,20 +590,6 @@ async def __aexit__(self, exc_type: object, exc: object, traceback: object) -> N 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: @@ -685,141 +668,38 @@ def _run_request_payload( return payload -def _inline_adapter_entrypoint(plan: dict[str, Any]) -> tuple[str, str] | None: - descriptor = ((plan.get("adapter_descriptor") or {}).get("descriptor") or {}) - if descriptor.get("adapter_kind") != "python": - return None - runner = descriptor.get("runner") or {} - module = runner.get("module") - callable_name = runner.get("callable") - if not module or not callable_name: - return None - return str(module), str(callable_name) - - -async def _run_inline_adapter( +async def _run_native_lifecycle( + native: Any, plan: dict[str, Any], request: dict[str, Any], - entrypoint: tuple[str, str], ) -> dict[str, Any]: - runtime_id = _new_id("runtime") - invocation_id = _new_id("invocation") - environment = _environment_handle(plan) - artifacts = _artifact_manifest(plan) - relay_runtime = _prepare_relay_runtime_config( - plan, runtime_id, invocation_id, request, artifacts - ) - payload = _fabric_adapter_payload( - plan, - runtime_id, - invocation_id, - environment, - artifacts, - relay_runtime, - request, - ) - module_name, callable_name = entrypoint - events = [ - _event( - "runtime_start", - f"started runtime {runtime_id}", - { - "runtime_id": runtime_id, - "environment_id": environment["environment_id"], - "environment_provider": environment["provider"], - }, - ), - _event( - "invocation_start", - f"starting inline python adapter for {_harness_type(plan)}", - { - "runtime_id": runtime_id, - "invocation_id": invocation_id, - "module": module_name, - "callable": callable_name, - }, - ), - ] - metadata = { - "adapter_runner": "python_inline", - "module": module_name, - "callable": callable_name, - "environment_provider": environment["provider"], - } - try: - output = await _call_inline_adapter(entrypoint, plan, payload, relay_runtime["env"]) - status = "failed" if isinstance(output, dict) and output.get("failed") else "succeeded" - error = _output_error(output, metadata) if status == "failed" else None - except Exception as exc: # noqa: BLE001 - normalize adapter failures for consumers. - output = {} - status = "failed" - error = { - "stage": "invoke", - "code": "python_inline_adapter_error", - "message": str(exc), - "retryable": False, - "metadata": { - **metadata, - "exception_type": type(exc).__name__, - }, - } - _collect_inline_adapter_artifacts(output, artifacts) - events.extend( - [ - _event( - "invocation_end", - f"inline python adapter completed with status {status}", - { - "runtime_id": runtime_id, - "invocation_id": invocation_id, - }, - ), - _event( - "runtime_stop", - f"stopped runtime {runtime_id}", - {"runtime_id": runtime_id}, - ), - ] - ) - result = { - "agent_name": plan["agent_name"], - "profile": plan.get("profile"), - "harness_type": _harness_type(plan), - "adapter_kind": _adapter_kind(plan), - "adapter_id": _adapter_id(plan), - "runtime_id": runtime_id, - "invocation_id": invocation_id, - "request_id": request["request_id"], - "status": status, - "output": output, - "artifacts": artifacts, - "telemetry": _telemetry_ref(plan, relay_runtime), - "events": events, - "metadata": metadata, - } - if error is not None: - result["error"] = error - return result - - -async def _call_inline_adapter( - entrypoint: tuple[str, str], - plan: dict[str, Any], - payload: dict[str, Any], - relay_env: dict[str, str], -) -> Any: - module_name, callable_name = entrypoint - adapter_root = ((plan.get("adapter_descriptor") or {}).get("root")) - added_paths = _prepend_adapter_paths(adapter_root) - try: - module = importlib.import_module(module_name) - func = _resolve_attr(module, callable_name) - with _patched_environ(relay_env): - if inspect.iscoroutinefunction(func): - return await func(payload) - return await _call_blocking(lambda: func(payload)) - finally: - _restore_sys_path(added_paths) + def _run() -> dict[str, Any]: + plan_json = json.dumps(plan) + runtime = json.loads(native.start_runtime(plan_json)) + runtime_json = json.dumps(runtime) + result: dict[str, Any] | None = None + invoke_error: Exception | None = None + try: + try: + result = json.loads( + native.invoke_runtime(plan_json, runtime_json, json.dumps(request)) + ) + except Exception as error: + invoke_error = error + raise + return result + finally: + try: + stop_events = json.loads(native.stop_runtime(plan_json, runtime_json)) + except Exception: + if invoke_error is not None: + stop_events = [] + else: + raise + if isinstance(result, dict) and isinstance(stop_events, list): + result.setdefault("events", []).extend(stop_events) + + return await _call_blocking(_run) async def _call_blocking(func: Any) -> Any: @@ -831,243 +711,6 @@ async def _call_blocking(func: Any) -> Any: return await loop.run_in_executor(executor, func) -def _fabric_adapter_payload( - plan: dict[str, Any], - runtime_id: str, - invocation_id: str, - environment: dict[str, Any], - artifacts: dict[str, Any], - relay_runtime: dict[str, Any], - request: dict[str, Any], -) -> dict[str, Any]: - effective_config = _effective_config(plan) - return { - "effective_config": effective_config, - "runtime_context": { - "runtime_id": runtime_id, - "invocation_id": invocation_id, - "request_id": request["request_id"], - "environment": environment, - "artifacts": artifacts, - "telemetry": _runtime_telemetry_context(plan, relay_runtime), - }, - "request": request, - "capability_plan": plan.get("capability_plan") or {}, - "telemetry_plan": plan.get("telemetry_plan"), - } - - -def _effective_config(plan: dict[str, Any]) -> dict[str, Any]: - if isinstance(plan.get("effective_config"), dict): - effective = json.loads(json.dumps(plan["effective_config"])) - else: - effective = { - "agent_name": plan.get("agent_name"), - "profile": plan.get("profile"), - "profiles": plan.get("profiles") or [], - "agent_root": plan.get("agent_root"), - "config_path": plan.get("config_path"), - "config_root": plan.get("config_root"), - "config": plan.get("config") or {}, - } - effective["agent_root"] = _absolute_plan_path(effective.get("agent_root")) - effective["config_path"] = _absolute_plan_path(effective.get("config_path")) - effective["config_root"] = _absolute_plan_path(effective.get("config_root")) - return effective - - -def _runtime_telemetry_context( - plan: dict[str, Any], relay_runtime: dict[str, Any] -) -> dict[str, Any] | None: - telemetry = plan.get("telemetry_plan") - if telemetry is None: - return None - metadata: dict[str, Any] = {} - for source, target in ( - ("relay_mode", "relay_mode"), - ("relay_project", "relay_project"), - ("relay_output_dir", "relay_output_dir"), - ("adapter_outputs", "adapter_outputs"), - ): - if source in telemetry and telemetry[source] is not None: - metadata[target] = telemetry[source] - return { - "relay_enabled": bool(telemetry.get("relay_enabled")), - "config_path": relay_runtime.get("config_path"), - "env": relay_runtime.get("env") or {}, - "metadata": metadata, - } - - -def _environment_handle(plan: dict[str, Any]) -> dict[str, Any]: - environment_plan = plan.get("environment_plan") or {} - config = plan.get("config") or {} - runtime = config.get("runtime") or {} - return { - "environment_id": _new_id("environment"), - "provider": environment_plan.get("provider", "local"), - "control_location": environment_plan.get("control_location", "external_control"), - "workspace": _absolute_plan_path(environment_plan.get("workspace") or plan.get("agent_root")), - "artifacts": environment_plan.get("artifacts") or _runtime_artifact_path(plan, runtime), - "ownership": environment_plan.get("ownership", "caller_owned"), - "connection": environment_plan.get("connection", {}), - "metadata": { - **(environment_plan.get("settings") or {}), - **(environment_plan.get("metadata") or {}), - }, - } - - -def _artifact_manifest(plan: dict[str, Any]) -> dict[str, Any]: - root = _runtime_artifact_path(plan, ((plan.get("config") or {}).get("runtime") or {})) - if root is not None: - Path(root).mkdir(parents=True, exist_ok=True) - return { - "root": root, - "artifacts": [], - } - - -def _runtime_artifact_path(plan: dict[str, Any], runtime: dict[str, Any]) -> str | None: - artifacts = runtime.get("artifacts") - if artifacts: - return str(_resolve_plan_path(plan.get("config_root"), artifacts)) - environment_artifacts = (plan.get("environment_plan") or {}).get("artifacts") - if environment_artifacts: - return str(environment_artifacts) - return None - - -def _prepare_relay_runtime_config( - plan: dict[str, Any], - runtime_id: str, - invocation_id: str, - request: dict[str, Any], - artifacts: dict[str, Any], -) -> dict[str, Any]: - telemetry = plan.get("telemetry_plan") - if not telemetry or not telemetry.get("relay_enabled"): - return {"config_path": None, "env": {}} - root = artifacts.get("root") - if not root: - return {"config_path": None, "env": {}} - relay_config = { - "schema_version": "fabric.relay/v1alpha1", - "relay": { - "enabled": True, - "mode": telemetry.get("relay_mode") or "sdk", - "project": telemetry.get("relay_project"), - "output_dir": telemetry.get("relay_output_dir"), - "config": telemetry.get("relay_config") or {}, - }, - "fabric": { - "agent_name": plan.get("agent_name"), - "profile": plan.get("profile"), - "harness_type": _harness_type(plan), - "adapter_id": _adapter_id(plan), - "runtime_id": runtime_id, - "invocation_id": invocation_id, - "request_id": request["request_id"], - "adapter_outputs": telemetry.get("adapter_outputs") or [], - }, - } - path = Path(root) / "relay-config.json" - path.write_text(json.dumps(relay_config, indent=2, sort_keys=True), encoding="utf-8") - _add_artifact(artifacts, "relay_config", "telemetry_config", path, "application/json") - return { - "config_path": str(path.resolve()), - "env": { - "FABRIC_RELAY_ENABLED": "true", - "FABRIC_RELAY_MODE": telemetry.get("relay_mode") or "sdk", - "FABRIC_RELAY_CONFIG_PATH": str(path.resolve()), - }, - } - - -def _collect_inline_adapter_artifacts(output: Any, artifacts: dict[str, Any]) -> None: - if not isinstance(output, dict): - return - for index, artifact in enumerate(output.get("relay_artifacts") or []): - path = artifact.get("path") - if not path: - continue - _add_artifact( - artifacts, - f"relay_{artifact.get('kind', 'artifact')}_{index}", - artifact.get("kind", "telemetry"), - Path(path), - "application/json", - ) - - -def _add_artifact( - manifest: dict[str, Any], - name: str, - kind: str, - path: Path, - media_type: str | None = None, -) -> None: - manifest.setdefault("artifacts", []).append( - { - "name": name, - "kind": kind, - "path": str(path), - "media_type": media_type, - } - ) - - -def _telemetry_ref(plan: dict[str, Any], relay_runtime: dict[str, Any]) -> dict[str, Any] | None: - telemetry = plan.get("telemetry_plan") - if telemetry is None: - return None - metadata: dict[str, Any] = {} - for source, target in ( - ("relay_mode", "relay_mode"), - ("relay_project", "relay_project"), - ("relay_output_dir", "relay_output_dir"), - ("relay_config", "relay_config"), - ("adapter_outputs", "adapter_outputs"), - ): - if source in telemetry and telemetry[source] is not None: - metadata[target] = telemetry[source] - if relay_runtime.get("config_path"): - metadata["relay_config_path"] = relay_runtime["config_path"] - return { - "relay_enabled": bool(telemetry.get("relay_enabled")), - "metadata": metadata, - } - - -def _output_error(output: Any, metadata: dict[str, Any]) -> dict[str, Any]: - message = "inline python adapter returned failed status" - if isinstance(output, dict) and output.get("error"): - message = str(output["error"]) - return { - "stage": "invoke", - "code": "python_inline_adapter_failed", - "message": message, - "retryable": False, - "metadata": metadata, - } - - -def _event(kind: str, message: str, metadata: dict[str, Any]) -> dict[str, Any]: - return { - "event_id": _new_id("event"), - "timestamp_millis": int(time.time() * 1000), - "kind": kind, - "message": message, - "metadata": metadata, - } - - -def _adapter_id(plan: dict[str, Any]) -> str | None: - descriptor = ((plan.get("adapter_descriptor") or {}).get("descriptor") or {}) - harness = (plan.get("config") or {}).get("harness") or {} - return descriptor.get("adapter_id") or harness.get("adapter_id") - - def _adapter_kind(plan: dict[str, Any]) -> str: descriptor = ((plan.get("adapter_descriptor") or {}).get("descriptor") or {}) return descriptor.get("adapter_kind", "process") @@ -1076,67 +719,3 @@ def _adapter_kind(plan: dict[str, Any]) -> str: def _harness_type(plan: dict[str, Any]) -> str: descriptor = ((plan.get("adapter_descriptor") or {}).get("descriptor") or {}) return descriptor.get("adapter_id", "unknown") - - -def _new_id(prefix: str) -> str: - return f"{prefix}-{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}" - - -def _absolute_plan_path(path: Any) -> str | None: - if path is None: - return None - return str(Path(path).resolve()) - - -def _resolve_plan_path(root: Any, path: Any) -> Path: - path_obj = Path(path) - if path_obj.is_absolute(): - return path_obj - if root is None: - return path_obj - return Path(root) / path_obj - - -def _prepend_adapter_paths(adapter_root: Any) -> list[str]: - if not adapter_root: - return [] - root = Path(adapter_root) - candidates = [root / "src", root / "python"] - added: list[str] = [] - for candidate in candidates: - if candidate.is_dir(): - value = str(candidate) - sys.path.insert(0, value) - added.append(value) - return added - - -def _restore_sys_path(added_paths: list[str]) -> None: - for value in added_paths: - try: - sys.path.remove(value) - except ValueError: - pass - - -def _resolve_attr(module: Any, dotted_name: str) -> Any: - value = module - for part in dotted_name.split("."): - value = getattr(value, part) - return value - - -@contextmanager -def _patched_environ(updates: Mapping[str, str]): - previous: dict[str, str | None] = {} - for key, value in updates.items(): - previous[key] = os.environ.get(key) - os.environ[key] = value - try: - yield - finally: - for key, value in previous.items(): - if value is None: - os.environ.pop(key, None) - else: - os.environ[key] = value diff --git a/python/tests/smoke_environment_handle.py b/python/tests/smoke_environment_handle.py index 628f72f6b..3155c305f 100644 --- a/python/tests/smoke_environment_handle.py +++ b/python/tests/smoke_environment_handle.py @@ -1,31 +1,30 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Regression: the inline-path environment handle absolutizes the workspace. - -A relative workspace (config-root-relative) must be resolved to an absolute path -so an adapter does not re-join it onto the absolute config_root and double it -(e.g. examples/agent/examples/agent/...). Dependency-free; no native extension. -""" +"""Regression: started runtime handles absolutize the workspace path.""" from __future__ import annotations +import asyncio import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -from nemo_fabric.client import _environment_handle +from nemo_fabric import FabricClient + +ROOT = Path(__file__).resolve().parents[2] + +async def main() -> None: + async with FabricClient() as client: + session = await client.start(ROOT / "examples" / "code-review-agent", profile="env_local") + try: + workspace = session.runtime["environment"]["workspace"] + finally: + await session.stop() -def main() -> None: - plan = { - "environment_plan": {"workspace": "examples/code-review-agent/repos/my-service"}, - "config": {"runtime": {}}, - "agent_root": "examples/code-review-agent", - } - workspace = _environment_handle(plan)["workspace"] assert os.path.isabs(workspace), f"workspace not absolute: {workspace}" assert ( "code-review-agent/examples/code-review-agent" not in workspace @@ -35,4 +34,4 @@ def main() -> None: if __name__ == "__main__": - main() + asyncio.run(main()) diff --git a/python/tests/smoke_native_sdk.py b/python/tests/smoke_native_sdk.py index c10686620..2983df34c 100644 --- a/python/tests/smoke_native_sdk.py +++ b/python/tests/smoke_native_sdk.py @@ -111,13 +111,20 @@ async def smoke(client: FabricClient) -> None: temp_agent = Path(tmpdir) / "hermes-shim-agent" copytree(fixture_agent, temp_agent) result = await client.run(temp_agent, profile="env_local", input_text="hello native") + async with await client.start(temp_agent, profile="env_local") as session: + first = await session.invoke("hello session one") + second = await session.invoke("hello session two") assert result["status"] == "succeeded" assert result["adapter_kind"] == "python" - assert result["metadata"]["adapter_runner"] == "python_inline" + assert result["metadata"]["adapter_runner"] == "python" assert result["output"]["received"] == "hello native" assert result["output"]["native_mcp_servers"] == ["github"] - assert not any(artifact["name"] == "stdout" for artifact in result["artifacts"]["artifacts"]) + assert any(artifact["name"] == "stdout" for artifact in result["artifacts"]["artifacts"]) + assert first["status"] == "succeeded" + assert second["status"] == "succeeded" + assert first["runtime_id"] == second["runtime_id"] + assert session.runtime["runtime_id"] == first["runtime_id"] if __name__ == "__main__": diff --git a/python/tests/smoke_sdk_sessions.py b/python/tests/smoke_sdk_sessions.py index fb5c0ecd9..896f727d6 100644 --- a/python/tests/smoke_sdk_sessions.py +++ b/python/tests/smoke_sdk_sessions.py @@ -1,111 +1,119 @@ # 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. -""" +"""Smoke: the SDK Session boundary over the native RuntimeHandle lifecycle.""" from __future__ import annotations import asyncio +import json import sys from pathlib import Path +from typing import Any 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 +from nemo_fabric import FabricClient, Session, SessionStatus -seen_history: list[list] = [] +def _plan() -> dict[str, Any]: + return { + "agent_name": "demo", + "profile": "hermes_sdk", + "adapter_descriptor": { + "descriptor": {"adapter_kind": "python", "adapter_id": "test.fabric.shim"} + }, + } -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}"}, - ] +def _runtime() -> dict[str, Any]: 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-fake", + "runtime_id": "runtime-1", + "agent_name": "demo", + "harness_type": "test.fabric.shim", + "mode": "session", + "adapter_kind": "python", + "adapter_id": "test.fabric.shim", + "environment": { + "environment_id": "environment-1", + "provider": "local", + "control_location": "external_control", + "ownership": "caller_owned", }, } -def _session() -> Session: - return Session( - client=FabricClient(), - plan={"agent_name": "demo", "profile": "hermes_sdk"}, - entrypoint=("fake.module", "run"), - ) +class FakeNative: + def __init__(self) -> None: + self.requests: list[dict[str, Any]] = [] + self.stopped = 0 + + def invoke_runtime(self, plan_json: str, runtime_json: str, request_json: str) -> str: + request = json.loads(request_json) + self.requests.append(request) + turn = len(self.requests) + return json.dumps( + { + "status": "succeeded", + "request_id": request["request_id"], + "runtime_id": json.loads(runtime_json)["runtime_id"], + "invocation_id": f"invocation-{turn}", + "events": [{"event_id": f"evt-{turn}", "kind": "log", "message": "ok"}], + "output": { + "messages": [ + {"role": "user", "content": request.get("input")}, + {"role": "assistant", "content": f"reply-{turn}"}, + ], + }, + } + ) + + def stop_runtime(self, plan_json: str, runtime_json: str) -> str: + self.stopped += 1 + return "[]" + + +class NativeClient(FabricClient): + def __init__(self, native: FakeNative) -> None: + super().__init__() + self.native = native + + def _require_native_module(self, method: str) -> FakeNative: + return self.native -async def multi_turn_threads_history() -> None: - seen_history.clear() - client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] - session = _session() +def _session(native: FakeNative) -> Session: + return Session(client=NativeClient(native), plan=_plan(), runtime=_runtime()) + + +async def stable_runtime_across_turns() -> None: + native = FakeNative() + session = _session(native) assert session.status is SessionStatus.ACTIVE - assert session.messages == [] + assert session.runtime_id == "runtime-1" + assert "session_id" not in session.info + assert not hasattr(session, "id") 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 - - # 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] - 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 + assert [inv["runtime_id"] for inv in session.invocations] == ["runtime-1", "runtime-1"] + assert "history" not in native.requests[0]["context"] + assert "history" not in native.requests[1]["context"] + assert session.runtime_id == "runtime-1" -async def stop_is_idempotent_and_blocks_invoke() -> None: - client_mod._run_inline_adapter = _fake_inline # type: ignore[assignment] - session = _session() + +async def stream_and_lifecycle() -> None: + native = FakeNative() + session = _session(native) + items = [item async for item in session.stream("hello")] + assert items[-1]["status"] == "succeeded" + assert items[:-1] and all(e.get("kind") == "log" for e in items[:-1]) + + await session.stop() await session.stop() assert session.status is SessionStatus.STOPPED - await session.stop() # idempotent + assert native.stopped == 1 try: await session.invoke("too late") except RuntimeError: @@ -113,18 +121,13 @@ async def stop_is_idempotent_and_blocks_invoke() -> None: 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() + native = FakeNative() + session = _session(native) await session.cancel() assert session.status is SessionStatus.CANCELLED - await session.cancel() # idempotent + assert native.stopped == 1 try: await session.invoke("after cancel") except RuntimeError: @@ -133,53 +136,10 @@ async def cancel_when_idle_marks_cancelled() -> None: 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] + await stable_runtime_across_turns() + await stream_and_lifecycle() + await cancel_when_idle_marks_cancelled() print("smoke_sdk_sessions ok") diff --git a/python/tests/smoke_typed_config.py b/python/tests/smoke_typed_config.py index 84e9ed396..f304b7874 100644 --- a/python/tests/smoke_typed_config.py +++ b/python/tests/smoke_typed_config.py @@ -8,7 +8,7 @@ * ``plan_config`` / ``doctor_config`` resolve a maintained (repository) adapter with ``base_dir=None`` -- zero filesystem layout, no ``agent.yaml``. -* ``run_config`` drives a real inline run using only a local adapter directory +* ``run_config`` drives a real core runtime run using only a local adapter directory (still no agent package). * the ``*_config`` methods are native-only; the CLI fallback raises a clear, documented error rather than silently degrading. @@ -27,7 +27,7 @@ from nemo_fabric import FabricClient, FabricNativeUnavailableError ROOT = Path(__file__).resolve().parents[2] -# The inline test adapter (needs only python3, no secrets), shipped as a fixture. +# The test adapter (needs only python3, no secrets), shipped as a fixture. SHIM_ADAPTERS = ROOT / "tests" / "fixtures" / "hermes-shim-agent" / "adapters" @@ -61,7 +61,7 @@ def _repository_adapter_config() -> dict: def _shim_adapter_config() -> dict: - """Config referencing the inline test adapter (runs without secrets).""" + """Config referencing the test adapter (runs without secrets).""" config = _repository_adapter_config() config["metadata"] = {"name": "typed-only-run"} @@ -103,7 +103,7 @@ async def resolves_and_diagnoses_without_a_directory(client: FabricClient) -> No async def runs_without_an_agent_package(client: FabricClient) -> None: - """run_config drives an inline run with only an adapter dir (no agent.yaml).""" + """run_config drives a core run with only an adapter dir (no agent.yaml).""" config = _shim_adapter_config() with tempfile.TemporaryDirectory(prefix="typed-run-") as tmpdir: @@ -111,12 +111,13 @@ async def runs_without_an_agent_package(client: FabricClient) -> None: # Only the adapter lives here -- this is deliberately NOT an agent # package (no agent.yaml / profiles / repos / skills). copytree(SHIM_ADAPTERS, base / "adapters") + (base / "ws").mkdir() assert not (base / "agent.yaml").exists() result = await client.run_config(config, base_dir=base, input_text="hello typed") assert result["status"] == "succeeded", result.get("status") assert result["adapter_kind"] == "python" - assert result["metadata"]["adapter_runner"] == "python_inline" + assert result["metadata"]["adapter_runner"] == "python" assert result["output"]["received"] == "hello typed" diff --git a/tests/fixtures/hermes-shim-agent/adapters/hermes-shim/src/nemo_fabric_test_adapters/hermes_shim/adapter.py b/tests/fixtures/hermes-shim-agent/adapters/hermes-shim/src/nemo_fabric_test_adapters/hermes_shim/adapter.py index 44945eec5..511ff6c09 100644 --- a/tests/fixtures/hermes-shim-agent/adapters/hermes-shim/src/nemo_fabric_test_adapters/hermes_shim/adapter.py +++ b/tests/fixtures/hermes-shim-agent/adapters/hermes-shim/src/nemo_fabric_test_adapters/hermes_shim/adapter.py @@ -21,7 +21,7 @@ def main() -> None: def run(payload: dict[str, Any]) -> dict[str, Any]: - """Inline test adapter entrypoint used by SDK smoke tests.""" + """Test adapter entrypoint used by SDK smoke tests.""" return run_selected_mode(payload) diff --git a/tests/smoke_hermes_session.py b/tests/smoke_hermes_session.py index 0eaacfc9f..05fa3819c 100644 --- a/tests/smoke_hermes_session.py +++ b/tests/smoke_hermes_session.py @@ -4,12 +4,13 @@ """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). +adapter and asserts the session carries conversation memory across turns through +the same Fabric runtime handle. 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: +SDK-only and runs through the native Fabric runtime lifecycle, 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 @@ -62,7 +63,8 @@ async def _run() -> None: 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 r2["runtime_id"] == r1["runtime_id"], (r1, r2) + # Hermes should return a transcript that includes the prior turn. 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() diff --git a/tests/test_hermes_sdk_adapter.py b/tests/test_hermes_sdk_adapter.py new file mode 100644 index 000000000..e573d7138 --- /dev/null +++ b/tests/test_hermes_sdk_adapter.py @@ -0,0 +1,158 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the Hermes SDK adapter's Fabric runtime mapping.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from types import ModuleType +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +HERMES_SDK_SRC = ROOT / "adapters" / "hermes-sdk" / "src" +if str(HERMES_SDK_SRC) not in sys.path: + sys.path.insert(0, str(HERMES_SDK_SRC)) + +from nemo_fabric_adapters.hermes_sdk import adapter # noqa: E402 + + +def test_runtime_id_drives_hermes_session_id_and_hermes_db_history( + monkeypatch, + tmp_path: Path, +) -> None: + captured: dict[str, Any] = {} + db_history = [{"role": "user", "content": "from hermes db"}] + + class FakeSessionDB: + def get_session(self, session_id: str) -> dict[str, str] | None: + captured.setdefault("db_get_session", []).append(session_id) + if session_id == "runtime-resolved-456": + return {"id": session_id} + return None + + def resolve_resume_session_id(self, session_id: str) -> str: + captured["db_resolve_session"] = session_id + return "runtime-resolved-456" + + def get_messages_as_conversation(self, session_id: str) -> list[dict[str, str]]: + captured["db_get_messages"] = session_id + return list(db_history) + + class FakeAIAgent: + def __init__( + self, + *, + base_url: str | None = None, + api_key: str | None = None, + provider: str | None = None, + model: str = "", + max_iterations: int = 1, + enabled_toolsets: list[str] | None = None, + quiet_mode: bool = True, + skip_context_files: bool = True, + skip_memory: bool = True, + save_trajectories: bool = False, + max_tokens: int = 512, + temperature: float = 0.0, + reasoning_config: dict[str, Any] | None = None, + insert_reasoning: bool = False, + platform: str | None = None, + session_id: str | None = None, + session_db: Any | None = None, + ) -> None: + captured["init"] = { + "session_id": session_id, + "session_db": session_db, + "platform": platform, + "model": model, + "provider": provider, + } + self.session_id = session_id or "generated-session" + self.model = model + self.platform = platform + + def run_conversation( + self, + user_message: str, + *, + system_message: str | None = None, + conversation_history: list[dict[str, str]] | None = None, + sync_honcho: bool = False, + dont_review: bool = True, + ) -> dict[str, Any]: + captured["conversation"] = { + "user_message": user_message, + "system_message": system_message, + "conversation_history": conversation_history, + "sync_honcho": sync_honcho, + "dont_review": dont_review, + } + return { + "response": "ok", + "completed": True, + "failed": False, + "messages": [{"role": "assistant", "content": "ok"}], + } + + hermes_cli = ModuleType("hermes_cli") + hermes_config = ModuleType("hermes_cli.config") + hermes_config.load_config = lambda: {} # type: ignore[attr-defined] + hermes_plugins = ModuleType("hermes_cli.plugins") + hermes_plugins.discover_plugins = lambda force=False: None # type: ignore[attr-defined] + hermes_plugins.invoke_hook = lambda *args, **kwargs: None # type: ignore[attr-defined] + hermes_state = ModuleType("hermes_state") + hermes_state.SessionDB = FakeSessionDB # type: ignore[attr-defined] + run_agent = ModuleType("run_agent") + run_agent.AIAgent = FakeAIAgent # type: ignore[attr-defined] + + monkeypatch.setitem(sys.modules, "hermes_cli", hermes_cli) + monkeypatch.setitem(sys.modules, "hermes_cli.config", hermes_config) + monkeypatch.setitem(sys.modules, "hermes_cli.plugins", hermes_plugins) + monkeypatch.setitem(sys.modules, "hermes_state", hermes_state) + monkeypatch.setitem(sys.modules, "run_agent", run_agent) + monkeypatch.setenv("TEST_API_KEY", "secret") + + payload = { + "effective_config": { + "agent_name": "demo", + "config_root": str(tmp_path), + "config": { + "harness": { + "settings": { + "hermes_home": "./hermes-home", + "enabled_toolsets": [], + "system_prompt": "system", + } + }, + "models": { + "default": { + "provider": "test-provider", + "model": "test-model", + "api_key_env": "TEST_API_KEY", + } + }, + }, + }, + "runtime_context": { + "runtime_id": "runtime-fabric-123", + "environment": {"workspace": str(tmp_path)}, + }, + "request": { + "input": "hello", + "context": {"history": [{"role": "user", "content": "stale"}]}, + }, + "capability_plan": {"native": {}}, + } + + output = adapter.run_hermes_sdk(payload) + + assert captured["db_resolve_session"] == "runtime-fabric-123" + assert captured["db_get_session"] == ["runtime-resolved-456"] + assert captured["db_get_messages"] == "runtime-resolved-456" + assert captured["init"]["session_id"] == "runtime-fabric-123" + assert isinstance(captured["init"]["session_db"], FakeSessionDB) + assert captured["init"]["platform"] == "fabric" + assert captured["conversation"]["conversation_history"] == db_history + assert "session_id" not in output diff --git a/tests/test_session.py b/tests/test_session.py index 24196976a..56a1e9fbb 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -3,291 +3,537 @@ """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. +Dependency-free: a fake native lifecycle module stands in for the Rust binding, +so these exercise the Python orchestration without Hermes or a built extension. """ from __future__ import annotations import asyncio +import json +import time +from typing import Any import pytest -from nemo_fabric import ( - FabricClient, - FabricNativeUnavailableError, - FabricSessionUnsupportedError, - Session, - SessionStatus, -) +from nemo_fabric import FabricClient, FabricNativeUnavailableError, Session, SessionStatus from nemo_fabric import client as client_mod -def _plan(adapter_kind: str = "python") -> dict: +def _plan(adapter_kind: str = "python") -> dict[str, Any]: return { "agent_name": "demo", "profile": "hermes_sdk", "adapter_descriptor": { - "descriptor": {"adapter_kind": adapter_kind, "adapter_id": "test.fabric.shim"} + "descriptor": { + "adapter_kind": adapter_kind, + "adapter_id": "test.fabric.shim", + "runner": {"module": "fake.module", "callable": "run"}, + } }, } -def _session(overrides: dict | None = None) -> Session: +def _runtime() -> dict[str, Any]: + return { + "runtime_id": "runtime-1", + "agent_name": "demo", + "harness_type": "test.fabric.shim", + "mode": "session", + "adapter_kind": "python", + "adapter_id": "test.fabric.shim", + "environment": { + "environment_id": "environment-1", + "provider": "local", + "control_location": "external_control", + "ownership": "caller_owned", + }, + } + + +class FakeNative: + def __init__(self) -> None: + self.plans: list[dict[str, Any]] = [] + self.requests: list[dict[str, Any]] = [] + self.stopped = 0 + self.block_invoke = False + self.fail_invoke = False + self.fail_stop = False + + def plan(self, path: str, profile: Any = None) -> str: + self.plans.append({"path": path, "profile": profile}) + assert path == "agent" + if profile is not None: + assert profile == "hermes_sdk" + return json.dumps(_plan()) + + def plan_config( + self, + config_json: str, + profiles_json: str | None = None, + base_dir: str | None = None, + ) -> str: + assert json.loads(config_json)["metadata"]["name"] == "demo" + return json.dumps(_plan()) + + def start_runtime(self, plan_json: str) -> str: + assert json.loads(plan_json)["agent_name"] == "demo" + return json.dumps(_runtime()) + + def invoke_runtime( + self, plan_json: str, runtime_json: str, request_json: str + ) -> str: + if self.block_invoke: + time.sleep(0.2) + if self.fail_invoke: + raise RuntimeError("invoke failed") + plan = json.loads(plan_json) + runtime = json.loads(runtime_json) + request = json.loads(request_json) + self.requests.append(request) + turn = len(self.requests) + return json.dumps( + { + "agent_name": plan["agent_name"], + "profile": plan.get("profile"), + "harness_type": "test.fabric.shim", + "adapter_kind": "python", + "adapter_id": "test.fabric.shim", + "runtime_id": runtime["runtime_id"], + "invocation_id": f"invocation-{turn}", + "request_id": request["request_id"], + "status": "succeeded", + "events": [ + { + "event_id": f"evt-{turn}", + "kind": "log", + "message": f"turn {turn}", + } + ], + "output": { + "messages": [ + {"role": "user", "content": request.get("input")}, + {"role": "assistant", "content": f"reply-{turn}"}, + ], + "response": f"reply-{turn}", + }, + "artifacts": {"artifacts": []}, + } + ) + + def stop_runtime(self, plan_json: str, runtime_json: str) -> str: + assert json.loads(plan_json)["agent_name"] == "demo" + assert json.loads(runtime_json)["runtime_id"] == "runtime-1" + self.stopped += 1 + if self.fail_stop: + raise RuntimeError("stop failed") + return json.dumps([]) + + +class NativeClient(FabricClient): + def __init__(self, native: FakeNative) -> None: + super().__init__() + self.native = native + + def plan(self, path, *, profile=None): # type: ignore[no-untyped-def,override] + return json.loads(self.native.plan(str(path), profile)) + + def _native_module(self) -> FakeNative: + return self.native + + def _require_native_module(self, method: str) -> FakeNative: + return self.native + + +def _session(native: FakeNative | None = None, overrides: dict | None = None) -> Session: return Session( - client=FabricClient(), + client=NativeClient(native or FakeNative()), plan=_plan(), - entrypoint=("fake.module", "run"), + runtime=_runtime(), 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() +async def test_start_creates_session_from_core_runtime_handle() -> None: + native = FakeNative() + session = await NativeClient(native).start("agent", profile="hermes_sdk") + + assert native.plans == [{"path": "agent", "profile": "hermes_sdk"}] assert session.status is SessionStatus.ACTIVE - assert session.messages == [] - assert session.invocations == [] + assert session.runtime_id == "runtime-1" + assert session.runtime["runtime_id"] == "runtime-1" + assert session.info["runtime_id"] == "runtime-1" + assert "session_id" not in session.info + assert not hasattr(session, "id") -async def test_invoke_threads_accumulated_history(seen_history: list[list]) -> None: - session = _session() +async def test_invoke_uses_stable_runtime_and_does_not_replay_history() -> None: + native = FakeNative() + session = _session(native) 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 + assert [inv["runtime_id"] for inv in session.invocations] == [ + "runtime-1", + "runtime-1", + ] + assert "history" not in native.requests[0]["context"] + assert "history" not in native.requests[1]["context"] + assert session.runtime_id == "runtime-1" + assert len(session.messages) == 2 -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_request_level_overrides_are_merged() -> None: + native = FakeNative() + session = _session(native, overrides={"a": "session"}) + await session.invoke( + request={"input": "x", "overrides": {"b": "request"}}, + overrides={"c": "turn"}, + ) + assert native.requests[0]["overrides"] == { + "a": "session", + "b": "request", + "c": "turn", + } -async def test_invoke_adopts_session_id_from_output(seen_history: list[list]) -> None: + +async def test_stream_yields_events_then_result() -> None: session = _session() - await session.invoke("hi") - assert session.id == "sess-1" + items = [item async for item in session.stream("hi")] + + assert items[-1]["status"] == "succeeded" + assert items[:-1] and all(event.get("kind") == "log" for event in items[:-1]) + +async def test_stop_is_idempotent_and_blocks_invoke() -> None: + native = FakeNative() + session = _session(native) -async def test_invoke_merges_session_and_turn_overrides( + await session.stop() + await session.stop() + + assert session.status is SessionStatus.STOPPED + assert native.stopped == 1 + with pytest.raises(RuntimeError): + await session.invoke("too late") + + +async def test_stop_rejects_in_flight_turn(monkeypatch: pytest.MonkeyPatch) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def _blocking(func): # type: ignore[no-untyped-def] + started.set() + await release.wait() + return func() + + monkeypatch.setattr(client_mod, "_call_blocking", _blocking) + native = FakeNative() + session = _session(native) + first = asyncio.create_task(session.invoke("turn one")) + await started.wait() + + with pytest.raises(RuntimeError, match="turn is in flight"): + await session.stop() + assert session.status is SessionStatus.ACTIVE + assert native.stopped == 0 + + release.set() + await first + + +async def test_stop_blocks_new_turns_while_shutdown_is_in_progress( monkeypatch: pytest.MonkeyPatch, ) -> None: - captured: dict = {} + started = asyncio.Event() + release = asyncio.Event() - async def _fake(plan, request, entrypoint): - captured["overrides"] = request.get("overrides") - return {"status": "succeeded", "output": {}} + async def _blocking(func): # type: ignore[no-untyped-def] + if session._closing: # noqa: SLF001 - state-machine regression test + started.set() + await release.wait() + return func() - monkeypatch.setattr(client_mod, "_run_inline_adapter", _fake) - session = _session(overrides={"model": "a"}) - await session.invoke("x", overrides={"temperature": 0.0}) + monkeypatch.setattr(client_mod, "_call_blocking", _blocking) + native = FakeNative() + session = _session(native) + stop_task = asyncio.create_task(session.stop()) + await started.wait() - assert captured["overrides"] == {"model": "a", "temperature": 0.0} + with pytest.raises(RuntimeError, match="shutdown is in progress"): + await session.invoke("too late") + release.set() + await stop_task + assert session.status is SessionStatus.STOPPED + assert native.stopped == 1 -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_failure_clears_shutdown_guard_for_retry() -> None: + native = FakeNative() + native.fail_stop = True + session = _session(native) + with pytest.raises(RuntimeError, match="stop failed"): + await session.stop() -async def test_stop_is_idempotent_and_blocks_invoke(seen_history: list[list]) -> None: - session = _session() + assert session.status is SessionStatus.ACTIVE + assert session._closing is False # noqa: SLF001 - state-machine regression test + + native.fail_stop = False await session.stop() + assert session.status is SessionStatus.STOPPED - await session.stop() # idempotent - with pytest.raises(RuntimeError): - await session.invoke("too late") + assert native.stopped == 2 -async def test_context_manager_auto_stops(seen_history: list[list]) -> None: - async with _session() as session: +async def test_context_manager_auto_stops() -> None: + native = FakeNative() + async with _session(native) as session: await session.invoke("hi") assert session.status is SessionStatus.ACTIVE + assert session.status is SessionStatus.STOPPED + assert native.stopped == 1 -async def test_cancel_when_idle_marks_cancelled(seen_history: list[list]) -> None: - session = _session() +async def test_cancel_when_idle_marks_cancelled() -> None: + native = FakeNative() + session = _session(native) await session.cancel() + await session.cancel() + assert session.status is SessionStatus.CANCELLED - await session.cancel() # idempotent + assert native.stopped == 1 with pytest.raises(RuntimeError): await session.invoke("after cancel") -async def test_cancel_aborts_in_flight_turn(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_cancel_stop_failure_keeps_session_retryable() -> None: + native = FakeNative() + native.fail_stop = True + session = _session(native) + + with pytest.raises(RuntimeError, match="stop failed"): + await session.cancel() + + assert session.status is SessionStatus.ACTIVE + assert native.stopped == 1 + + native.fail_stop = False + await session.cancel() + + assert session.status is SessionStatus.CANCELLED + assert native.stopped == 2 + + +async def test_cancel_blocks_new_turns_while_shutdown_is_in_progress( + 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": {}} + async def _blocking(func): # type: ignore[no-untyped-def] + if session._closing: # noqa: SLF001 - state-machine regression test + started.set() + await release.wait() + return func() - monkeypatch.setattr(client_mod, "_run_inline_adapter", _blocking) - session = _session() - turn = asyncio.create_task(session.invoke("long running")) + monkeypatch.setattr(client_mod, "_call_blocking", _blocking) + native = FakeNative() + session = _session(native) + cancel_task = asyncio.create_task(session.cancel()) await started.wait() + with pytest.raises(RuntimeError, match="shutdown is in progress"): + await session.invoke("too late") + + release.set() + await cancel_task + assert session.status is SessionStatus.CANCELLED + assert native.stopped == 1 + + +async def test_cancel_aborts_in_flight_turn() -> None: + native = FakeNative() + native.block_invoke = True + session = _session(native) + turn = asyncio.create_task(session.invoke("long running")) + await asyncio.sleep(0) + await session.cancel() + assert session.status is SessionStatus.CANCELLED + assert native.stopped == 1 with pytest.raises(asyncio.CancelledError): await turn -async def test_info_summarizes_the_session(seen_history: list[list]) -> None: +async def test_info_summarizes_the_session() -> None: session = _session() info = session.info - assert info["session_id"] == session.id + + assert info["runtime_id"] == "runtime-1" 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: +async def test_messages_invocations_and_runtime_return_copies() -> None: session = _session() await session.invoke("hi") - snapshot = session.messages - snapshot.append({"role": "user", "content": "tampered"}) - snapshot[0]["content"] = "mutated" # deep mutation of a returned message object + messages = session.messages + messages[0]["content"] = "mutated" invocations = session.invocations invocations.clear() + runtime = session.runtime + runtime["runtime_id"] = "mutated" - # Mutating the returned lists or their items must not affect session state. - assert len(session.messages) == 2 - assert session.messages[0]["content"] != "mutated" + assert session.messages[0]["content"] == "hi" 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() + assert session.runtime["runtime_id"] == "runtime-1" + + +async def test_invoke_without_output_messages_keeps_transcript() -> None: + class NoMessageNative(FakeNative): + def invoke_runtime(self, plan_json, runtime_json, request_json): # type: ignore[no-untyped-def] + self.requests.append(json.loads(request_json)) + return json.dumps( + { + "status": "succeeded", + "runtime_id": "runtime-1", + "invocation_id": "invocation-1", + "request_id": self.requests[-1]["request_id"], + "output": {}, + } + ) + + session = _session(NoMessageNative()) 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_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 = {} + assert session.messages == [] + assert len(session.invocations) == 1 - 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"}, - ) +async def test_empty_output_messages_replaces_existing_transcript() -> None: + class EmptyMessageNative(FakeNative): + def invoke_runtime(self, plan_json, runtime_json, request_json): # type: ignore[no-untyped-def] + if not self.requests: + return super().invoke_runtime(plan_json, runtime_json, request_json) + request = json.loads(request_json) + self.requests.append(request) + return json.dumps( + { + "status": "succeeded", + "runtime_id": "runtime-1", + "invocation_id": f"invocation-{len(self.requests)}", + "request_id": request["request_id"], + "output": {"messages": []}, + } + ) + + native = EmptyMessageNative() + session = _session(native) + await session.invoke("hi") + assert session.messages - # session < request < per-turn, all merged (none bypassed). - assert captured["overrides"] == {"a": "session", "b": "request", "c": "turn"} + await session.invoke("reset") + assert session.messages == [] async def test_concurrent_invokes_are_rejected(monkeypatch: pytest.MonkeyPatch) -> None: started = asyncio.Event() release = asyncio.Event() - async def _blocking(plan, request, entrypoint): + async def _blocking(func): # type: ignore[no-untyped-def] started.set() await release.wait() - return {"status": "succeeded", "output": {}} + return func() - monkeypatch.setattr(client_mod, "_run_inline_adapter", _blocking) + monkeypatch.setattr(client_mod, "_call_blocking", _blocking) session = _session() first = asyncio.create_task(session.invoke("turn one")) - await started.wait() # first turn is in flight + await started.wait() - # 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) + await first 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") + + +async def test_run_collapses_through_core_runtime_lifecycle() -> None: + native = FakeNative() + + result = await NativeClient(native).run("agent", input_text="hello") + + assert result["status"] == "succeeded" + assert result["runtime_id"] == "runtime-1" + assert native.requests[0]["input"] == "hello" + assert native.stopped == 1 + + +async def test_run_stops_runtime_when_invoke_raises() -> None: + native = FakeNative() + native.fail_invoke = True + + with pytest.raises(RuntimeError, match="invoke failed"): + await NativeClient(native).run("agent", input_text="hello") + + assert native.stopped == 1 + + +async def test_run_preserves_invoke_error_when_stop_also_raises() -> None: + native = FakeNative() + native.fail_invoke = True + native.fail_stop = True + + with pytest.raises(RuntimeError, match="invoke failed"): + await NativeClient(native).run("agent", input_text="hello") + + assert native.stopped == 1 + + +async def test_run_surfaces_stop_error_after_successful_invoke() -> None: + native = FakeNative() + native.fail_stop = True + + with pytest.raises(RuntimeError, match="stop failed"): + await NativeClient(native).run("agent", input_text="hello") + + assert native.stopped == 1 + + +async def test_run_config_collapses_through_core_runtime_lifecycle() -> None: + native = FakeNative() + config = {"schema_version": "fabric.agent/v1alpha1", "metadata": {"name": "demo"}} + + result = await NativeClient(native).run_config(config, input_text="hello typed") + + assert result["status"] == "succeeded" + assert result["runtime_id"] == "runtime-1" + assert native.requests[0]["input"] == "hello typed" + assert native.stopped == 1 + + +async def test_start_config_creates_session_from_core_runtime_handle() -> None: + native = FakeNative() + config = {"schema_version": "fabric.agent/v1alpha1", "metadata": {"name": "demo"}} + + session = await NativeClient(native).start_config(config) + result = await session.invoke("hello typed session") + + assert session.status is SessionStatus.ACTIVE + assert session.runtime_id == "runtime-1" + assert result["runtime_id"] == "runtime-1" + assert native.requests[0]["input"] == "hello typed session"