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",