Add OutputScope for output message tracing

- Add Response model class for agent execution response details
- Add spans_scopes directory with OutputScope implementation
- Add unit tests for OutputScope functionality

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

Author: Copilot

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/models/response.py

@@ -0,0 +1,12 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Response:
+    """Response details from agent execution."""
+
+    messages: list[str]
+    """The list of response messages."""

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

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.

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

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from ..agent_details import AgentDetails
+from ..constants import GEN_AI_OUTPUT_MESSAGES_KEY
+from ..models.response import Response
+from ..opentelemetry_scope import OpenTelemetryScope
+from ..tenant_details import TenantDetails
+from ..utils import safe_json_dumps
+
+OUTPUT_OPERATION_NAME = "output_messages"
+
+
+class OutputScope(OpenTelemetryScope):
+    """Provides OpenTelemetry tracing scope for output messages."""
+
+    @staticmethod
+    def start(
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        response: Response,
+    ) -> "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
+
+        Returns:
+            A new OutputScope instance
+        """
+        return OutputScope(agent_details, tenant_details, response)
+
+    def __init__(
+        self,
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        response: Response,
+    ):
+        """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
+        """
+        super().__init__(
+            kind="Client",
+            operation_name=OUTPUT_OPERATION_NAME,
+            activity_name=(f"{OUTPUT_OPERATION_NAME} {agent_details.agent_id}"),
+            agent_details=agent_details,
+            tenant_details=tenant_details,
+        )
+
+        # Set response messages
+        self.set_tag_maybe(GEN_AI_OUTPUT_MESSAGES_KEY, safe_json_dumps(response.messages))
+
+    def record_output_messages(self, messages: list[str]) -> None:
+        """Records the output messages for telemetry tracking.
+
+        Args:
+            messages: List of output messages
+        """
+        self.set_tag_maybe(GEN_AI_OUTPUT_MESSAGES_KEY, safe_json_dumps(messages))

tests/observability/core/test_output_scope.py

@@ -0,0 +1,198 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import os
+import sys
+import unittest
+from pathlib import Path
+
+import pytest
+from microsoft_agents_a365.observability.core import (
+    AgentDetails,
+    TenantDetails,
+    configure,
+    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.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
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+
+class TestOutputScope(unittest.TestCase):
+    """Unit tests for OutputScope and its methods."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Set up test environment once for all tests."""
+        # Configure Microsoft Agent 365 for testing
+        os.environ["ENABLE_A365_OBSERVABILITY"] = "true"
+
+        configure(
+            service_name="test-output-scope-service",
+            service_namespace="test-namespace",
+        )
+        # Create test data
+        cls.tenant_details = TenantDetails(tenant_id="12345678-1234-5678-1234-567812345678")
+        cls.agent_details = AgentDetails(
+            agent_id="test-agent-123",
+            agent_name="Test Agent",
+            agent_description="A test agent for output scope testing",
+        )
+
+    def setUp(self):
+        super().setUp()
+
+        # Reset TelemetryManager state to ensure fresh configuration
+        _telemetry_manager._tracer_provider = None
+        _telemetry_manager._span_processors = {}
+        OpenTelemetryScope._tracer = None
+
+        # Reconfigure to get a fresh TracerProvider
+        configure(
+            service_name="test-output-scope-service",
+            service_namespace="test-namespace",
+        )
+
+        # Set up tracer to capture spans
+        self.span_exporter = InMemorySpanExporter()
+        tracer_provider = get_tracer_provider()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(self.span_exporter))
+
+    def tearDown(self):
+        super().tearDown()
+        self.span_exporter.clear()
+
+    def test_output_scope_creation(self):
+        """Test that OutputScope can be created successfully."""
+        response = Response(messages=["Hello, how can I help you?"])
+
+        scope = OutputScope.start(self.agent_details, self.tenant_details, response)
+
+        self.assertIsNotNone(scope)
+        scope.dispose()
+
+    def test_record_output_messages_method_exists(self):
+        """Test that record_output_messages method exists on OutputScope."""
+        response = Response(messages=["Initial message"])
+        scope = OutputScope.start(self.agent_details, self.tenant_details, response)
+
+        if scope is not None:
+            # Test that the method exists
+            self.assertTrue(hasattr(scope, "record_output_messages"))
+            self.assertTrue(callable(scope.record_output_messages))
+            scope.dispose()
+
+    def test_output_messages_set_on_span(self):
+        """Test that output messages are set on span attributes."""
+        response = Response(messages=["This is the agent response"])
+
+        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 {}
+
+        self.assertIn(
+            GEN_AI_OUTPUT_MESSAGES_KEY,
+            span_attributes,
+            "Expected output messages key to be set on span",
+        )
+
+        # Verify the message content is in the serialized output
+        output_value = span_attributes[GEN_AI_OUTPUT_MESSAGES_KEY]
+        self.assertIn("This is the agent response", output_value)
+
+    def test_multiple_output_messages(self):
+        """Test that multiple output messages are properly recorded."""
+        response = Response(messages=["First response", "Second response", "Third response"])
+
+        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 {}
+
+        self.assertIn(
+            GEN_AI_OUTPUT_MESSAGES_KEY,
+            span_attributes,
+            "Expected output messages key to be set on span",
+        )
+
+        output_value = span_attributes[GEN_AI_OUTPUT_MESSAGES_KEY]
+        self.assertIn("First response", output_value)
+        self.assertIn("Second response", output_value)
+        self.assertIn("Third response", output_value)
+
+    def test_record_output_messages_updates_span(self):
+        """Test that record_output_messages updates the span with new messages."""
+        response = Response(messages=["Initial message"])
+
+        scope = OutputScope.start(self.agent_details, self.tenant_details, response)
+
+        if scope is not None:
+            # Record updated messages
+            scope.record_output_messages(["Updated message 1", "Updated message 2"])
+            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 {}
+
+        self.assertIn(
+            GEN_AI_OUTPUT_MESSAGES_KEY,
+            span_attributes,
+            "Expected output messages key to be set on span",
+        )
+
+        # The span should have the updated messages
+        output_value = span_attributes[GEN_AI_OUTPUT_MESSAGES_KEY]
+        self.assertIn("Updated message 1", output_value)
+        self.assertIn("Updated message 2", output_value)
+
+    def test_output_scope_context_manager(self):
+        """Test that OutputScope works as a context manager."""
+        response = Response(messages=["Context manager test"])
+
+        with OutputScope.start(self.agent_details, self.tenant_details, response) as scope:
+            self.assertIsNotNone(scope)
+
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
+
+    def test_output_scope_span_name(self):
+        """Test that OutputScope creates spans with correct operation name."""
+        response = Response(messages=["Test message"])
+
+        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]
+        # The activity name should contain "output_messages" and the agent id
+        self.assertIn("output_messages", span.name)
+        self.assertIn(self.agent_details.agent_id, span.name)
+
+
+if __name__ == "__main__":
+    # Run pytest only on the current file
+    sys.exit(pytest.main([str(Path(__file__))] + sys.argv[1:]))