From 51dc35679ae26179197b6f670401aa892bf54a5f Mon Sep 17 00:00:00 2001 From: liangcheng <10039911+slcnx@user.noreply.gitee.com> Date: Fri, 7 Aug 2026 15:53:32 +0800 Subject: [PATCH 1/2] Python: capture workflow telemetry input and output --- .../agent_framework/_workflows/_executor.py | 22 +++- .../agent_framework/_workflows/_workflow.py | 30 ++++- .../_workflows/_workflow_context.py | 18 ++- .../core/agent_framework/observability.py | 38 +++++++ .../workflow/test_workflow_observability.py | 106 ++++++++++++++++++ 5 files changed, 209 insertions(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_executor.py b/python/packages/core/agent_framework/_workflows/_executor.py index 3571df2557c..869841a18df 100644 --- a/python/packages/core/agent_framework/_workflows/_executor.py +++ b/python/packages/core/agent_framework/_workflows/_executor.py @@ -11,7 +11,11 @@ from collections.abc import Awaitable, Callable from typing import Any, TypeVar, overload -from ..observability import create_processing_span +from ..observability import ( + OtelAttr, + _set_sensitive_span_attributes, # pyright: ignore[reportPrivateUsage] + create_processing_span, +) from ._events import ( WorkflowErrorDetails, WorkflowEvent, @@ -277,7 +281,7 @@ async def execute( type(message).__name__, source_trace_contexts=trace_contexts, source_span_ids=source_span_ids, - ): + ) as span: # Find the handler and handler spec that matches the message type. handler = self._find_handler(message) @@ -286,6 +290,13 @@ async def execute( # Unwrap raw data for handler call message = message.data + _set_sensitive_span_attributes( + span, + message, + (OtelAttr.EXECUTOR_INPUT, OtelAttr.INPUT_VALUE), + (OtelAttr.INPUT_MIME_TYPE,), + ) + # Create the appropriate WorkflowContext based on handler specs context = self._create_context_for_handler( source_executor_ids=source_executor_ids, @@ -316,6 +327,13 @@ async def execute( sent_messages = context.get_sent_messages() yielded_outputs = context.get_yielded_outputs() completion_data = sent_messages + yielded_outputs + if completion_data: + _set_sensitive_span_attributes( + span, + completion_data, + (OtelAttr.EXECUTOR_OUTPUT, OtelAttr.OUTPUT_VALUE), + (OtelAttr.OUTPUT_MIME_TYPE,), + ) completed_event = WorkflowEvent.executor_completed( self.id, completion_data if completion_data else None ) diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 77060b93a91..dbad56c2104 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -19,7 +19,12 @@ from .._sessions import ContextProvider from .._types import ResponseStream from ..exceptions import WorkflowException -from ..observability import OtelAttr, capture_exception, create_workflow_span +from ..observability import ( + OtelAttr, + _set_sensitive_span_attributes, # pyright: ignore[reportPrivateUsage] + capture_exception, + create_workflow_span, +) from ._checkpoint import CheckpointStorage from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, INTERNAL_SOURCE_ID, WORKFLOW_RUN_KWARGS_KEY from ._edge import ( @@ -478,6 +483,7 @@ def get_executors_list(self) -> list[Executor]: async def _run_workflow_with_tracing( self, initial_executor_fn: Callable[[], Awaitable[None]] | None = None, + telemetry_input: Any | None = None, is_continuation: bool = False, streaming: bool = False, function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, @@ -490,6 +496,7 @@ async def _run_workflow_with_tracing( Args: initial_executor_fn: Optional function to execute initial executor. + telemetry_input: Input payload for this run, captured only when sensitive telemetry is enabled. is_continuation: True when this run is a continuation of prior work (a checkpoint restore or a responses-only replay) rather than a fresh new turn delivered via the start executor with @@ -517,9 +524,20 @@ async def _run_workflow_with_tracing( OtelAttr.WORKFLOW_RUN_SPAN, attributes, ) as span: + from ..observability import OBSERVABILITY_SETTINGS + + capture_workflow_io = OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and span.is_recording() saw_request = False emitted_in_progress_pending = False + workflow_outputs: list[Any] | None = [] if capture_workflow_io else None try: + if capture_workflow_io and telemetry_input is not None: + _set_sensitive_span_attributes( + span, + telemetry_input, + (OtelAttr.INPUT_VALUE,), + (OtelAttr.INPUT_MIME_TYPE,), + ) # Add workflow started event (telemetry + surface state to consumers) span.add_event(OtelAttr.WORKFLOW_STARTED) # Emit explicit start/status events to the stream @@ -577,6 +595,8 @@ async def _run_workflow_with_tracing( # Track request events for final status determination if event.type == "request_info": saw_request = True + elif workflow_outputs is not None and event.type == "output": + workflow_outputs.append(event.data) yield event if event.type == "request_info" and not emitted_in_progress_pending: @@ -597,6 +617,13 @@ async def _run_workflow_with_tracing( terminal_status = WorkflowEvent.status(self._status) yield terminal_status + if workflow_outputs: + _set_sensitive_span_attributes( + span, + workflow_outputs, + (OtelAttr.OUTPUT_VALUE,), + (OtelAttr.OUTPUT_MIME_TYPE,), + ) span.add_event(OtelAttr.WORKFLOW_COMPLETED) except Exception as exc: # Drain any pending events (for example, executor_failed) before yielding failed event @@ -875,6 +902,7 @@ async def _run_core( async for event in self._run_workflow_with_tracing( initial_executor_fn=initial_executor_fn, + telemetry_input=message if message is not None else responses, is_continuation=(message is None), streaming=streaming, function_invocation_kwargs=function_invocation_kwargs, diff --git a/python/packages/core/agent_framework/_workflows/_workflow_context.py b/python/packages/core/agent_framework/_workflows/_workflow_context.py index 18a53e0bfa6..0fdf7c96983 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_context.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_context.py @@ -13,7 +13,11 @@ from opentelemetry.trace import SpanKind from typing_extensions import Never, TypeVar -from ..observability import OtelAttr, create_workflow_span +from ..observability import ( + OtelAttr, + _set_sensitive_span_attributes, # pyright: ignore[reportPrivateUsage] + create_workflow_span, +) from ._events import ( WorkflowEvent, WorkflowEventSource, @@ -327,10 +331,20 @@ async def send_message(self, message: OutT, target_id: str | None = None) -> Non from ..observability import OBSERVABILITY_SETTINGS # Create publishing span (inherits current trace context automatically) - attributes: dict[str, str] = {OtelAttr.MESSAGE_TYPE: type(message).__name__} + attributes: dict[str, str] = { + OtelAttr.MESSAGE_TYPE: type(message).__name__, + OtelAttr.MESSAGE_SOURCE_ID: self._executor_id, + } if target_id: attributes[OtelAttr.MESSAGE_DESTINATION_EXECUTOR_ID] = target_id + attributes[OtelAttr.MESSAGE_TARGET_ID] = target_id with create_workflow_span(OtelAttr.MESSAGE_SEND_SPAN, attributes, kind=SpanKind.PRODUCER) as span: + _set_sensitive_span_attributes( + span, + message, + (OtelAttr.MESSAGE_CONTENT, OtelAttr.INPUT_VALUE), + (OtelAttr.INPUT_MIME_TYPE,), + ) # Create Message wrapper msg = WorkflowMessage(data=message, source_id=self._executor_id, target_id=target_id) diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 5b40626875f..37677a16cd4 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -44,6 +44,7 @@ from . import __version__ as version_info from ._serialization import ( _is_serialization_protocol, # pyright: ignore[reportPrivateUsage] + make_json_safe, ) from ._settings import load_settings @@ -304,6 +305,8 @@ class OtelAttr(str, Enum): EXECUTOR_PROCESS_SPAN = "executor.process" EXECUTOR_ID = "executor.id" EXECUTOR_TYPE = "executor.type" + EXECUTOR_INPUT = "executor.input" + EXECUTOR_OUTPUT = "executor.output" # Edge group attributes EDGE_GROUP_PROCESS_SPAN = "edge_group.process" EDGE_GROUP_TYPE = "edge_group.type" @@ -317,6 +320,15 @@ class OtelAttr(str, Enum): MESSAGE_TYPE = "message.type" MESSAGE_PAYLOAD_TYPE = "message.payload_type" MESSAGE_DESTINATION_EXECUTOR_ID = "message.destination_executor_id" + MESSAGE_CONTENT = "message.content" + + # OpenInference attributes for vendor-neutral input/output ingestion. + # https://arize-ai.github.io/openinference/spec/semantic_conventions.html + INPUT_VALUE = "input.value" + INPUT_MIME_TYPE = "input.mime_type" + OUTPUT_VALUE = "output.value" + OUTPUT_MIME_TYPE = "output.mime_type" + JSON_MIME_TYPE = "application/json" # Activity events EVENT_NAME = "event.name" @@ -356,6 +368,32 @@ def __str__(self) -> str: return self.value +def _serialize_for_telemetry(value: Any) -> str: + """Serialize heterogeneous telemetry payloads without affecting application execution.""" + try: + return json.dumps(make_json_safe(value), ensure_ascii=False) + except Exception: + value_type = f"{type(value).__module__}.{type(value).__qualname__}" + return json.dumps(f"[Unserializable: {value_type}]", ensure_ascii=False) + + +def _set_sensitive_span_attributes( # pyright: ignore[reportUnusedFunction] + span: trace.Span, + value: Any, + value_attributes: Sequence[str | OtelAttr], + mime_type_attributes: Sequence[str | OtelAttr] = (), +) -> None: + """Set serialized payload attributes only when sensitive telemetry is enabled.""" + if not OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED or not span.is_recording(): + return + + serialized_value = _serialize_for_telemetry(value) + for attribute in value_attributes: + span.set_attribute(attribute, serialized_value) + for attribute in mime_type_attributes: + span.set_attribute(attribute, OtelAttr.JSON_MIME_TYPE) + + ROLE_EVENT_MAP = { "system": OtelAttr.SYSTEM_MESSAGE, "user": OtelAttr.USER_MESSAGE, diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py index eb2772be124..f69c0db9b62 100644 --- a/python/packages/core/tests/workflow/test_workflow_observability.py +++ b/python/packages/core/tests/workflow/test_workflow_observability.py @@ -98,6 +98,17 @@ def processed_messages(self) -> list[Any]: return self._processed_messages +class OutputExecutor(Executor): + """Executor that yields a structured workflow output for telemetry tests.""" + + def __init__(self, id: str = "output_executor") -> None: + super().__init__(id=id) + + @handler + async def handle_message(self, message: dict[str, int], ctx: WorkflowContext[Any, dict[str, int]]) -> None: + await ctx.yield_output({"result": message["value"] + 1}) + + async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None: """Test creation and attributes of all span types (workflow, processing, sending).""" # Create a mock workflow object @@ -229,6 +240,101 @@ async def test_trace_context_handling(span_exporter: InMemorySpanExporter) -> No assert processing_span.attributes.get("message.payload_type") == "str" +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_workflow_payloads_are_captured_when_sensitive_data_enabled( + span_exporter: InMemorySpanExporter, +) -> None: + workflow = WorkflowBuilder(start_executor=OutputExecutor()).build() + span_exporter.clear() + + async for _ in workflow.run({"value": 1}, stream=True): + pass + + spans = span_exporter.get_finished_spans() + workflow_span = next(span for span in spans if span.name == OtelAttr.WORKFLOW_RUN_SPAN) + executor_span = next(span for span in spans if span.name == "executor.process output_executor") + + assert workflow_span.attributes is not None + assert workflow_span.attributes[OtelAttr.INPUT_VALUE] == '{"value": 1}' + assert workflow_span.attributes[OtelAttr.OUTPUT_VALUE] == '[{"result": 2}]' + assert workflow_span.attributes[OtelAttr.INPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE + assert workflow_span.attributes[OtelAttr.OUTPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE + + assert executor_span.attributes is not None + assert executor_span.attributes[OtelAttr.EXECUTOR_INPUT] == '{"value": 1}' + assert executor_span.attributes[OtelAttr.EXECUTOR_OUTPUT] == '[{"result": 2}]' + assert executor_span.attributes[OtelAttr.INPUT_VALUE] == '{"value": 1}' + assert executor_span.attributes[OtelAttr.OUTPUT_VALUE] == '[{"result": 2}]' + + +@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True) +async def test_workflow_payloads_are_omitted_when_sensitive_data_disabled( + span_exporter: InMemorySpanExporter, +) -> None: + workflow = WorkflowBuilder(start_executor=OutputExecutor()).build() + span_exporter.clear() + + async for _ in workflow.run({"value": 1}, stream=True): + pass + + payload_attributes = { + OtelAttr.EXECUTOR_INPUT, + OtelAttr.EXECUTOR_OUTPUT, + OtelAttr.MESSAGE_CONTENT, + OtelAttr.INPUT_VALUE, + OtelAttr.OUTPUT_VALUE, + OtelAttr.INPUT_MIME_TYPE, + OtelAttr.OUTPUT_MIME_TYPE, + } + for span in span_exporter.get_finished_spans(): + assert span.attributes is not None + assert payload_attributes.isdisjoint(span.attributes) + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_message_send_captures_content_and_routing(span_exporter: InMemorySpanExporter) -> None: + workflow_ctx: WorkflowContext[str] = WorkflowContext( + MockExecutor("source_executor"), + ["source"], + State(), + InProcRunnerContext(), + ) + + await workflow_ctx.send_message("hello", target_id="target_executor") + + sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN) + assert sending_span.attributes is not None + assert sending_span.attributes[OtelAttr.MESSAGE_SOURCE_ID] == "source_executor" + assert sending_span.attributes[OtelAttr.MESSAGE_TARGET_ID] == "target_executor" + assert sending_span.attributes[OtelAttr.MESSAGE_CONTENT] == '"hello"' + assert sending_span.attributes[OtelAttr.INPUT_VALUE] == '"hello"' + assert sending_span.attributes[OtelAttr.INPUT_MIME_TYPE] == OtelAttr.JSON_MIME_TYPE + + +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_message_send_telemetry_serialization_cannot_break_workflow( + span_exporter: InMemorySpanExporter, +) -> None: + class UnserializablePayload: + __slots__ = () + + def __str__(self) -> str: + raise RuntimeError("cannot stringify") + + workflow_ctx: WorkflowContext[Any] = WorkflowContext( + MockExecutor("source_executor"), + ["source"], + State(), + InProcRunnerContext(), + ) + + await workflow_ctx.send_message(UnserializablePayload()) + + sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN) + assert sending_span.attributes is not None + assert "[Unserializable:" in str(sending_span.attributes[OtelAttr.MESSAGE_CONTENT]) + + @pytest.mark.parametrize("enable_instrumentation", [False], indirect=True) async def test_trace_context_disabled_when_tracing_disabled( enable_instrumentation: bool, span_exporter: InMemorySpanExporter From 7f0c8faa98201456788538f5a50588c829b1e647 Mon Sep 17 00:00:00 2001 From: liangcheng <10039911+slcnx@user.noreply.gitee.com> Date: Fri, 7 Aug 2026 16:25:20 +0800 Subject: [PATCH 2/2] Python: harden workflow telemetry serialization --- .../agent_framework/_workflows/_workflow.py | 5 +++- .../core/agent_framework/observability.py | 2 +- .../workflow/test_workflow_observability.py | 23 ++++++++++++++++++- 3 files changed, 27 insertions(+), 3 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index dbad56c2104..ba6760b8f17 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -899,10 +899,13 @@ async def _run_core( ) initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage) + telemetry_input = message.data if isinstance(message, WorkflowMessage) else message + if telemetry_input is None: + telemetry_input = responses async for event in self._run_workflow_with_tracing( initial_executor_fn=initial_executor_fn, - telemetry_input=message if message is not None else responses, + telemetry_input=telemetry_input, is_continuation=(message is None), streaming=streaming, function_invocation_kwargs=function_invocation_kwargs, diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 37677a16cd4..6953ed71e5e 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -371,7 +371,7 @@ def __str__(self) -> str: def _serialize_for_telemetry(value: Any) -> str: """Serialize heterogeneous telemetry payloads without affecting application execution.""" try: - return json.dumps(make_json_safe(value), ensure_ascii=False) + return json.dumps(make_json_safe(value), ensure_ascii=False, allow_nan=False) except Exception: value_type = f"{type(value).__module__}.{type(value).__qualname__}" return json.dumps(f"[Unserializable: {value_type}]", ensure_ascii=False) diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py index f69c0db9b62..05005459810 100644 --- a/python/packages/core/tests/workflow/test_workflow_observability.py +++ b/python/packages/core/tests/workflow/test_workflow_observability.py @@ -247,7 +247,8 @@ async def test_workflow_payloads_are_captured_when_sensitive_data_enabled( workflow = WorkflowBuilder(start_executor=OutputExecutor()).build() span_exporter.clear() - async for _ in workflow.run({"value": 1}, stream=True): + workflow_message = WorkflowMessage(data={"value": 1}, source_id="external", target_id=None) + async for _ in workflow.run(workflow_message, stream=True): pass spans = span_exporter.get_finished_spans() @@ -335,6 +336,26 @@ def __str__(self) -> str: assert "[Unserializable:" in str(sending_span.attributes[OtelAttr.MESSAGE_CONTENT]) +@pytest.mark.parametrize("non_finite_value", [float("nan"), float("inf"), float("-inf")]) +@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True) +async def test_message_send_telemetry_rejects_non_finite_json_values( + span_exporter: InMemorySpanExporter, + non_finite_value: float, +) -> None: + workflow_ctx: WorkflowContext[float] = WorkflowContext( + MockExecutor("source_executor"), + ["source"], + State(), + InProcRunnerContext(), + ) + + await workflow_ctx.send_message(non_finite_value) + + sending_span = next(span for span in span_exporter.get_finished_spans() if span.name == OtelAttr.MESSAGE_SEND_SPAN) + assert sending_span.attributes is not None + assert sending_span.attributes[OtelAttr.MESSAGE_CONTENT] == '"[Unserializable: builtins.float]"' + + @pytest.mark.parametrize("enable_instrumentation", [False], indirect=True) async def test_trace_context_disabled_when_tracing_disabled( enable_instrumentation: bool, span_exporter: InMemorySpanExporter