Python: Fix: AgentTelemetryLayer captures context-provider-extended instructions in gen_ai.system_instructions

python/packages/core/agent_framework/_agents.py

@@ -1447,6 +1447,13 @@ async def _prepare_session_and_messages(
             else:
                 chat_options["instructions"] = combined_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
+
+        if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED:
+            AGENT_MERGED_INSTRUCTIONS.set(chat_options.get("instructions"))
+
         return session_context, chat_options
 
     def as_mcp_server(

python/packages/core/agent_framework/observability.py

@@ -105,6 +105,32 @@
     "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
+)
+
+
+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, ...]] = (
@@ -1541,6 +1567,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 +1629,10 @@ 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.
+                    _recapture_system_instructions(span)
+
                     if (
                         OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED
                         and isinstance(response, AgentResponse)
@@ -1618,12 +1649,20 @@ 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(
                 _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]:
@@ -1642,6 +1681,13 @@ 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.
+                    _recapture_system_instructions(span)
+
                     duration = perf_counter() - start_time_stamp
                     if response:
                         response_attributes = _get_response_attributes(
@@ -1664,6 +1710,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()