From 2448a77dc6a4a41525a1a958e6f57648b67a2e7a Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Wed, 15 Apr 2026 17:03:58 -0700 Subject: [PATCH 1/2] Fix: AgentTelemetryLayer now captures context-provider-extended instructions AgentTelemetryLayer._trace_agent_invocation() recorded gen_ai.system_instructions from merged_options before calling execute(), which runs before context providers extend instructions via context.extend_instructions(). This meant the invoke_agent span only showed base instructions, not the full instructions sent to the LLM. Fix: introduce AGENT_MERGED_INSTRUCTIONS ContextVar that is set by _prepare_session_and_messages() after context providers run and instructions are merged. AgentTelemetryLayer reads this after execution and re-captures gen_ai.system_instructions with the complete instructions. Both streaming and non-streaming paths are fixed. Fixes #5291 --- .../packages/core/agent_framework/_agents.py | 6 +++ .../core/agent_framework/observability.py | 43 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index 585898ae523..a8288bfd982 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -1447,6 +1447,12 @@ async def _prepare_session_and_messages( else: chat_options["instructions"] = combined_instructions + # Publish the final merged instructions so AgentTelemetryLayer can + # re-capture gen_ai.system_instructions with provider-extended content. + from .observability import AGENT_MERGED_INSTRUCTIONS + + AGENT_MERGED_INSTRUCTIONS.set(chat_options.get("instructions")) + return session_context, chat_options def as_mcp_server( diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 6998e5994f8..454e51a6573 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -105,6 +105,13 @@ "inner_accumulated_usage", default=None ) +# Tracks the final merged instructions (base + context-provider-extended) for the current agent invocation. +# Set by BaseAgent._prepare_session_and_messages() after context providers run, +# read by AgentTelemetryLayer to re-capture gen_ai.system_instructions with the full instructions. +AGENT_MERGED_INSTRUCTIONS: Final[contextvars.ContextVar[str | list[str] | None]] = contextvars.ContextVar( + "agent_merged_instructions", default=None +) + OTEL_METRICS: Final[str] = "__otel_metrics__" TOKEN_USAGE_BUCKET_BOUNDARIES: Final[tuple[float, ...]] = ( @@ -1541,6 +1548,7 @@ def _trace_agent_invocation( inner_response_telemetry_captured_fields ) inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({}) + agent_merged_instructions_token = AGENT_MERGED_INSTRUCTIONS.set(None) if stream: try: @@ -1602,6 +1610,21 @@ async def _finalize_stream() -> None: ) _apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields) _capture_response(span=span, attributes=response_attributes, duration=duration) + + # Re-capture system_instructions if context providers extended them. + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + merged_instructions = AGENT_MERGED_INSTRUCTIONS.get() + if merged_instructions is not None: + if not isinstance(merged_instructions, list): + merged_instructions = [merged_instructions] + otel_sys_instructions = [ + {"type": "text", "content": instruction} for instruction in merged_instructions + ] + span.set_attribute( + OtelAttr.SYSTEM_INSTRUCTIONS, + json.dumps(otel_sys_instructions, ensure_ascii=False), + ) + if ( OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and isinstance(response, AgentResponse) @@ -1618,6 +1641,7 @@ async def _finalize_stream() -> None: finally: INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) + AGENT_MERGED_INSTRUCTIONS.reset(agent_merged_instructions_token) _close_span() wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = result_stream.with_cleanup_hook( @@ -1642,6 +1666,24 @@ async def _run() -> AgentResponse[Any]: except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) raise + + # Re-capture system_instructions if context providers extended them. + # The initial capture above only sees base instructions from merged_options; + # AGENT_MERGED_INSTRUCTIONS is set by _prepare_session_and_messages() + # after context providers have run and merged their contributions. + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + merged_instructions = AGENT_MERGED_INSTRUCTIONS.get() + if merged_instructions is not None: + if not isinstance(merged_instructions, list): + merged_instructions = [merged_instructions] + otel_sys_instructions = [ + {"type": "text", "content": instruction} for instruction in merged_instructions + ] + span.set_attribute( + OtelAttr.SYSTEM_INSTRUCTIONS, + json.dumps(otel_sys_instructions, ensure_ascii=False), + ) + duration = perf_counter() - start_time_stamp if response: response_attributes = _get_response_attributes( @@ -1664,6 +1706,7 @@ async def _run() -> AgentResponse[Any]: finally: INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) + AGENT_MERGED_INSTRUCTIONS.reset(agent_merged_instructions_token) return _run() From bcdd7f53b622bd1bcfc370fb9e591d5102975b5b Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Thu, 16 Apr 2026 11:04:33 -0700 Subject: [PATCH 2/2] Address Copilot review: extract helper, fix GC leak, guard sensitive data --- .../packages/core/agent_framework/_agents.py | 9 ++-- .../core/agent_framework/observability.py | 54 ++++++++++--------- 2 files changed, 34 insertions(+), 29 deletions(-) diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index a8288bfd982..b94b6ed8828 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -1447,11 +1447,12 @@ async def _prepare_session_and_messages( else: chat_options["instructions"] = combined_instructions - # Publish the final merged instructions so AgentTelemetryLayer can - # re-capture gen_ai.system_instructions with provider-extended content. - from .observability import AGENT_MERGED_INSTRUCTIONS + # Only store merged instructions when sensitive-data capture is enabled, + # so provider-extended instruction text isn't retained unnecessarily. + from .observability import AGENT_MERGED_INSTRUCTIONS, OBSERVABILITY_SETTINGS - AGENT_MERGED_INSTRUCTIONS.set(chat_options.get("instructions")) + if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + AGENT_MERGED_INSTRUCTIONS.set(chat_options.get("instructions")) return session_context, chat_options diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 454e51a6573..829e38a283c 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -113,6 +113,25 @@ ) +def _recapture_system_instructions(span: trace.Span) -> None: + """Re-capture gen_ai.system_instructions from the AGENT_MERGED_INSTRUCTIONS ContextVar. + + Called after execute() completes so the span reflects provider-extended instructions + rather than just the base instructions captured before execution. + """ + if not OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: + return + merged_instructions = AGENT_MERGED_INSTRUCTIONS.get() + if merged_instructions is not None: + if not isinstance(merged_instructions, list): + merged_instructions = [merged_instructions] + otel_sys_instructions = [{"type": "text", "content": instruction} for instruction in merged_instructions] + span.set_attribute( + OtelAttr.SYSTEM_INSTRUCTIONS, + json.dumps(otel_sys_instructions, ensure_ascii=False), + ) + + OTEL_METRICS: Final[str] = "__otel_metrics__" TOKEN_USAGE_BUCKET_BOUNDARIES: Final[tuple[float, ...]] = ( 1, @@ -1612,18 +1631,7 @@ async def _finalize_stream() -> None: _capture_response(span=span, attributes=response_attributes, duration=duration) # Re-capture system_instructions if context providers extended them. - if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: - merged_instructions = AGENT_MERGED_INSTRUCTIONS.get() - if merged_instructions is not None: - if not isinstance(merged_instructions, list): - merged_instructions = [merged_instructions] - otel_sys_instructions = [ - {"type": "text", "content": instruction} for instruction in merged_instructions - ] - span.set_attribute( - OtelAttr.SYSTEM_INSTRUCTIONS, - json.dumps(otel_sys_instructions, ensure_ascii=False), - ) + _recapture_system_instructions(span) if ( OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED @@ -1647,7 +1655,14 @@ async def _finalize_stream() -> None: wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = result_stream.with_cleanup_hook( _record_duration ).with_cleanup_hook(_finalize_stream) - weakref.finalize(wrapped_stream, _close_span) + + def _gc_cleanup() -> None: + INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) + INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) + AGENT_MERGED_INSTRUCTIONS.reset(agent_merged_instructions_token) + _close_span() + + weakref.finalize(wrapped_stream, _gc_cleanup) return wrapped_stream async def _run() -> AgentResponse[Any]: @@ -1671,18 +1686,7 @@ async def _run() -> AgentResponse[Any]: # The initial capture above only sees base instructions from merged_options; # AGENT_MERGED_INSTRUCTIONS is set by _prepare_session_and_messages() # after context providers have run and merged their contributions. - if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED: - merged_instructions = AGENT_MERGED_INSTRUCTIONS.get() - if merged_instructions is not None: - if not isinstance(merged_instructions, list): - merged_instructions = [merged_instructions] - otel_sys_instructions = [ - {"type": "text", "content": instruction} for instruction in merged_instructions - ] - span.set_attribute( - OtelAttr.SYSTEM_INSTRUCTIONS, - json.dumps(otel_sys_instructions, ensure_ascii=False), - ) + _recapture_system_instructions(span) duration = perf_counter() - start_time_stamp if response: