diff --git a/README.md b/README.md index 91e9242e2..69c51d2d7 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ All instrumentations use [opentelemetry-util-genai](./util/opentelemetry-util-ge | [opentelemetry-instrumentation-genai-openai-agents](./instrumentation/opentelemetry-instrumentation-genai-openai-agents) | openai-agents >= 0.3.3, < 1 | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-genai-openai-agents/) | | [opentelemetry-instrumentation-genai-portkey](./instrumentation/opentelemetry-instrumentation-genai-portkey) | portkey-ai >= 1.0.0, < 3 | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-genai-portkey/) | | [opentelemetry-instrumentation-genai-qwen-agent](./instrumentation/opentelemetry-instrumentation-genai-qwen-agent) | qwen-agent >= 0.0.20, < 1 | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-genai-qwen-agent/) | -| [opentelemetry-instrumentation-genai-qwenpaw](./instrumentation/opentelemetry-instrumentation-genai-qwenpaw) | qwenpaw >= 1.1.0, < 2.0.0; python_version < '3.14' | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-genai-qwenpaw/) | +| [opentelemetry-instrumentation-genai-qwenpaw](./instrumentation/opentelemetry-instrumentation-genai-qwenpaw) | qwenpaw >= 1.1.0, < 3.0.0; python_version < '3.14' | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-genai-qwenpaw/) | | [opentelemetry-instrumentation-genai-smolagents](./instrumentation/opentelemetry-instrumentation-genai-smolagents) | smolagents >= 1.24.0, < 2 | [1.2b0](https://pypi.org/project/opentelemetry-instrumentation-genai-smolagents/) | | [opentelemetry-instrumentation-google-genai](./instrumentation/opentelemetry-instrumentation-google-genai) | google-genai >= 1.32.0, <3 | [1.3b0](https://pypi.org/project/opentelemetry-instrumentation-google-genai/) | diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/.changelog/994.added b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/.changelog/994.added new file mode 100644 index 000000000..e70d5448d --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/.changelog/994.added @@ -0,0 +1 @@ +Support QwenPaw 2.x runtime turns while retaining compatibility with 1.x. diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/README.rst b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/README.rst index db997fc52..f7e6a643d 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/README.rst @@ -9,7 +9,7 @@ OpenTelemetry QwenPaw Instrumentation This library traces user turns handled by `QwenPaw `_, a personal assistant application built on AgentScope. Each turn that goes through -``AgentRunner.query_handler`` produces one ``invoke_agent`` span following +``AgentRunner.query_handler`` (1.x) or ``Runtime.run`` (2.x) produces one ``invoke_agent`` span following the OpenTelemetry GenAI semantic conventions, carrying the agent display name (``gen_ai.agent.name``, when already cached by the runner), the session id (``gen_ai.conversation.id``), and — when content capture is enabled — the @@ -40,11 +40,15 @@ under the ``invoke_agent`` span. Supported versions ------------------ -QwenPaw 1.1.x only. The 2.x `runtime architecture upgrade -`_ -refactored away ``AgentRunner.query_handler``, the seam this package -patches, so the dependency is capped at ``< 2.0.0`` and 2.x support will -be added separately against the new runtime entry point. +QwenPaw >= 1.1.0, < 3.0.0. Version 1.x uses the agent runner; version +2.x uses the runtime request entry point and captures text, reasoning, +tool calls/results, and image/audio/video/file content from request messages and +streamed SSE envelopes. Request normalization and span creation remain lazy; +SDK-generated session IDs propagate to nested GenAI invocations. +Version 2.x support requires ``opentelemetry-util-genai`` 1.4b0 or newer. +QwenPaw 2.x requires Python >= 3.11, +< 3.14. Both entry points produce one internal ``invoke_agent`` span per +turn; model and tool execution remain owned by the underlying libraries. Installation ------------ @@ -113,6 +117,33 @@ Conformance Semantic-convention conformance scenarios live in `tests/conformance `_. +The same ``QwenPawInstrumentor`` selects the entry point for the installed +QwenPaw major version. The 1.x runner instrumentation remains supported. +Tests cover both the oldest and latest releases of each runtime generation: + +.. list-table:: + :header-rows: 1 + + * - QwenPaw + - Entry point + - Conformance tox environment + * - 1.1.0 + - ``AgentRunner.query_handler`` + - ``py311-test-instrumentation-genai-qwenpaw-oldest-conformance`` + * - Latest 1.x + - ``AgentRunner.query_handler`` + - ``py313-test-instrumentation-genai-qwenpaw-v1-conformance`` + * - 2.0.0 + - ``Runtime.run`` + - ``py311-test-instrumentation-genai-qwenpaw-v2-conformance`` + * - Latest 2.x + - ``Runtime.run`` + - ``py313-test-instrumentation-genai-qwenpaw-conformance`` + +Each version runs successful and failed turns through Weaver live-check. +The 2.x scenarios additionally cover reasoning, tool calls/results, and +multimodal request messages. + References ---------- diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/manual/requirements.txt b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/manual/requirements.txt index 6b90c3d39..2d9b2a133 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/manual/requirements.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/manual/requirements.txt @@ -1,4 +1,4 @@ -qwenpaw>=1.1.12.post3,<2.0.0 +qwenpaw>=2.0,<3.0 opentelemetry-sdk~=1.45.2 opentelemetry-exporter-otlp-proto-grpc~=1.45.2 diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/zero-code/requirements.txt b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/zero-code/requirements.txt index 2948ebf87..95460986b 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/zero-code/requirements.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/examples/zero-code/requirements.txt @@ -1,4 +1,4 @@ -qwenpaw>=1.1.12.post3,<2.0.0 +qwenpaw>=2.0,<3.0 opentelemetry-sdk~=1.45.2 opentelemetry-distro~=0.64b0 diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/pyproject.toml index d63172b05..ddf4e2684 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ [project.optional-dependencies] # qwenpaw itself requires Python <3.14, so the marker keeps the extra # installable (as a no-op) on newer interpreters used for tooling. -instruments = ["qwenpaw >= 1.1.0, < 2.0.0; python_version < '3.14'"] +instruments = ["qwenpaw >= 1.1.0, < 3.0.0; python_version < '3.14'"] [project.entry-points.opentelemetry_instrumentor] qwenpaw = "opentelemetry.instrumentation.genai.qwenpaw:QwenPawInstrumentor" diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/__init__.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/__init__.py index ed6186fe0..11f438a90 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/__init__.py @@ -8,7 +8,8 @@ Instrumentation for `QwenPaw `_, a personal assistant application built on AgentScope. -Each user turn handled by ``AgentRunner.query_handler`` is traced as one +Each user turn handled by ``AgentRunner.query_handler`` (1.x) or +``Runtime.run`` (2.x) is traced as one ``invoke_agent`` span following the OpenTelemetry GenAI semantic conventions. @@ -51,6 +52,7 @@ import logging from collections.abc import Collection from importlib import import_module +from importlib.util import find_spec from typing import Any from wrapt import wrap_function_wrapper @@ -62,6 +64,7 @@ from .package import _instruments from .patch import make_query_handler_wrapper +from .runtime import make_runtime_wrapper, observe_context_build from .version import __version__ logger = logging.getLogger(__name__) @@ -69,12 +72,22 @@ __all__ = ["QwenPawInstrumentor"] _RUNNER_MODULE = "qwenpaw.app.runner.runner" +_RUNTIME_MODULE = "qwenpaw.runtime.runtime" + + +def _runtime_v2() -> bool: + try: + return find_spec(_RUNTIME_MODULE) is not None + except ModuleNotFoundError as error: + if error.name not in ("qwenpaw", "qwenpaw.runtime"): + raise + return False class QwenPawInstrumentor(BaseInstrumentor): """An instrumentor for the QwenPaw application runner. - Traces ``AgentRunner.query_handler`` as an ``invoke_agent`` span and + Traces the 1.x agent runner or 2.x runtime as an ``invoke_agent`` span and optionally captures the turn's input/output messages. Model and tool calls are delegated to AgentScope and are not instrumented here. """ @@ -83,7 +96,7 @@ def instrumentation_dependencies(self) -> Collection[str]: return _instruments def _instrument(self, **kwargs: Any) -> None: - """Enable the ``AgentRunner.query_handler`` instrumentation. + """Enable instrumentation for the installed QwenPaw runtime. Args: **kwargs: Optional arguments @@ -102,20 +115,28 @@ def _instrument(self, **kwargs: Any) -> None: completion_hook=kwargs.get("completion_hook") or load_completion_hook(), ) - wrap_function_wrapper( - _RUNNER_MODULE, - "AgentRunner.query_handler", - make_query_handler_wrapper(handler), - ) - logger.debug( - "Instrumented %s.AgentRunner.query_handler", _RUNNER_MODULE - ) + if _runtime_v2(): + wrap_function_wrapper( + _RUNTIME_MODULE, + "Runtime._build_context", + observe_context_build, + ) + wrap_function_wrapper( + _RUNTIME_MODULE, "Runtime.run", make_runtime_wrapper(handler) + ) + else: + wrap_function_wrapper( + _RUNNER_MODULE, + "AgentRunner.query_handler", + make_query_handler_wrapper(handler), + ) def _uninstrument(self, **kwargs: Any) -> None: - """Disable the ``AgentRunner.query_handler`` instrumentation.""" + """Disable instrumentation for the installed QwenPaw runtime.""" del kwargs - runner_module = import_module(_RUNNER_MODULE) - unwrap(runner_module.AgentRunner, "query_handler") - logger.debug( - "Uninstrumented %s.AgentRunner.query_handler", _RUNNER_MODULE - ) + if _runtime_v2(): + runtime = import_module(_RUNTIME_MODULE).Runtime + unwrap(runtime, "run") + unwrap(runtime, "_build_context") + else: + unwrap(import_module(_RUNNER_MODULE).AgentRunner, "query_handler") diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/package.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/package.py index c0c71f1dd..e632575ae 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/package.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/package.py @@ -1,6 +1,6 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -_instruments = ("qwenpaw >= 1.1.0, < 2.0.0; python_version < '3.14'",) +_instruments = ("qwenpaw >= 1.1.0, < 3.0.0; python_version < '3.14'",) _supports_metrics = True diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/runtime.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/runtime.py new file mode 100644 index 000000000..b847f86e5 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/src/opentelemetry/instrumentation/genai/qwenpaw/runtime.py @@ -0,0 +1,362 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Observe QwenPaw 2.x's request and SSE envelope boundary.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncGenerator, Callable, Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from enum import Enum +from typing import cast +from urllib.parse import urlsplit + +from opentelemetry.context import Context, attach, detach, get_current +from opentelemetry.util.genai.handler import TelemetryHandler +from opentelemetry.util.genai.invocation import LocalAgentInvocation +from opentelemetry.util.genai.stream import ( + AbandonedStreamError, + AsyncStreamWrapper, +) +from opentelemetry.util.genai.types import ( + Error, + InputMessage, + MessagePart, + OutputMessage, + ReasoningPart, + TextPart, + ToolCallRequestPart, + ToolCallResponsePart, +) +from opentelemetry.util.genai.utils import image_from_url + +from .utils import non_empty_str + + +def _field(value: object, name: str, default: object = None) -> object: + if isinstance(value, dict): + result = cast("dict[str, object]", value).get(name, default) + else: + result = getattr(value, name, default) + return result.value if isinstance(result, Enum) else result + + +def _items(value: object) -> list[object]: + if isinstance(value, (list, tuple)): + return list(cast("list[object]", value)) + return [] + + +def _json_value(value: object) -> object: + if isinstance(value, str): + try: + return json.loads(value) + except ValueError: + pass + return value + + +def _part( + block: object, message_type: object, *, parse_arguments: bool = True +) -> MessagePart | None: + kind = _field(block, "type") + text = _field(block, "text") + if kind == "text" and isinstance(text, str): + if message_type == "reasoning": + return ReasoningPart(content=text) + return TextPart(content=text) + if kind == "data": + data = _field(block, "data") + call_id = non_empty_str(_field(data, "call_id")) + if message_type in ("function_call", "mcp_tool_call", "plugin_call"): + name = non_empty_str(_field(data, "name")) + if name: + arguments = _field(data, "arguments") + return ToolCallRequestPart( + name=name, + id=call_id, + arguments=_json_value(arguments) + if parse_arguments + else arguments, + ) + if message_type in ( + "function_call_output", + "mcp_tool_call_output", + "plugin_call_output", + ): + return ToolCallResponsePart( + id=call_id, response=_field(data, "output") + ) + uri_field = { + "image": "image_url", + "audio": "data", + "video": "video_url", + "file": "file_url", + }.get(str(kind)) + if uri_field: + uri = _field(block, uri_field) + if isinstance(uri, str) and uri: + audio_format = _field(block, "format") + if ( + kind == "audio" + and isinstance(audio_format, str) + and audio_format + and not urlsplit(uri).scheme + ): + uri = f"data:audio/{audio_format};base64,{uri}" + return image_from_url( + uri, + modality="document" if kind == "file" else str(kind), + ) + return None + + +def _parts(message: object) -> list[MessagePart]: + return [ + part + for block in _items(_field(message, "content")) + if (part := _part(block, _field(message, "type"))) is not None + ] + + +def input_messages_from_request(request: object) -> list[InputMessage]: + """Read envelope messages without normalizing or mutating the request.""" + return [ + InputMessage(role=str(_field(message, "role") or "user"), parts=parts) + for message in _items(_field(request, "input")) + if (parts := _parts(message)) + ] + + +_current_runtime: ContextVar[_RuntimeInvocation | None] = ContextVar( + "qwenpaw_runtime_invocation", default=None +) + + +class _RuntimeInvocation: + """Bind the invocation after the SDK lazily normalizes its request.""" + + _suspended = True + + def __init__( + self, handler: TelemetryHandler, runtime: object, request: object + ) -> None: + self.handler = handler + self.runtime = runtime + self.request = request + self.invocation: LocalAgentInvocation | None = None + self._context: Context | None = None + + def start(self, request: object) -> LocalAgentInvocation: + if self.invocation is None: + workspace = getattr(self.runtime, "workspace", None) + # Workspace.config can load from disk; use only the cached configuration. + config = getattr(workspace, "_config", None) + self.invocation = self.handler.invoke_local_agent( + agent_name=non_empty_str(getattr(config, "name", None)), + conversation_id=non_empty_str(_field(request, "session_id")), + ) + if self.invocation.should_capture_content: + self.invocation.input_messages = input_messages_from_request( + request + ) + return self.invocation + + @contextmanager + def activate(self) -> Iterator[None]: + token = _current_runtime.set(self) + context_token = ( + attach(self._context) if self._context is not None else None + ) + try: + yield + finally: + # Runtime hooks can keep a child span open across SSE yields. + self._context = get_current() + if self.invocation is not None: + self.invocation.suspend() + if context_token is not None: + detach(context_token) + _current_runtime.reset(token) + + +def observe_context_build( + wrapped: Callable[..., object], + instance: object, + args: tuple[object, ...], + kwargs: dict[str, object], +) -> object: + current = _current_runtime.get() + if current is not None: + current.start(args[0] if args else kwargs.get("request")) + return wrapped(*args, **kwargs) + + +class _ContentBlock: + def __init__(self, part: MessagePart) -> None: + self.part: MessagePart = part + self.deltas: list[str] = [] + + def finish(self) -> MessagePart: + part = self.part + if isinstance(part, ToolCallRequestPart): + arguments = part.arguments + if self.deltas: + prefix = arguments if isinstance(arguments, str) else "" + arguments = prefix + "".join(self.deltas) + return ToolCallRequestPart( + name=part.name, id=part.id, arguments=_json_value(arguments) + ) + if self.deltas and isinstance(part, (TextPart, ReasoningPart)): + return type(part)(content=part.content + "".join(self.deltas)) + return part + + +class RuntimeStreamWrapper(AsyncStreamWrapper[object]): + """Track envelope snapshots and deltas while preserving the lazy stream.""" + + def __init__( + self, + stream: AsyncGenerator[object, None], + pending: _RuntimeInvocation, + ) -> None: + super().__init__(stream, pending) + self._self_pending = pending + self._self_blocks: dict[ + str, dict[tuple[object, object], _ContentBlock] + ] = {} + self._self_message_types: dict[str, object] = {} + self._self_message_roles: dict[str, str] = {} + self._self_complete = False + self._self_error: Error | None = None + + def _process_chunk(self, chunk: object) -> None: + invocation = self._self_pending.start(self._self_pending.request) + if _field(chunk, "object") == "response": + session_id = non_empty_str(_field(chunk, "session_id")) + if session_id: + invocation.conversation_id = session_id + status = _field(chunk, "status") + self._self_complete = status == "completed" + if status in ("failed", "cancelled"): + error = _field(chunk, "error") + self._self_error = Error( + type=non_empty_str(_field(error, "code")) or str(status), + message=non_empty_str(_field(error, "message")), + ) + if invocation.should_capture_content: + for message in _items(_field(chunk, "output")): + self._capture_message(message) + elif invocation.should_capture_content: + if _field(chunk, "object") == "message": + self._capture_message(chunk) + elif _field(chunk, "object") == "content": + self._capture_content(chunk) + + def _capture_message(self, message: object) -> None: + role = _field(message, "role") + if role not in ("assistant", "tool"): + return + message_id = non_empty_str(_field(message, "id")) + if message_id is None: + return + message_type = _field(message, "type") + blocks: dict[tuple[object, object], _ContentBlock] = {} + for position, block in enumerate(_items(_field(message, "content"))): + part = _part(block, message_type, parse_arguments=False) + if part is not None: + index = _field(block, "index") + key = ( + _field(block, "type"), + position if index is None else index, + ) + blocks[key] = _ContentBlock(part) + # Snapshots order completed blocks but can omit an unfinished block. + for key, part in self._self_blocks.get(message_id, {}).items(): + blocks.setdefault(key, part) + if blocks: + self._self_blocks[message_id] = blocks + self._self_message_types[message_id] = message_type + self._self_message_roles[message_id] = str(role) + + def _capture_content(self, content: object) -> None: + message_id = non_empty_str(_field(content, "msg_id")) + if message_id is None: + return + # QwenPaw numbers text separately from media, so their indexes can collide. + key = (_field(content, "type"), _field(content, "index", 0)) + blocks = self._self_blocks.get(message_id) + previous = blocks.get(key) if blocks is not None else None + if previous is not None and _field(content, "delta") is True: + delta = None + if isinstance(previous.part, ToolCallRequestPart): + delta = _field(_field(content, "data"), "arguments") + elif isinstance(previous.part, (TextPart, ReasoningPart)): + delta = _field(content, "text") + if isinstance(delta, str): + previous.deltas.append(delta) + return + part = _part( + content, + self._self_message_types.get(message_id), + parse_arguments=False, + ) + if part is not None: + self._self_blocks.setdefault(message_id, {})[key] = _ContentBlock( + part + ) + + def _output_messages(self) -> list[OutputMessage]: + return [ + OutputMessage( + role=self._self_message_roles.get(message_id, "assistant"), + parts=[block.finish() for block in blocks.values()], + ) + for message_id, blocks in self._self_blocks.items() + ] + + def _on_stream_end(self) -> None: + invocation = self._self_pending.invocation + if invocation is None: + return + invocation.output_messages = self._output_messages() + if self._self_error is not None: + invocation.fail(self._self_error) + else: + if self._self_complete: + invocation.finish_reasons = ["stop"] + invocation.stop() + + def _on_stream_error(self, error: BaseException) -> None: + if self._self_pending.invocation is None and isinstance( + error, AbandonedStreamError + ): + return + invocation = self._self_pending.start(self._self_pending.request) + invocation.output_messages = self._output_messages() + invocation.fail(error) + + +def make_runtime_wrapper( + handler: TelemetryHandler, +) -> Callable[..., RuntimeStreamWrapper]: + """Build the wrapt wrapper for ``Runtime.run(request)``.""" + + def runtime_wrapper( + wrapped: Callable[..., AsyncGenerator[object, None]], + instance: object, + args: tuple[object, ...], + kwargs: dict[str, object], + ) -> RuntimeStreamWrapper: + request = args[0] if args else kwargs.get("request") + pending = _RuntimeInvocation(handler, instance, request) + try: + stream = wrapped(*args, **kwargs) + except BaseException as error: + pending.start(request).fail(error) + raise + return RuntimeStreamWrapper(stream, pending) + + return runtime_wrapper diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/invoke_agent.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/invoke_agent.py index 4e7223cc5..48a79edd1 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/invoke_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/invoke_agent.py @@ -4,24 +4,26 @@ """Conformance scenario: one QwenPaw turn traced as ``invoke_agent``. This package emits exactly one semconv operation — ``invoke_agent`` for -each ``AgentRunner.query_handler`` turn. QwenPaw delegates model (LLM) and +each runner (1.x) or runtime (2.x) turn. QwenPaw delegates model (LLM) and tool execution to AgentScope, so no ``chat`` or ``execute_tool`` spans are produced by this package and no scenario covers them — those operations -belong to the AgentScope instrumentation. The runner's command path is -stubbed with a canned assistant reply (the package's unit tests mock the -same seam; no HTTP is involved and no cassette is needed). +belong to the AgentScope instrumentation. A registered slash command drives the real 2.x runtime; +the 1.x command path uses a canned reply. Neither requires HTTP or a cassette. """ from __future__ import annotations import asyncio import importlib +from importlib.metadata import version from typing import Any from tests.harness import ( fake_run_command_path, make_request, + make_runtime, patched_command_path, + runtime_request, user_command_msgs, ) @@ -34,9 +36,6 @@ class InvokeAgentScenario(Scenario): - # The specific attribute values (agent name, conversation id, message - # content) are covered by unit tests; conformance only validates the - # telemetry shape against the semconv registry. expected_spans = {"invoke_agent": 1} expected_metrics = ("gen_ai.invoke_agent.duration",) @@ -48,8 +47,7 @@ def run( logger_provider: LoggerProvider, vcr: Any, ) -> None: - del vcr # the command path is stubbed; no HTTP is involved - runner_module = importlib.import_module("qwenpaw.app.runner.runner") + del vcr with instrument( QwenPawInstrumentor(), tracer_provider=tracer_provider, @@ -57,6 +55,12 @@ def run( meter_provider=meter_provider, content_capture="SPAN_ONLY", ): + if int(version("qwenpaw").split(".")[0]) >= 2: + asyncio.run(_drive_runtime()) + return + runner_module = importlib.import_module( + "qwenpaw.app.runner.runner" + ) with patched_command_path( runner_module, fake_run_command_path("Conformance reply.") ): @@ -69,3 +73,9 @@ async def _drive_one_turn(runner_module: Any) -> None: user_command_msgs(), make_request(session_id="sess-conformance") ): pass + + +async def _drive_runtime() -> None: + runtime = make_runtime("Conformance reply.") + async for _ in runtime.run(runtime_request(session_id="sess-conformance")): + pass diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_error.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_error.py new file mode 100644 index 000000000..b0909cc84 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_error.py @@ -0,0 +1,102 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Validate errors from the 1.x runner and 2.x runtime command paths.""" + +from __future__ import annotations + +import asyncio +import importlib +from importlib.metadata import version +from typing import Any + +from tests.harness import ( + failing_run_command_path, + make_request, + make_runtime, + patched_command_path, + runtime_request, + user_command_msgs, +) + +from opentelemetry.instrumentation.genai.qwenpaw import QwenPawInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test_util_genai.conformance import ( + ExpectedViolation, + Scenario, +) +from opentelemetry.test_util_genai.instrumentor import instrument + + +class RuntimeErrorScenario(Scenario): + expected_spans = {"invoke_agent": 1} + expected_metrics = ("gen_ai.invoke_agent.duration",) + # The pinned registry does not yet define the util's agent exception event. + expected_violations = ( + ExpectedViolation( + advice_id="missing_event", + message_substring="Event 'gen_ai.invoke_agent.exception' does not exist", + ), + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + del vcr + with instrument( + QwenPawInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + asyncio.run(_drive_error()) + + +async def _drive_error() -> None: + if int(version("qwenpaw").split(".")[0]) < 2: + await _drive_runner_error() + return + + from qwenpaw.runtime.slash_command_registry import CommandSpec + + error = ConnectionError("conformance failure") + + async def fail(ctx: Any, args: str) -> None: + del ctx, args + raise error + + runtime = make_runtime() + runtime.workspace.plugins.slash_command_registry.register( + CommandSpec(name="fail", handler=fail) + ) + try: + async for _ in runtime.run(runtime_request("/fail")): + pass + except ConnectionError as caught: + assert caught is error + else: + raise AssertionError("Runtime did not re-raise the command error") + + +async def _drive_runner_error() -> None: + runner_module = importlib.import_module("qwenpaw.app.runner.runner") + error = ConnectionError("conformance failure") + runner = runner_module.AgentRunner(agent_id="conformance-agent") + with patched_command_path(runner_module, failing_run_command_path(error)): + try: + async for _ in runner.query_handler( + user_command_msgs(), make_request() + ): + pass + except ConnectionError as caught: + assert caught is error + else: + raise AssertionError("Runner did not re-raise the command error") diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_messages.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_messages.py new file mode 100644 index 000000000..a3e81f858 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conformance/runtime_messages.py @@ -0,0 +1,107 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Drive rich AgentScope events through the 2.x runtime's envelope encoder.""" + +from __future__ import annotations + +import asyncio +import importlib +from contextlib import ExitStack +from types import SimpleNamespace +from typing import Any +from unittest import mock + +from tests.harness import make_runtime, runtime_request + +from opentelemetry.instrumentation.genai.qwenpaw import QwenPawInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test_util_genai.conformance import Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + + +class RuntimeMessagesScenario(Scenario): + expected_spans = {"invoke_agent": 1} + expected_metrics = ("gen_ai.invoke_agent.duration",) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + del vcr + with instrument( + QwenPawInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + content_capture="SPAN_ONLY", + ): + asyncio.run(_drive_events()) + + +async def _drive_events() -> None: + from agentscope.event import EventType + from qwenpaw.schemas import FileContent, ImageContent, VideoContent + + async def reply_stream(*, inputs: Any) -> Any: + del inputs + events = [ + (EventType.THINKING_BLOCK_START, {}), + (EventType.THINKING_BLOCK_DELTA, {"delta": "Thinking."}), + (EventType.THINKING_BLOCK_END, {}), + (EventType.TOOL_CALL_START, {"tool_call_name": "weather"}), + (EventType.TOOL_CALL_DELTA, {"delta": '{"city":"Paris"}'}), + (EventType.TOOL_CALL_END, {}), + (EventType.TOOL_RESULT_START, {"tool_call_name": "weather"}), + (EventType.TOOL_RESULT_TEXT_DELTA, {"delta": "sunny"}), + (EventType.TOOL_RESULT_END, {}), + (EventType.TEXT_BLOCK_START, {}), + (EventType.TEXT_BLOCK_DELTA, {"delta": "It is sunny."}), + (EventType.TEXT_BLOCK_END, {}), + ] + for kind, extra in events: + yield SimpleNamespace( + type=kind, + block_id="block-1", + tool_call_id="call-1", + metadata={}, + **extra, + ) + + async def close() -> None: + pass + + agent = SimpleNamespace(reply_stream=reply_stream, close=close) + runtime_module = importlib.import_module("qwenpaw.runtime.runtime") + runtime = make_runtime() + runtime.workspace.plugins.modes = [] + request = runtime_request("Describe the weather.") + request.input[0].content.extend( + [ + ImageContent(image_url="https://example.invalid/image.png"), + VideoContent(video_url="https://example.invalid/video.mp4"), + FileContent(file_url="https://example.invalid/file.pdf"), + ] + ) + with ExitStack() as stack: + stack.enter_context( + mock.patch.object( + runtime_module.AgentBuilder, + "build", + mock.AsyncMock(return_value=agent), + ) + ) + if hasattr(runtime_module, "record_agent_activity"): + stack.enter_context( + mock.patch.object( + runtime_module, "record_agent_activity", mock.AsyncMock() + ) + ) + async for _ in runtime.run(request): + pass diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conftest.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conftest.py index 26d43ad5d..17a26ecaf 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conftest.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/conftest.py @@ -7,6 +7,7 @@ from __future__ import annotations import importlib +from importlib.metadata import version import pytest @@ -20,10 +21,16 @@ @pytest.fixture(name="runner_module") def fixture_runner_module(): - try: - return importlib.import_module("qwenpaw.app.runner.runner") - except ImportError: - pytest.skip("qwenpaw is not installed") + if int(version("qwenpaw").split(".")[0]) >= 2: + pytest.skip("AgentRunner is the QwenPaw 1.x entry point") + return importlib.import_module("qwenpaw.app.runner.runner") + + +@pytest.fixture(name="runtime_module") +def fixture_runtime_module(): + if int(version("qwenpaw").split(".")[0]) < 2: + pytest.skip("Runtime is the QwenPaw 2.x entry point") + return importlib.import_module("qwenpaw.runtime.runtime") @pytest.fixture diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/harness.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/harness.py index 4887576cd..d8525fe4f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/harness.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/harness.py @@ -28,9 +28,15 @@ def assistant_reply(text: str) -> Any: def user_command_msgs(text: str = "/stop") -> list[Any]: - from agentscope.message import Msg + from agentscope.message import Msg, TextBlock - return [Msg(name="user", role="user", content=text)] + return [ + Msg( + name="user", + role="user", + content=[TextBlock(type="text", text=text)], + ) + ] def make_request( @@ -96,3 +102,40 @@ async def fake_resolve(self: Any, session_id: Any, query: Any): mock.patch.object(runner_module, "run_command_path", fake) ) yield + + +def runtime_request( + text: str = "/test", session_id: str | None = "sess-2" +) -> Any: + from qwenpaw.schemas import AgentRequest, Message, Role, TextContent + + return AgentRequest( + input=[Message(role=Role.USER, content=[TextContent(text=text)])], + session_id=session_id, + ) + + +def make_runtime(response_text: str = "Runtime reply.") -> Any: + """Use the real 2.x command registry, hooks, runtime and envelope encoder.""" + from qwenpaw.runtime.hooks import HookRegistry + from qwenpaw.runtime.runtime import Runtime + from qwenpaw.runtime.slash_command_registry import ( + CommandSpec, + SlashCommandRegistry, + ) + + async def reply(ctx: Any, args: str) -> Any: + del ctx, args + return assistant_reply(response_text) + + commands = SlashCommandRegistry() + commands.register(CommandSpec(name="test", handler=reply)) + workspace = SimpleNamespace( + agent_id="test-agent", + workspace_dir=None, + _config=SimpleNamespace(name="Friday"), + plugins=SimpleNamespace( + hook_registry=HookRegistry(), slash_command_registry=commands + ), + ) + return Runtime(workspace=workspace, app_services=None) diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt index 4d429b23c..2d9da7dd3 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt @@ -12,11 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -# Latest release of the 1.x line; 2.0 removed `AgentRunner.query_handler` -# and is outside this package's supported range. -qwenpaw==1.1.12.post3 -# AgentScope 1.0.x imports streamablehttp_client, removed in MCP 2. -mcp<2 +qwenpaw==2.2.2.post1 wrapt>=2.5.2 -e util/opentelemetry-util-genai[wrapt] -e instrumentation/opentelemetry-instrumentation-genai-qwenpaw diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v1.txt b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v1.txt new file mode 100644 index 000000000..4153ebbe2 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v1.txt @@ -0,0 +1,21 @@ +# 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. + +# Latest release of the legacy AgentRunner-based runtime. +qwenpaw==1.1.12.post3 +# AgentScope 1.0.x imports streamablehttp_client, removed in MCP 2. +mcp<2 +wrapt>=2.2.2 +-e util/opentelemetry-util-genai +-e instrumentation/opentelemetry-instrumentation-genai-qwenpaw diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v2.txt b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v2.txt new file mode 100644 index 000000000..fe5e103e4 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v2.txt @@ -0,0 +1,18 @@ +# 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. + +qwenpaw==2.0.0 +wrapt>=2.2.2 +-e util/opentelemetry-util-genai +-e instrumentation/opentelemetry-instrumentation-genai-qwenpaw diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py index 6da32fbac..c0b77d333 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py @@ -5,6 +5,7 @@ from __future__ import annotations +from importlib.metadata import version from typing import Any import pytest @@ -21,12 +22,20 @@ ) from .conformance.invoke_agent import InvokeAgentScenario +from .conformance.runtime_error import RuntimeErrorScenario +from .conformance.runtime_messages import RuntimeMessagesScenario @pytest.mark.parametrize( "scenario", [ InvokeAgentScenario(), + RuntimeErrorScenario(), + *( + [RuntimeMessagesScenario()] + if int(version("qwenpaw").split(".")[0]) >= 2 + else [] + ), ], ids=lambda s: type(s).__name__, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentation_scope.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentation_scope.py index c97ff23ca..ab4e5e23c 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentation_scope.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentation_scope.py @@ -1,14 +1,11 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -import pytest - from opentelemetry.instrumentation.genai.qwenpaw import QwenPawInstrumentor from opentelemetry.instrumentation.genai.qwenpaw.version import __version__ from opentelemetry.test_util_genai.scope import TelemetryHandlerScopeTest -@pytest.mark.usefixtures("runner_module") class TestInstrumentationScope(TelemetryHandlerScopeTest): instrumentor_class = QwenPawInstrumentor instrumentation_scope_name = "opentelemetry.instrumentation.genai.qwenpaw" diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentor.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentor.py index 2af3caa20..c7caa3319 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentor.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_instrumentor.py @@ -5,18 +5,25 @@ from __future__ import annotations +from importlib.metadata import Distribution, PackageNotFoundError from types import ModuleType +import pytest + from opentelemetry.instrumentation.genai.qwenpaw import QwenPawInstrumentor def test_instrumentor_targets_qwenpaw_distribution(): assert QwenPawInstrumentor().instrumentation_dependencies() == ( - "qwenpaw >= 1.1.0, < 2.0.0; python_version < '3.14'", + "qwenpaw >= 1.1.0, < 3.0.0; python_version < '3.14'", ) def test_uninstrument_unwraps_query_handler(monkeypatch): + monkeypatch.setattr( + "opentelemetry.instrumentation.genai.qwenpaw._runtime_v2", + lambda: False, + ) runner_module = ModuleType("qwenpaw.app.runner.runner") runner_module.AgentRunner = type("AgentRunner", (), {}) unwrap_calls = [] @@ -38,9 +45,8 @@ def test_uninstrument_unwraps_query_handler(monkeypatch): def test_instrument_uninstrument_roundtrip( - runner_module, tracer_provider, logger_provider, meter_provider + tracer_provider, logger_provider, meter_provider ): - del runner_module instrumentor = QwenPawInstrumentor() instrumentor.instrument( tracer_provider=tracer_provider, @@ -48,3 +54,23 @@ def test_instrument_uninstrument_roundtrip( meter_provider=meter_provider, ) instrumentor.uninstrument() + + +def test_instrument_uninstrument_without_qwenpaw_metadata( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from_name = Distribution.from_name + + def from_name_without_qwenpaw(name: str) -> Distribution: + if name == "qwenpaw": + raise PackageNotFoundError(name) + return from_name(name) + + monkeypatch.setattr(Distribution, "from_name", from_name_without_qwenpaw) + + instrumentor = QwenPawInstrumentor() + instrumentor.instrument(skip_dep_check=True) + try: + assert instrumentor.is_instrumented_by_opentelemetry + finally: + instrumentor.uninstrument() diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_message_mapping.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_message_mapping.py index 973c0aaa3..e3ba1dc75 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_message_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_message_mapping.py @@ -76,7 +76,9 @@ def test_output_message_from_yield_item(): assert output.finish_reason is None assert output.parts[0].content == "done" - user = Msg(name="user", role="user", content="hi") + user = Msg( + name="user", role="user", content=[TextBlock(type="text", text="hi")] + ) assert output_message_from_yield_item((user, True)) is None assert output_message_from_yield_item("not-a-tuple") is None assert output_message_from_yield_item((None, True)) is None diff --git a/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_runtime.py b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_runtime.py new file mode 100644 index 000000000..5a5ac1a3e --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_runtime.py @@ -0,0 +1,944 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Drive the installed QwenPaw 2.x runtime and its real envelope protocol.""" + +from __future__ import annotations + +import asyncio +import gc +import json +from contextlib import suppress +from importlib.metadata import version +from types import AsyncGeneratorType +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from opentelemetry.instrumentation.genai.qwenpaw import QwenPawInstrumentor +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) +from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE +from opentelemetry.test_util_genai.instrumentor import instrument +from opentelemetry.trace import SpanKind, StatusCode, get_current_span +from opentelemetry.util.genai.handler import TelemetryHandler + +from .harness import make_runtime, runtime_request + +pytestmark = pytest.mark.usefixtures("runtime_module") + + +async def drain(stream: Any) -> list[Any]: + return [event async for event in stream] + + +def output_text(span: Any) -> list[str]: + return [ + part["content"] + for message in json.loads( + span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES] + ) + for part in message["parts"] + if part["type"] == "text" + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("as_dict", [False, True]) +@pytest.mark.parametrize("keyword", [False, True]) +async def test_runtime_turn( + instrument_with_content: Any, + span_exporter: Any, + as_dict: Any, + keyword: Any, + metric_reader: Any, +) -> None: + runtime = make_runtime() + request = runtime_request() + if as_dict: + request = request.model_dump(mode="json") + stream = runtime.run(request=request) if keyword else runtime.run(request) + events = await drain(stream) + assert events[-1].status == "completed" + (span,) = span_exporter.get_finished_spans() + assert span.name == "invoke_agent Friday" + assert span.kind is SpanKind.INTERNAL + assert span.status.status_code is StatusCode.UNSET + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "invoke_agent" + assert span.attributes[GenAI.GEN_AI_AGENT_NAME] == "Friday" + assert span.attributes[GenAI.GEN_AI_CONVERSATION_ID] == "sess-2" + assert GenAI.GEN_AI_AGENT_ID not in span.attributes + assert isinstance(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES], str) + assert isinstance(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES], str) + assert json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES])[0][ + "parts" + ] == [{"type": "text", "content": "/test"}] + assert output_text(span) == ["Runtime reply."] + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ("stop",) + assert GenAI.GEN_AI_REQUEST_STREAM not in span.attributes + metrics = [ + metric + for resource in metric_reader.get_metrics_data().resource_metrics + for scope in resource.scope_metrics + for metric in scope.metrics + ] + assert [metric.name for metric in metrics] == [ + "gen_ai.invoke_agent.duration" + ] + (point,) = metrics[0].data.data_points + assert point.count == 1 + assert point.sum >= 0 + assert point.attributes[GenAI.GEN_AI_AGENT_NAME] == "Friday" + + +@pytest.mark.asyncio +async def test_disabled_capture( + instrument_no_content: Any, span_exporter: Any +) -> None: + await drain(make_runtime().run(runtime_request())) + (span,) = span_exporter.get_finished_spans() + assert GenAI.GEN_AI_INPUT_MESSAGES not in span.attributes + assert GenAI.GEN_AI_OUTPUT_MESSAGES not in span.attributes + + +@pytest.mark.asyncio +@pytest.mark.parametrize("session_id", [None, "sess-2"]) +async def test_lazy_context_and_nested_invocations( + instrument_with_content: Any, + span_exporter: Any, + tracer_provider: Any, + session_id: str | None, +) -> None: + runtime = make_runtime() + from qwenpaw.runtime.slash_command_registry import CommandSpec + + handler = TelemetryHandler(tracer_provider=tracer_provider) + + async def nested(ctx: Any, args: Any) -> Any: + from .harness import assistant_reply + + with handler.invoke_local_agent(agent_name="nested"): + pass + return assistant_reply("nested reply") + + runtime.workspace.plugins.slash_command_registry.register( + CommandSpec(name="nested", handler=nested) + ) + tracer = tracer_provider.get_tracer(__name__) + with tracer.start_as_current_span("parent") as parent: + request = runtime_request("/nested", session_id=session_id) + stream = runtime.run(request) + assert request.session_id == session_id + assert isinstance(stream, AsyncGeneratorType) + assert stream.__class__ is AsyncGeneratorType + assert span_exporter.get_finished_spans() == () + assert get_current_span() is parent + async for _ in stream: + assert get_current_span() is parent + assert get_current_span() is parent + spans = {span.name: span for span in span_exporter.get_finished_spans()} + assert spans["invoke_agent Friday"].parent == parent.get_span_context() + assert ( + spans["invoke_agent nested"].parent + == spans["invoke_agent Friday"].context + ) + assert ( + spans["invoke_agent nested"].attributes[GenAI.GEN_AI_CONVERSATION_ID] + == spans["invoke_agent Friday"].attributes[ + GenAI.GEN_AI_CONVERSATION_ID + ] + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error,error_type", + [ + (ConnectionError("transport failed"), "ConnectionError"), + (asyncio.CancelledError(), "asyncio.exceptions.CancelledError"), + ], +) +async def test_runtime_stream_error( + instrument_with_content: Any, + span_exporter: Any, + error: Any, + error_type: Any, +) -> None: + from qwenpaw.runtime.hooks import HookBase + from qwenpaw.runtime.phases import Phase + + class FailAfterResponse(HookBase): + phase = Phase.POST_RESPONSE + name = "fail-after-response" + + async def run(self: Any, ctx: Any) -> None: + raise error + + runtime = make_runtime("partial") + runtime.workspace.plugins.hook_registry.register(FailAfterResponse()) + with pytest.raises(type(error)) as raised: + await drain(runtime.run(runtime_request())) + assert raised.value is error + (span,) = span_exporter.get_finished_spans() + assert span.status.status_code is StatusCode.ERROR + assert span.attributes[ERROR_TYPE] == error_type + assert isinstance(span.attributes[ERROR_TYPE], str) + assert GenAI.GEN_AI_RESPONSE_FINISH_REASONS not in span.attributes + assert "partial" in output_text(span) + + +@pytest.mark.asyncio +async def test_caller_error( + instrument_with_content: Any, span_exporter: Any +) -> None: + error = ValueError("caller failed") + stream = make_runtime().run(runtime_request()) + with pytest.raises(ValueError) as raised: + async with stream: + await anext(stream) + await anext(stream) + raise error + assert raised.value is error + if version("qwenpaw") == "2.0.0": + with suppress(GeneratorExit): + await anext(stream.__wrapped__) + assert stream.ag_frame is None + (span,) = span_exporter.get_finished_spans() + assert span.attributes[ERROR_TYPE] == "ValueError" + assert GenAI.GEN_AI_RESPONSE_FINISH_REASONS not in span.attributes + assert output_text(span) == ["Runtime reply."] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("read", [False, True]) +async def test_early_close( + instrument_with_content: Any, span_exporter: Any, read: Any +) -> None: + stream = make_runtime().run(runtime_request()) + if read: + await anext(stream) + await anext(stream) + if read and version("qwenpaw") == "2.0.0": + # QwenPaw 2.0.0 catches GeneratorExit and yields an error envelope. + # Preserve its close failure rather than masking it in instrumentation. + with pytest.raises(RuntimeError, match="ignored GeneratorExit"): + await stream.aclose() + with suppress(GeneratorExit): + await anext(stream.__wrapped__) + else: + await stream.aclose() + await stream.aclose() + if not read: + assert span_exporter.get_finished_spans() == () + return + (span,) = span_exporter.get_finished_spans() + assert span.status.status_code is ( + StatusCode.ERROR + if read and version("qwenpaw") == "2.0.0" + else StatusCode.UNSET + ) + assert GenAI.GEN_AI_RESPONSE_FINISH_REASONS not in span.attributes + assert stream.ag_frame is None + if read: + assert output_text(span) == ["Runtime reply."] + + +@pytest.mark.asyncio +async def test_abandoned_stream( + instrument_with_content: Any, span_exporter: Any +) -> None: + stream = make_runtime().run(runtime_request()) + await anext(stream) + await anext(stream) + underlying = stream.__wrapped__ + del stream + gc.collect() + (span,) = span_exporter.get_finished_spans() + assert span.attributes[ERROR_TYPE] == "_OTHER" + assert output_text(span) == ["Runtime reply."] + if version("qwenpaw") == "2.0.0": + with pytest.raises(RuntimeError, match="ignored GeneratorExit"): + await underlying.aclose() + with suppress(GeneratorExit): + await anext(underlying) + + +@pytest.mark.asyncio +async def test_runtime_generated_session( + instrument_no_content: Any, span_exporter: Any +) -> None: + events = await drain(make_runtime().run(runtime_request(session_id=None))) + (span,) = span_exporter.get_finished_spans() + assert isinstance(span.attributes[GenAI.GEN_AI_CONVERSATION_ID], str) + assert ( + span.attributes[GenAI.GEN_AI_CONVERSATION_ID] == events[-1].session_id + ) + + +@pytest.mark.asyncio +async def test_completion_hook( + tracer_provider: Any, logger_provider: Any, meter_provider: Any +) -> None: + hook = MagicMock() + with instrument( + QwenPawInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="NO_CONTENT", + completion_hook=hook, + ): + await drain(make_runtime().run(runtime_request())) + hook.on_completion.assert_called_once() + assert ( + hook.on_completion.call_args.kwargs["inputs"][0].parts[0].content + == "/test" + ) + assert ( + hook.on_completion.call_args.kwargs["outputs"][0].parts[0].content + == "Runtime reply." + ) + + +@pytest.mark.asyncio +async def test_partial_deltas_and_message_snapshots( + tracer_provider: Any, span_exporter: Any, monkeypatch: Any +) -> None: + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + make_runtime_wrapper, + ) + + monkeypatch.setenv( + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "SPAN_ONLY" + ) + handler = TelemetryHandler(tracer_provider=tracer_provider) + + async def events(request: object) -> Any: + yield { + "object": "message", + "id": "reply", + "type": "message", + "role": "assistant", + "content": [], + } + yield { + "object": "content", + "msg_id": "reply", + "index": 0, + "type": "text", + "text": "hello ", + "delta": True, + } + yield { + "object": "content", + "msg_id": "reply", + "index": 0, + "type": "text", + "text": "world", + "delta": True, + } + yield { + "object": "content", + "msg_id": "reply", + "index": 0, + "type": "text", + "text": "hello world", + "delta": False, + } + + stream = make_runtime_wrapper(handler)( + events, make_runtime(), (runtime_request(),), {} + ) + await anext(stream) + await anext(stream) + await anext(stream) + await anext(stream) + await stream.aclose() + (span,) = span_exporter.get_finished_spans() + assert output_text(span) == ["hello world"] + assert GenAI.GEN_AI_RESPONSE_FINISH_REASONS not in span.attributes + + +def test_request_message_parts() -> None: + from qwenpaw.schemas import ( + AgentRequest, + FileContent, + ImageContent, + Message, + Role, + TextContent, + VideoContent, + ) + + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + input_messages_from_request, + ) + + request = AgentRequest( + input=[ + Message( + role=Role.USER, + content=[ + TextContent(text="describe"), + ImageContent( + image_url="https://example.invalid/image.png" + ), + VideoContent( + video_url="https://example.invalid/video.mp4" + ), + FileContent(file_url="https://example.invalid/file.pdf"), + ], + ) + ] + ) + (message,) = input_messages_from_request(request) + assert message.role == "user" + assert [ + (part.type, getattr(part, "modality", None)) for part in message.parts + ] == [ + ("text", None), + ("uri", "image"), + ("uri", "video"), + ("uri", "document"), + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "message_type,data,part_type", + [ + ("reasoning", "thinking", "reasoning"), + ( + "function_call", + { + "name": "weather", + "call_id": "call-1", + "arguments": '{"city":"Paris"}', + }, + "tool_call", + ), + ( + "function_call_output", + {"call_id": "call-1", "output": "sunny"}, + "tool_call_response", + ), + ], +) +async def test_output_message_parts( + tracer_provider: Any, + span_exporter: Any, + monkeypatch: Any, + message_type: str, + data: Any, + part_type: str, +) -> None: + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + make_runtime_wrapper, + ) + + monkeypatch.setenv( + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "SPAN_ONLY" + ) + handler = TelemetryHandler(tracer_provider=tracer_provider) + block = ( + {"type": "text", "text": data} + if message_type == "reasoning" + else {"type": "data", "data": data} + ) + + async def events(request: object) -> Any: + yield { + "object": "message", + "id": "reply", + "type": message_type, + "role": "assistant", + "content": [], + } + yield {"object": "content", "msg_id": "reply", "index": 0, **block} + + stream = make_runtime_wrapper(handler)( + events, make_runtime(), (runtime_request(),), {} + ) + await drain(stream) + (span,) = span_exporter.get_finished_spans() + (message,) = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert message["parts"][0]["type"] == part_type + if message_type == "function_call": + assert message["parts"][0]["arguments"] == {"city": "Paris"} + if message_type == "function_call_output": + assert message["parts"][0]["response"] == "sunny" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "part_kind", ["reasoning", "partial_tool_call", "tool_call", "tool_result"] +) +async def test_real_envelope_parts( + instrument_with_content: Any, + tracer_provider: Any, + span_exporter: Any, + monkeypatch: pytest.MonkeyPatch, + part_kind: str, +) -> None: + from types import SimpleNamespace + + from agentscope.event import EventType + from qwenpaw.runtime.envelope import Envelope + + from opentelemetry.instrumentation.genai.qwenpaw import runtime + + parse_arguments = MagicMock(wraps=runtime._json_value) + monkeypatch.setattr(runtime, "_json_value", parse_arguments) + + if part_kind == "reasoning": + event_fields = [ + (EventType.THINKING_BLOCK_START, {}), + (EventType.THINKING_BLOCK_DELTA, {"delta": "hello "}), + (EventType.THINKING_BLOCK_DELTA, {"delta": "world"}), + ] + else: + event_fields = [ + (EventType.TOOL_CALL_START, {"tool_call_name": "weather"}), + (EventType.TOOL_CALL_DELTA, {"delta": '{"city":'}), + (EventType.TOOL_CALL_DELTA, {"delta": '"Paris"}'}), + ] + if part_kind == "tool_result": + event_fields += [ + (EventType.TOOL_CALL_END, {}), + (EventType.TOOL_RESULT_START, {"tool_call_name": "weather"}), + (EventType.TOOL_RESULT_TEXT_DELTA, {"delta": "sunny"}), + (EventType.TOOL_RESULT_END, {}), + ] + + async def events(request: object) -> Any: + envelope = Envelope("sess-2") + for kind, extra in event_fields: + event = SimpleNamespace( + type=kind, + tool_call_id="call-1", + block_id="thinking-1", + metadata={}, + **extra, + ) + async for chunk in envelope.translate_event(event): + yield chunk + if part_kind == "tool_result": + async for chunk in envelope.finalize(): + yield chunk + + handler = TelemetryHandler(tracer_provider=tracer_provider) + stream = runtime.make_runtime_wrapper(handler)( + events, make_runtime(), (runtime_request(),), {} + ) + if part_kind == "tool_result": + await drain(stream) + else: + for _ in range(4 if part_kind == "tool_call" else 3): + await anext(stream) + parse_arguments.assert_not_called() + await stream.aclose() + (span,) = span_exporter.get_finished_spans() + messages = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + parts = [part for msg in messages for part in msg["parts"]] + if part_kind == "reasoning": + assert parts == [{"type": "reasoning", "content": "hello world"}] + else: + arguments = ( + '{"city":' + if part_kind == "partial_tool_call" + else '{"city":"Paris"}' + ) + parse_arguments.assert_called_once_with(arguments) + assert parts[0] == { + "type": "tool_call", + "name": "weather", + "id": "call-1", + "arguments": {"city": "Paris"} + if part_kind != "partial_tool_call" + else '{"city":', + } + if part_kind == "tool_result": + assert messages[-1]["role"] == "tool" + assert parts[-1] == { + "type": "tool_call_response", + "id": "call-1", + "response": "sunny", + } + + +@pytest.mark.asyncio +async def test_normalization_error_restores_context( + instrument_with_content: Any, span_exporter: Any, tracer_provider: Any +) -> None: + from pydantic import ValidationError + + tracer = tracer_provider.get_tracer(__name__) + with tracer.start_as_current_span("parent") as parent: + stream = make_runtime().run({"input": "invalid"}) + with pytest.raises(ValidationError): + await anext(stream) + assert get_current_span() is parent + (span,) = span_exporter.get_finished_spans() + assert span.parent == parent.get_span_context() + assert ( + span.attributes[ERROR_TYPE] + == "pydantic_core._pydantic_core.ValidationError" + ) + + +@pytest.mark.asyncio +async def test_unread_abandoned_stream_does_not_invoke_agent( + instrument_no_content: Any, span_exporter: Any +) -> None: + stream = make_runtime().run(runtime_request()) + del stream + gc.collect() + assert span_exporter.get_finished_spans() == () + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fail", [False, True]) +async def test_hook_context_survives_yields( + instrument_with_content: Any, + tracer_provider: Any, + span_exporter: Any, + fail: bool, +) -> None: + from qwenpaw.runtime.hooks import HookBase, HookResult + from qwenpaw.runtime.phases import Phase + + from opentelemetry import baggage + from opentelemetry.context import attach, detach + + tracer = tracer_provider.get_tracer(__name__) + error = ConnectionError("after response") + + class OpenScope(HookBase): + phase = Phase.PRE_DISPATCH + name = "open-scope" + + async def run(self, ctx: Any) -> HookResult: + scope = tracer.start_as_current_span("hook") + ctx.extras["scope"] = scope + ctx.extras["span"] = scope.__enter__() + ctx.extras["token"] = attach(baggage.set_baggage("hook", "active")) + return HookResult() + + class AfterResponse(HookBase): + phase = Phase.POST_RESPONSE + name = "after-response" + + async def run(self, ctx: Any) -> HookResult: + assert get_current_span() is ctx.extras["span"] + assert baggage.get_baggage("hook") == "active" + with tracer.start_as_current_span("after-response"): + pass + if fail: + raise error + return HookResult() + + class CloseScope(HookBase): + phase = Phase.FINALLY + name = "close-scope" + + async def run(self, ctx: Any) -> HookResult: + assert get_current_span() is ctx.extras["span"] + detach(ctx.extras["token"]) + ctx.extras["scope"].__exit__(None, None, None) + return HookResult() + + runtime = make_runtime() + for hook in (OpenScope(), AfterResponse(), CloseScope()): + runtime.workspace.plugins.hook_registry.register(hook) + with tracer.start_as_current_span("parent") as parent: + try: + async for _ in runtime.run(runtime_request()): + assert get_current_span() is parent + assert baggage.get_baggage("hook") is None + except ConnectionError as caught: + assert fail + assert caught is error + else: + assert not fail + assert get_current_span() is parent + assert baggage.get_baggage("hook") is None + spans = {span.name: span for span in span_exporter.get_finished_spans()} + assert spans["hook"].parent == spans["invoke_agent Friday"].context + assert spans["after-response"].parent == spans["hook"].context + assert spans["invoke_agent Friday"].parent == parent.get_span_context() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", [ConnectionError("failed"), asyncio.CancelledError()] +) +async def test_error_preserves_unfinished_second_block( + instrument_with_content: Any, + tracer_provider: Any, + span_exporter: Any, + error: BaseException, +) -> None: + from types import SimpleNamespace + + from agentscope.event import EventType + from qwenpaw.runtime.envelope import Envelope + + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + make_runtime_wrapper, + ) + + async def events(request: object) -> Any: + envelope = Envelope("sess-2") + for kind, block_id, delta in [ + (EventType.TEXT_BLOCK_START, "first", ""), + (EventType.TEXT_BLOCK_DELTA, "first", "ready"), + (EventType.TEXT_BLOCK_END, "first", ""), + (EventType.TEXT_BLOCK_START, "second", ""), + (EventType.TEXT_BLOCK_DELTA, "second", "unfinished"), + ]: + async for chunk in envelope.translate_event( + SimpleNamespace( + type=kind, + block_id=block_id, + delta=delta, + metadata={}, + ) + ): + yield chunk + async for chunk in envelope.error_envelope( + "failed", "ConnectionError" + ): + yield chunk + raise error + + stream = make_runtime_wrapper( + TelemetryHandler(tracer_provider=tracer_provider) + )( + events, + make_runtime(), + (runtime_request(),), + {}, + ) + with pytest.raises(type(error)) as caught: + await drain(stream) + assert caught.value is error + (span,) = span_exporter.get_finished_spans() + assert output_text(span) == ["ready", "unfinished"] + assert span.status.status_code is StatusCode.ERROR + + +@pytest.mark.parametrize("kind", ["image", "video", "file", "audio"]) +def test_inline_request_media(kind: str) -> None: + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + input_messages_from_request, + ) + from opentelemetry.util.genai.types import BlobPart + + block = {"type": kind} + if kind == "audio": + block.update(data="YWJj", format="wav") + else: + block[f"{kind}_url"] = "data:application/octet-stream;base64,YWJj" + (message,) = input_messages_from_request( + {"input": [{"role": "user", "content": [block]}]} + ) + (part,) = message.parts + assert isinstance(part, BlobPart) + assert part.content == b"abc" + assert part.modality == ("document" if kind == "file" else kind) + assert part.mime_type == ( + "audio/wav" if kind == "audio" else "application/octet-stream" + ) + + +@pytest.mark.parametrize("audio_format", [None, "ogg"]) +def test_audio_only_request(audio_format: str | None) -> None: + from qwenpaw.schemas import AgentRequest, AudioContent, Message, Role + + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + input_messages_from_request, + ) + from opentelemetry.util.genai.types import UriPart + + url = "https://example.invalid/voice.ogg" + (message,) = input_messages_from_request( + AgentRequest( + input=[ + Message( + role=Role.USER, + content=[AudioContent(data=url, format=audio_format)], + ) + ] + ) + ) + assert message.role == "user" + assert message.parts == [ + UriPart(uri=url, modality="audio", mime_type=None) + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mime_type", ["image/png", "video/mp4", "audio/wav"]) +async def test_real_envelope_inline_media( + tracer_provider: Any, + span_exporter: Any, + mime_type: str, +) -> None: + from types import SimpleNamespace + + from agentscope.event import EventType + from qwenpaw.runtime.envelope import Envelope + + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + make_runtime_wrapper, + ) + from opentelemetry.util.genai.types import BlobPart + + async def events(request: object) -> Any: + envelope = Envelope("sess-2") + for kind in ( + EventType.DATA_BLOCK_START, + EventType.DATA_BLOCK_DELTA, + EventType.DATA_BLOCK_END, + ): + async for chunk in envelope.translate_event( + SimpleNamespace( + type=kind, + block_id="media", + media_type=mime_type, + data="YWJj", + metadata={}, + ) + ): + yield chunk + async for chunk in envelope.finalize(): + yield chunk + + hook = MagicMock() + handler = TelemetryHandler( + tracer_provider=tracer_provider, completion_hook=hook + ) + stream = make_runtime_wrapper(handler)( + events, make_runtime(), (runtime_request(),), {} + ) + await drain(stream) + hook.on_completion.assert_called_once() + (message,) = hook.on_completion.call_args.kwargs["outputs"] + assert message.parts == [ + BlobPart( + content=b"abc", + mime_type=mime_type, + modality=mime_type.split("/")[0], + ) + ] + (span,) = span_exporter.get_finished_spans() + assert span.status.status_code is StatusCode.UNSET + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mime_type", ["image/png", "video/mp4", "audio/wav"]) +@pytest.mark.parametrize("ending", ["complete", "close", "error"]) +async def test_real_envelope_mixed_content( + tracer_provider: Any, + span_exporter: Any, + monkeypatch: Any, + mime_type: str, + ending: str, +) -> None: + from types import SimpleNamespace + + from agentscope.event import EventType + from qwenpaw.runtime.envelope import Envelope + + from opentelemetry.instrumentation.genai.qwenpaw.runtime import ( + make_runtime_wrapper, + ) + from opentelemetry.util.genai.types import BlobPart, TextPart + + error = ConnectionError("mixed content interrupted") + + async def events(request: object) -> Any: + envelope = Envelope("sess-2") + for kind, block_id, fields in [ + (EventType.TEXT_BLOCK_START, "before", {}), + (EventType.TEXT_BLOCK_DELTA, "before", {"delta": "before"}), + (EventType.TEXT_BLOCK_END, "before", {}), + (EventType.DATA_BLOCK_START, "media", {"media_type": mime_type}), + (EventType.DATA_BLOCK_DELTA, "media", {"data": "YWJj"}), + (EventType.DATA_BLOCK_END, "media", {}), + (EventType.TEXT_BLOCK_START, "after", {}), + (EventType.TEXT_BLOCK_DELTA, "after", {"delta": "af"}), + (EventType.TEXT_BLOCK_DELTA, "after", {"delta": "ter"}), + ]: + async for chunk in envelope.translate_event( + SimpleNamespace( + type=kind, block_id=block_id, metadata={}, **fields + ) + ): + yield chunk + if ending == "error": + async for chunk in envelope.error_envelope( + "interrupted", "ConnectionError" + ): + yield chunk + raise error + async for chunk in envelope.translate_event( + SimpleNamespace( + type=EventType.TEXT_BLOCK_END, block_id="after", metadata={} + ) + ): + yield chunk + async for chunk in envelope.finalize(): + yield chunk + + monkeypatch.setenv( + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "SPAN_ONLY" + ) + hook = MagicMock() + handler = TelemetryHandler( + tracer_provider=tracer_provider, completion_hook=hook + ) + stream = make_runtime_wrapper(handler)( + events, make_runtime(), (runtime_request(),), {} + ) + if ending == "close": + async for chunk in stream: + if getattr(chunk, "text", None) == "ter": + break + await stream.aclose() + elif ending == "error": + with pytest.raises(ConnectionError) as caught: + await drain(stream) + assert caught.value is error + else: + await drain(stream) + expected = [ + TextPart(content="before"), + BlobPart( + content=b"abc", + mime_type=mime_type, + modality=mime_type.split("/")[0], + ), + TextPart(content="after"), + ] + hook.on_completion.assert_called_once() + (message,) = hook.on_completion.call_args.kwargs["outputs"] + assert message.parts == expected + (span,) = span_exporter.get_finished_spans() + (recorded,) = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert recorded["role"] == "assistant" + assert recorded["parts"] == [ + {"type": "text", "content": "before"}, + { + "type": "blob", + "content": "YWJj", + "mime_type": mime_type, + "modality": mime_type.split("/")[0], + }, + {"type": "text", "content": "after"}, + ] + assert span.status.status_code is ( + StatusCode.ERROR if ending == "error" else StatusCode.UNSET + ) diff --git a/tox.ini b/tox.ini index 47a82e2bd..afab5cfd1 100644 --- a/tox.ini +++ b/tox.ini @@ -89,7 +89,12 @@ envlist = ; qwenpaw supports Python <3.14. py3{11,13}-test-instrumentation-genai-qwenpaw-latest py311-test-instrumentation-genai-qwenpaw-oldest + py313-test-instrumentation-genai-qwenpaw-v1 + py311-test-instrumentation-genai-qwenpaw-v2 py313-test-instrumentation-genai-qwenpaw-conformance + py311-test-instrumentation-genai-qwenpaw-oldest-conformance + py313-test-instrumentation-genai-qwenpaw-v1-conformance + py311-test-instrumentation-genai-qwenpaw-v2-conformance ; instrumentation-genai-weaviate-client ; TODO: write tests (tests/ is empty), add tests/requirements.{oldest,latest}.txt @@ -293,8 +298,14 @@ deps = qwenpaw-latest: {[testenv]test_deps} qwenpaw-latest: {[testenv]pytest_deps} qwenpaw-latest: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt + qwenpaw-v1: {[testenv]test_deps} + qwenpaw-v1: {[testenv]pytest_deps} + qwenpaw-v1: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v1.txt + qwenpaw-v2: {[testenv]test_deps} + qwenpaw-v2: {[testenv]pytest_deps} + qwenpaw-v2: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.v2.txt qwenpaw-conformance: {[testenv]pytest_deps} - qwenpaw-conformance: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt + qwenpaw-conformance-!oldest-!v1-!v2: -r {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/requirements.latest.txt util-genai: {[testenv]test_deps} util-genai: {[testenv]pytest_deps} @@ -362,7 +373,7 @@ commands = test-instrumentation-genai-ollama-{oldest,latest}: pytest --ignore={toxinidir}/instrumentation/opentelemetry-instrumentation-genai-ollama/tests/test_conformance.py {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-ollama/tests --vcr-record=none {posargs} test-instrumentation-genai-ollama-conformance: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-ollama/tests/test_conformance.py --vcr-record=none {posargs} - test-instrumentation-genai-qwenpaw-{oldest,latest}: pytest --ignore={toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests --vcr-record=none {posargs} + test-instrumentation-genai-qwenpaw-{oldest,latest,v1,v2}-!conformance: pytest --ignore={toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests --vcr-record=none {posargs} test-instrumentation-genai-qwenpaw-conformance: pytest {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw/tests/test_conformance.py --vcr-record=none {posargs} test-util-genai: pytest {toxinidir}/util/opentelemetry-util-genai/tests {posargs} @@ -468,6 +479,9 @@ deps = -e {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-openai-agents[instruments] -e {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwen-agent[instruments] -e {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-qwenpaw[instruments] + # QwenPaw 2.x and CrewAI have incompatible transitive dependency ranges. + # Check the shared environment with 1.x; the isolated QwenPaw envs test 2.x. + qwenpaw<2; python_version < "3.14" -e {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-portkey[instruments] -e {toxinidir}/instrumentation/opentelemetry-instrumentation-genai-ollama[instruments] diff --git a/uv.lock b/uv.lock index 3dda14e48..0002b4f9d 100644 --- a/uv.lock +++ b/uv.lock @@ -5195,7 +5195,7 @@ requires-dist = [ { name = "opentelemetry-instrumentation", specifier = ">=0.64b0,<1" }, { name = "opentelemetry-semantic-conventions", specifier = ">=0.64b0,<1" }, { name = "opentelemetry-util-genai", extras = ["wrapt"], editable = "util/opentelemetry-util-genai" }, - { name = "qwenpaw", marker = "python_full_version < '3.14' and extra == 'instruments'", specifier = ">=1.1.0,<2.0.0" }, + { name = "qwenpaw", marker = "python_full_version < '3.14' and extra == 'instruments'", specifier = ">=1.1.0,<3.0.0" }, ] provides-extras = ["instruments"]