From db7ed22ba370d63e3f67527f21229ba7c4d8245b Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Fri, 30 Oct 2026 15:02:56 -0400
Subject: [PATCH 1/6] Python: align workflow span name with OpenTelemetry
 conventions

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: fa592ce6-05c6-4421-801b-59c77df4b2df
---
 .../ag-ui/tests/ag_ui/test_workflow_run.py    |  4 +-
 .../core/agent_framework/observability.py     |  2 +-
 .../workflow/test_functional_workflow.py      |  2 +-
 .../core/tests/workflow/test_workflow.py      |  2 +-
 .../workflow/test_workflow_observability.py   | 42 ++++++++++++++++---
 5 files changed, 42 insertions(+), 10 deletions(-)

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 7a418c6bdb6..0f131e76407 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
@@ -601,7 +601,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
             "ag-ui-workflow-thread",
         ]
 
-    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == "workflow.run"]
+    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == "invoke_workflow"]
     assert len(workflow_spans) == 2
     workflow_conversation_ids = []
     for span in workflow_spans:
@@ -631,7 +631,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
     assert fallback_agent_span.attributes is not None
     assert fallback_agent_span.attributes.get("gen_ai.conversation.id") == "provider-conversation"
 
-    fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == "workflow.run")
+    fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == "invoke_workflow")
     assert fallback_workflow_span.attributes is not None
     assert "gen_ai.conversation.id" not in fallback_workflow_span.attributes
 
diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py
index 9fec610e248..05af40e2ada 100644
--- a/python/packages/core/agent_framework/observability.py
+++ b/python/packages/core/agent_framework/observability.py
@@ -299,7 +299,7 @@ class OtelAttr(str, Enum):
     WORKFLOW_DESCRIPTION = "workflow.description"
     WORKFLOW_DEFINITION = "workflow.definition"
     WORKFLOW_BUILD_SPAN = "workflow.build"
-    WORKFLOW_RUN_SPAN = "workflow.run"
+    WORKFLOW_RUN_SPAN = "invoke_workflow"
     WORKFLOW_STARTED = "workflow.started"
     WORKFLOW_COMPLETED = "workflow.completed"
     WORKFLOW_ERROR = "workflow.error"
diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py
index 08b02ea84a8..f54fcf0d614 100644
--- a/python/packages/core/tests/workflow/test_functional_workflow.py
+++ b/python/packages/core/tests/workflow/test_functional_workflow.py
@@ -631,7 +631,7 @@ async def pipeline(x: int) -> int:
         assert follow_up.get_outputs() == [7]
 
     async def test_nested_processing_span_parents_under_workflow_run(self, span_exporter: Any) -> None:
-        """Spans opened during ``_execute`` must parent under the unattached ``workflow.run`` span."""
+        """Spans opened during ``_execute`` must parent under the unattached ``invoke_workflow`` span."""
         from agent_framework.observability import OtelAttr, create_processing_span
 
         @step
diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py
index 2ecc00c9d35..84df5e5001c 100644
--- a/python/packages/core/tests/workflow/test_workflow.py
+++ b/python/packages/core/tests/workflow/test_workflow.py
@@ -1434,7 +1434,7 @@ def capture_loop_exception(_loop: asyncio.AbstractEventLoop, context: dict[str,
         if "Failed to detach context" in rec.getMessage() or "was created in a different Context" in rec.getMessage()
     ]
     assert otel_errors == [], f"Abandoned stream leaked OpenTelemetry errors: {otel_errors!r}"
-    assert any(span.name == "workflow.run" for span in span_exporter.get_finished_spans())
+    assert any(span.name == "invoke_workflow" for span in span_exporter.get_finished_spans())
 
 
 async def test_workflow_unawaited_run_coroutine_releases_run_lock() -> None:
diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index b6a8082c8f7..01f62164eba 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -120,6 +120,38 @@ def processed_messages(self) -> list[Any]:
         return self._processed_messages
 
 
+@pytest.mark.parametrize("streaming", [False, True])
+@pytest.mark.parametrize("functional", [False, True])
+async def test_workflow_run_span_uses_semantic_convention_name(
+    span_exporter: InMemorySpanExporter, streaming: bool, functional: bool
+) -> None:
+    """Graph and functional workflow invocations use the standard span name."""
+    if functional:
+
+        @workflow
+        async def simple_workflow(message: str) -> str:
+            return message
+
+        workflow_instance = simple_workflow.build()
+    else:
+        workflow_instance = WorkflowBuilder(start_executor=MockExecutor()).build()
+
+    span_exporter.clear()
+    if streaming:
+        async for _ in workflow_instance.run("test input", stream=True):
+            pass
+    else:
+        await workflow_instance.run("test input")
+
+    run_spans = [
+        span
+        for span in span_exporter.get_finished_spans()
+        if any(event.name == "workflow.started" for event in span.events)
+    ]
+    assert len(run_spans) == 1
+    assert run_spans[0].name == "invoke_workflow"
+
+
 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
@@ -165,7 +197,7 @@ async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter)
     assert len(spans) == 3
 
     # Check workflow span
-    workflow_span = next(s for s in spans if s.name == "workflow.run")  # type: ignore[assignment, misc]
+    workflow_span = next(s for s in spans if s.name == "invoke_workflow")  # type: ignore[assignment, misc]
     assert workflow_span.kind == trace.SpanKind.INTERNAL  # type: ignore[attr-defined]
     assert workflow_span.attributes is not None  # type: ignore[attr-defined]
     assert workflow_span.attributes.get(OtelAttr.WORKFLOW_ID) == "test-workflow-123"  # type: ignore[attr-defined]
@@ -359,7 +391,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter)
 
     # Should have workflow span, processing spans, and sending spans
     # Processing spans now use executor_id as the span name, filter by executor.id attribute
-    workflow_spans = [s for s in spans if s.name == "workflow.run"]
+    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
     processing_spans = [s for s in spans if _span_attributes(s).get("executor.id") is not None]
     sending_spans = [s for s in spans if s.name == "message.send"]
     build_spans_after_run = [s for s in spans if s.name == "workflow.build"]
@@ -377,7 +409,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter)
     assert all(
         processing_span.parent is not None and processing_span.parent.span_id == workflow_span.context.span_id
         for processing_span in processing_spans
-    ), "Executor processing spans should be children of workflow.run"
+    ), "Executor processing spans should be children of invoke_workflow"
     assert workflow_span.events is not None
     event_names = [event.name for event in workflow_span.events]
     assert "workflow.started" in event_names
@@ -464,7 +496,7 @@ async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
     spans = span_exporter.get_finished_spans()
 
     # Find workflow span
-    workflow_spans = [s for s in spans if s.name == "workflow.run"]
+    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
     assert len(workflow_spans) == 1
 
     workflow_span = workflow_spans[0]
@@ -542,7 +574,7 @@ async def failing_workflow(message: str) -> None:
     _assert_workflow_error_details(executor_failed_events[0].details, private_error_detail, enable_sensitive_data)
 
     spans = span_exporter.get_finished_spans()
-    workflow_spans = [span for span in spans if span.name == "workflow.run"]
+    workflow_spans = [span for span in spans if span.name == "invoke_workflow"]
     assert len(workflow_spans) == 1
     span = workflow_spans[0]
     assert span.attributes is not None

From c4c2dd58b83b3cdadc610a7169ce6ce5605349ae Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Fri, 30 Oct 2026 15:16:52 -0400
Subject: [PATCH 2/6] Python: type graph and functional workflow regression
 test explicitly

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: fa592ce6-05c6-4421-801b-59c77df4b2df
---
 .../core/tests/workflow/test_workflow_observability.py         | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index 01f62164eba..22ecb233869 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -7,7 +7,7 @@
 from opentelemetry import trace
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
-from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder, step, workflow
+from agent_framework import FunctionalWorkflow, InMemoryCheckpointStorage, WorkflowBuilder, step, workflow
 from agent_framework._workflows._events import WorkflowErrorDetails, WorkflowEvent
 from agent_framework._workflows._executor import Executor, handler
 from agent_framework._workflows._runner_context import InProcRunnerContext, MessageType, WorkflowMessage
@@ -126,6 +126,7 @@ async def test_workflow_run_span_uses_semantic_convention_name(
     span_exporter: InMemorySpanExporter, streaming: bool, functional: bool
 ) -> None:
     """Graph and functional workflow invocations use the standard span name."""
+    workflow_instance: Workflow | FunctionalWorkflow
     if functional:
 
         @workflow

From 3b311d25a9e9110a58dcfab44844d289387e22f7 Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Mon, 2 Nov 2026 09:51:45 -0500
Subject: [PATCH 3/6] Python: align workflow name telemetry attribute with
 OpenTelemetry

Rename workflow.name to gen_ai.workflow.name and cover graph and functional workflows in streaming and non-streaming runs.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: aa4dae7a-fdbd-475c-8ec6-fa8333e77cc4
---
 .../core/agent_framework/observability.py       |  2 +-
 .../workflow/test_workflow_observability.py     | 17 ++++++++++++-----
 2 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py
index 05af40e2ada..5d414b4d915 100644
--- a/python/packages/core/agent_framework/observability.py
+++ b/python/packages/core/agent_framework/observability.py
@@ -295,7 +295,7 @@ class OtelAttr(str, Enum):
     WORKFLOW_ID = "workflow.id"
     WORKFLOW_BUILDER_NAME = "workflow_builder.name"
     WORKFLOW_BUILDER_DESCRIPTION = "workflow_builder.description"
-    WORKFLOW_NAME = "workflow.name"
+    WORKFLOW_NAME = "gen_ai.workflow.name"
     WORKFLOW_DESCRIPTION = "workflow.description"
     WORKFLOW_DEFINITION = "workflow.definition"
     WORKFLOW_BUILD_SPAN = "workflow.build"
diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index 22ecb233869..a42927b7c27 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -121,21 +121,24 @@ def processed_messages(self) -> list[Any]:
 
 
 @pytest.mark.parametrize("streaming", [False, True])
-@pytest.mark.parametrize("functional", [False, True])
+@pytest.mark.parametrize(
+    ("functional", "workflow_name"),
+    [(False, "customer_support_pipeline"), (True, "customer_support_pipeline"), (True, None)],
+)
 async def test_workflow_run_span_uses_semantic_convention_name(
-    span_exporter: InMemorySpanExporter, streaming: bool, functional: bool
+    span_exporter: InMemorySpanExporter, streaming: bool, functional: bool, workflow_name: str | None
 ) -> None:
-    """Graph and functional workflow invocations use the standard span name."""
+    """Graph and functional workflow invocations use standard span and workflow attribute names."""
     workflow_instance: Workflow | FunctionalWorkflow
     if functional:
 
-        @workflow
+        @workflow(name=workflow_name)
         async def simple_workflow(message: str) -> str:
             return message
 
         workflow_instance = simple_workflow.build()
     else:
-        workflow_instance = WorkflowBuilder(start_executor=MockExecutor()).build()
+        workflow_instance = WorkflowBuilder(start_executor=MockExecutor(), name=workflow_name).build()
 
     span_exporter.clear()
     if streaming:
@@ -151,6 +154,10 @@ async def simple_workflow(message: str) -> str:
     ]
     assert len(run_spans) == 1
     assert run_spans[0].name == "invoke_workflow"
+    attributes = run_spans[0].attributes
+    assert attributes is not None
+    assert "workflow.name" not in attributes
+    assert attributes["gen_ai.workflow.name"] == (workflow_name or "simple_workflow")
 
 
 async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None:

From 23760625da70a830fe0b280827b30e3d148bf38f Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Mon, 2 Nov 2026 09:53:53 -0500
Subject: [PATCH 4/6] Python: record workflow invocation operation attribute

Set gen_ai.operation.name to invoke_workflow when workflow invocation spans are created. Cover graph and functional workflows, attached span creation, and unchanged executor/message spans.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: aa4dae7a-fdbd-475c-8ec6-fa8333e77cc4
---
 python/packages/core/agent_framework/observability.py       | 6 ++++--
 .../core/tests/workflow/test_workflow_observability.py      | 4 ++++
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py
index 5d414b4d915..bc067187bf8 100644
--- a/python/packages/core/agent_framework/observability.py
+++ b/python/packages/core/agent_framework/observability.py
@@ -3663,8 +3663,10 @@ def _workflow_span_attributes(
 ) -> dict[str, str | int] | None:
     span_attributes: dict[str, str | int] = dict(attributes) if attributes is not None else {}
     conversation_id = _TELEMETRY_CONVERSATION_ID.get()
-    if name == OtelAttr.WORKFLOW_RUN_SPAN and conversation_id is not None:
-        span_attributes.setdefault(OtelAttr.CONVERSATION_ID, conversation_id)
+    if name == OtelAttr.WORKFLOW_RUN_SPAN:
+        span_attributes[OtelAttr.OPERATION] = "invoke_workflow"
+        if conversation_id is not None:
+            span_attributes.setdefault(OtelAttr.CONVERSATION_ID, conversation_id)
     return span_attributes or None
 
 
diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index a42927b7c27..c2e96d6a02c 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -158,6 +158,7 @@ async def simple_workflow(message: str) -> str:
     assert attributes is not None
     assert "workflow.name" not in attributes
     assert attributes["gen_ai.workflow.name"] == (workflow_name or "simple_workflow")
+    assert attributes["gen_ai.operation.name"] == "invoke_workflow"
 
 
 async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None:
@@ -183,6 +184,7 @@ async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter)
             OtelAttr.WORKFLOW_ID: mock_workflow.id,
         },
     ) as workflow_span:
+        assert _span_attributes(workflow_span)["gen_ai.operation.name"] == "invoke_workflow"
         workflow_span.add_event(OtelAttr.WORKFLOW_STARTED)
         sending_attributes: dict[str, str | int] = {
             OtelAttr.MESSAGE_TYPE: "ResponseMessage",
@@ -200,6 +202,8 @@ async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter)
             assert workflow_span is not None and workflow_span.is_recording()
             assert processing_span is not None and processing_span.is_recording()
             assert sending_span is not None and sending_span.is_recording()
+            assert "gen_ai.operation.name" not in _span_attributes(processing_span)
+            assert "gen_ai.operation.name" not in _span_attributes(sending_span)
 
     spans = span_exporter.get_finished_spans()
     assert len(spans) == 3

From 4fae4981648da922bbcc0d2c1f07122dcf8de5fb Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Mon, 2 Nov 2026 09:58:23 -0500
Subject: [PATCH 5/6] Python: include workflow names in invocation spans

Prepare workflow span names and attributes together so named invocation spans preserve operation and scoped conversation attributes. Cover both span factories, graph and functional workflows, AG-UI propagation, and unnamed fallback behavior.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: aa4dae7a-fdbd-475c-8ec6-fa8333e77cc4
---
 .../ag-ui/tests/ag_ui/test_workflow_run.py    |  6 ++--
 .../core/agent_framework/observability.py     | 14 +++++---
 .../core/tests/core/test_observability.py     | 32 ++++++++++++++-----
 .../workflow/test_functional_workflow.py      |  2 +-
 .../core/tests/workflow/test_workflow.py      |  2 +-
 .../workflow/test_workflow_observability.py   | 11 ++++---
 .../references/observability.md               |  4 +++
 7 files changed, 49 insertions(+), 22 deletions(-)

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 0f131e76407..d434d1d4dbf 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
@@ -601,7 +601,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
             "ag-ui-workflow-thread",
         ]
 
-    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == "invoke_workflow"]
+    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == f"invoke_workflow {workflow.name}"]
     assert len(workflow_spans) == 2
     workflow_conversation_ids = []
     for span in workflow_spans:
@@ -631,7 +631,9 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
     assert fallback_agent_span.attributes is not None
     assert fallback_agent_span.attributes.get("gen_ai.conversation.id") == "provider-conversation"
 
-    fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == "invoke_workflow")
+    fallback_workflow_span = next(
+        span for span in exporter.get_finished_spans() if span.name == f"invoke_workflow {workflow.name}"
+    )
     assert fallback_workflow_span.attributes is not None
     assert "gen_ai.conversation.id" not in fallback_workflow_span.attributes
 
diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py
index bc067187bf8..71afdf94bbd 100644
--- a/python/packages/core/agent_framework/observability.py
+++ b/python/packages/core/agent_framework/observability.py
@@ -3657,17 +3657,19 @@ def workflow_tracer() -> Tracer:
     return get_tracer() if OBSERVABILITY_SETTINGS.ENABLED else trace.NoOpTracer()
 
 
-def _workflow_span_attributes(
+def _prepare_workflow_span(
     name: str,
     attributes: Mapping[str, str | int] | None = None,
-) -> dict[str, str | int] | None:
+) -> tuple[str, dict[str, str | int] | None]:
     span_attributes: dict[str, str | int] = dict(attributes) if attributes is not None else {}
     conversation_id = _TELEMETRY_CONVERSATION_ID.get()
     if name == OtelAttr.WORKFLOW_RUN_SPAN:
         span_attributes[OtelAttr.OPERATION] = "invoke_workflow"
         if conversation_id is not None:
             span_attributes.setdefault(OtelAttr.CONVERSATION_ID, conversation_id)
-    return span_attributes or None
+        if workflow_name := span_attributes.get(OtelAttr.WORKFLOW_NAME):
+            name = f"{name} {workflow_name}"
+    return name, span_attributes or None
 
 
 @contextlib.contextmanager
@@ -3695,11 +3697,12 @@ def create_workflow_span(
     generator is later closed on GC from a different ``Context``. Streaming
     workflow runs should use :func:`start_workflow_span` instead.
     """
+    name, span_attributes = _prepare_workflow_span(name, attributes)
     return _capture_span_exceptions(
         workflow_tracer().start_as_current_span(
             name,
             kind=kind,
-            attributes=_workflow_span_attributes(name, attributes),
+            attributes=span_attributes,
             record_exception=False,
             set_status_on_exception=False,
         )
@@ -3719,7 +3722,8 @@ def start_workflow_span(
     across a yield causes ``ValueError: Token was created in a different
     Context`` when the generator is garbage-collected.
     """
-    return workflow_tracer().start_span(name, kind=kind, attributes=_workflow_span_attributes(name, attributes))
+    name, span_attributes = _prepare_workflow_span(name, attributes)
+    return workflow_tracer().start_span(name, kind=kind, attributes=span_attributes)
 
 
 def create_processing_span(
diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py
index fab710a7c53..a3e8c8b4002 100644
--- a/python/packages/core/tests/core/test_observability.py
+++ b/python/packages/core/tests/core/test_observability.py
@@ -3074,31 +3074,44 @@ def test_start_workflow_span_does_not_attach_as_current(span_exporter: InMemoryS
     assert spans[0].attributes["key"] == "value"  # type: ignore[index]  # pyrefly: ignore[unsupported-operation]  # ty: ignore[not-subscriptable]
 
 
-def test_create_workflow_span_uses_scoped_conversation_id(span_exporter: InMemorySpanExporter) -> None:
-    """An ambient conversation id is applied only within its workflow execution scope."""
+@pytest.mark.parametrize("workflow_name", [None, "", "customer_support_pipeline"])
+@pytest.mark.parametrize("unattached", [False, True])
+def test_workflow_span_uses_scoped_conversation_id(
+    span_exporter: InMemorySpanExporter, workflow_name: str | None, unattached: bool
+) -> None:
+    """Named and unnamed workflow spans retain scoped conversation ids with either span factory."""
     from agent_framework.observability import (
         OtelAttr,
         _use_telemetry_conversation_id,  # pyright: ignore[reportPrivateUsage]
         create_workflow_span,
+        start_workflow_span,
     )
 
+    span_factory = start_workflow_span if unattached else create_workflow_span
+    attributes: dict[str, str | int] = {OtelAttr.WORKFLOW_NAME: workflow_name} if workflow_name is not None else {}
     span_exporter.clear()  # type: ignore[attr-defined]
     with _use_telemetry_conversation_id("application-thread"):
-        with create_workflow_span(OtelAttr.WORKFLOW_RUN_SPAN):
+        with span_factory(OtelAttr.WORKFLOW_RUN_SPAN, attributes=attributes):
             pass
-        with create_workflow_span(
+        with span_factory(
             OtelAttr.WORKFLOW_RUN_SPAN,
-            attributes={OtelAttr.CONVERSATION_ID: "explicit-thread"},
+            attributes={**attributes, OtelAttr.CONVERSATION_ID: "explicit-thread"},
         ):
             pass
-        with create_workflow_span(OtelAttr.MESSAGE_SEND_SPAN):
+        with span_factory(OtelAttr.MESSAGE_SEND_SPAN, attributes=attributes):
             pass
-    with create_workflow_span(OtelAttr.WORKFLOW_RUN_SPAN):
+    with span_factory(OtelAttr.WORKFLOW_RUN_SPAN, attributes=attributes):
         pass
 
     spans = span_exporter.get_finished_spans()  # type: ignore[attr-defined]
-    workflow_spans = [span for span in spans if span.name == OtelAttr.WORKFLOW_RUN_SPAN]
+    workflow_spans = [
+        span
+        for span in spans
+        if span.attributes is not None and span.attributes.get(OtelAttr.OPERATION) == "invoke_workflow"
+    ]
     assert len(workflow_spans) == 3
+    expected_name = f"invoke_workflow {workflow_name}" if workflow_name else "invoke_workflow"
+    assert all(span.name == expected_name for span in workflow_spans)
     ambient_attributes = workflow_spans[0].attributes
     explicit_attributes = workflow_spans[1].attributes
     unscoped_attributes = workflow_spans[2].attributes
@@ -3112,6 +3125,9 @@ def test_create_workflow_span_uses_scoped_conversation_id(span_exporter: InMemor
     message_send_attributes = message_send_span.attributes
     assert message_send_attributes is not None
     assert OtelAttr.CONVERSATION_ID not in message_send_attributes
+    assert OtelAttr.OPERATION not in message_send_attributes
+    assert OtelAttr.CONVERSATION_ID not in attributes
+    assert OtelAttr.OPERATION not in attributes
 
 
 def test_create_processing_span(span_exporter):
diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py
index f54fcf0d614..3709be62374 100644
--- a/python/packages/core/tests/workflow/test_functional_workflow.py
+++ b/python/packages/core/tests/workflow/test_functional_workflow.py
@@ -648,7 +648,7 @@ async def pipeline(x: int) -> int:
         assert result.get_outputs() == [6]
 
         spans = span_exporter.get_finished_spans()  # type: ignore[attr-defined]
-        run_spans = [s for s in spans if s.name == OtelAttr.WORKFLOW_RUN_SPAN]
+        run_spans = [s for s in spans if s.name == f"invoke_workflow {pipeline.name}"]
         process_spans = [s for s in spans if s.name == f"{OtelAttr.EXECUTOR_PROCESS_SPAN} traced_add"]
         assert len(run_spans) == 1
         assert len(process_spans) == 1
diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py
index 84df5e5001c..493e83f078f 100644
--- a/python/packages/core/tests/workflow/test_workflow.py
+++ b/python/packages/core/tests/workflow/test_workflow.py
@@ -1434,7 +1434,7 @@ def capture_loop_exception(_loop: asyncio.AbstractEventLoop, context: dict[str,
         if "Failed to detach context" in rec.getMessage() or "was created in a different Context" in rec.getMessage()
     ]
     assert otel_errors == [], f"Abandoned stream leaked OpenTelemetry errors: {otel_errors!r}"
-    assert any(span.name == "invoke_workflow" for span in span_exporter.get_finished_spans())
+    assert any(span.name == f"invoke_workflow {workflow.name}" for span in span_exporter.get_finished_spans())
 
 
 async def test_workflow_unawaited_run_coroutine_releases_run_lock() -> None:
diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index c2e96d6a02c..4cd5a860549 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -153,11 +153,12 @@ async def simple_workflow(message: str) -> str:
         if any(event.name == "workflow.started" for event in span.events)
     ]
     assert len(run_spans) == 1
-    assert run_spans[0].name == "invoke_workflow"
+    expected_name = workflow_name or "simple_workflow"
+    assert run_spans[0].name == f"invoke_workflow {expected_name}"
     attributes = run_spans[0].attributes
     assert attributes is not None
     assert "workflow.name" not in attributes
-    assert attributes["gen_ai.workflow.name"] == (workflow_name or "simple_workflow")
+    assert attributes["gen_ai.workflow.name"] == expected_name
     assert attributes["gen_ai.operation.name"] == "invoke_workflow"
 
 
@@ -403,7 +404,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter)
 
     # Should have workflow span, processing spans, and sending spans
     # Processing spans now use executor_id as the span name, filter by executor.id attribute
-    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
+    workflow_spans = [s for s in spans if s.name == f"invoke_workflow {workflow.name}"]
     processing_spans = [s for s in spans if _span_attributes(s).get("executor.id") is not None]
     sending_spans = [s for s in spans if s.name == "message.send"]
     build_spans_after_run = [s for s in spans if s.name == "workflow.build"]
@@ -508,7 +509,7 @@ async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
     spans = span_exporter.get_finished_spans()
 
     # Find workflow span
-    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
+    workflow_spans = [s for s in spans if s.name == f"invoke_workflow {workflow.name}"]
     assert len(workflow_spans) == 1
 
     workflow_span = workflow_spans[0]
@@ -586,7 +587,7 @@ async def failing_workflow(message: str) -> None:
     _assert_workflow_error_details(executor_failed_events[0].details, private_error_detail, enable_sensitive_data)
 
     spans = span_exporter.get_finished_spans()
-    workflow_spans = [span for span in spans if span.name == "invoke_workflow"]
+    workflow_spans = [span for span in spans if span.name == f"invoke_workflow {workflow_instance.name}"]
     assert len(workflow_spans) == 1
     span = workflow_spans[0]
     assert span.attributes is not None
diff --git a/skills/agent-framework-builder-python/references/observability.md b/skills/agent-framework-builder-python/references/observability.md
index b3f7b3e1abe..d6be8b8e3e7 100644
--- a/skills/agent-framework-builder-python/references/observability.md
+++ b/skills/agent-framework-builder-python/references/observability.md
@@ -17,6 +17,10 @@ runtime evidence; it does not determine whether output is correct.
 Inspect run boundaries, model calls, tool and MCP calls, workflow transitions, checkpoints, approvals, retries, usage,
 and hosted-service continuations. Use stable agent, workflow, and service identifiers.
 
+Workflow invocation spans are named `invoke_workflow {gen_ai.workflow.name}` when the workflow-name attribute is
+non-empty, or `invoke_workflow` otherwise. Their `gen_ai.operation.name` is always `invoke_workflow`; supplied
+conversation identifiers remain available as `gen_ai.conversation.id`.
+
 ## Learn more
 
 - [Agent observability](https://learn.microsoft.com/agent-framework/agents/observability)

From 38792a7c530f30d07201aeae2c27b8cc57e3e6de Mon Sep 17 00:00:00 2001
From: Vincent Biret <vibiret@microsoft.com>
Date: Mon, 2 Nov 2026 13:11:15 -0500
Subject: [PATCH 6/6] Python: omit generated workflow names from invocation
 telemetry

Track generated-name provenance while retaining runtime names and checkpoint identity. Cover unnamed and explicitly named graph workflows in streaming and non-streaming modes and preserve scoped conversation propagation through core and AG-UI paths.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: aa4dae7a-fdbd-475c-8ec6-fa8333e77cc4
---
 .../ag-ui/tests/ag_ui/test_workflow_run.py    | 18 +++--
 .../agent_framework/_workflows/_workflow.py   |  3 +-
 .../_workflows/_workflow_builder.py           |  7 +-
 .../core/tests/workflow/test_workflow.py      |  2 +-
 .../workflow/test_workflow_observability.py   | 69 ++++++++++++++++---
 .../references/observability.md               |  2 +
 6 files changed, 81 insertions(+), 20 deletions(-)

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 d434d1d4dbf..54a14bd3113 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
@@ -517,7 +517,10 @@ async def finalizer(message: str, ctx: WorkflowContext[None, str]) -> None:
     assert not [event for event in events if event.type == "CUSTOM" and event.name == "workflow_output"]  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
 
 
-async def test_workflow_and_agent_spans_use_supplied_agui_thread_id(monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.mark.parametrize("workflow_name", [None, "customer_support_pipeline"])
+async def test_workflow_and_agent_spans_use_supplied_agui_thread_id(
+    monkeypatch: pytest.MonkeyPatch, workflow_name: str | None
+) -> None:
     """Workflow spans use supplied AG-UI threads without replacing provider fallback behavior."""
     import agent_framework.observability as observability
     from opentelemetry.sdk.trace import TracerProvider
@@ -561,7 +564,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
         name="workflow-agent",
         middleware=[passthrough_middleware],
     )
-    workflow = WorkflowBuilder(start_executor=participant, output_from="all").build()
+    workflow = WorkflowBuilder(start_executor=participant, output_from="all", name=workflow_name).build()
 
     for run_number in (1, 2):
         events = [
@@ -601,11 +604,16 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
             "ag-ui-workflow-thread",
         ]
 
-    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == f"invoke_workflow {workflow.name}"]
+    expected_span_name = f"invoke_workflow {workflow_name}" if workflow_name else "invoke_workflow"
+    workflow_spans = [span for span in exporter.get_finished_spans() if span.name == expected_span_name]
     assert len(workflow_spans) == 2
     workflow_conversation_ids = []
     for span in workflow_spans:
         assert span.attributes is not None
+        if workflow_name:
+            assert span.attributes["gen_ai.workflow.name"] == workflow_name
+        else:
+            assert "gen_ai.workflow.name" not in span.attributes
         workflow_conversation_ids.append(span.attributes.get("gen_ai.conversation.id"))
 
     assert workflow_conversation_ids == [
@@ -631,9 +639,7 @@ async def passthrough_middleware(_context: AgentContext, call_next: Any) -> None
     assert fallback_agent_span.attributes is not None
     assert fallback_agent_span.attributes.get("gen_ai.conversation.id") == "provider-conversation"
 
-    fallback_workflow_span = next(
-        span for span in exporter.get_finished_spans() if span.name == f"invoke_workflow {workflow.name}"
-    )
+    fallback_workflow_span = next(span for span in exporter.get_finished_spans() if span.name == expected_span_name)
     assert fallback_workflow_span.attributes is not None
     assert "gen_ai.conversation.id" not in fallback_workflow_span.attributes
 
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index c60617dabd5..cd566778484 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -366,6 +366,7 @@ def __init__(
         self.start_executor_id = start_executor.id
         self.max_iterations = max_iterations
         self.name = name
+        self._generated_name: str | None = None
         self.description = description
         # Generate a unique ID for the workflow instance for monitoring purposes. This is not intended to be a
         # stable identifier across instances created from the same builder, for that, use the name field.
@@ -559,7 +560,7 @@ async def _run_workflow_with_tracing(
         """
         # Create workflow span that encompasses the entire execution
         attributes: dict[str, Any] = {OtelAttr.WORKFLOW_ID: self.id}
-        if self.name:
+        if self.name and self.name != self._generated_name:
             attributes[OtelAttr.WORKFLOW_NAME] = self.name
         if self.description:
             attributes[OtelAttr.WORKFLOW_DESCRIPTION] = self.description
diff --git a/python/packages/core/agent_framework/_workflows/_workflow_builder.py b/python/packages/core/agent_framework/_workflows/_workflow_builder.py
index 3eed9ca800a..084ddcf8fc4 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow_builder.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow_builder.py
@@ -1,5 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
 import logging
 import sys
 import uuid
@@ -110,7 +112,8 @@ def __init__(
                 for all workflow instances created from this builder. If not provided, a unique name
                 will be generated. This will be useful for versioning, monitoring, checkpointing, and
                 debugging workflows. Keeping this name unique across versions of your workflow definitions
-                is recommended for better observability and management.
+                is recommended for better observability and management. Generated names are omitted from
+                workflow invocation span names and the ``gen_ai.workflow.name`` attribute.
             description: Optional description of what the workflow does.
             start_executor: The starting executor for the workflow. Can be an Executor instance
                 or SupportsAgentRun instance.
@@ -153,6 +156,7 @@ def __init__(
         self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage
         self._max_iterations: int = max_iterations
         self._name: str = name or f"WorkflowBuilder-{uuid.uuid4()!s}"
+        self._generated_name: str | None = self._name if not name else None
         self._description: str | None = description
         # Maps underlying SupportsAgentRun object id -> wrapped Executor so we reuse the same wrapper
         # across start_executor / add_edge calls. This avoids multiple AgentExecutor instances
@@ -866,6 +870,7 @@ async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
                     output_from=output_for_workflow,
                     intermediate_output_from=intermediate_output_for_workflow,
                 )
+                workflow._generated_name = self._generated_name  # pyright: ignore[reportPrivateUsage]
                 build_attributes: dict[str, Any] = {
                     OtelAttr.WORKFLOW_BUILDER_NAME: self._name,
                     OtelAttr.WORKFLOW_ID: workflow.id,
diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py
index 493e83f078f..84df5e5001c 100644
--- a/python/packages/core/tests/workflow/test_workflow.py
+++ b/python/packages/core/tests/workflow/test_workflow.py
@@ -1434,7 +1434,7 @@ def capture_loop_exception(_loop: asyncio.AbstractEventLoop, context: dict[str,
         if "Failed to detach context" in rec.getMessage() or "was created in a different Context" in rec.getMessage()
     ]
     assert otel_errors == [], f"Abandoned stream leaked OpenTelemetry errors: {otel_errors!r}"
-    assert any(span.name == f"invoke_workflow {workflow.name}" for span in span_exporter.get_finished_spans())
+    assert any(span.name == "invoke_workflow" for span in span_exporter.get_finished_spans())
 
 
 async def test_workflow_unawaited_run_coroutine_releases_run_lock() -> None:
diff --git a/python/packages/core/tests/workflow/test_workflow_observability.py b/python/packages/core/tests/workflow/test_workflow_observability.py
index 4cd5a860549..676e678531f 100644
--- a/python/packages/core/tests/workflow/test_workflow_observability.py
+++ b/python/packages/core/tests/workflow/test_workflow_observability.py
@@ -16,6 +16,7 @@
 from agent_framework._workflows._workflow_context import WorkflowContext
 from agent_framework.observability import (
     OtelAttr,
+    _use_telemetry_conversation_id,  # pyright: ignore[reportPrivateUsage]
     create_processing_span,
     create_workflow_span,
 )
@@ -123,7 +124,14 @@ def processed_messages(self) -> list[Any]:
 @pytest.mark.parametrize("streaming", [False, True])
 @pytest.mark.parametrize(
     ("functional", "workflow_name"),
-    [(False, "customer_support_pipeline"), (True, "customer_support_pipeline"), (True, None)],
+    [
+        (False, "customer_support_pipeline"),
+        (False, None),
+        (False, ""),
+        (False, "WorkflowBuilder-00000000-0000-0000-0000-000000000000"),
+        (True, "customer_support_pipeline"),
+        (True, None),
+    ],
 )
 async def test_workflow_run_span_uses_semantic_convention_name(
     span_exporter: InMemorySpanExporter, streaming: bool, functional: bool, workflow_name: str | None
@@ -141,11 +149,12 @@ async def simple_workflow(message: str) -> str:
         workflow_instance = WorkflowBuilder(start_executor=MockExecutor(), name=workflow_name).build()
 
     span_exporter.clear()
-    if streaming:
-        async for _ in workflow_instance.run("test input", stream=True):
-            pass
-    else:
-        await workflow_instance.run("test input")
+    with _use_telemetry_conversation_id("workflow-thread"):
+        if streaming:
+            async for _ in workflow_instance.run("test input", stream=True):
+                pass
+        else:
+            await workflow_instance.run("test input")
 
     run_spans = [
         span
@@ -153,13 +162,51 @@ async def simple_workflow(message: str) -> str:
         if any(event.name == "workflow.started" for event in span.events)
     ]
     assert len(run_spans) == 1
-    expected_name = workflow_name or "simple_workflow"
-    assert run_spans[0].name == f"invoke_workflow {expected_name}"
+    expected_name = workflow_name or ("simple_workflow" if functional else None)
+    assert run_spans[0].name == (f"invoke_workflow {expected_name}" if expected_name else "invoke_workflow")
     attributes = run_spans[0].attributes
     assert attributes is not None
     assert "workflow.name" not in attributes
-    assert attributes["gen_ai.workflow.name"] == expected_name
+    if expected_name:
+        assert attributes["gen_ai.workflow.name"] == expected_name
+    else:
+        assert "gen_ai.workflow.name" not in attributes
     assert attributes["gen_ai.operation.name"] == "invoke_workflow"
+    assert attributes["gen_ai.conversation.id"] == "workflow-thread"
+
+
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_unnamed_workflow_preserves_runtime_name_and_checkpoint_identity(
+    span_exporter: InMemorySpanExporter, streaming: bool
+) -> None:
+    """Omitting generated telemetry names does not change runtime or checkpoint identity."""
+    storage = InMemoryCheckpointStorage()
+    workflow_instance = WorkflowBuilder(start_executor=MockExecutor(), checkpoint_storage=storage).build()
+    runtime_name = workflow_instance.name
+    assert runtime_name.startswith("WorkflowBuilder-")
+    other_workflow = WorkflowBuilder(start_executor=MockExecutor()).build()
+    assert other_workflow.name != runtime_name
+
+    span_exporter.clear()
+    if streaming:
+        async for _ in workflow_instance.run("test input", stream=True):
+            pass
+    else:
+        await workflow_instance.run("test input")
+
+    assert workflow_instance.name == runtime_name
+    assert workflow_instance.to_dict()["name"] == runtime_name
+    checkpoint = await storage.get_latest(workflow_name=runtime_name)
+    assert checkpoint is not None
+    assert checkpoint.workflow_name == runtime_name
+    run_span = next(
+        span
+        for span in span_exporter.get_finished_spans()
+        if span.attributes is not None and span.attributes.get("gen_ai.operation.name") == "invoke_workflow"
+    )
+    assert run_span.name == "invoke_workflow"
+    assert run_span.attributes is not None
+    assert "gen_ai.workflow.name" not in run_span.attributes
 
 
 async def test_span_creation_and_attributes(span_exporter: InMemorySpanExporter) -> None:
@@ -404,7 +451,7 @@ async def test_end_to_end_workflow_tracing(span_exporter: InMemorySpanExporter)
 
     # Should have workflow span, processing spans, and sending spans
     # Processing spans now use executor_id as the span name, filter by executor.id attribute
-    workflow_spans = [s for s in spans if s.name == f"invoke_workflow {workflow.name}"]
+    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
     processing_spans = [s for s in spans if _span_attributes(s).get("executor.id") is not None]
     sending_spans = [s for s in spans if s.name == "message.send"]
     build_spans_after_run = [s for s in spans if s.name == "workflow.build"]
@@ -509,7 +556,7 @@ async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
     spans = span_exporter.get_finished_spans()
 
     # Find workflow span
-    workflow_spans = [s for s in spans if s.name == f"invoke_workflow {workflow.name}"]
+    workflow_spans = [s for s in spans if s.name == "invoke_workflow"]
     assert len(workflow_spans) == 1
 
     workflow_span = workflow_spans[0]
diff --git a/skills/agent-framework-builder-python/references/observability.md b/skills/agent-framework-builder-python/references/observability.md
index d6be8b8e3e7..5337e27e9b4 100644
--- a/skills/agent-framework-builder-python/references/observability.md
+++ b/skills/agent-framework-builder-python/references/observability.md
@@ -20,6 +20,8 @@ and hosted-service continuations. Use stable agent, workflow, and service identi
 Workflow invocation spans are named `invoke_workflow {gen_ai.workflow.name}` when the workflow-name attribute is
 non-empty, or `invoke_workflow` otherwise. Their `gen_ai.operation.name` is always `invoke_workflow`; supplied
 conversation identifiers remain available as `gen_ai.conversation.id`.
+Graph workflows without an application-provided name retain their generated runtime/checkpoint name, but omit it
+from `gen_ai.workflow.name` and use the bare invocation span name. Functional workflows default to their function name.
 
 ## Learn more