Add parent_id parameter to OutputScope

Similar to the .NET ExecuteToolScope, the OutputScope now accepts an optional
parent_id parameter to link spans to upstream operations.

Co-authored-by: nikhilNava <211831449+nikhilNava@users.noreply.github.com>

Author: Copilot

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/output_scope.py

@@ -2,7 +2,7 @@
 # Licensed under the MIT License.
 
 from ..agent_details import AgentDetails
-from ..constants import GEN_AI_OUTPUT_MESSAGES_KEY
+from ..constants import CUSTOM_PARENT_SPAN_ID_KEY, GEN_AI_OUTPUT_MESSAGES_KEY
 from ..models.response import Response
 from ..opentelemetry_scope import OpenTelemetryScope
 from ..tenant_details import TenantDetails
@@ -19,31 +19,37 @@ def start(
         agent_details: AgentDetails,
         tenant_details: TenantDetails,
         response: Response,
+        parent_id: str | None = None,
     ) -> "OutputScope":
         """Creates and starts a new scope for output tracing.
 
         Args:
             agent_details: The details of the agent
             tenant_details: The details of the tenant
             response: The response details from the agent
+            parent_id: Optional parent Activity ID used to link this span to an upstream
+                operation
 
         Returns:
             A new OutputScope instance
         """
-        return OutputScope(agent_details, tenant_details, response)
+        return OutputScope(agent_details, tenant_details, response, parent_id)
 
     def __init__(
         self,
         agent_details: AgentDetails,
         tenant_details: TenantDetails,
         response: Response,
+        parent_id: str | None = None,
     ):
         """Initialize the output scope.
 
         Args:
             agent_details: The details of the agent
             tenant_details: The details of the tenant
             response: The response details from the agent
+            parent_id: Optional parent Activity ID used to link this span to an upstream
+                operation
         """
         super().__init__(
             kind="Client",
@@ -53,6 +59,9 @@ def __init__(
             tenant_details=tenant_details,
         )
 
+        # Set parent ID if provided
+        self.set_tag_maybe(CUSTOM_PARENT_SPAN_ID_KEY, parent_id)
+
         # Set response messages
         self.set_tag_maybe(GEN_AI_OUTPUT_MESSAGES_KEY, safe_json_dumps(response.messages))
 

tests/observability/core/test_output_scope.py

@@ -14,7 +14,10 @@
     get_tracer_provider,
 )
 from microsoft_agents_a365.observability.core.config import _telemetry_manager
-from microsoft_agents_a365.observability.core.constants import GEN_AI_OUTPUT_MESSAGES_KEY
+from microsoft_agents_a365.observability.core.constants import (
+    CUSTOM_PARENT_SPAN_ID_KEY,
+    GEN_AI_OUTPUT_MESSAGES_KEY,
+)
 from microsoft_agents_a365.observability.core.models.response import Response
 from microsoft_agents_a365.observability.core.opentelemetry_scope import OpenTelemetryScope
 from microsoft_agents_a365.observability.core.spans_scopes.output_scope import OutputScope
@@ -192,6 +195,54 @@ def test_output_scope_span_name(self):
         self.assertIn("output_messages", span.name)
         self.assertIn(self.agent_details.agent_id, span.name)
 
+    def test_output_scope_with_parent_id(self):
+        """Test that OutputScope records parent_id when provided."""
+        response = Response(messages=["Test message with parent"])
+        parent_id = "00-1234567890abcdef1234567890abcdef-abcdefabcdef1234-01"
+
+        scope = OutputScope.start(
+            self.agent_details, self.tenant_details, response, parent_id=parent_id
+        )
+
+        if scope is not None:
+            scope.dispose()
+
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
+
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
+
+        # Verify the parent ID is set as a span attribute
+        self.assertIn(
+            CUSTOM_PARENT_SPAN_ID_KEY,
+            span_attributes,
+            "Expected custom parent span ID to be set on span",
+        )
+        self.assertEqual(span_attributes[CUSTOM_PARENT_SPAN_ID_KEY], parent_id)
+
+    def test_output_scope_without_parent_id(self):
+        """Test that OutputScope doesn't set parent_id attribute when not provided."""
+        response = Response(messages=["Test message without parent"])
+
+        scope = OutputScope.start(self.agent_details, self.tenant_details, response)
+
+        if scope is not None:
+            scope.dispose()
+
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
+
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
+
+        # Verify the parent ID attribute is NOT set when not provided
+        self.assertNotIn(
+            CUSTOM_PARENT_SPAN_ID_KEY,
+            span_attributes,
+            "Expected custom parent span ID NOT to be set when not provided",
+        )
+
 
 if __name__ == "__main__":
     # Run pytest only on the current file