From a1c02cf7c7cf2820ab87c807b354c20a3f1f0313 Mon Sep 17 00:00:00 2001 From: Sandy Chapman Date: Mon, 10 Aug 2026 14:21:22 -0300 Subject: [PATCH] fix: keep a completed Deep Agents turn completed when telemetry fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `DeepAgentsRuntime.invoke()` wrapped the Relay plugin/scope context managers and the agent call in one `try/except Exception`. Because a context manager's `__exit__` runs inside that `try`, an exception raised purely while *closing* the telemetry scope was assigned to the invocation `error`, and `normalize_output` reported `completed=false` / `failed=true` for a turn the agent had already finished — final response, tool calls, and workspace changes all intact. This is reachable today. NeMo Relay keeps one process-global LIFO scope stack in a `ContextVar`; LangGraph schedules child tasks with `copy_context()`, which shares that same mutable stack, so concurrent chain callbacks can close out of LIFO order and Relay's validator rejects the pop with RuntimeError: invalid argument: scope handle is not at the top of the stack The stranded child scope then makes the outer `deepagents-request` scope raise on exit. Downstream (NVBug 6562846) this scored 31 otherwise-successful agent-eval trials as adapter invocation failures. Deep Agents is where this is reachable because it is the only adapter that installs a LangChain callback handler, whose nested per-chain scopes are what LangGraph's concurrent tasks close out of order. Separate the two failure domains structurally. `TurnOutcome` carries `error` (the agent failed) and `telemetry_error` (recording the turn failed) as independent fields. `_invoke_agent` owns the invocation domain; `_invoke_with_telemetry` owns the telemetry one and guards nothing but the context managers, so an exception it catches can only be a telemetry fault. Whether the agent got to run is answered by whether an outcome exists, not by a side flag: no outcome means telemetry failed on the way in and there is nothing to preserve, so that stays an invocation failure and the split cannot widen into general failure suppression. Scope and plugin teardown are caught separately, because an exception crossing the plugin's `__aexit__` is replaced by any fault the plugin raises in turn. Artifact collection is covered too. It walks the filesystem after the turn is over, so letting it raise would discard a completed invocation — the same failure mode this change exists to stop. `_telemetry_output` returns that fault instead of raising, and faults from multiple stages are joined rather than the first winning. Relay's scope stack outlives an invocation, so preserving the turn is not enough on its own: a fault that leaves a scope current poisons the runtime, and a later, properly nested turn would otherwise report itself telemetry-clean while running under the stale scope. Compare the current scope handle against the one captured before the turn; a mismatch quarantines the runtime. Quarantined turns keep running and stay completed, but open no request scope, reference no artifacts of their own, and always report degraded. The quarantine survives `stop()`/`start()`, because restarting does not clean the process's scope stack. This is containment, not repair: the Relay middleware attached at start still emits under the stale scope, and the real fix belongs in NeMo Relay. Report a fault as `telemetry.degraded: true` alongside `telemetry.error`, so a consumer has a machine-readable signal rather than a message to parse. On the turn the fault happened `telemetry.error` carries it verbatim; turns that inherit the quarantine report it as `telemetry.quarantine_cause` instead, so a consumer matching per-turn errors does not see one fault reported once per remaining turn. Both keys are absent on a clean run. Signed-off-by: Sandy Chapman --- adapters/deepagents/README.md | 35 ++ .../deepagents/adapter.py | 240 +++++++-- tests/adapters/test_deepagents.py | 508 +++++++++++++++++- tests/integrations/test_relay_scope_leak.py | 185 +++++++ 4 files changed, 922 insertions(+), 46 deletions(-) create mode 100644 tests/integrations/test_relay_scope_leak.py 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"]