Set gen_ai.workflow.name span attribute on workflow invocations

WorkflowInvocation now sets gen_ai.workflow.name at span creation when the
workflow name is known, so it is available for sampling.

Also add a rego live-check manifest for invoke_workflow (expected attributes
and span-name format) that enforces gen_ai.workflow.name is emitted.

Author: lmolkova

policies/genai_span_validation.rego

@@ -68,6 +68,7 @@ _span_name_keyed_attr["embeddings"]        := "gen_ai.request.model"
 _span_name_keyed_attr["execute_tool"]      := "gen_ai.tool.name"
 _span_name_keyed_attr["invoke_agent"]      := "gen_ai.agent.name"
 _span_name_keyed_attr["create_agent"]      := "gen_ai.agent.name"
+_span_name_keyed_attr["invoke_workflow"]   := "gen_ai.workflow.name"
 
 # Span name SHOULD be `{op}` (when the keyed attribute is absent) or
 # `{op} {value}` (when present). Mirrors the "SHOULD append when known"
@@ -123,6 +124,8 @@ _expected_for_op("create_agent", _) := _create_agent_expected
 
 _expected_for_op("retrieval", _) := _retrieval_expected
 
+_expected_for_op("invoke_workflow", _) := _invoke_workflow_expected
+
 # Inference (chat / generate_content / text_completion).
 # Required: gen_ai.operation.name, gen_ai.provider.name.
 # Always-emit Recommended: response model/id, finish reasons, token usage,
@@ -191,6 +194,14 @@ _retrieval_expected := {
 	"server.address",
 }
 
+# Invoke workflow. Only gen_ai.operation.name is unconditionally required;
+# gen_ai.workflow.name is conditionally required "when available" but is
+# effectively always known where a workflow is instrumented, so flag it.
+_invoke_workflow_expected := {
+	"gen_ai.operation.name",
+	"gen_ai.workflow.name",
+}
+
 # Per expected attribute, one violation if missing.
 deny contains _span_finding(
 	"genai_expected_attribute_missing",

util/opentelemetry-util-genai/.changelog/+workflow-name-attribute.fixed

@@ -0,0 +1 @@
+Set the `gen_ai.workflow.name` span attribute on workflow invocations when the workflow name is known.

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py

@@ -62,6 +62,8 @@ def _get_base_attributes(self) -> dict[str, AttributeValue]:
         attrs: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
         }
+        if self.name:
+            attrs[GenAI.GEN_AI_WORKFLOW_NAME] = self.name
         return attrs
 
     def _get_messages_for_span(self) -> dict[str, AttributeValue]:

util/opentelemetry-util-genai/tests/test_handler_workflow.py

@@ -86,6 +86,24 @@ def test_start_workflow_records_monotonic_start(self) -> None:
         self.assertEqual(invocation._monotonic_start_s, 500.0)
         invocation.stop()
 
+    def test_start_workflow_sets_workflow_name_attribute(self) -> None:
+        invocation = self.handler.workflow(name="my_pipeline")
+        invocation.stop()
+
+        spans = self._get_finished_spans()
+        value = spans[0].attributes[GenAI.GEN_AI_WORKFLOW_NAME]
+        self.assertEqual(value, "my_pipeline")
+        self.assertIsInstance(value, str)
+
+    def test_start_workflow_without_name_omits_workflow_name_attribute(
+        self,
+    ) -> None:
+        invocation = self.handler.workflow(name=None)
+        invocation.stop()
+
+        spans = self._get_finished_spans()
+        self.assertNotIn(GenAI.GEN_AI_WORKFLOW_NAME, spans[0].attributes)
+
     # ------------------------------------------------------------------
     # stop_workflow
     # ------------------------------------------------------------------
@@ -217,6 +235,9 @@ def get_description(self):
         self.assertEqual(
             captured_attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow"
         )
+        self.assertEqual(
+            captured_attributes[GenAI.GEN_AI_WORKFLOW_NAME], "my-workflow"
+        )
 
         spans = self._get_finished_spans()
         self.assertEqual(len(spans), 1)