From e36e46cd1070bb24009455034a9524528962b4dd Mon Sep 17 00:00:00 2001 From: RichardoMu <44485717+RichardoMrMu@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:25:30 +0000 Subject: [PATCH] fix(strands): tolerate the invocation baggage detach on cross-context close strands-agents 1.60.0 attaches an OpenTelemetry baggage context in Agent.stream_async and detaches it from the async generator's finally block. That token is only valid in the context that started the stream, so an application that closes the stream from another context - for example asyncio.create_task(stream.aclose()), which copies the current context - makes opentelemetry.context.detach log "Failed to detach context" and raise ValueError. The instrumentation now keeps the framework's module-level context_api binding behind a proxy that pairs every attach with a private context probe. When the probe shows that the token belongs to another context the impossible reset is skipped, so closing a stream no longer records that error; when the token is still usable it is reset exactly as before, and attach/detach of tokens the proxy did not hand out are delegated unchanged. Frameworks without the baggage attach (strands-agents < 1.60.0) keep the untouched context API. Verified locally with strands-agents 1.60.0 (37 passed, previously the stream cleanup test failed with the reported ValueError), 1.59.0 and 1.50.2 (35 passed, 2 skipped each). --- .../CHANGELOG.md | 7 + .../README.md | 9 + .../instrumentation/strands/_compat.py | 146 +++++++++++++++ .../instrumentation/strands/_instrumentor.py | 17 ++ .../tests/test_instrumentor.py | 167 +++++++++++++++--- 5 files changed, 324 insertions(+), 22 deletions(-) create mode 100644 instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_compat.py diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md index 06c0d53a3..f382ff34d 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Fixed + +- Tolerate the framework-owned baggage context detach that runs when an + application closes a strand stream from another context, so `strands-agents` + 1.60.0 and later no longer surface a spurious `Failed to detach context` + error while the instrumentation's own context hygiene is unchanged. + ## Version 0.9.0 (2026-09-07) ### Added diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md b/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md index f795d2317..c77347092 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/README.md @@ -62,3 +62,12 @@ invoke_agent (INTERNAL, gen_ai.span.kind=AGENT) the reporter error while its parent spans retain the original business error. - Strands-native `strands.*` metrics remain available, while the LoongSuite OSS product contract derives its unified GenAI metrics from the emitted spans. +- Strands 1.60.0 and later attach an OpenTelemetry baggage context for each + invocation inside `Agent.stream_async` and detach it when the stream is + closed. That baggage only exists in the context that started the stream, so + closing a stream from another context (for example + `asyncio.create_task(stream.aclose())`) cannot restore it. The + instrumentation tolerates that framework-owned detach and keeps its own + context tokens strictly paired, so closing a stream does not surface a + spurious `Failed to detach context` error. + diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_compat.py b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_compat.py new file mode 100644 index 000000000..2fc2536db --- /dev/null +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_compat.py @@ -0,0 +1,146 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Compat shims for Strands Agents behaviour that spans OTel contexts.""" + +import inspect +import logging +from contextvars import ContextVar +from typing import Any + +logger = logging.getLogger(__name__) + +# Private probe used to tell whether a context token can still be reset in the +# current context. A dedicated ContextVar keeps that check off the +# ``opentelemetry.context`` logging path. +_CONTEXT_PROBE: ContextVar = ContextVar("loongsuite_strands_context_probe") + + +class _TolerantContextApi: + """Delegate to the OTel context API and keep one framework detach safe. + + ``strands.agent.agent`` keeps a module level ``context_api`` binding and, + from Strands 1.60.0 on, wraps every invocation in an OpenTelemetry baggage + attach/detach pair. ``context_api.attach`` returns a token that only the + context that created it can reset, and ``opentelemetry.context.detach`` + logs ``Failed to detach context`` *before* it raises ``ValueError`` when the + ambient context changed. An application that closes the stream from another + context - ``asyncio.create_task(stream.aclose())`` copies the current + context - resumes the framework's ``finally`` somewhere else, so tolerating + the exception would still leave that error record behind. + + This proxy pairs every framework attach with a private context probe. When + the probe proves the token can still be reset, the proxy resets it as + before; when it cannot, the impossible reset is skipped, so nothing is + logged and no exception escapes. Attach and detach of tokens the proxy did + not hand out are delegated unchanged. + """ + + def __init__(self, delegate: Any) -> None: + self._delegate = delegate + # Context tokens are unhashable, so pair them by identity and keep the + # token alive for as long as the pairing exists. + self._probes: dict[int, tuple[Any, Any]] = {} + + def __getattr__(self, name: str) -> Any: + return getattr(self._delegate, name) + + def attach(self, context: Any) -> Any: + probe_token = _CONTEXT_PROBE.set(object()) + token = self._delegate.attach(context) + self._probes[id(token)] = (token, probe_token) + return token + + def detach(self, token: Any) -> None: + paired = self._probes.pop(id(token), None) + if paired is None or paired[0] is not token: + self._delegate.detach(token) + return + if _resets_in_current_context(paired[1]): + self._delegate.detach(token) + return + logger.debug( + "Skipped a Strands baggage detach: the token belongs to the " + "context that started the stream", + ) + + +def _resets_in_current_context(probe_token: Any) -> bool: + """Whether the ambient context is the one that created ``probe_token``.""" + try: + _CONTEXT_PROBE.reset(probe_token) + except ValueError: + return False + return True + + +class BaggageContextCompat: + """Keep the framework owned baggage detach from failing on stream close. + + Strands 1.60.0 attaches an OpenTelemetry baggage context inside + ``Agent.stream_async`` and detaches it in the async generator ``finally``. + The baggage travels with the context that starts the stream, so closing a + stream from another context (for example + ``asyncio.create_task(stream.aclose())``) cannot restore it. This shim makes + only that framework owned detach tolerant, keeps the instrumentation's own + ContextVar hygiene intact, and leaves frameworks without the baggage attach + (Strands < 1.60.0) untouched. + """ + + def __init__(self) -> None: + self._module: Any = None + self._original: Any = None + self._proxy: Any = None + + def install(self, module: Any = None) -> None: + """Patch ``module.context_api`` when the framework attaches baggage. + + ``module`` defaults to ``strands.agent.agent`` and is injectable so the + shim can be tested against a stub module. + """ + agent_module = module if module is not None else _agent_module() + original = getattr(agent_module, "context_api", None) + if original is None or not _attaches_invocation_baggage(agent_module): + return + proxy = _TolerantContextApi(original) + agent_module.context_api = proxy + self._module = agent_module + self._original = original + self._proxy = proxy + + def restore(self) -> None: + if ( + self._module is not None + and self._module.context_api is self._proxy + ): + self._module.context_api = self._original + self._module = None + self._original = None + self._proxy = None + + +def _agent_module() -> Any: + from strands.agent import agent as agent_module # noqa: PLC0415 + + return agent_module + + +def _attaches_invocation_baggage(agent_module: Any) -> bool: + """Whether ``Agent`` accepts the 1.60.0 ``baggage_attributes`` argument.""" + agent = getattr(agent_module, "Agent", None) + try: + parameters = inspect.signature(agent.__init__).parameters + except (AttributeError, TypeError, ValueError): + return False + return "baggage_attributes" in parameters diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_instrumentor.py b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_instrumentor.py index 52fdb8e12..8a365c021 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_instrumentor.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/src/opentelemetry/instrumentation/strands/_instrumentor.py @@ -20,6 +20,9 @@ from opentelemetry import context as otel_context from opentelemetry.instrumentation.instrumentor import BaseInstrumentor +from opentelemetry.instrumentation.strands._compat import ( + BaggageContextCompat, +) from opentelemetry.instrumentation.strands._hooks import LoongsuiteHook from opentelemetry.instrumentation.strands._native_telemetry import ( NativeTelemetrySuppression, @@ -48,6 +51,17 @@ def _instrument(self, **kwargs): self._hook = LoongsuiteHook(handler) self._native_telemetry = NativeTelemetrySuppression() + self._baggage_context_compat = BaggageContextCompat() + + try: + self._baggage_context_compat.install() + except Exception: + self._baggage_context_compat.restore() + logger.warning( + "Failed to tolerate the Strands baggage context detach; " + "LoongSuite instrumentation will continue", + exc_info=True, + ) try: self._native_telemetry.install() @@ -160,6 +174,9 @@ def _safe_hook_call(callback: Any, *args: Any) -> None: def _uninstrument(self, **kwargs): self._unwrap_agent_lifecycle() + baggage_context_compat = getattr(self, "_baggage_context_compat", None) + if baggage_context_compat is not None: + baggage_context_compat.restore() native_telemetry = getattr(self, "_native_telemetry", None) if native_telemetry is not None: native_telemetry.restore() diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py b/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py index 294684bac..f2c0bdbf7 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-strands/tests/test_instrumentor.py @@ -14,9 +14,13 @@ import asyncio import builtins +import contextvars +import importlib +import inspect import json import os from collections.abc import AsyncGenerator, Sequence +from types import SimpleNamespace from typing import Any import pytest @@ -25,6 +29,8 @@ from strands.models import Model from strands.telemetry.tracer import get_tracer +from opentelemetry import baggage +from opentelemetry import context as otel_context from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( @@ -42,6 +48,10 @@ from opentelemetry.instrumentation.strands import ( _instrumentor as instrumentor_module, ) +from opentelemetry.instrumentation.strands._compat import ( + BaggageContextCompat, + _TolerantContextApi, +) from opentelemetry.instrumentation.strands._hooks import ( _agent_invocation_usage, _model_provider, @@ -235,28 +245,31 @@ def content_telemetry(monkeypatch): native.tracer = original_native_tracer -def _agent() -> Agent: - return Agent( - model=DeterministicModel( - [ - { - "content": [ - { - "toolUse": { - "toolUseId": "tool-1", - "name": "calculator", - "input": {"expression": "2+2"}, - } - } - ] - }, - {"content": [{"text": "The answer is 4."}]}, - ] - ), - tools=[calculator], - name="test_agent", - retry_strategy=None, - ) +_TOOL_THEN_ANSWER: list[dict[str, Any]] = [ + { + "content": [ + { + "toolUse": { + "toolUseId": "tool-1", + "name": "calculator", + "input": {"expression": "2+2"}, + } + } + ] + }, + {"content": [{"text": "The answer is 4."}]}, +] + + +def _agent(**kwargs: Any) -> Agent: + options: dict[str, Any] = { + "model": DeterministicModel(_TOOL_THEN_ANSWER), + "tools": [calculator], + "name": "test_agent", + "retry_strategy": None, + } + options.update(kwargs) + return Agent(**options) @pytest.mark.asyncio @@ -955,6 +968,116 @@ def test_native_tracer_is_restored_on_uninstrument(telemetry): assert native.tracer is not replacement +def _stub_agent_module() -> Any: + def _init_with_baggage(self, *, baggage_attributes=None) -> None: + return None + + return SimpleNamespace( + context_api=otel_context, + Agent=type("Agent", (), {"__init__": _init_with_baggage}), + ) + + +def _supports_baggage_attributes() -> bool: + return "baggage_attributes" in inspect.signature(Agent.__init__).parameters + + +def test_baggage_context_compat_tolerates_cross_context_detach(): + module = _stub_agent_module() + compat = BaggageContextCompat() + compat.install(module) + previous = otel_context.get_current() + try: + assert module.context_api is not otel_context + token = module.context_api.attach( + otel_context.set_value("strands.compat.probe", "value") + ) + assert otel_context.get_value("strands.compat.probe") == "value" + contextvars.copy_context().run(module.context_api.detach, token) + assert otel_context.get_value("strands.compat.probe") == "value" + finally: + # the skipped detach leaves its value attached; restore the context + otel_context.attach(previous) + compat.restore() + assert module.context_api is otel_context + + +def test_tolerant_context_api_only_tolerates_cross_context_detach(): + class _RaisingDelegate: + def __getattr__(self, name: str) -> Any: + return getattr(otel_context, name) + + @staticmethod + def detach(token: Any) -> None: + raise ValueError("unrelated detach failure") + + proxy = _TolerantContextApi(_RaisingDelegate()) + assert proxy.get_value("strands.compat.delegate.probe") is None + with pytest.raises(ValueError, match="unrelated detach failure"): + proxy.detach(object()) + + +def test_baggage_context_compat_skips_frameworks_without_baggage(): + def _init_without_baggage(self) -> None: + return None + + with_context_api = SimpleNamespace( + context_api=otel_context, + Agent=type("Agent", (), {"__init__": _init_without_baggage}), + ) + without_context_api = SimpleNamespace( + Agent=type("Agent", (), {"__init__": _init_without_baggage}) + ) + + compat = BaggageContextCompat() + try: + compat.install(with_context_api) + assert with_context_api.context_api is otel_context + compat.install(without_context_api) + finally: + compat.restore() + + +def test_instrumentor_patches_and_restores_framework_context_api(telemetry): + instrumentor, _, _ = telemetry + module = importlib.import_module("strands.agent.agent") + if getattr(module, "context_api", None) is None: + pytest.skip("installed strands-agents has no context_api binding") + assert module.context_api is not otel_context + instrumentor.uninstrument() + assert module.context_api is otel_context + + +@pytest.mark.asyncio +async def test_invocation_baggage_attach_is_preserved( + telemetry, caplog, monkeypatch +): + if not _supports_baggage_attributes(): + pytest.skip("installed strands-agents has no baggage_attributes") + module = importlib.import_module("strands.agent.agent") + attached: list[dict[str, str]] = [] + proxy = module.context_api + original_attach = proxy.attach + + def spy_attach(context: Any) -> Any: + attached.append(dict(baggage.get_all(context))) + return original_attach(context) + + monkeypatch.setattr(proxy, "attach", spy_attach) + agent = _agent(baggage_attributes={"tenant": "acme"}) + + async for _ in agent.stream_async("Calculate 2+2"): + pass + + assert attached == [{"tenant": "acme"}] + assert baggage.get_all() == {} + assert not any( + "different Context" in record.getMessage() + or "Failed to detach context" in record.getMessage() + for record in caplog.records + ) + + def test_instrumentation_dependencies(): assert StrandsInstrumentor().instrumentation_dependencies() == ( "strands-agents >= 1.50.2, < 2.0.0",