diff --git a/adapters/deepagents/README.md b/adapters/deepagents/README.md index 378dc700a..7d1fc7a20 100644 --- a/adapters/deepagents/README.md +++ b/adapters/deepagents/README.md @@ -207,6 +207,41 @@ includes the NeMo Relay Python package. OTel/OpenInference export is available through the relay plugin config; the example provides `with_relay_otel(...)` and `with_relay_openinference(...)` variants. + + Telemetry is a separate failure domain from the agent turn. After the agent has + been invoked, no telemetry fault — a failed scope close, a failed export flush, + or a failed artifact scan — changes the functional outcome: it is reported in the + `telemetry` block instead, as `telemetry.degraded: true` plus a `telemetry.error` + message. A turn the agent completed therefore stays `completed`, and a turn the + agent failed stays failed with its own `error`; the telemetry fault never + overwrites either. Faults from more than one stage are joined into that one + message rather than the first one winning. Both keys are absent on a clean run. + + `telemetry.degraded` is the machine-readable signal to branch on. After a scope + or flush fault the run is degraded but `relay_artifacts` is still populated, + because a partial trajectory is usually worth reading — treat it as untrusted + rather than absent. When artifact collection itself is what failed there is + nothing to reference, so `relay_artifacts` is absent entirely. + + A telemetry failure that happens *before* the agent runs leaves no functional + outcome to preserve, so it is reported as an invocation `error` as well. + + Relay's scope stack lives in the process and outlives a single invocation, so a + fault that leaves a scope current poisons the runtime rather than just the turn. + When that happens the runtime is quarantined: every later turn keeps running and + stays `completed`, but is no longer wrapped in a request scope, reports + `telemetry.degraded: true` with a sticky message, and references no + `relay_artifacts` of its own — the artifacts on disk belong to the earlier turns. + This contains the damage rather than repairing it: the Relay middleware attached + to the agent at start still emits, and those events nest under the stale scope, so + a quarantined runtime's trajectory is untrustworthy rather than empty. The + quarantine deliberately survives `stop()`/`start()`, because restarting the + runtime does not clean the process's scope stack. + + On the turn the fault happened, `telemetry.error` carries it verbatim. On the + turns that inherit the quarantine it appears as `telemetry.quarantine_cause` + instead, so a consumer counting or matching per-turn errors does not see the same + fault reported once per remaining turn. - **Native** (`telemetry.providers.native.config`): the provider config OpenTelemetry/OpenInference exporter is applied and spans export directly to the configured collector, without writing ATOF/ATIF relay artifacts. diff --git a/adapters/deepagents/src/nemo_fabric_adapters/deepagents/adapter.py b/adapters/deepagents/src/nemo_fabric_adapters/deepagents/adapter.py index 8b16617b5..6a8433fc9 100644 --- a/adapters/deepagents/src/nemo_fabric_adapters/deepagents/adapter.py +++ b/adapters/deepagents/src/nemo_fabric_adapters/deepagents/adapter.py @@ -50,6 +50,19 @@ # through harness.settings.deepagents. Executable objects (AgentMiddleware, BaseTool, # Python callables) cannot cross the SDK->JSON->payload boundary and are excluded. DEEPAGENTS_PASSTHROUGH_KEYS = frozenset({"subagents", "interrupt_on"}) +# Appended to the fault that poisoned Relay's scope stack, and then reported on every +# later turn of the same runtime so none of them can look telemetry-clean. Deliberately +# does not claim later turns are untraced: the Relay middleware is attached to the +# compiled agent at start and keeps emitting, so what is actually lost is trustworthy +# nesting, not all telemetry. +_QUARANTINE_NOTE = ( + "telemetry unreliable for the rest of this runtime: an earlier turn left the Relay " + "scope stack dirty, so this turn is not wrapped in a request scope and any events " + "the agent middleware still emits are nested under a stale scope" +) +# Sentinel for "this handle carries no identity", kept distinct from a real ``None`` +# attribute value so an unreadable handle can never compare equal to another one. +_UNREADABLE = object() class AdapterConfigError(RuntimeError): @@ -471,6 +484,8 @@ def __init__(self) -> None: self._relay_scope_type: Any = None self._relay_plugin_config: dict[str, Any] | None = None self._callback_handler_type: Any = None + self._telemetry_quarantine: str | None = None + self._telemetry_quarantine_cause: str | None = None async def start(self, payload: dict[str, Any]) -> None: if self._started: @@ -570,73 +585,138 @@ async def invoke(self, invocation: dict[str, Any]) -> dict[str, Any]: user_message = json.dumps(user_message, sort_keys=True) request_id = request.get("request_id") - result_state: Any = None - events: list[dict[str, Any]] = [] - turn_messages: list[dict[str, Any]] = [] - error: str | None = None resumed = self._completed_invocations > 0 + inherited_quarantine = self._telemetry_quarantine is not None + if self._observability is None: + outcome = await self._invoke_agent(user_message) + else: + outcome = await self._invoke_with_telemetry(user_message, request_id) + + if outcome.error is None: + self._completed_invocations += 1 + + telemetry_runtime, relay_artifacts, collect_error = self._telemetry_output() + return normalize_output( + model_name=self._model_name, + base_url=self._base_url, + runtime_id=self._runtime_id, + thread_id=self._thread_id, + resumed=resumed, + result_state=outcome.result_state, + events=outcome.events or [], + turn_messages=outcome.turn_messages or [], + error=outcome.error, + telemetry_runtime=telemetry_runtime, + relay_artifacts=relay_artifacts, + telemetry_error=_join_faults(outcome.telemetry_error, collect_error), + telemetry_quarantine_cause=( + self._telemetry_quarantine_cause if inherited_quarantine else None + ), + ) + + async def _invoke_with_telemetry( + self, + user_message: str, + request_id: str | None, + ) -> TurnOutcome: + """Run one turn inside the Relay plugin/scope, isolating telemetry faults. + + ``_invoke_agent`` has already absorbed any invocation failure, so an exception + caught here can only have come from telemetry setup or teardown. + """ + + if self._telemetry_quarantine is not None: + # Relay's scope stack is still dirty from an earlier turn. Skip the request + # scope: pushing onto that stack would nest this turn under a stale scope and + # invite another failed pop. + outcome = await self._invoke_agent(user_message) + return outcome._replace(telemetry_error=self._telemetry_quarantine) + + baseline = _current_scope_handle() + outcome: TurnOutcome | None = None + scope_error: str | None = None try: - if self._observability is not None: - callback_handler = self._callback_handler_type() - async with self._relay_plugin.plugin(self._relay_plugin_config): + callback_handler = self._callback_handler_type() + async with self._relay_plugin.plugin(self._relay_plugin_config): + # Caught here rather than left to propagate: an exception crossing the + # plugin's ``__aexit__`` is replaced by any fault the plugin raises in + # turn, which would lose one of the two. + try: with self._relay_scope.scope( "deepagents-request", self._relay_scope_type.Agent, metadata={"nemo_fabric_request_id": request_id}, ): - ( - result_state, - events, - turn_messages, - ) = await invoke_compiled_agent( - self._agent, + outcome = await self._invoke_agent( user_message, - self._thread_id, callbacks=[callback_handler], ) - else: - result_state, events, turn_messages = await invoke_compiled_agent( - self._agent, - user_message, - self._thread_id, - ) - except Exception as exc: # normalized adapter failure - error = f"{type(exc).__name__}: {exc}" - - if error is None: - self._completed_invocations += 1 + except Exception as exc: + scope_error = _error_text(exc) + except Exception as exc: # telemetry lifecycle fault + telemetry_error = _join_faults(scope_error, _error_text(exc)) + else: + telemetry_error = scope_error + + if telemetry_error is not None and not _scope_top_unchanged(baseline): + self._telemetry_quarantine = _QUARANTINE_NOTE + self._telemetry_quarantine_cause = telemetry_error + telemetry_error = _join_faults(telemetry_error, _QUARANTINE_NOTE) + + if outcome is None: + # No outcome means the agent never ran, so there is nothing to preserve. + return TurnOutcome(error=telemetry_error, telemetry_error=telemetry_error) + return outcome._replace(telemetry_error=telemetry_error) + + async def _invoke_agent( + self, + user_message: str, + callbacks: list[Any] | None = None, + ) -> TurnOutcome: + """Run one agent turn, normalizing an invocation failure into an error string.""" - telemetry_runtime, relay_artifacts = self._telemetry_output() - return normalize_output( - model_name=self._model_name, - base_url=self._base_url, - runtime_id=self._runtime_id, - thread_id=self._thread_id, - resumed=resumed, + try: + result_state, events, turn_messages = await invoke_compiled_agent( + self._agent, + user_message, + self._thread_id, + callbacks=callbacks, + ) + except Exception as exc: # normalized adapter failure + return TurnOutcome(error=_error_text(exc)) + return TurnOutcome( result_state=result_state, events=events, turn_messages=turn_messages, - error=error, - telemetry_runtime=telemetry_runtime, - relay_artifacts=relay_artifacts, ) def _telemetry_output( self, - ) -> tuple[dict[str, Any] | None, list[dict[str, str]] | None]: + ) -> tuple[dict[str, Any] | None, list[dict[str, str]] | None, str | None]: + """Return the telemetry block, artifact references, and any collection fault. + + Collecting references walks the filesystem, so it is returned as a fault rather + than raised: raising here would discard an already-completed turn. + """ + if self._observability is None: - return None, None + return None, None, None telemetry_runtime = { "enabled": True, "provider": self._telemetry_provider, "emitter": self._observability.emitter, } - relay_artifacts = ( - common_utils.collect_relay_artifacts(self._observability.plugin_config) - if self._observability.collect_artifacts - else None - ) - return telemetry_runtime, relay_artifacts + if not self._observability.collect_artifacts: + return telemetry_runtime, None, None + if self._telemetry_quarantine is not None: + return telemetry_runtime, None, None + try: + relay_artifacts = common_utils.collect_relay_artifacts( + self._observability.plugin_config + ) + except Exception as exc: + return telemetry_runtime, None, _error_text(exc) + return telemetry_runtime, relay_artifacts, None async def stop(self) -> None: checkpointer = self._checkpointer @@ -736,6 +816,66 @@ class Observability(NamedTuple): collect_artifacts: bool +class TurnOutcome(NamedTuple): + """One agent turn, with its two failure domains kept apart: ``error`` means the + agent failed, ``telemetry_error`` means recording it did. + """ + + result_state: Any = None + events: list[dict[str, Any]] | None = None + turn_messages: list[dict[str, Any]] | None = None + error: str | None = None + telemetry_error: str | None = None + + +def _error_text(exc: BaseException) -> str: + return f"{type(exc).__name__}: {exc}" + + +def _current_scope_handle() -> Any: + """Return Relay's current scope handle, or ``None`` when it cannot be read.""" + + try: + import nemo_relay + + return nemo_relay.scope.get_handle() + except Exception: + return None + + +def _scope_top_unchanged(baseline: Any) -> bool: + """Report whether the scope current now is the one current before the turn. + + This checks the top of the stack, not the whole stack: Relay exposes no depth, so a + fault that left the stack deeper while restoring the top would read as unchanged. + The observed failure strands a child scope on top, which this does catch. Anything + unreadable — a missing handle, or a handle without the identity attribute — counts + as changed, so a Relay rename cannot silently turn the check off. + """ + + if baseline is None: + return False + current = _current_scope_handle() + if current is None: + return False + baseline_uuid = getattr(baseline, "uuid", _UNREADABLE) + current_uuid = getattr(current, "uuid", _UNREADABLE) + if baseline_uuid is _UNREADABLE or current_uuid is _UNREADABLE: + return False + return bool(current_uuid == baseline_uuid) + + +def _join_faults(*faults: str | None) -> str | None: + """Combine telemetry faults into one message; teardown and artifact collection can + both fail in the same turn. + """ + + present = [fault for fault in faults if fault] + if not present: + return None + return "; ".join(present) + + def _relay_dependency_error() -> RuntimeError: return RuntimeError( "telemetry is enabled but a compatible 'nemo-relay' package is not installed; " @@ -786,6 +926,8 @@ def normalize_output( error: str | None, telemetry_runtime: dict[str, Any] | None, relay_artifacts: list[dict[str, str]] | None, + telemetry_error: str | None = None, + telemetry_quarantine_cause: str | None = None, ) -> dict[str, Any]: messages = _extract_messages(result_state) response = _final_response(messages) @@ -812,8 +954,16 @@ def normalize_output( "failed": error is not None, "error": error, } - if telemetry_runtime is not None: - output["telemetry"] = telemetry_runtime + if telemetry_runtime is not None or telemetry_error is not None: + # ``degraded`` marks the referenced artifacts as possibly truncated; both keys + # are absent on a clean run, so the telemetry block keeps its existing shape. + telemetry: dict[str, Any] = dict(telemetry_runtime or {}) + if telemetry_error is not None: + telemetry["degraded"] = True + telemetry["error"] = telemetry_error + if telemetry_quarantine_cause is not None: + telemetry["quarantine_cause"] = telemetry_quarantine_cause + output["telemetry"] = telemetry if relay_artifacts is not None: output["relay_artifacts"] = relay_artifacts return output diff --git a/tests/adapters/test_deepagents.py b/tests/adapters/test_deepagents.py index e43c41314..31f5403a8 100644 --- a/tests/adapters/test_deepagents.py +++ b/tests/adapters/test_deepagents.py @@ -15,6 +15,7 @@ import os import sys import types +import uuid from collections.abc import AsyncIterator from collections.abc import Iterator from pathlib import Path @@ -46,6 +47,21 @@ async def invoke_once(payload: dict[str, Any]) -> dict[str, Any]: await runtime.stop() +async def invoke_twice( + payload: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + """Two ordered invocations on one runtime, which is how Relay state outlives a turn.""" + + runtime = adapter.DeepAgentsRuntime() + await runtime.start(lifecycle_start_payload(payload)) + try: + first = await runtime.invoke(lifecycle_invocation(payload)) + second = await runtime.invoke(lifecycle_invocation(payload)) + return first, second + finally: + await runtime.stop() + + @pytest.fixture(name="fake_sdks", autouse=True) def fake_sdks_fixture(monkeypatch): """Stub the deepagents/langchain/langgraph SDKs with mocks. @@ -204,6 +220,20 @@ async def plugin_ctx(config: object) -> AsyncIterator[None]: class ScopeType: Agent = "agent" + class _Handle: + """Stand-in for nemo_relay ScopeHandle; the adapter compares ``uuid``.""" + + def __init__(self, name: str) -> None: + self.name = name + self.uuid = str(uuid.uuid4()) + + # Relay keeps one LIFO scope stack that outlives an invocation, so the fake keeps + # one too: a scope that fails to unwind stays current for later turns, which is the + # state the adapter has to detect. + stack: list[_Handle] = [_Handle("root")] + calls["stack"] = stack + calls["handle_type"] = _Handle + @contextlib.contextmanager def scope_ctx( name: str, @@ -214,7 +244,14 @@ def scope_ctx( # ``deepagents-request`` Agent scope wraps the invocation. calls.setdefault("scopes", []).append((name, scope_type)) calls.setdefault("scope_metadata", []).append(kwargs.get("metadata")) - yield + stack.append(_Handle(name)) + try: + yield + finally: + stack.pop() + + def get_handle() -> _Handle: + return stack[-1] class NemoRelayDeepAgentsCallbackHandler: def __init__(self, *_args: object, **_kwargs: object) -> None: @@ -228,6 +265,7 @@ def __init__(self, *_args: object, **_kwargs: object) -> None: plugin_mod.plugin = plugin_ctx scope_mod = types.ModuleType("nemo_relay.scope") scope_mod.scope = scope_ctx + scope_mod.get_handle = get_handle relay_root.plugin = plugin_mod relay_root.scope = scope_mod relay_root.ScopeType = ScopeType @@ -386,6 +424,474 @@ async def test_relay_telemetry_wraps_agent_and_reports_artifacts( ) +@pytest.fixture(name="relay_payload") +def relay_payload_fixture(make_payload, monkeypatch): + """Build a payload with Relay telemetry enabled and its plugin config stubbed.""" + + monkeypatch.setattr( + adapter.common_utils, + "load_relay_plugin_config", + lambda _p: {"version": 1, "components": []}, + ) + + def build(tmp_path) -> dict[str, Any]: + payload = make_payload(tmp_path) + payload["telemetry_plan"] = { + "providers": ["relay"], + "relay_enabled": True, + "relay_project": None, + "relay_output_dir": None, + "relay_config": {}, + "native_config": None, + "adapter_outputs": [], + } + return payload + + return build + + +async def test_relay_scope_teardown_failure_keeps_the_invocation_completed( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """A telemetry teardown fault must not rewrite a completed turn into a failure. + + Relay closes scopes in LIFO order, but LangGraph runs concurrent chain callbacks + that can finish out of that order, so closing the outer ``deepagents-request`` + scope can raise after the agent has already produced its final response. + """ + + import contextlib + + @contextlib.contextmanager + def exploding_scope(name: str, scope_type: object, **kwargs: object): + yield + raise RuntimeError( + "invalid argument: scope handle is not at the top of the stack" + ) + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", exploding_scope) + + output = await invoke_once(relay_payload(tmp_path)) + + # The functional outcome is preserved in full. + assert output["completed"] is True + assert output["failed"] is False + assert output["error"] is None + assert output["response"] == "reply to hello" + assert output["message_count"] == 2 + # ...and the telemetry fault is still reported, so a consumer can tell that + # observability degraded and the referenced trajectory may be truncated. + assert output["telemetry"]["degraded"] is True + assert "scope handle is not at the top of the stack" in output["telemetry"]["error"] + + +async def test_relay_plugin_teardown_failure_keeps_the_invocation_completed( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """The plugin context manager is the other half of the telemetry lifecycle. + + A failed export flush on the way out is as much an observability fault as a failed + scope pop, and must be treated the same way. + """ + + import contextlib + + @contextlib.asynccontextmanager + async def exploding_plugin(config: object): + yield + raise RuntimeError("relay plugin flush failed") + + monkeypatch.setattr(sys.modules["nemo_relay.plugin"], "plugin", exploding_plugin) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is True + assert output["failed"] is False + assert output["error"] is None + assert output["response"] == "reply to hello" + assert output["telemetry"]["degraded"] is True + assert output["telemetry"]["error"] == "RuntimeError: relay plugin flush failed" + + +async def test_a_dirty_scope_stack_quarantines_telemetry_for_later_turns( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Relay's scope stack outlives an invocation, so a fault poisons the runtime. + + Preserving the completed turn is not enough on its own: the scope left current by + turn 1 is still current for turn 2, which would otherwise report itself clean while + nesting its trajectory under a stale scope. + """ + + import contextlib + + entered: list[str] = [] + + @contextlib.contextmanager + def leaking_scope(name: str, scope_type: object, **kwargs: object): + # Mirrors the real failure: the child scope is never popped, so the outer close + # raises and the scope pushed here stays current for every later turn. + entered.append(name) + fake_relay["stack"].append(fake_relay["handle_type"](name)) + yield + raise RuntimeError( + "invalid argument: scope handle is not at the top of the stack" + ) + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", leaking_scope) + + first, second = await invoke_twice(relay_payload(tmp_path)) + + assert first["completed"] is True + assert first["telemetry"]["degraded"] is True + + # Turn 2 is functionally fine and must stay completed, but it ran on a dirty stack + # and must not claim clean telemetry. + assert second["completed"] is True + assert second["error"] is None + assert second["response"] == "reply to hello" + assert second["telemetry"]["degraded"] is True + assert "unreliable" in second["telemetry"]["error"] + # Only turn 1 opened a request scope; turn 2 must not push onto the dirty stack. + assert entered == ["deepagents-request"] + # Artifacts on disk belong to turn 1, so turn 2 must not reference them as its own. + assert "relay_artifacts" not in second + + # Turn 1 owns the fault, so it reports it verbatim and needs no separate cause. + assert "not at the top of the stack" in first["telemetry"]["error"] + assert "quarantine_cause" not in first["telemetry"] + # Turn 2 did not fail this way, so the fault appears only as provenance. Repeating + # it in ``error`` would make a consumer count one fault per turn and blame the wrong + # turn for it. + assert "not at the top of the stack" not in second["telemetry"]["error"] + assert "not at the top of the stack" in second["telemetry"]["quarantine_cause"] + + +async def test_an_agent_failure_on_a_quarantined_turn_keeps_both_domains( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Quarantine must not swallow, or be swallowed by, a real agent failure.""" + + import contextlib + + @contextlib.contextmanager + def leaking_scope(name: str, scope_type: object, **kwargs: object): + fake_relay["stack"].append(fake_relay["handle_type"](name)) + yield + raise RuntimeError( + "invalid argument: scope handle is not at the top of the stack" + ) + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", leaking_scope) + payload = relay_payload(tmp_path) + + runtime = adapter.DeepAgentsRuntime() + await runtime.start(lifecycle_start_payload(payload)) + try: + await runtime.invoke(lifecycle_invocation(payload)) + + async def boom(*_args: object, **_kwargs: object): + raise RuntimeError("model call failed") + + monkeypatch.setattr(adapter, "invoke_compiled_agent", boom) + quarantined = await runtime.invoke(lifecycle_invocation(payload)) + finally: + await runtime.stop() + + assert quarantined["completed"] is False + assert quarantined["failed"] is True + assert quarantined["error"] == "RuntimeError: model call failed" + assert quarantined["telemetry"]["degraded"] is True + assert "unreliable" in quarantined["telemetry"]["error"] + + +async def test_quarantine_survives_a_stop_and_restart( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """The dirty stack lives in the process, so a restart inherits it. + + Clearing the quarantine in ``stop()`` alongside the rest of the runtime state would + hand the next runtime a clean bill of health on a stack that is still corrupt. + """ + + import contextlib + + entries = {"count": 0} + + @contextlib.contextmanager + def leaks_once(name: str, scope_type: object, **kwargs: object): + # Only the first turn leaks. A later turn would unwind cleanly and report itself + # clean, so this test fails unless the quarantine itself carried across the + # restart — the stack is still dirty even though nothing new damages it. + entries["count"] += 1 + handle = fake_relay["handle_type"](name) + fake_relay["stack"].append(handle) + if entries["count"] == 1: + yield + raise RuntimeError( + "invalid argument: scope handle is not at the top of the stack" + ) + try: + yield + finally: + fake_relay["stack"].remove(handle) + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", leaks_once) + payload = relay_payload(tmp_path) + + runtime = adapter.DeepAgentsRuntime() + await runtime.start(lifecycle_start_payload(payload)) + first = await runtime.invoke(lifecycle_invocation(payload)) + await runtime.stop() + + await runtime.start(lifecycle_start_payload(payload)) + try: + after_restart = await runtime.invoke(lifecycle_invocation(payload)) + finally: + await runtime.stop() + + assert first["telemetry"]["degraded"] is True + assert after_restart["completed"] is True + assert after_restart["telemetry"]["degraded"] is True + assert "unreliable" in after_restart["telemetry"]["error"] + # The restarted runtime never opened a scope of its own; it inherited the verdict. + assert entries["count"] == 1 + + +async def test_an_unreadable_scope_handle_counts_as_dirty(monkeypatch): + """A safety check must fail closed when it cannot read the state it guards. + + Comparing missing attributes with a ``None`` default made two unreadable handles + look equal, which silently disabled the quarantine. + """ + + class Bare: + """A handle carrying no identity, as a future Relay rename would produce.""" + + monkeypatch.setattr(adapter, "_current_scope_handle", lambda: Bare()) + + assert adapter._scope_top_unchanged(Bare()) is False + + +async def test_a_fault_that_unwinds_cleanly_does_not_quarantine_later_turns( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Quarantine is for a dirty stack, not for any telemetry fault. + + A flush that fails after the scope unwound leaves nothing stale behind, so the next + turn is genuinely clean and must be reported that way. + """ + + import contextlib + + flushes = {"count": 0} + + @contextlib.asynccontextmanager + async def flaky_plugin(config: object): + yield + flushes["count"] += 1 + if flushes["count"] == 1: + raise RuntimeError("relay plugin flush failed") + + monkeypatch.setattr(sys.modules["nemo_relay.plugin"], "plugin", flaky_plugin) + + first, second = await invoke_twice(relay_payload(tmp_path)) + + assert first["telemetry"]["degraded"] is True + assert "degraded" not in second["telemetry"] + assert "error" not in second["telemetry"] + + +async def test_scope_and_plugin_teardown_faults_are_both_reported( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """A scope fault crossing the plugin's ``__aexit__`` is replaced by the plugin's. + + Both stages must survive that, or one lifecycle fault silently disappears. + """ + + import contextlib + + @contextlib.contextmanager + def exploding_scope(name: str, scope_type: object, **kwargs: object): + yield + raise RuntimeError("scope handle is not at the top of the stack") + + @contextlib.asynccontextmanager + async def exploding_plugin(config: object): + yield + raise RuntimeError("relay plugin flush failed") + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", exploding_scope) + monkeypatch.setattr(sys.modules["nemo_relay.plugin"], "plugin", exploding_plugin) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is True + assert output["error"] is None + assert output["telemetry"]["degraded"] is True + assert "scope handle is not at the top of the stack" in output["telemetry"]["error"] + assert "relay plugin flush failed" in output["telemetry"]["error"] + + +async def test_callback_handler_construction_failure_is_a_normalized_failure( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Building the callback handler is telemetry setup, so it must normalize too. + + Constructing it outside the guarded region would let the error escape ``invoke`` + instead of returning the required invocation-failure response. + """ + + def exploding_handler(*_args: object, **_kwargs: object): + raise RuntimeError("callback handler construction failed") + + monkeypatch.setattr( + sys.modules["nemo_relay.integrations.deepagents"], + "NemoRelayDeepAgentsCallbackHandler", + exploding_handler, + ) + invoke_agent = AsyncMock() + monkeypatch.setattr(adapter, "invoke_compiled_agent", invoke_agent) + + output = await invoke_once(relay_payload(tmp_path)) + + invoke_agent.assert_not_awaited() + assert output["completed"] is False + assert output["failed"] is True + assert output["error"] == "RuntimeError: callback handler construction failed" + # The same fault is reported in both domains, as it is for any other setup failure. + assert output["telemetry"]["degraded"] is True + assert ( + output["telemetry"]["error"] + == "RuntimeError: callback handler construction failed" + ) + + +async def test_artifact_collection_failure_does_not_discard_a_completed_turn( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Collecting artifact references walks the filesystem, so it can fail on its own. + + It runs after the turn is over, so letting it raise would throw away a completed + invocation — the same failure mode the invocation/telemetry split exists to stop. + """ + + def boom(_config: object) -> list[dict[str, str]]: + raise OSError("artifact directory disappeared") + + monkeypatch.setattr(adapter.common_utils, "collect_relay_artifacts", boom) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is True + assert output["failed"] is False + assert output["error"] is None + assert output["telemetry"]["degraded"] is True + assert output["telemetry"]["error"] == "OSError: artifact directory disappeared" + + +async def test_teardown_and_artifact_faults_are_both_reported( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Two telemetry faults in one turn: neither may silently swallow the other.""" + + import contextlib + + @contextlib.contextmanager + def exploding_scope(name: str, scope_type: object, **kwargs: object): + yield + raise RuntimeError("scope handle is not at the top of the stack") + + def boom(_config: object) -> list[dict[str, str]]: + raise OSError("artifact directory disappeared") + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", exploding_scope) + monkeypatch.setattr(adapter.common_utils, "collect_relay_artifacts", boom) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is True + assert output["telemetry"]["degraded"] is True + assert "scope handle is not at the top of the stack" in output["telemetry"]["error"] + assert "artifact directory disappeared" in output["telemetry"]["error"] + + +async def test_relay_setup_failure_before_the_agent_runs_stays_an_invocation_failure( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Telemetry that fails on the way *in* leaves no functional outcome to preserve.""" + + import contextlib + + @contextlib.contextmanager + def failing_scope(name: str, scope_type: object, **kwargs: object): + raise RuntimeError("relay scope push failed") + yield # pragma: no cover - unreachable, keeps this a generator + + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", failing_scope) + invoke_agent = AsyncMock() + monkeypatch.setattr(adapter, "invoke_compiled_agent", invoke_agent) + + output = await invoke_once(relay_payload(tmp_path)) + + # The setup fault must short-circuit before any agent request is consumed. + invoke_agent.assert_not_awaited() + assert output["completed"] is False + assert output["failed"] is True + assert output["error"] == "RuntimeError: relay scope push failed" + # The same fault is also a telemetry fault, so it is reported in both domains. + assert output["telemetry"]["degraded"] is True + assert output["telemetry"]["error"] == "RuntimeError: relay scope push failed" + + +async def test_agent_failure_under_relay_is_still_an_invocation_failure( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """Guard against the split widening: a real agent failure must still fail.""" + + async def boom(*_args: object, **_kwargs: object): + raise RuntimeError("model call failed") + + monkeypatch.setattr(adapter, "invoke_compiled_agent", boom) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is False + assert output["failed"] is True + assert output["error"] == "RuntimeError: model call failed" + # A clean telemetry lifecycle leaves the telemetry block untouched. + assert "error" not in output["telemetry"] + assert "degraded" not in output["telemetry"] + + +async def test_agent_failure_and_teardown_failure_keep_their_own_domains( + tmp_path, relay_payload, monkeypatch, fake_sdks, fake_relay +): + """When both fail, the invocation error stays the agent's, not the telemetry one.""" + + import contextlib + + async def boom(*_args: object, **_kwargs: object): + raise RuntimeError("model call failed") + + @contextlib.contextmanager + def exploding_scope(name: str, scope_type: object, **kwargs: object): + yield + raise RuntimeError("scope handle is not at the top of the stack") + + monkeypatch.setattr(adapter, "invoke_compiled_agent", boom) + monkeypatch.setattr(sys.modules["nemo_relay.scope"], "scope", exploding_scope) + + output = await invoke_once(relay_payload(tmp_path)) + + assert output["completed"] is False + assert output["failed"] is True + assert output["error"] == "RuntimeError: model call failed" + assert output["telemetry"]["degraded"] is True + assert "scope handle is not at the top of the stack" in output["telemetry"]["error"] + + async def test_native_telemetry_exports_without_artifacts( tmp_path, make_payload, monkeypatch, fake_sdks, fake_relay ): diff --git a/tests/integrations/test_relay_scope_leak.py b/tests/integrations/test_relay_scope_leak.py new file mode 100644 index 000000000..5d3693de5 --- /dev/null +++ b/tests/integrations/test_relay_scope_leak.py @@ -0,0 +1,185 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration coverage for the Relay scope leak the Deep Agents adapter contains. + +The Deep Agents unit tests monkeypatch a context manager that raises, which is enough to +pin result normalization but cannot reproduce the state leak: the real callback drops a +run from ``_scope_handles`` *before* its pop fails, so the scope stays on Relay's shared +stack and outlives the invocation. These tests drive the actual Relay callback and scope +so that behavior is covered rather than assumed. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import uuid + +import pytest +from nemo_fabric_adapters.deepagents import adapter + +nemo_relay = pytest.importorskip("nemo_relay", reason="requires the nemo-relay extra") + +from nemo_relay.integrations.langchain.callbacks import ( # noqa: E402 + NemoRelayCallbackHandler, +) + + +@pytest.fixture(autouse=True) +def isolated_scope_stack(): + """Give each test its own Relay scope stack. + + These tests strand a scope on purpose, and the stack is process-global, so without + isolation the damage would leak into every test that runs afterwards. ``ContextVar`` + assignment is the only way to install a stack for the current context on Relay 0.6. + """ + + token = nemo_relay._scope_stack_var.set(nemo_relay.create_scope_stack()) + try: + yield + finally: + nemo_relay._scope_stack_var.reset(token) + + +async def _overlapping_chain_runs(handler: NemoRelayCallbackHandler) -> None: + """Close two sibling chain runs out of LIFO order, as LangGraph's tasks do. + + A starts, B starts, A ends, B ends. Relay's stack requires B to close before A, so + A's pop is rejected. + """ + + run_a, run_b = uuid.uuid4(), uuid.uuid4() + a_started, b_started, allow_b_end = (asyncio.Event() for _ in range(3)) + + async def drive_a() -> None: + handler.on_chain_start({}, {"task": "A"}, run_id=run_a, name="A") + a_started.set() + await b_started.wait() + handler.on_chain_end({"done": "A"}, run_id=run_a) + allow_b_end.set() + + async def drive_b() -> None: + await a_started.wait() + handler.on_chain_start({}, {"task": "B"}, run_id=run_b, name="B") + b_started.set() + await allow_b_end.wait() + handler.on_chain_end({"done": "B"}, run_id=run_b) + + await asyncio.gather(asyncio.create_task(drive_a()), asyncio.create_task(drive_b())) + + +async def test_overlapping_chain_runs_strand_a_scope_on_the_shared_stack(caplog): + """The documented failure, reproduced against the installed Relay.""" + + handler = NemoRelayCallbackHandler() + baseline = nemo_relay.scope.get_handle() + + with caplog.at_level(logging.ERROR): + with pytest.raises(RuntimeError, match="not at the top of the stack"): + with nemo_relay.scope.scope( + "deepagents-request", nemo_relay.ScopeType.Agent + ): + await _overlapping_chain_runs(handler) + + # The callback stopped tracking the run before its pop failed... + assert handler._scope_handles == {} + # ...but the scope is still on Relay's stack, and stays current after the turn. + current = nemo_relay.scope.get_handle() + assert current.uuid != baseline.uuid + assert current.name == "A" + + +async def test_the_adapter_detects_the_stranded_scope(): + """``_scope_top_unchanged`` is what turns that leak into a sticky quarantine. + + Without a real stranded scope this cannot be exercised, which is why it lives here + rather than beside the monkeypatched unit tests. + """ + + handler = NemoRelayCallbackHandler() + baseline = adapter._current_scope_handle() + assert baseline is not None + assert adapter._scope_top_unchanged(baseline) is True + + with pytest.raises(RuntimeError, match="not at the top of the stack"): + with nemo_relay.scope.scope("deepagents-request", nemo_relay.ScopeType.Agent): + await _overlapping_chain_runs(handler) + + assert adapter._scope_top_unchanged(baseline) is False + + +async def test_a_clean_turn_leaves_the_stack_restored(): + """The detector must not report damage for a properly nested turn.""" + + baseline = adapter._current_scope_handle() + with nemo_relay.scope.scope("deepagents-request", nemo_relay.ScopeType.Agent): + await asyncio.sleep(0) + + assert adapter._scope_top_unchanged(baseline) is True + + +class _RecordingScope: + """The real Relay scope, with a note of which turns opened one.""" + + def __init__(self) -> None: + self.opened: list[str] = [] + + @contextlib.contextmanager + def scope(self, name: str, scope_type: object, **kwargs: object): + self.opened.append(name) + with nemo_relay.scope.scope(name, scope_type, **kwargs): + yield + + +class _NoopPlugin: + """Stand-in for the Relay plugin, which needs a live gateway config to start.""" + + @contextlib.asynccontextmanager + async def plugin(self, config: object): + yield + + +async def test_a_poisoned_runtime_quarantines_its_next_turn(monkeypatch): + """Two ordered turns end to end: real scope, real callback, real stack. + + The unit tests prove the quarantine against a monkeypatched context manager, which + by construction cannot show that the adapter reads the *same* state Relay actually + leaves behind. This drives the real callback overlap through the adapter's telemetry + path twice, so baseline capture, detection, and the sticky verdict are exercised + against Relay rather than against a stub. Only the plugin and the agent are stubbed: + the plugin needs a live gateway, and the agent is irrelevant to scope bookkeeping. + """ + + async def fake_invoke(agent, user_message, thread_id, callbacks=None): + if callbacks: + # Turn 1 runs the overlapping chain callbacks that strand a scope. + await _overlapping_chain_runs(callbacks[0]) + return {"messages": []}, [], [] + + monkeypatch.setattr(adapter, "invoke_compiled_agent", fake_invoke) + + recording_scope = _RecordingScope() + runtime = adapter.DeepAgentsRuntime() + runtime._agent = object() + runtime._relay_plugin = _NoopPlugin() + runtime._relay_plugin_config = {} + runtime._relay_scope = recording_scope + runtime._relay_scope_type = nemo_relay.ScopeType + runtime._callback_handler_type = NemoRelayCallbackHandler + + first = await runtime._invoke_with_telemetry("hello", "request-1") + second = await runtime._invoke_with_telemetry("hello again", "request-2") + + # Turn 1 completed, reported the real Relay fault, and poisoned the runtime. + assert first.error is None + assert "not at the top of the stack" in first.telemetry_error + assert runtime._telemetry_quarantine is not None + + # Turn 2 is functionally fine, never opens a scope on the dirty stack, and cannot + # report itself telemetry-clean. + assert second.error is None + assert second.telemetry_error == runtime._telemetry_quarantine + assert "not at the top of the stack" not in second.telemetry_error + assert recording_scope.opened == ["deepagents-request"]