diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index 64b0270e73c..a7e680f927f 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -5472,7 +5472,11 @@ async def test_agent_endpoint_correlates_gen_ai_spans_with_supplied_thread_id( monkeypatch.setattr( observability, "OBSERVABILITY_SETTINGS", - SimpleNamespace(ENABLED=True, SENSITIVE_DATA_ENABLED=False), + SimpleNamespace( + ENABLED=True, + SENSITIVE_DATA_ENABLED=False, + use_latest_experimental_gen_ai_semconv=True, + ), ) monkeypatch.setattr(observability, "get_tracer", lambda *args, **kwargs: tracer_provider.get_tracer("test")) diff --git a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py index 1953d4cc7e4..7ee5d835c5f 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_workflow_run.py @@ -133,7 +133,11 @@ async def test_workflow_and_agent_spans_use_supplied_agui_thread_id(monkeypatch: monkeypatch.setattr( observability, "OBSERVABILITY_SETTINGS", - SimpleNamespace(ENABLED=True, SENSITIVE_DATA_ENABLED=False), + SimpleNamespace( + ENABLED=True, + SENSITIVE_DATA_ENABLED=False, + use_latest_experimental_gen_ai_semconv=True, + ), ) monkeypatch.setattr(observability, "get_tracer", lambda *args, **kwargs: tracer_provider.get_tracer("test")) diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index c757f20ba61..55d1ebdbb83 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -35,7 +35,7 @@ from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeAlias, TypeVar, cast, TypeGuard +from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeAlias, TypeVar, cast import msgspec @@ -251,6 +251,7 @@ class _StateTypeRegistration: encoder: StateEncoder decoder: StateDecoder + _STATE_TYPE_REGISTRY: dict[str, _StateTypeRegistration] = {} _STATE_CLASS_REGISTRY: dict[type[Any], _StateTypeRegistration] = {} diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 45fd29864e9..1e4089808ad 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -746,7 +746,10 @@ async def invoke( "response_format", } } - if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + # gen_ai.tool.call.arguments/result were introduced above v1.36.0; only emit them + # as span attributes when that semconv version is active. + emit_tool_call_attrs = OBSERVABILITY_SETTINGS.emit_tool_call_attributes + if emit_tool_call_attrs: attributes.update({ OtelAttr.TOOL_ARGUMENTS: ( json.dumps(serializable_kwargs, default=str, ensure_ascii=False) if serializable_kwargs else "None" @@ -773,8 +776,9 @@ async def invoke( logger.info(f"Function {self.name} succeeded.") if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: result_str = str(result) - span.set_attribute(OtelAttr.TOOL_RESULT, result_str) logger.debug(f"Function result: {result_str}") + if emit_tool_call_attrs: + span.set_attribute(OtelAttr.TOOL_RESULT, result_str) return result try: parsed = parser(result) @@ -786,8 +790,9 @@ async def invoke( logger.info(f"Function {self.name} succeeded.") if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: result_str = "\n".join(c.text or "" for c in parsed if c.type == "text") or str(parsed) - span.set_attribute(OtelAttr.TOOL_RESULT, result_str) logger.debug(f"Function result: {result_str}") + if emit_tool_call_attrs: + span.set_attribute(OtelAttr.TOOL_RESULT, result_str) return parsed finally: duration = (end_time_stamp or perf_counter()) - start_time_stamp diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index fd23410eee5..d95c0660ae1 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -39,6 +39,7 @@ from dotenv import load_dotenv from opentelemetry import metrics, trace +from opentelemetry._logs import get_logger as get_otel_logger from typing_extensions import Sentinel from . import __version__ as version_info @@ -114,6 +115,7 @@ logger = logging.getLogger("agent_framework") +otel_event_logger = get_otel_logger("agent_framework", version_info) INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS: Final[contextvars.ContextVar[set[str] | None]] = contextvars.ContextVar( @@ -195,6 +197,13 @@ def _use_telemetry_conversation_id( # pyright: ignore[reportUnusedFunction] # # This is a workaround, we'll find a generic and better solution - see # https://github.com/open-telemetry/semantic-conventions/issues/1701 +# +# ``_capture_message_events_v1_36`` applies the same 1-microsecond-per-event step directly to the +# timestamps it passes to the OTel event logger, since those events bypass the stdlib ``logging`` +# pipeline and therefore the MessageListTimestampFilter entirely. +MESSAGE_EVENT_TIMESTAMP_STEP_NS: Final[int] = 1_000 + + class MessageListTimestampFilter(logging.Filter): """A filter to increment the timestamp of INFO logs by 1 microsecond.""" @@ -362,6 +371,7 @@ def __str__(self) -> str: "assistant": OtelAttr.ASSISTANT_MESSAGE, "tool": OtelAttr.TOOL_MESSAGE, } + FINISH_REASON_MAP = { "stop": "stop", "content_filter": "content_filter", @@ -385,6 +395,13 @@ def __str__(self) -> str: ("reasoning_tokens", OtelAttr.REASONING_OUTPUT_TOKENS), ) +LATEST_EXPERIMENTAL_GEN_AI_ATTRIBUTES: Final[frozenset[OtelAttr]] = frozenset({ + OtelAttr.CACHE_CREATION_INPUT_TOKENS, + OtelAttr.CACHE_READ_INPUT_TOKENS, + OtelAttr.REASONING_OUTPUT_TOKENS, + OtelAttr.TOOL_DEFINITIONS, +}) + # region Telemetry utils @@ -715,12 +732,21 @@ def create_metric_views() -> list[View]: ] +# Token recognized in the OTEL_SEMCONV_STABILITY_OPT_IN env var that opts into the GenAI +# conventions above the v1.36.0 baseline (referred to here as "latest", since even the +# baseline is not itself a stable release; see +# https://github.com/open-telemetry/semantic-conventions/blob/v1.37.0/docs/gen-ai). +GEN_AI_LATEST_EXPERIMENTAL_OPT_IN: Final[str] = "gen_ai_latest_experimental" + + class _ObservabilitySettingsData(TypedDict, total=False): """TypedDict schema for observability settings fields.""" enable_instrumentation: bool | None enable_sensitive_data: bool | None enable_console_exporters: bool | None + enable_message_events: bool | None + otel_semconv_stability_opt_in: str | None vs_code_extension_port: int | None @@ -754,6 +780,18 @@ class ObservabilitySettings: Can be set via environment variable ENABLE_SENSITIVE_DATA. enable_console_exporters: Enable console exporters for traces, logs, and metrics. Default is False. Can be set via environment variable ENABLE_CONSOLE_EXPORTERS. + enable_message_events: Emit the baseline v1.36.0 GenAI message events (``gen_ai.system.message``, + ``gen_ai.user.message``, ``gen_ai.assistant.message``, ``gen_ai.tool.message``, ``gen_ai.choice``) + for model invocation. Default is True. Can be set via environment variable ENABLE_MESSAGE_EVENTS. + Only takes effect when sensitive data capture is enabled. + otel_semconv_stability_opt_in: A comma-separated list of category-specific values, following the + standard OpenTelemetry comma-separated opt-in list format, currently only containing a single + token ``"gen_ai_latest_experimental"``. v1.36.0 is the OTel-recommended baseline; every + version above it is referred to here as "latest" (per OTel's own stability warning, even the + baseline is not a stable release of the GenAI conventions). The default, unlike upstream + OpenTelemetry which defaults to the baseline, ``"gen_ai_latest_experimental"`` selects the latest + conventions above v1.36.0; a list that omits that token (e.g. ``""``) selects the v1.36.0 + conventions instead. Can be set via environment variable OTEL_SEMCONV_STABILITY_OPT_IN. vs_code_extension_port: The port the AI Toolkit or Microsoft Foundry VS Code extensions are listening on. Default is None. Can be set via environment variable VS_CODE_EXTENSION_PORT. @@ -800,6 +838,9 @@ def __init__(self, **kwargs: Any) -> None: ) self.enable_console_exporters: bool = data.get("enable_console_exporters") or False + message_events_value = data.get("enable_message_events") + self.enable_message_events: bool = True if message_events_value is None else message_events_value + self.otel_semconv_stability_opt_in: str | None = data.get("otel_semconv_stability_opt_in") self.vs_code_extension_port: int | None = data.get("vs_code_extension_port") self.env_file_path = env_file_path self.env_file_encoding = env_file_encoding @@ -850,6 +891,23 @@ def enable_sensitive_data(self, value: bool) -> None: return self._enable_sensitive_data = value + @property + def use_latest_experimental_gen_ai_semconv(self) -> bool: + """Whether to emit the GenAI semantic conventions above the v1.36.0 baseline. + + v1.36.0 is the OTel-recommended baseline; every version above it is referred to here as + "latest". + + Computed from ``otel_semconv_stability_opt_in`` (env var ``OTEL_SEMCONV_STABILITY_OPT_IN``), a + comma-separated opt-in list per the standard OpenTelemetry format. Agent Framework defaults this + to True (opted into the conventions above v1.36.0) when the setting is unset, which differs from + upstream OpenTelemetry's default of retaining the baseline conventions. + """ + if self.otel_semconv_stability_opt_in is None: + return True + tokens = {token.strip() for token in self.otel_semconv_stability_opt_in.split(",")} + return GEN_AI_LATEST_EXPERIMENTAL_OPT_IN in tokens + @property def ENABLED(self) -> bool: """Check if model diagnostics are enabled. @@ -866,6 +924,15 @@ def SENSITIVE_DATA_ENABLED(self) -> bool: """ return self.enable_instrumentation and self.enable_sensitive_data + @property + def emit_tool_call_attributes(self) -> bool: + """Whether to emit gen_ai.tool.call.arguments/result on execute_tool spans. + + These attributes were introduced above v1.36.0, so they require both sensitive-data + capture and the semconv version that supports them. + """ + return self.SENSITIVE_DATA_ENABLED and self.use_latest_experimental_gen_ai_semconv + @property def is_setup(self) -> bool: """Check if the setup has been executed.""" @@ -1234,6 +1301,8 @@ def configure_otel_providers( *, enable_sensitive_data: bool | None = None, enable_console_exporters: bool | None = None, + enable_message_events: bool | None = None, + otel_semconv_stability_opt_in: str | None = None, exporters: list[LogRecordExporter | SpanExporter | MetricExporter] | None = None, views: list[View] | None = None, vs_code_extension_port: int | None = None, @@ -1274,6 +1343,13 @@ def configure_otel_providers( the environment variable ENABLE_SENSITIVE_DATA if set. Default is None. enable_console_exporters: Enable console exporters for traces, logs, and metrics. Overrides the environment variable ENABLE_CONSOLE_EXPORTERS if set. Default is None. + enable_message_events: Emit the baseline v1.36.0 GenAI message events (``gen_ai.system.message``, etc.) + for model invocation. Overrides the environment variable ENABLE_MESSAGE_EVENTS if set. Default is + None, which resolves to True (events enabled). + otel_semconv_stability_opt_in: a comma-separated list of category-specific values (see + ``ObservabilitySettings.otel_semconv_stability_opt_in`` for the full explanation). Overrides the + environment variable OTEL_SEMCONV_STABILITY_OPT_IN if set. Default is None, which resolves to the + conventions above the v1.36.0 baseline. exporters: A list of custom exporters for logs, metrics or spans, or any combination. These will be added in addition to exporters configured via environment variables. Default is None. @@ -1370,6 +1446,10 @@ def configure_otel_providers( settings_kwargs["enable_sensitive_data"] = enable_sensitive_data if enable_console_exporters is not None: settings_kwargs["enable_console_exporters"] = enable_console_exporters + if enable_message_events is not None: + settings_kwargs["enable_message_events"] = enable_message_events + if otel_semconv_stability_opt_in is not None: + settings_kwargs["otel_semconv_stability_opt_in"] = otel_semconv_stability_opt_in if vs_code_extension_port is not None: settings_kwargs["vs_code_extension_port"] = vs_code_extension_port @@ -1377,6 +1457,8 @@ def configure_otel_providers( OBSERVABILITY_SETTINGS.enable_instrumentation = updated_settings.enable_instrumentation OBSERVABILITY_SETTINGS.enable_sensitive_data = updated_settings.enable_sensitive_data OBSERVABILITY_SETTINGS.enable_console_exporters = updated_settings.enable_console_exporters + OBSERVABILITY_SETTINGS.enable_message_events = updated_settings.enable_message_events + OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = updated_settings.otel_semconv_stability_opt_in OBSERVABILITY_SETTINGS.vs_code_extension_port = updated_settings.vs_code_extension_port OBSERVABILITY_SETTINGS.env_file_path = updated_settings.env_file_path OBSERVABILITY_SETTINGS.env_file_encoding = updated_settings.env_file_encoding @@ -1393,6 +1475,16 @@ def configure_otel_providers( if enable_console_exporters is not None else _read_bool_env("ENABLE_CONSOLE_EXPORTERS") ) + OBSERVABILITY_SETTINGS.enable_message_events = ( + enable_message_events + if enable_message_events is not None + else _read_bool_env("ENABLE_MESSAGE_EVENTS", default=True) + ) + OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = ( + otel_semconv_stability_opt_in + if otel_semconv_stability_opt_in is not None + else os.getenv("OTEL_SEMCONV_STABILITY_OPT_IN") + ) OBSERVABILITY_SETTINGS.vs_code_extension_port = ( vs_code_extension_port if vs_code_extension_port is not None else _read_int_env("VS_CODE_EXTENSION_PORT") ) @@ -1562,14 +1654,21 @@ def get_response( if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages and span.is_recording(): system_instructions = _get_instructions_from_options(opts) - _capture_current_agent_system_instructions( + _capture_current_agent_system_instructions_latest_experimental( agent_span, span, system_instructions, ) - _capture_messages( + # Activate the span so the OTel event logger correlates these input events + # with this chat span's trace/span id rather than the ambient parent context. + with _activate_span(span): + _capture_message_events_v1_36( + provider_name=provider_name, + messages=messages, + system_instructions=system_instructions, + ) + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=messages, system_instructions=system_instructions, ) @@ -1645,13 +1744,18 @@ async def _finalize_stream() -> None: and response.messages and span.is_recording() ): - finish_reason = cast( - "FinishReason | None", - response.finish_reason if response.finish_reason in FINISH_REASON_MAP else None, - ) - _capture_messages( + finish_reason = _get_response_finish_reason(response) + # Activate the span: this cleanup hook runs after the final pull has + # exited its _activate_span context, so it wouldn't otherwise be current. + with _activate_span(span): + _capture_message_events_v1_36( + provider_name=provider_name, + messages=response.messages, + finish_reason=finish_reason, + output=True, + ) + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=response.messages, finish_reason=finish_reason, output=True, @@ -1681,17 +1785,21 @@ async def _get_response() -> ChatResponse: with _get_span(attributes=attributes, span_name_attribute=OtelAttr.REQUEST_MODEL) as span: if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages and span.is_recording(): system_instructions = _get_instructions_from_options(opts) - _capture_current_agent_system_instructions( + _capture_current_agent_system_instructions_latest_experimental( agent_span, span, system_instructions, ) - _capture_messages( - span=span, + _capture_message_events_v1_36( provider_name=provider_name, messages=messages, system_instructions=system_instructions, ) + _capture_message_span_attributes_latest_experimental( + span=span, + messages=messages, + system_instructions=system_instructions, + ) start_time_stamp = perf_counter() try: response = cast( @@ -1721,13 +1829,15 @@ async def _get_response() -> ChatResponse: ) _mark_inner_response_telemetry_captured(response) if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and response.messages and span.is_recording(): - finish_reason = cast( - "FinishReason | None", - response.finish_reason if response.finish_reason in FINISH_REASON_MAP else None, + finish_reason = _get_response_finish_reason(response) + _capture_message_events_v1_36( + provider_name=provider_name, + messages=response.messages, + finish_reason=finish_reason, + output=True, ) - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=response.messages, finish_reason=finish_reason, output=True, @@ -1880,12 +1990,13 @@ def _trace_agent_invocation( inner_response_telemetry_captured_fields: set[str] = set() inner_response_telemetry_captured_fields_token: contextvars.Token[set[str] | None] | None = None inner_accumulated_usage_token: contextvars.Token[UsageDetails | None] | None = None + # Agent Framework's agents run in-process (the actual network call happens on a nested + # chat span), so invoke_agent spans use the default INTERNAL kind. span = _start_streaming_span(attributes, OtelAttr.AGENT_NAME) if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages and span.is_recording(): - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=messages, system_instructions=_get_instructions_from_options(dict(merged_options)), ) @@ -1957,9 +2068,8 @@ async def _finalize_stream() -> None: and response.messages and span.is_recording() ): - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=response.messages, output=True, ) @@ -2023,9 +2133,8 @@ async def _run() -> AgentResponse[Any]: with _get_span(attributes=attributes, span_name_attribute=OtelAttr.AGENT_NAME) as span: try: if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages and span.is_recording(): - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=messages, system_instructions=_get_instructions_from_options(dict(merged_options)), ) @@ -2056,9 +2165,8 @@ async def _run() -> AgentResponse[Any]: and response.messages and span.is_recording() ): - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name=provider_name, messages=response.messages, output=True, ) @@ -2296,14 +2404,18 @@ def _activate_span(span: trace.Span) -> Generator[None]: def _get_span( attributes: dict[str, Any], span_name_attribute: str, + kind: trace.SpanKind = trace.SpanKind.INTERNAL, ) -> Generator[trace.Span, Any, Any]: - """Start a span for a agent run. + """Start a span for an agent run. + + Agent Framework's agents run in-process (the actual network call happens on a nested + chat span), so invoke_agent spans use the default INTERNAL kind. Note: `attributes` must contain the `span_name_attribute` key. """ operation = attributes.get(OtelAttr.OPERATION, "operation") span_name = attributes.get(span_name_attribute, "unknown") - span = get_tracer().start_span(f"{operation} {span_name}") + span = get_tracer().start_span(f"{operation} {span_name}", kind=kind) span.set_attributes(attributes) with trace.use_span( span=span, @@ -2314,7 +2426,11 @@ def _get_span( yield current_span -def _start_streaming_span(attributes: dict[str, Any], span_name_attribute: str) -> trace.Span: +def _start_streaming_span( + attributes: dict[str, Any], + span_name_attribute: str, + kind: trace.SpanKind = trace.SpanKind.INTERNAL, +) -> trace.Span: """Start a non-current span for a streaming operation. Unlike :func:`_get_span`, the returned span is not attached to the current @@ -2332,7 +2448,7 @@ def _start_streaming_span(attributes: dict[str, Any], span_name_attribute: str) """ operation = attributes.get(OtelAttr.OPERATION, "operation") span_name = attributes.get(span_name_attribute, "unknown") - span = get_tracer().start_span(f"{operation} {span_name}") + span = get_tracer().start_span(f"{operation} {span_name}", kind=kind) span.set_attributes(attributes) return span @@ -2577,7 +2693,6 @@ def _otel_tool_definition(type_value: str, name_value: str, source: Mapping[str, OTEL_ATTR_MAP: dict[str | tuple[str, ...], tuple[str, Callable[[Any], Any] | None, bool, Any]] = { "choice_count": (OtelAttr.CHOICE_COUNT, None, False, 1), "operation_name": (OtelAttr.OPERATION, None, False, None), - "system_name": (OtelAttr.SYSTEM, None, False, None), "provider_name": (OtelAttr.PROVIDER_NAME, None, False, None), "service_url": (OtelAttr.ADDRESS, None, False, None), "conversation_id": (OtelAttr.CONVERSATION_ID, None, True, None), @@ -2613,6 +2728,14 @@ def _otel_tool_definition(type_value: str, name_value: str, source: Mapping[str, } +def _provider_name_attr() -> OtelAttr: + """Return the provider-identifying attribute for the active GenAI semconv version. + + ``gen_ai.system`` was renamed to ``gen_ai.provider.name`` in the conventions above v1.36.0. + """ + return OtelAttr.PROVIDER_NAME if OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv else OtelAttr.SYSTEM + + def _get_span_attributes(**kwargs: Any) -> dict[str, Any]: """Get the span attributes from a kwargs dictionary.""" attributes: dict[str, Any] = {} @@ -2625,6 +2748,12 @@ def _get_span_attributes(**kwargs: Any) -> dict[str, Any]: check_options, default_value, ) in OTEL_ATTR_MAP.items(): + if ( + otel_key in LATEST_EXPERIMENTAL_GEN_AI_ATTRIBUTES + and not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv + ): + continue + # Normalize to tuple of keys keys = (source_keys,) if isinstance(source_keys, str) else source_keys @@ -2647,6 +2776,11 @@ def _get_span_attributes(**kwargs: Any) -> dict[str, Any]: if result is not None: attributes[otel_key] = result + if OtelAttr.PROVIDER_NAME in attributes: + # Rename to the active semconv version's key; extend with a similar pop/rename if future + # OTel releases rename other attributes we emit. + attributes[_provider_name_attr()] = attributes.pop(OtelAttr.PROVIDER_NAME) + return attributes @@ -2657,9 +2791,9 @@ def capture_exception(span: trace.Span, exception: Exception, timestamp: int | N span.set_status(status=trace.StatusCode.ERROR, description=repr(exception)) -def _capture_system_instructions(span: trace.Span, system_instructions: str | list[str] | None) -> None: +def _capture_system_instructions_latest_experimental(span: trace.Span, system_instructions: str | list[str] | None) -> None: """Capture system instructions on a span.""" - if not system_instructions: + if not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv or not system_instructions: return otel_sys_instructions = [ {"type": "text", "content": instruction} for instruction in _normalize_instructions(system_instructions) @@ -2670,13 +2804,17 @@ def _capture_system_instructions(span: trace.Span, system_instructions: str | li ) -def _capture_current_agent_system_instructions( +def _capture_current_agent_system_instructions_latest_experimental( agent_span: trace.Span, chat_span: trace.Span, system_instructions: str | list[str] | None, ) -> None: """Capture final chat instructions on the current agent span when the chat span belongs to it.""" - if not system_instructions or not agent_span.is_recording(): + if ( + not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv + or not system_instructions + or not agent_span.is_recording() + ): return agent_attributes_obj = getattr(agent_span, "attributes", None) @@ -2698,7 +2836,7 @@ def _capture_current_agent_system_instructions( ): return - _capture_system_instructions(agent_span, system_instructions) + _capture_system_instructions_latest_experimental(agent_span, system_instructions) def _normalize_instructions(system_instructions: str | list[str]) -> list[str]: @@ -2737,50 +2875,162 @@ def _instructions_preserve_existing_agent_instructions( return new_text == existing_text or new_text.startswith(f"{existing_text}\n") -def _capture_messages( - span: trace.Span, +def _capture_message_events_v1_36( provider_name: str, messages: AgentRunInputs, + *, system_instructions: str | list[str] | None = None, output: bool = False, finish_reason: FinishReason | None = None, ) -> None: - """Log messages with extra information.""" + """Emit baseline v1.36.0 GenAI events for a model invocation.""" + if not OBSERVABILITY_SETTINGS.enable_message_events: + return + + # One wall-clock read, then a fixed step per event so order survives backends that + # truncate/collapse timestamps for tightly-emitted events (see + # https://github.com/open-telemetry/semantic-conventions/issues/1701). + timestamp = time_ns() + + if not output and system_instructions: + for instruction in _normalize_instructions(system_instructions): + _emit_otel_event_v1_36(OtelAttr.SYSTEM_MESSAGE, {"content": instruction}, provider_name, timestamp) + timestamp += MESSAGE_EVENT_TIMESTAMP_STEP_NS + from ._types import normalize_messages normalized_messages = normalize_messages(messages) - otel_messages: list[dict[str, Any]] = [] - for index, message in enumerate(normalized_messages): - # Reuse the otel message representation for logging instead of calling to_dict() - # to avoid expensive Pydantic serialization overhead - otel_message = _to_otel_message(message) - logger.info( - otel_message, - extra={ - OtelAttr.EVENT_NAME: OtelAttr.CHOICE if output else ROLE_EVENT_MAP.get(message.role), - OtelAttr.PROVIDER_NAME: provider_name, - MessageListTimestampFilter.INDEX_KEY: index, - }, - ) - otel_messages.append(otel_message) - if finish_reason: - otel_messages[-1]["finish_reason"] = FINISH_REASON_MAP[finish_reason] + + if output: + if not finish_reason: + # Finish reason is required for output events; if not provided, skip emitting choice events. + return + for index, message in enumerate(normalized_messages): + _emit_otel_event_v1_36( + OtelAttr.CHOICE, _to_otel_choice_v1_36(message, index, finish_reason), provider_name, timestamp + ) + timestamp += MESSAGE_EVENT_TIMESTAMP_STEP_NS + return + + for message in normalized_messages: + for event_name, body in _to_otel_input_events_v1_36(message): + _emit_otel_event_v1_36(event_name, body, provider_name, timestamp) + timestamp += MESSAGE_EVENT_TIMESTAMP_STEP_NS + + +def _emit_otel_event_v1_36( + event_name: OtelAttr, + body: dict[str, Any], + provider_name: str, + timestamp: int, +) -> None: + """Emit an OpenTelemetry event with a native structured body.""" + otel_event_logger.emit( + timestamp=timestamp, + body=body, + attributes={OtelAttr.SYSTEM.value: provider_name}, + event_name=event_name.value, + ) + + +def _capture_message_span_attributes_latest_experimental( + span: trace.Span, + messages: AgentRunInputs, + *, + system_instructions: str | list[str] | None = None, + output: bool = False, + finish_reason: FinishReason | None = None, +) -> None: + """Capture the latest (above-baseline) GenAI message span attributes.""" + if not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv: + return + + from ._types import normalize_messages + + otel_messages = [_to_otel_message_latest_experimental(message) for message in normalize_messages(messages)] + if finish_reason and otel_messages: + otel_messages[-1]["finish_reason"] = FINISH_REASON_MAP.get(finish_reason, finish_reason) span.set_attribute( OtelAttr.OUTPUT_MESSAGES if output else OtelAttr.INPUT_MESSAGES, json.dumps(otel_messages, ensure_ascii=False), ) - _capture_system_instructions(span, system_instructions) + _capture_system_instructions_latest_experimental(span, system_instructions) + + +def _to_otel_input_events_v1_36(message: Message) -> list[tuple[OtelAttr, dict[str, Any]]]: + """Create baseline v1.36.0 event names and bodies for an input message.""" + event_name = ROLE_EVENT_MAP.get(message.role) + if event_name is None: + return [] + + if message.role == "tool": + tool_events = [ + ( + OtelAttr.TOOL_MESSAGE, + { + "id": content.call_id, + "content": content.result if content.result is not None else "", + }, + ) + for content in message.contents + if content.type == "function_result" and content.call_id + ] + if tool_events: + return tool_events + return [] + + body: dict[str, Any] = {} + if message.text: + body["content"] = message.text + if message.role == "assistant": + tool_calls = _to_otel_tool_calls_v1_36(message) + if tool_calls: + body["tool_calls"] = tool_calls + return [(event_name, body)] + + +def _to_otel_choice_v1_36(message: Message, index: int, finish_reason: str) -> dict[str, Any]: + """Create a baseline v1.36.0 choice event body.""" + choice_message: dict[str, Any] = {} + if message.text: + choice_message["content"] = message.text + if message.role != "assistant": + choice_message["role"] = message.role + tool_calls = _to_otel_tool_calls_v1_36(message) + if tool_calls: + choice_message["tool_calls"] = tool_calls + return { + "index": index, + "finish_reason": finish_reason, + "message": choice_message, + } -def _to_otel_message(message: Message) -> dict[str, Any]: +def _to_otel_tool_calls_v1_36(message: Message) -> list[dict[str, Any]]: + """Create baseline v1.36.0 function-call structures for a message.""" + return [ + { + "id": content.call_id, + "type": "function", + "function": { + "name": content.name, + "arguments": content.arguments, + }, + } + for content in message.contents + if content.type == "function_call" and content.call_id and content.name + ] + + +def _to_otel_message_latest_experimental(message: Message) -> dict[str, Any]: """Create a otel representation of a message.""" return { "role": message.role, - "parts": [_to_otel_part(content) for content in message.contents], + "parts": [_to_otel_part_latest_experimental(content) for content in message.contents], } -def _to_otel_part(content: Content) -> dict[str, Any] | None: +def _to_otel_part_latest_experimental(content: Content) -> dict[str, Any] | None: """Create a otel representation of a Content.""" from ._types import _get_data_bytes_as_str # pyright: ignore[reportPrivateUsage] @@ -2854,12 +3104,31 @@ def _apply_accumulated_usage(attributes: dict[str, Any], captured_fields: set[st def _apply_usage_attributes(attributes: dict[str, Any], usage: Mapping[str, Any]) -> None: """Apply known usage details as standard OTel GenAI attributes.""" for usage_key, otel_attr in USAGE_DETAIL_TO_OTEL_ATTR: + if ( + otel_attr in LATEST_EXPERIMENTAL_GEN_AI_ATTRIBUTES + and not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv + ): + continue value = usage.get(usage_key) if value is None or isinstance(value, bool) or not isinstance(value, int): continue attributes.setdefault(otel_attr, value) +def _get_response_finish_reason(response: ChatResponse | AgentResponse) -> FinishReason | None: + """Get the finish reason from a response, falling back to the raw representation. + + Some providers only populate ``finish_reason`` on ``raw_representation`` rather than the + normalized response field. + """ + finish_reason = getattr(response, "finish_reason", None) + if not finish_reason and response.raw_representation is not None: + raw_finish_reason = getattr(response.raw_representation, "finish_reason", None) + if isinstance(raw_finish_reason, str): + finish_reason = raw_finish_reason + return cast("FinishReason | None", finish_reason) + + def _get_response_attributes( attributes: dict[str, Any], response: ChatResponse | AgentResponse, @@ -2870,11 +3139,7 @@ def _get_response_attributes( """Get the response attributes from a response.""" if capture_response_id and response.response_id: attributes[OtelAttr.RESPONSE_ID] = response.response_id - finish_reason = getattr(response, "finish_reason", None) - if not finish_reason: - finish_reason = ( - getattr(response.raw_representation, "finish_reason", None) if response.raw_representation else None - ) + finish_reason = _get_response_finish_reason(response) if isinstance(finish_reason, str) and finish_reason: attributes[OtelAttr.FINISH_REASONS] = json.dumps([finish_reason]) if model := getattr(response, "model", None): @@ -2887,6 +3152,7 @@ def _get_response_attributes( GEN_AI_METRIC_ATTRIBUTES = ( OtelAttr.OPERATION, OtelAttr.PROVIDER_NAME, + OtelAttr.SYSTEM, OtelAttr.REQUEST_MODEL, OtelAttr.RESPONSE_MODEL, OtelAttr.ADDRESS, diff --git a/python/packages/core/tests/conftest.py b/python/packages/core/tests/conftest.py index 1627da85cd2..d56bd724fef 100644 --- a/python/packages/core/tests/conftest.py +++ b/python/packages/core/tests/conftest.py @@ -4,6 +4,9 @@ from typing import Any from unittest.mock import patch +from opentelemetry._logs import get_logger_provider, set_logger_provider +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from pytest import fixture @@ -29,6 +32,8 @@ def span_exporter(monkeypatch, enable_instrumentation: bool, enable_sensitive_da "ENABLE_INSTRUMENTATION", "ENABLE_SENSITIVE_DATA", "ENABLE_CONSOLE_EXPORTERS", + "ENABLE_MESSAGE_EVENTS", + "OTEL_SEMCONV_STABILITY_OPT_IN", "OTEL_EXPORTER_OTLP_ENDPOINT", "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", @@ -87,3 +92,23 @@ def span_exporter(monkeypatch, enable_instrumentation: bool, enable_sensitive_da yield exporter # Clean up exporter.clear() + + +@fixture +def log_record_exporter(span_exporter: SpanExporter) -> Generator[InMemoryLogRecordExporter]: + """Fixture providing an in-memory exporter for OTel log records (e.g. gen_ai message events). + + Depends on ``span_exporter`` so ObservabilitySettings/env vars are configured first. The global + OTel LoggerProvider can only be set once per process, so on later test runs this just attaches + another processor to whichever LoggerProvider a previous test already installed. + """ + exporter = InMemoryLogRecordExporter() + set_logger_provider(LoggerProvider()) + provider = get_logger_provider() + if not hasattr(provider, "add_log_record_processor"): + raise RuntimeError("Logger provider does not support adding log record processors.") + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) # type: ignore + + yield exporter + # Clean up + exporter.clear() diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index 65b254b60c5..1e767e2d1fd 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -35,8 +35,11 @@ ChatTelemetryLayer, MessageListTimestampFilter, OtelAttr, - _capture_messages, + _capture_message_events_v1_36, + _capture_message_span_attributes_latest_experimental, _get_instructions_from_options, + _to_otel_choice_v1_36, + _to_otel_input_events_v1_36, get_function_span, ) @@ -365,6 +368,26 @@ async def test_chat_client_observability_with_instructions( assert [msg.get("role") for msg in input_messages] == ["user"] +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_baseline_semconv_omits_system_instructions( + mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data +): + """Test that baseline v1.36.0 telemetry omits the post-v1.36.0 system instructions attribute.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + client = mock_chat_client() + + await client.get_response( + messages=[Message(role="user", contents=["Test message"])], + options={"model": "Test", "instructions": "You are a helpful assistant."}, + ) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert OtelAttr.SYSTEM_INSTRUCTIONS not in spans[0].attributes # type: ignore[operator] # pyrefly: ignore[not-iterable] # ty: ignore[unsupported-operator] + + @pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) async def test_chat_client_streaming_observability_with_instructions( mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data @@ -477,6 +500,97 @@ def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: ) +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_streaming_input_events_correlated_to_chat_span( + mock_chat_client, + span_exporter: InMemorySpanExporter, + log_record_exporter, + enable_sensitive_data, +) -> None: + """Regression guard: streaming input events must carry the chat span's trace/span id. + + ``_capture_message_events_v1_36`` emits the baseline v1.36.0 GenAI message events via the + native OTel event logger, which derives trace/span correlation from whatever span is + current in the ambient context at emit time. In the streaming path the chat span is + started with ``_start_streaming_span`` (not attached as current), so those events must be + emitted while the chat span is explicitly activated -- otherwise they get correlated with + the caller's (parent) span instead of this chat operation. + """ + client = mock_chat_client() + messages = [Message(role="user", contents=["Test"])] + + stream = client.get_response(stream=True, messages=messages, options={"model": "Test"}) + async for _update in stream: + pass + await stream.get_final_response() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + chat_span = spans[0] + assert chat_span.context is not None + + user_message_records = [ + record.log_record + for record in log_record_exporter.get_finished_logs() + if record.log_record.event_name == OtelAttr.USER_MESSAGE.value + ] + assert len(user_message_records) == 1 + user_message_record = user_message_records[0] + + assert user_message_record.trace_id == chat_span.context.trace_id, ( + "input event was not correlated with the chat span's trace" + ) + assert user_message_record.span_id == chat_span.context.span_id, ( + "input event was not correlated with the chat span; it must be emitted while the " + "chat span is activated as current" + ) + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_streaming_output_events_correlated_to_chat_span( + mock_chat_client, + span_exporter: InMemorySpanExporter, + log_record_exporter, + enable_sensitive_data, +) -> None: + """Regression guard: streaming output (choice) events must carry the chat span's trace/span id. + + ``_finalize_stream`` runs as a cleanup hook after the final iterator pull has already exited + its ``_activate_span(span)`` context, so the chat span is no longer current by the time output + events are emitted there. They must therefore be emitted inside an explicit + ``_activate_span(span)`` block, otherwise they get correlated with whatever span happens to be + current in the consuming context instead of this chat operation. + """ + client = mock_chat_client() + messages = [Message(role="user", contents=["Test"])] + + stream = client.get_response(stream=True, messages=messages, options={"model": "Test"}) + async for _update in stream: + pass + await stream.get_final_response() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + chat_span = spans[0] + assert chat_span.context is not None + + choice_records = [ + record.log_record + for record in log_record_exporter.get_finished_logs() + if record.log_record.event_name == OtelAttr.CHOICE.value + ] + assert len(choice_records) == 1 + choice_record = choice_records[0] + + assert choice_record.trace_id == chat_span.context.trace_id, ( + "output event was not correlated with the chat span's trace" + ) + assert choice_record.span_id == chat_span.context.span_id, ( + "output event was not correlated with the chat span; it must be emitted while the " + "chat span is activated as current during stream finalization" + ) + + @pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) async def test_chat_client_observability_with_system_message_and_instructions( mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data @@ -1832,6 +1946,152 @@ def test_enable_instrumentation_reads_env_sensitive_data(monkeypatch): assert observability.OBSERVABILITY_SETTINGS.enable_sensitive_data is True +# region Test GenAI semconv stability opt-in + + +def test_semconv_defaults_to_latest_experimental_when_unset(monkeypatch): + """OTEL_SEMCONV_STABILITY_OPT_IN unset → MAF defaults to the latest conventions.""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.delenv("OTEL_SEMCONV_STABILITY_OPT_IN", raising=False) + settings = ObservabilitySettings() + + assert settings.otel_semconv_stability_opt_in is None + assert settings.use_latest_experimental_gen_ai_semconv is True + + +def test_semconv_explicit_empty_opts_into_baseline(monkeypatch): + """Explicitly setting OTEL_SEMCONV_STABILITY_OPT_IN='' opts into the baseline v1.36.0 conventions.""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "") + settings = ObservabilitySettings() + + assert settings.use_latest_experimental_gen_ai_semconv is False + + +def test_semconv_explicit_token_opts_into_latest_experimental(monkeypatch): + """Explicitly including 'gen_ai_latest_experimental' opts into the latest conventions.""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental") + settings = ObservabilitySettings() + + assert settings.use_latest_experimental_gen_ai_semconv is True + + +def test_semconv_multi_value_list_checks_for_gen_ai_token(monkeypatch): + """OTEL_SEMCONV_STABILITY_OPT_IN supports the standard comma-separated multi-value list format.""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "database, gen_ai_latest_experimental") + settings = ObservabilitySettings() + assert settings.use_latest_experimental_gen_ai_semconv is True + + monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "database,messaging") + settings = ObservabilitySettings() + assert settings.use_latest_experimental_gen_ai_semconv is False + + +def test_baseline_semconv_skips_current_agent_system_instruction_checks(span_exporter: InMemorySpanExporter): + """Baseline v1.36.0 returns before inspecting spans for the unsupported system instructions attribute.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + agent_span = Mock() + + observability._capture_current_agent_system_instructions_latest_experimental( # pyright: ignore[reportPrivateUsage] + agent_span, + Mock(), + "You are a helpful assistant.", + ) + + agent_span.is_recording.assert_not_called() + + +def test_enable_message_events_defaults_true(monkeypatch): + """ENABLE_MESSAGE_EVENTS unset → defaults to True (backward-compatible with pre-versioning behavior).""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.delenv("ENABLE_MESSAGE_EVENTS", raising=False) + settings = ObservabilitySettings() + + assert settings.enable_message_events is True + + +def test_enable_message_events_can_be_disabled(monkeypatch): + """ENABLE_MESSAGE_EVENTS=false disables the baseline v1.36.0 message events.""" + from agent_framework.observability import ObservabilitySettings + + monkeypatch.setenv("ENABLE_MESSAGE_EVENTS", "false") + settings = ObservabilitySettings() + + assert settings.enable_message_events is False + + +@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True) +def test_get_span_attributes_uses_provider_name_under_latest_semconv(span_exporter: InMemorySpanExporter): + """Under the default (latest) semconv, the provider attribute is gen_ai.provider.name.""" + from agent_framework.observability import _get_span_attributes # pyright: ignore[reportPrivateUsage] + + attributes = _get_span_attributes(provider_name="test_provider") + + assert attributes[OtelAttr.PROVIDER_NAME] == "test_provider" + assert OtelAttr.SYSTEM not in attributes + + +@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True) +def test_get_span_attributes_uses_system_under_baseline_semconv(span_exporter: InMemorySpanExporter): + """Under the baseline v1.36.0 semconv, the provider attribute reverts to gen_ai.system.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + attributes = observability._get_span_attributes(provider_name="test_provider") # pyright: ignore[reportPrivateUsage] + + assert attributes[OtelAttr.SYSTEM] == "test_provider" + assert OtelAttr.PROVIDER_NAME not in attributes + + +@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True) +def test_get_span_attributes_omits_post_v1_36_attributes_under_baseline_semconv( + span_exporter: InMemorySpanExporter, +): + """Baseline v1.36.0 omits attributes introduced by later GenAI conventions.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + attributes = observability._get_span_attributes( # pyright: ignore[reportPrivateUsage] + provider_name="test_provider", + tools=[{"type": "web_search", "name": "web_search"}], + ) + + assert attributes == { + OtelAttr.CHOICE_COUNT: 1, + OtelAttr.SYSTEM: "test_provider", + } + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_observability_provider_name_under_baseline_semconv( + mock_chat_client, span_exporter: InMemorySpanExporter +): + """Chat spans report gen_ai.system (not gen_ai.provider.name) under the baseline v1.36.0 semconv.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + client = mock_chat_client() + + messages = [Message(role="user", contents=["Test message"])] + span_exporter.clear() + await client.get_response(messages=messages, options={"model": "Test"}) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + attributes = spans[0].attributes or {} + assert OtelAttr.SYSTEM in attributes + assert OtelAttr.PROVIDER_NAME not in attributes + + # region Test disable_instrumentation sticky behavior @@ -1988,10 +2248,10 @@ def test_disable_instrumentation_in_all(monkeypatch): def test_to_otel_part_text(): """Test _to_otel_part with text content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="text", text="Hello world") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result == {"type": "text", "content": "Hello world"} @@ -1999,10 +2259,10 @@ def test_to_otel_part_text(): def test_to_otel_part_text_reasoning(): """Test _to_otel_part with text_reasoning content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="text_reasoning", text="Thinking about this...") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result == {"type": "reasoning", "content": "Thinking about this..."} @@ -2010,10 +2270,10 @@ def test_to_otel_part_text_reasoning(): def test_to_otel_part_uri(): """Test _to_otel_part with uri content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="uri", uri="https://example.com/image.png", media_type="image/png") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result == { "type": "uri", @@ -2026,10 +2286,10 @@ def test_to_otel_part_uri(): def test_to_otel_part_uri_no_media_type(): """Test _to_otel_part with uri content without media_type.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="uri", uri="https://example.com/file") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result == { "type": "uri", @@ -2042,11 +2302,11 @@ def test_to_otel_part_uri_no_media_type(): def test_to_otel_part_data(): """Test _to_otel_part with data content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental data = b"binary data" content = Content.from_data(data=data, media_type="application/octet-stream") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result["type"] == "blob" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert result["mime_type"] == "application/octet-stream" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] @@ -2056,10 +2316,10 @@ def test_to_otel_part_data(): def test_to_otel_part_function_call(): """Test _to_otel_part with function_call content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="function_call", call_id="call_123", name="test_function", arguments='{"arg1": "value1"}') - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result == { "type": "tool_call", @@ -2072,11 +2332,11 @@ def test_to_otel_part_function_call(): def test_to_otel_part_function_call_reuses_prepared_arguments(): """Test _to_otel_part does not re-serialize function-call arguments in the observability hot path.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental arguments = {"payload": object()} content = Content(type="function_call", call_id="call_789", name="handoff", arguments=arguments) - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result is not None assert result["arguments"] is arguments @@ -2131,15 +2391,90 @@ def test_make_json_safe_dict_with_non_string_keys(): def test_to_otel_part_function_result(): """Test _to_otel_part with function_result content.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental content = Content(type="function_result", call_id="call_123", result="Success") - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) assert result["type"] == "tool_call_response" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert result["id"] == "call_123" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] +# region Test baseline v1.36.0 event conversion + + +def test_to_otel_v1_36_user_message_body(): + """Baseline user events use content rather than the experimental parts shape.""" + events = _to_otel_input_events_v1_36(Message(role="user", contents=["Hello", "world"])) + + assert events == [(OtelAttr.USER_MESSAGE, {"content": "Hello world"})] + + +def test_to_otel_v1_36_assistant_tool_call_body(): + """Baseline assistant events use the v1.36 function-call nesting.""" + message = Message( + role="assistant", + contents=[Content.from_function_call(call_id="call_123", name="get_weather", arguments='{"city":"Paris"}')], + ) + + events = _to_otel_input_events_v1_36(message) + + assert events == [ + ( + OtelAttr.ASSISTANT_MESSAGE, + { + "tool_calls": [ + { + "id": "call_123", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city":"Paris"}'}, + } + ] + }, + ) + ] + + +def test_to_otel_v1_36_tool_results_emit_one_event_per_call(): + """Each baseline tool event has the one required tool-call id.""" + message = Message( + role="tool", + contents=[ + Content.from_function_result(call_id="call_1", result="first"), + Content.from_function_result(call_id="call_2", result={"value": 2}), + ], + ) + + events = _to_otel_input_events_v1_36(message) + + assert events == [ + (OtelAttr.TOOL_MESSAGE, {"id": "call_1", "content": "first"}), + (OtelAttr.TOOL_MESSAGE, {"id": "call_2", "content": '{"value": 2}'}), + ] + + +def test_to_otel_v1_36_tool_message_without_call_id_is_skipped(): + """A baseline tool event is not emitted when its required call id is unavailable.""" + events = _to_otel_input_events_v1_36(Message(role="tool", contents=["Uncorrelated result"])) + + assert events == [] + + +def test_to_otel_v1_36_choice_body(): + """Baseline choices contain index, finish reason, and a nested message.""" + body = _to_otel_choice_v1_36( + Message(role="assistant", contents=["Done"]), + index=1, + finish_reason="tool_calls", + ) + + assert body == { + "index": 1, + "finish_reason": "tool_calls", + "message": {"content": "Done"}, + } + + # region Test workflow observability functions @@ -2490,6 +2825,35 @@ def test_get_response_attributes_with_additional_usage(): assert result[OtelAttr.REASONING_OUTPUT_TOKENS] == 30 +def test_get_response_attributes_omits_post_v1_36_usage_under_baseline_semconv(monkeypatch: pytest.MonkeyPatch): + """Baseline v1.36.0 keeps total usage while omitting newer token breakdowns.""" + from unittest.mock import Mock + + import agent_framework.observability as observability + + monkeypatch.setattr(observability.OBSERVABILITY_SETTINGS, "otel_semconv_stability_opt_in", "") + response = Mock( + response_id=None, + finish_reason=None, + raw_representation=None, + model=None, + usage_details={ + "input_token_count": 100, + "output_token_count": 50, + "cache_creation_input_token_count": 10, + "cache_read_input_token_count": 20, + "reasoning_output_token_count": 30, + }, + ) + + result = observability._get_response_attributes({}, response) # pyright: ignore[reportPrivateUsage] + + assert result == { + OtelAttr.INPUT_TOKENS: 100, + OtelAttr.OUTPUT_TOKENS: 50, + } + + def test_get_response_attributes_maps_legacy_usage_keys(): """Test _get_response_attributes maps legacy provider usage keys to standard OTel attributes.""" from unittest.mock import Mock @@ -2647,11 +3011,11 @@ def test_observability_settings_configure_already_setup(monkeypatch): def test_to_otel_part_generic(): """Test _to_otel_part with unknown content type uses to_dict fallback.""" from agent_framework import Content - from agent_framework.observability import _to_otel_part + from agent_framework.observability import _to_otel_part_latest_experimental # Create a content with type that falls to default case content = Content(type="annotations", text="some text") # type: ignore[arg-type] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] - result = _to_otel_part(content) + result = _to_otel_part_latest_experimental(content) # Should return result from to_dict assert result is not None @@ -2682,6 +3046,104 @@ def test_get_response_attributes_finish_reason_from_raw(): assert OtelAttr.FINISH_REASONS in result +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_choice_event_uses_raw_representation_finish_reason( + span_exporter: InMemorySpanExporter, + log_record_exporter, + enable_sensitive_data, +) -> None: + """Regression guard: choice events must use the raw_representation finish_reason fallback. + + Some providers only populate ``finish_reason`` on ``raw_representation`` rather than the + normalized ``ChatResponse.finish_reason`` field. ``_capture_message_events_v1_36`` skips + emitting choice events entirely when no finish_reason is available, so callers must resolve + the same fallback as ``_get_response_attributes`` before deciding whether to emit. + """ + from unittest.mock import Mock + + class RawFinishReasonChatClient(ChatTelemetryLayer, BaseChatClient[Any]): + def service_url(self): + return "https://test.example.com" + + def _inner_get_response( # pyrefly: ignore[bad-override] + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse]: + async def _get() -> ChatResponse: + raw_rep = Mock() + raw_rep.finish_reason = "stop" + return ChatResponse( + messages=[Message("assistant", ["Hello"])], + finish_reason=None, + raw_representation=raw_rep, + ) + + return _get() + + client = RawFinishReasonChatClient() + await client.get_response(messages=[Message(role="user", contents=["Test"])], options={"model": "Test"}) + + choice_records = [ + record.log_record + for record in log_record_exporter.get_finished_logs() + if record.log_record.event_name == OtelAttr.CHOICE.value + ] + assert len(choice_records) == 1 + assert choice_records[0].body["finish_reason"] == "stop" # type: ignore + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_chat_client_ignores_non_string_raw_representation_finish_reason( + span_exporter: InMemorySpanExporter, + log_record_exporter, + enable_sensitive_data, +) -> None: + """Regression guard: an unconfigured raw_representation attribute must not crash telemetry. + + ``raw_representation`` is frequently a test double (e.g. ``unittest.mock.Mock()``) whose + unset ``finish_reason`` attribute auto-vivifies to a ``Mock`` rather than raising + ``AttributeError``. The raw_representation finish_reason fallback must reject non-string + values so this never leaks a non-JSON-serializable object into `json.dumps` calls. + """ + from unittest.mock import Mock + + class UnconfiguredRawChatClient(ChatTelemetryLayer, BaseChatClient[Any]): + def service_url(self): + return "https://test.example.com" + + def _inner_get_response( # pyrefly: ignore[bad-override] + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse]: + async def _get() -> ChatResponse: + return ChatResponse( + messages=[Message("assistant", ["Hello"])], + finish_reason=None, + raw_representation=Mock(), + ) + + return _get() + + client = UnconfiguredRawChatClient() + response = await client.get_response(messages=[Message(role="user", contents=["Test"])], options={"model": "Test"}) + + assert response.text == "Hello" + choice_records = [ + record.log_record + for record in log_record_exporter.get_finished_logs() + if record.log_record.event_name == OtelAttr.CHOICE.value + ] + assert choice_records == [] + + # region Test agent instrumentation @@ -2746,11 +3208,13 @@ class MockAgent(AgentTelemetryLayer, _MockAgent): # type: ignore[misc] # pyref agent = MockAgent() span_exporter.clear() - response = await agent.run(messages="Hello") + with patch("agent_framework.observability.otel_event_logger.emit") as mock_emit: + response = await agent.run(messages="Hello") assert response is not None spans = span_exporter.get_finished_spans() assert len(spans) == 1 + assert mock_emit.call_count == 0 @pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) @@ -2944,6 +3408,209 @@ async def _inner_get_response(self, *, messages, options, **kwargs): assert output_messages[-1].get("finish_reason") == "stop" +# region Test _capture_messages GenAI semconv versioning + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_baseline_semconv_emits_events_only(span_exporter: InMemorySpanExporter): + """Baseline v1.36.0 conventions (opt-in list without the experimental token): events only, no span attribute.""" + from opentelemetry import trace + + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + tracer = trace.get_tracer("test") + span_exporter.clear() + + with ( + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + tracer.start_as_current_span("test_span"), + ): + observability._capture_message_events_v1_36( # type: ignore[reportPrivateUsage] + provider_name="test_provider", + messages=[Message(role="user", contents=["Test"])], + ) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + attributes = spans[0].attributes or {} + assert OtelAttr.INPUT_MESSAGES not in attributes + mock_emit.assert_called_once() + assert mock_emit.call_args.kwargs["event_name"] == "gen_ai.user.message" + assert mock_emit.call_args.kwargs["body"] == {"content": "Test"} + assert mock_emit.call_args.kwargs["attributes"] == {"gen_ai.system": "test_provider"} + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_latest_experimental_emits_span_attribute_only_when_events_disabled( + span_exporter: InMemorySpanExporter, +): + """Latest experimental conventions with ENABLE_MESSAGE_EVENTS=false: span attribute only, no events.""" + import json + + from opentelemetry import trace + + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "gen_ai_latest_experimental" + observability.OBSERVABILITY_SETTINGS.enable_message_events = False + tracer = trace.get_tracer("test") + span_exporter.clear() + + with ( + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + tracer.start_as_current_span("test_span") as span, + ): + observability._capture_message_span_attributes_latest_experimental( # type: ignore[reportPrivateUsage] + span=span, + messages=[Message(role="user", contents=["Test"])], + ) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + attributes = spans[0].attributes or {} + input_messages = json.loads(cast(str, attributes[OtelAttr.INPUT_MESSAGES])) + assert [msg.get("role") for msg in input_messages] == ["user"] + assert mock_emit.call_count == 0 + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_defaults_emit_both_events_and_span_attribute(span_exporter: InMemorySpanExporter): + """Default settings (nothing configured) preserve pre-versioning behavior: both events and span attribute.""" + import json + + from opentelemetry import trace + + import agent_framework.observability as observability + + tracer = trace.get_tracer("test") + span_exporter.clear() + + with ( + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + tracer.start_as_current_span("test_span") as span, + ): + observability._capture_message_events_v1_36( # type: ignore[reportPrivateUsage] + provider_name="test_provider", + messages=[Message(role="user", contents=["Test"])], + ) + observability._capture_message_span_attributes_latest_experimental( # type: ignore[reportPrivateUsage] + span=span, + messages=[Message(role="user", contents=["Test"])], + ) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + attributes = spans[0].attributes or {} + input_messages = json.loads(cast(str, attributes[OtelAttr.INPUT_MESSAGES])) + assert [msg.get("role") for msg in input_messages] == ["user"] + mock_emit.assert_called_once() + assert mock_emit.call_args.kwargs["event_name"] == "gen_ai.user.message" + assert mock_emit.call_args.kwargs["body"] == {"content": "Test"} + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_baseline_semconv_emits_one_event_per_choice(span_exporter: InMemorySpanExporter): + """Baseline v1.36 emits each model choice with its required index and finish reason.""" + from opentelemetry import trace + + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + tracer = trace.get_tracer("test") + + with ( + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + tracer.start_as_current_span("test_span"), + ): + observability._capture_message_events_v1_36( # type: ignore[reportPrivateUsage] + provider_name="test_provider", + messages=[ + Message(role="assistant", contents=["First"]), + Message(role="assistant", contents=["Second"]), + ], + output=True, + finish_reason=cast(Any, "tool_calls"), + ) + + assert [call.kwargs["event_name"] for call in mock_emit.call_args_list] == ["gen_ai.choice", "gen_ai.choice"] + assert [call.kwargs["body"] for call in mock_emit.call_args_list] == [ + {"index": 0, "finish_reason": "tool_calls", "message": {"content": "First"}}, + {"index": 1, "finish_reason": "tool_calls", "message": {"content": "Second"}}, + ] + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_preserves_custom_finish_reason(span_exporter: InMemorySpanExporter): + """Custom finish reasons remain available to baseline events and experimental attributes.""" + import json + + from opentelemetry import trace + + import agent_framework.observability as observability + + tracer = trace.get_tracer("test") + with ( + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + tracer.start_as_current_span("test_span") as span, + ): + observability._capture_message_events_v1_36( # type: ignore[reportPrivateUsage] + provider_name="test_provider", + messages=[Message(role="assistant", contents=["Done"])], + output=True, + finish_reason=cast(Any, "guardrail"), + ) + observability._capture_message_span_attributes_latest_experimental( # type: ignore[reportPrivateUsage] + span=span, + messages=[Message(role="assistant", contents=["Done"])], + output=True, + finish_reason=cast(Any, "guardrail"), + ) + + assert mock_emit.call_args.kwargs["body"]["finish_reason"] == "guardrail" + spans = span_exporter.get_finished_spans() + attributes = spans[0].attributes or {} + output_messages = json.loads(cast(str, attributes[OtelAttr.OUTPUT_MESSAGES])) + assert output_messages[-1]["finish_reason"] == "guardrail" + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +def test_capture_messages_exports_native_otel_event(span_exporter: InMemorySpanExporter, monkeypatch): + """The SDK receives a native body, dedicated event name, and current span context.""" + from opentelemetry import trace + from opentelemetry._logs import get_logger + from opentelemetry.sdk._logs import LoggerProvider + from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor + + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + exporter = InMemoryLogRecordExporter() + logger_provider = LoggerProvider(shutdown_on_exit=False) + logger_provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + monkeypatch.setattr( + observability, + "otel_event_logger", + get_logger("agent_framework.test", logger_provider=logger_provider), + ) + tracer = trace.get_tracer("test") + + with tracer.start_as_current_span("test_span") as span: + observability._capture_message_events_v1_36( # type: ignore[reportPrivateUsage] + provider_name="test_provider", + messages=[Message(role="user", contents=["Test"])], + ) + span_context = span.get_span_context() + + (readable_record,) = exporter.get_finished_logs() + record = readable_record.log_record + assert record.event_name == "gen_ai.user.message" + assert record.body == {"content": "Test"} + assert dict(record.attributes or {}) == {"gen_ai.system": "test_provider"} + assert record.trace_id == span_context.trace_id + assert record.span_id == span_context.span_id + + # region Test agent streaming exception @@ -3969,9 +4636,8 @@ async def test_system_instructions_preserves_non_ascii_characters(span_exporter: span_exporter.clear() with tracer.start_as_current_span("test_span") as span: - _capture_messages( + _capture_message_span_attributes_latest_experimental( span=span, - provider_name="test_provider", messages=[Message(role="user", contents=["Test"])], system_instructions=chinese_text, ) @@ -4025,7 +4691,7 @@ class HandoffRequest: span_exporter.clear() tracer = trace.get_tracer("test") with tracer.start_as_current_span("test_span") as span: - _capture_messages(span=span, provider_name="test_provider", messages=[msg]) + _capture_message_span_attributes_latest_experimental(span=span, messages=[msg]) spans = span_exporter.get_finished_spans() span = spans[0] # type: ignore[assignment] @@ -4035,10 +4701,10 @@ class HandoffRequest: assert tool_part["arguments"]["data"] == {"target_agent": "helper", "reason": "overflow"} -def test_capture_messages_keeps_framework_instructions_out_of_logs_and_span_messages( +def test_capture_messages_emits_framework_instructions_separately_from_history( span_exporter: InMemorySpanExporter, ): - """Test separate framework instructions do not appear in chat-history logs or span messages.""" + """Test separate framework instructions use their own baseline event and experimental span attribute.""" import json from opentelemetry import trace @@ -4047,33 +4713,39 @@ def test_capture_messages_keeps_framework_instructions_out_of_logs_and_span_mess span_exporter.clear() with ( - patch("agent_framework.observability.logger.info") as mock_logger_info, + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, tracer.start_as_current_span("test_span") as span, ): - _capture_messages( - span=span, + _capture_message_events_v1_36( provider_name="test_provider", messages=[Message(role="user", contents=["Test"])], system_instructions="Framework system instruction", ) + _capture_message_span_attributes_latest_experimental( + span=span, + messages=[Message(role="user", contents=["Test"])], + system_instructions="Framework system instruction", + ) spans = span_exporter.get_finished_spans() assert len(spans) == 1 input_messages = json.loads(spans[0].attributes[OtelAttr.INPUT_MESSAGES]) # type: ignore[arg-type, index] # pyrefly: ignore[bad-argument-type, unsupported-operation] # ty: ignore[invalid-argument-type, not-subscriptable] assert [msg.get("role") for msg in input_messages] == ["user"] - assert mock_logger_info.call_count == 1, f"Expected 1 log call, got {mock_logger_info.call_count}" - (first_call,) = mock_logger_info.call_args_list - assert first_call.args - logged_message = first_call.args[0] - assert logged_message["role"] == "user" - assert logged_message["parts"][0]["content"] == "Test" + assert [call.kwargs["event_name"] for call in mock_emit.call_args_list] == [ + "gen_ai.system.message", + "gen_ai.user.message", + ] + assert [call.kwargs["body"] for call in mock_emit.call_args_list] == [ + {"content": "Framework system instruction"}, + {"content": "Test"}, + ] -def test_capture_messages_logs_only_chat_history_when_framework_instructions_are_separate( +def test_capture_messages_preserves_framework_instructions_and_system_history( span_exporter: InMemorySpanExporter, ): - """Test chat-history logging preserves original system messages without prepending framework instructions.""" + """Test baseline events preserve separate instructions and original system history.""" import json from opentelemetry import trace @@ -4082,11 +4754,10 @@ def test_capture_messages_logs_only_chat_history_when_framework_instructions_are span_exporter.clear() with ( - patch("agent_framework.observability.logger.info") as mock_logger_info, + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, tracer.start_as_current_span("test_span") as span, ): - _capture_messages( - span=span, + _capture_message_events_v1_36( provider_name="test_provider", messages=[ Message(role="system", contents=["Original system message"]), @@ -4094,17 +4765,72 @@ def test_capture_messages_logs_only_chat_history_when_framework_instructions_are ], system_instructions="Framework system instruction", ) + _capture_message_span_attributes_latest_experimental( + span=span, + messages=[ + Message(role="system", contents=["Original system message"]), + Message(role="user", contents=["Test"]), + ], + system_instructions="Framework system instruction", + ) spans = span_exporter.get_finished_spans() assert len(spans) == 1 input_messages = json.loads(spans[0].attributes[OtelAttr.INPUT_MESSAGES]) # type: ignore[arg-type, index] # pyrefly: ignore[bad-argument-type, unsupported-operation] # ty: ignore[invalid-argument-type, not-subscriptable] assert [msg.get("role") for msg in input_messages] == ["system", "user"] - assert mock_logger_info.call_count == 2, f"Expected 2 log calls, got {mock_logger_info.call_count}" - logged_messages = [call.args[0] for call in mock_logger_info.call_args_list] - assert [msg["role"] for msg in logged_messages] == ["system", "user"] - assert logged_messages[0]["parts"][0]["content"] == "Original system message" - assert logged_messages[1]["parts"][0]["content"] == "Test" + assert [call.kwargs["event_name"] for call in mock_emit.call_args_list] == [ + "gen_ai.system.message", + "gen_ai.system.message", + "gen_ai.user.message", + ] + assert [call.kwargs["body"] for call in mock_emit.call_args_list] == [ + {"content": "Framework system instruction"}, + {"content": "Original system message"}, + {"content": "Test"}, + ] + + +def test_capture_messages_reads_time_once_then_steps_per_event(): + """Test the timestamp is read once, then stepped by a fixed amount for each subsequent event.""" + from agent_framework.observability import MESSAGE_EVENT_TIMESTAMP_STEP_NS + + with ( + patch("agent_framework.observability.time_ns", return_value=1_000) as mock_time_ns, + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + ): + _capture_message_events_v1_36( + provider_name="test_provider", + messages=[Message(role="user", contents=["Test"])], + system_instructions="Framework system instruction", + ) + + mock_time_ns.assert_called_once() + assert [call.kwargs["timestamp"] for call in mock_emit.call_args_list] == [ + 1_000, + 1_000 + MESSAGE_EVENT_TIMESTAMP_STEP_NS, + ] + + +def test_capture_messages_stepped_timestamps_preserve_order_when_clock_collapses(): + """Test the stepped timestamps stay strictly increasing even when the clock reads a single value.""" + with ( + patch("agent_framework.observability.time_ns", return_value=1_000), + patch("agent_framework.observability.otel_event_logger.emit") as mock_emit, + ): + _capture_message_events_v1_36( + provider_name="test_provider", + messages=[ + Message(role="user", contents=["First"]), + Message(role="user", contents=["Second"]), + Message(role="user", contents=["Third"]), + ], + system_instructions="Framework system instruction", + ) + + timestamps = [call.kwargs["timestamp"] for call in mock_emit.call_args_list] + assert timestamps == sorted(set(timestamps)) + assert len(timestamps) == 4 @pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) @@ -5911,7 +6637,7 @@ async def _consume(s): # When ``ENABLE_INSTRUMENTATION`` is on (the default) but no OpenTelemetry # tracer provider has been configured, the global provider is the # ``ProxyTracerProvider`` which returns non-recording spans. The telemetry -# layers gate sensitive-data serialization (``_capture_messages``) on +# layers gate sensitive-data serialization on # ``span.is_recording()`` so that we don't pay the JSON-serialization cost # when the span is going to be dropped anyway. The tests below verify that # behavior by patching ``get_tracer`` to return a ``NoOpTracer``. @@ -5930,14 +6656,18 @@ async def test_chat_capture_messages_skipped_when_span_not_recording( with ( patch("agent_framework.observability.get_tracer", return_value=NoOpTracer()), - patch("agent_framework.observability._capture_messages") as mock_capture_messages, + patch("agent_framework.observability._capture_message_events_v1_36") as mock_capture_v1_36, + patch( + "agent_framework.observability._capture_message_span_attributes_latest_experimental" + ) as mock_capture_experimental, patch("agent_framework.observability._capture_response") as mock_capture_response, ): response = await client.get_response(messages=messages, options={"model": "Test"}) assert response is not None # Sensitive-data serialization must be skipped because span.is_recording() is False. - assert mock_capture_messages.call_count == 0 + assert mock_capture_v1_36.call_count == 0 + assert mock_capture_experimental.call_count == 0 # _capture_response still runs so that metric histograms continue to record. assert mock_capture_response.call_count == 1 @@ -5955,7 +6685,10 @@ async def test_chat_streaming_capture_messages_skipped_when_span_not_recording( with ( patch("agent_framework.observability.get_tracer", return_value=NoOpTracer()), - patch("agent_framework.observability._capture_messages") as mock_capture_messages, + patch("agent_framework.observability._capture_message_events_v1_36") as mock_capture_v1_36, + patch( + "agent_framework.observability._capture_message_span_attributes_latest_experimental" + ) as mock_capture_experimental, patch("agent_framework.observability._capture_response") as mock_capture_response, ): updates: list[ChatResponseUpdate] = [] @@ -5965,7 +6698,8 @@ async def test_chat_streaming_capture_messages_skipped_when_span_not_recording( await stream.get_final_response() assert len(updates) == 2 - assert mock_capture_messages.call_count == 0 + assert mock_capture_v1_36.call_count == 0 + assert mock_capture_experimental.call_count == 0 assert mock_capture_response.call_count == 1 @@ -5981,13 +6715,13 @@ async def test_agent_capture_messages_skipped_when_span_not_recording( with ( patch("agent_framework.observability.get_tracer", return_value=NoOpTracer()), - patch("agent_framework.observability._capture_messages") as mock_capture_messages, + patch("agent_framework.observability._capture_message_span_attributes_latest_experimental") as mock_capture, patch("agent_framework.observability._capture_response") as mock_capture_response, ): response = await agent.run("Test message") assert response is not None - assert mock_capture_messages.call_count == 0 + assert mock_capture.call_count == 0 assert mock_capture_response.call_count == 1 @@ -6003,7 +6737,7 @@ async def test_agent_streaming_capture_messages_skipped_when_span_not_recording( with ( patch("agent_framework.observability.get_tracer", return_value=NoOpTracer()), - patch("agent_framework.observability._capture_messages") as mock_capture_messages, + patch("agent_framework.observability._capture_message_span_attributes_latest_experimental") as mock_capture, patch("agent_framework.observability._capture_response") as mock_capture_response, ): updates: list[Any] = [] @@ -6013,7 +6747,7 @@ async def test_agent_streaming_capture_messages_skipped_when_span_not_recording( await stream.get_final_response() assert len(updates) == 2 - assert mock_capture_messages.call_count == 0 + assert mock_capture.call_count == 0 assert mock_capture_response.call_count == 1 @@ -6027,12 +6761,16 @@ async def test_chat_capture_messages_called_when_span_recording( span_exporter.clear() with ( - patch("agent_framework.observability._capture_messages") as mock_capture_messages, + patch("agent_framework.observability._capture_message_events_v1_36") as mock_capture_v1_36, + patch( + "agent_framework.observability._capture_message_span_attributes_latest_experimental" + ) as mock_capture_experimental, patch("agent_framework.observability._capture_response") as mock_capture_response, ): response = await client.get_response(messages=messages, options={"model": "Test"}) assert response is not None - # Two _capture_messages calls: one for input, one for output messages. - assert mock_capture_messages.call_count == 2 + # Each representation is captured once for input and once for output messages. + assert mock_capture_v1_36.call_count == 2 + assert mock_capture_experimental.call_count == 2 assert mock_capture_response.call_count == 1 diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 6575cd0c118..1f418507ace 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/python/packages/core/tests/core/test_sessions.py @@ -1489,13 +1489,12 @@ async def test_save_messages_preserves_duplicate_content(self) -> None: assert state["messages"][0].text == "yes" assert state["messages"][1].text == "yes" - async def test_save_messages_handles_replayed_transcript_with_duplicates(self) -> None: provider = InMemoryHistoryProvider() state: dict[str, Any] = {} - msg_b = Message (role = "user", contents=["B"]) - await provider.save_messages("s1", [msg_b], state = state) + msg_b = Message(role="user", contents=["B"]) + await provider.save_messages("s1", [msg_b], state=state) assert len(state["messages"]) == 1 msg_a = Message(role="user", contents=["A"]) @@ -1503,7 +1502,7 @@ async def test_save_messages_handles_replayed_transcript_with_duplicates(self) - msg_b2 = Message(role="user", contents=["B"]) msg_d = Message(role="user", contents=["D"]) - await provider.save_messages("s1", [msg_a, msg_b, msg_c, msg_b2, msg_d], state = state) + await provider.save_messages("s1", [msg_a, msg_b, msg_c, msg_b2, msg_d], state=state) assert len(state["messages"]) == 4 texts = [m.text for m in state["messages"]] diff --git a/python/packages/core/tests/core/test_tools.py b/python/packages/core/tests/core/test_tools.py index a24e20cacd4..33fad82ddc6 100644 --- a/python/packages/core/tests/core/test_tools.py +++ b/python/packages/core/tests/core/test_tools.py @@ -652,6 +652,35 @@ def telemetry_test_tool(x: int, y: int) -> int: assert attributes[OtelAttr.TOOL_CALL_ID] == "test_call_id" +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_tool_invoke_telemetry_omits_tool_call_attrs_under_baseline_semconv(span_exporter: InMemorySpanExporter): + """gen_ai.tool.call.arguments/result were introduced above v1.36.0; omit them under the baseline semconv.""" + import agent_framework.observability as observability + + observability.OBSERVABILITY_SETTINGS.otel_semconv_stability_opt_in = "" + + @tool( + name="telemetry_test_tool", + description="A test tool for telemetry", + ) + def telemetry_test_tool(x: int, y: int) -> int: + """A function that adds two numbers for telemetry testing.""" + return x + y + + span_exporter.clear() + result = await telemetry_test_tool.invoke(x=1, y=2, tool_call_id="test_call_id") + + assert isinstance(result, list) + assert result[0].text == "3" + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes is not None + assert OtelAttr.TOOL_ARGUMENTS not in span.attributes + assert OtelAttr.TOOL_RESULT not in span.attributes + + async def test_tool_invoke_rejects_unexpected_runtime_kwargs() -> None: """Ensure invoke() requires runtime data to flow through FunctionInvocationContext.""" diff --git a/python/samples/02-agents/observability/README.md b/python/samples/02-agents/observability/README.md index 4ed36c0c4b6..201a717e784 100644 --- a/python/samples/02-agents/observability/README.md +++ b/python/samples/02-agents/observability/README.md @@ -20,7 +20,7 @@ For more information, please refer to the following resources: The Agent Framework Python SDK is **natively instrumented** to emit logs, traces, and metrics throughout agent/model invocation and tool execution, so you can monitor your AI application's performance and track token consumption. Instrumentation follows the OpenTelemetry [Semantic Conventions for GenAI](https://opentelemetry.io/docs/specs/semconv/gen-ai/), and workflows emit their own spans for end-to-end visibility. -Setting up observability is also easy: a single call to `configure_otel_providers()` from the `agent_framework.observability` module wires up the trace, log, and metric providers. It reads the standard OpenTelemetry environment variables to configure exporters automatically. +> See [GenAI semantic-conventions versioning](#genai-semantic-conventions-versioning) for details on how Agent Framework supports different versions of the conventions. ### Five patterns for configuring observability @@ -199,12 +199,43 @@ Agent Framework reads the following environment variables: | `ENABLE_INSTRUMENTATION` | `true` | Set to `false` to disable native instrumentation. See [Disabling instrumentation](#disabling-instrumentation) for the programmatic alternative with sticky semantics. | | `ENABLE_SENSITIVE_DATA` | `false` | Set to `true` to emit sensitive data (prompts, responses, etc.). | | `ENABLE_CONSOLE_EXPORTERS` | `false` | Set to `true` to add console exporters. Only used by `configure_otel_providers()`. | +| `ENABLE_MESSAGE_EVENTS` | `true` | Set to `false` to stop emitting the baseline v1.36.0 GenAI message events (`gen_ai.system.message`, etc.) for model invocation. **Has no effect unless `ENABLE_SENSITIVE_DATA=true`.** See [GenAI semantic-conventions versioning](#genai-semantic-conventions-versioning). | +| `OTEL_SEMCONV_STABILITY_OPT_IN` | unset (conventions above v1.36.0) | A comma-separated list of category-specific values, following the standard OpenTelemetry comma-separated opt-in list format, currently only containing a single token ``"gen_ai_latest_experimental"``. v1.36.0 is the OTel-recommended baseline; every version above it is referred to here as "latest" (even the baseline is an expeirmental release). The default, unlike upstream OpenTelemetry which retains the baseline conventions, ``"gen_ai_latest_experimental"`` selects the latest conventions above v1.36.0; a list that omits that token (e.g. ``""``) selects the v1.36.0 conventions instead. See [GenAI semantic-conventions versioning](#genai-semantic-conventions-versioning). | | `VS_CODE_EXTENSION_PORT` | unset | Port used by the [AI Toolkit for VS Code](https://marketplace.visualstudio.com/items?itemName=ms-windows-ai-studio.windows-ai-studio#tracing) tracing integration. Only used by `configure_otel_providers()`. | You can also call `enable_sensitive_telemetry()` from `agent_framework.observability` to opt in to sensitive-data capture programmatically. > **Note**: Sensitive data includes prompts, responses, and tool arguments. Only enable it in development or test environments — it may expose user or system secrets in production. +### GenAI semantic-conventions versioning + +[v1.36.0](https://github.com/open-telemetry/semantic-conventions/blob/v1.36.0/docs/gen-ai) is the OpenTelemetry-recommended **baseline** for existing GenAI instrumentations. Releases above it (v1.37.0 and later) are referred to as **latest** and, per OTel's own [stability warning](https://github.com/open-telemetry/semantic-conventions/blob/v1.37.0/docs/gen-ai), keep changing in more than one way. `OTEL_SEMCONV_STABILITY_OPT_IN` is the OTel-standard switch between these two rule sets, and Agent Framework applies it consistently across every attribute/representation it knows differs between the two: + +| Aspect | v1.36.0 (baseline) | Above v1.36.0 (latest, the default) | +|--------|------------------|--------------------------------------------| +| Input/output message representation | Log-record **events** (`gen_ai.system.message`, `gen_ai.user.message`, `gen_ai.assistant.message`, `gen_ai.tool.message`, `gen_ai.choice`) | `gen_ai.input.messages`/`gen_ai.output.messages` **span attributes** | +| Provider-identifying attribute | `gen_ai.system` | `gen_ai.provider.name` | +| Tool call arguments/results on `execute_tool` spans | Not emitted (introduced in v1.38.0) | `gen_ai.tool.call.arguments` / `gen_ai.tool.call.result` | + +`invoke_agent` spans always use `INTERNAL` span kind (the OTel default), regardless of semconv version. The v1.41.0 spec defines `CLIENT` for agents that are themselves a remote service and `INTERNAL` for agents that run in-process (no `server.address`/`server.port`/token-usage attributes, since the actual network call happens on a nested `chat` span instead). Agent Framework's own agents run in-process — `agent.run()` orchestrates a locally-running chat client, which creates its own nested `chat` span for the actual network call — so `INTERNAL` applies uniformly, without needing to classify each agent implementation across packages. What's **not yet covered** by this flag is the rest of the v1.41.0 attribute-group split: under the conventions above v1.36.0, the `invoke_agent` client span is defined to drop `gen_ai.response.id`, `gen_ai.response.model`, and `gen_ai.response.finish_reasons` and add `gen_ai.agent.version` instead. Agent Framework **still emits** the former **three** unconditionally on `invoke_agent` spans and **does not** emit `gen_ai.agent.version` at all under either semconv version. + +> **`ENABLE_SENSITIVE_DATA=true` is a prerequisite for the message-representation and tool-call-attribute rows above.** Chat content (prompts, responses, tool arguments/results) is only ever captured when sensitive-data capture is enabled (see [`ENABLE_SENSITIVE_DATA`](#environment-variables) above); the provider-attribute rename applies regardless, since `gen_ai.system`/`gen_ai.provider.name` is not sensitive data. If `ENABLE_SENSITIVE_DATA` is `false` (the default), `ENABLE_MESSAGE_EVENTS` has nothing to switch and is effectively ignored, and no `gen_ai.tool.call.*` attributes are emitted under either semconv version. + +Agent Framework defaults to the conventions above v1.36.0 (unlike upstream OpenTelemetry, which retains the baseline conventions) because most users already depend on them, and — to avoid a breaking change for anyone consuming the older message events for modelinvocation — also keeps emitting those events by default via `ENABLE_MESSAGE_EVENTS`. `ENABLE_MESSAGE_EVENTS` is controlled independently of `OTEL_SEMCONV_STABILITY_OPT_IN`: + +```bash +# Capture agent/chat client/tool input and output contents (default: false): +export ENABLE_SENSITIVE_DATA=true + +# Opt into the baseline v1.36.0 conventions only (default: "gen_ai_latest_experimental"): +export OTEL_SEMCONV_STABILITY_OPT_IN="" + +# Agent Framework still emits the baseline v1.36.0 message events for model invocations even +# when the semconv opt-in is set to latest for compatibility reasons. To stop emitting those +# events (default: true): +export ENABLE_MESSAGE_EVENTS=false +``` + ### Disabling instrumentation There are two ways to turn Agent Framework's native instrumentation off, and they have **different scopes**: