Instrument Proactive

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/proactive.py

@@ -16,6 +16,7 @@
 from .conversation import Conversation
 from .create_conversation_options import CreateConversationOptions
 from .proactive_options import ProactiveOptions
+from .telemetry import spans
 
 if TYPE_CHECKING:
     from microsoft_agents.hosting.core.turn_context import TurnContext
@@ -95,7 +96,7 @@ def _storage_key(conversation_id: str) -> str:
 
     async def store_conversation(
         self,
-        context_or_conversation: "TurnContext | Conversation",
+        context_or_conversation: TurnContext | Conversation,
     ) -> None:
         """
         Persist a :class:`~microsoft_agents.hosting.core.app.proactive.conversation.Conversation`
@@ -120,10 +121,13 @@ async def store_conversation(
         else:
             conversation = context_or_conversation
 
-        conversation.validate()
-        key = self._storage_key(conversation.conversation_reference.conversation.id)
-        logger.debug("Storing conversation with key: %s", key)
-        await self._storage.write({key: conversation})
+        with spans.ProactiveStoreConversation(
+            conversation.conversation_reference.conversation.id
+        ):
+            conversation.validate()
+            key = self._storage_key(conversation.conversation_reference.conversation.id)
+            logger.debug("Storing conversation with key: %s", key)
+            await self._storage.write({key: conversation})
 
     async def get_conversation(self, conversation_id: str) -> Optional[Conversation]:
         """
@@ -136,9 +140,12 @@ async def get_conversation(self, conversation_id: str) -> Optional[Conversation]
             or ``None`` if not found.
         :rtype: Optional[:class:`~microsoft_agents.hosting.core.app.proactive.conversation.Conversation`]
         """
-        key = self._storage_key(conversation_id)
-        results = await self._storage.read([key], target_cls=Conversation)
-        return results.get(key)
+        with spans.ProactiveGetConversation(conversation_id) as span:
+            key = self._storage_key(conversation_id)
+            results = await self._storage.read([key], target_cls=Conversation)
+            conversation = results.get(key)
+            span.share(found=conversation is not None)
+            return conversation
 
     async def delete_conversation(self, conversation_id: str) -> None:
         """
@@ -147,9 +154,10 @@ async def delete_conversation(self, conversation_id: str) -> None:
         :param conversation_id: The conversation ID to delete.
         :type conversation_id: str
         """
-        key = self._storage_key(conversation_id)
-        logger.debug("Deleting conversation with key: %s", key)
-        await self._storage.delete([key])
+        with spans.ProactiveDeleteConversation(conversation_id):
+            key = self._storage_key(conversation_id)
+            logger.debug("Deleting conversation with key: %s", key)
+            await self._storage.delete([key])
 
     # ------------------------------------------------------------------
     # Send a single activity
@@ -181,7 +189,9 @@ async def send_activity(
             conversation is not found in storage.
         """
         conversation = await self._resolve_conversation(conversation_id_or_conversation)
-        return await Proactive._send_activity_impl(adapter, conversation, activity)
+        conversation_id = conversation.conversation_reference.conversation.id
+        with spans.ProactiveSendActivity(conversation_id, activity):
+            return await Proactive._send_activity_impl(adapter, conversation, activity)
 
     @staticmethod
     async def _send_activity_impl(
@@ -252,6 +262,7 @@ async def continue_conversation(
             :attr:`~ProactiveOptions.fail_on_unsigned_in_connections` is ``True``.
         """
         conversation = await self._resolve_conversation(conversation_id_or_conversation)
+        conversation_id = conversation.conversation_reference.conversation.id
 
         captured_exc: Optional[BaseException] = None
         claims = Conversation.identity_from_claims(conversation.claims)
@@ -267,10 +278,14 @@ async def _callback(context: "TurnContext") -> None:
             except Exception as exc:  # noqa: BLE001
                 captured_exc = exc
 
-        await adapter.continue_conversation_with_claims(claims, continuation, _callback)
+        with spans.ProactiveContinueConversation(conversation_id, continuation):
 
-        if captured_exc is not None:
-            raise captured_exc
+            await adapter.continue_conversation_with_claims(
+                claims, continuation, _callback
+            )
+
+            if captured_exc is not None:
+                raise captured_exc
 
     # ------------------------------------------------------------------
     # Create a new conversation
@@ -303,40 +318,42 @@ async def create_conversation(
         new_conversation: Optional[Conversation] = None
         captured_exc: Optional[BaseException] = None
 
-        audience = options.audience or options.identity.get_token_audience()
+        with spans.ProactiveCreateConversation(options):
 
-        async def _callback(context: "TurnContext") -> None:
-            nonlocal new_conversation, captured_exc
-            try:
-                reference = context.activity.get_conversation_reference()
-                new_conversation = Conversation(
-                    claims=options.identity,
-                    conversation_reference=reference,
-                )
+            audience = options.audience or options.identity.get_token_audience()
 
-                if options.store_conversation:
-                    await self.store_conversation(new_conversation)
-
-                if handler is not None:
-                    state = await self._load_state(context)
-                    await handler(context, state)
-                    await state.save(context)
-            except Exception as exc:  # noqa: BLE001
-                captured_exc = exc
+            async def _callback(context: "TurnContext") -> None:
+                nonlocal new_conversation, captured_exc
+                try:
+                    reference = context.activity.get_conversation_reference()
+                    new_conversation = Conversation(
+                        claims=options.identity,
+                        conversation_reference=reference,
+                    )
 
-        await adapter.create_conversation(
-            options.identity.get_app_id() or "",
-            options.channel_id,
-            options.service_url,
-            audience,
-            options.parameters,
-            _callback,
-        )
+                    if options.store_conversation:
+                        await self.store_conversation(new_conversation)
+
+                    if handler is not None:
+                        state = await self._load_state(context)
+                        await handler(context, state)
+                        await state.save(context)
+                except Exception as exc:  # noqa: BLE001
+                    captured_exc = exc
+
+            await adapter.create_conversation(
+                options.identity.get_app_id() or "",
+                options.channel_id,
+                options.service_url,
+                audience,
+                options.parameters,
+                _callback,
+            )
 
-        if captured_exc is not None:
-            raise captured_exc
+            if captured_exc is not None:
+                raise captured_exc
 
-        return new_conversation
+            return new_conversation
 
     # ------------------------------------------------------------------
     # Internal helpers

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/telemetry/constants.py

@@ -0,0 +1,9 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+SPAN_STORE_CONVERSATION = "agents.proactive.store_conversation"
+SPAN_GET_CONVERSATION = "agents.proactive.get_conversation"
+SPAN_DELETE_CONVERSATION = "agents.proactive.delete_conversation"
+SPAN_SEND_ACTIVITY = "agents.proactive.send_activity"
+SPAN_CONTINUE_CONVERSATION = "agents.proactive.continue_conversation"
+SPAN_CREATE_CONVERSATION = "agents.proactive.create_conversation"

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/telemetry/spans.py

@@ -0,0 +1,142 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from __future__ import annotations
+
+from microsoft_agents.activity import Activity
+from microsoft_agents.hosting.core.telemetry import (
+    AttributeMap,
+    attributes,
+    SimpleSpanWrapper,
+)
+from . import constants
+from ..create_conversation_options import CreateConversationOptions
+
+
+class ProactiveStoreConversation(SimpleSpanWrapper):
+    """Span for storing a conversation reference in proactive scenarios, starting from when the store operation is initiated until it is completed. This span can be used to correlate telemetry related to storing conversation references in proactive scenarios."""
+
+    def __init__(self, conversation_id: str):
+        """Initializes the ProactiveStoreConversation SpanWrapper.
+
+        :param conversation_id: The ID of the conversation being stored, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_STORE_CONVERSATION)
+        self._conversation_id = conversation_id
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.CONVERSATION_ID: self._conversation_id or attributes.UNKNOWN,
+        }
+
+
+class ProactiveGetConversation(SimpleSpanWrapper):
+    """Span for getting a conversation reference in proactive scenarios."""
+
+    def __init__(self, conversation_id: str):
+        """Initializes the ProactiveGetConversation SpanWrapper.
+
+        :param conversation_id: The ID of the conversation being retrieved, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_GET_CONVERSATION)
+        self._conversation_id = conversation_id
+        self._found: bool | None = None
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.CONVERSATION_ID: self._conversation_id or attributes.UNKNOWN,
+            attributes.CONVERSATION_FOUND: (
+                self._found if self._found is not None else attributes.UNKNOWN
+            ),
+        }
+
+    def share(self, found: bool) -> None:
+        """Records to the span whether the conversation being retrieved was found.
+
+        :param found: Whether the conversation being retrieved was found
+        """
+        self._found = found
+
+
+class ProactiveDeleteConversation(SimpleSpanWrapper):
+    """Span for deleting a conversation reference in proactive scenarios, starting from when the delete operation is initiated until it is completed. This span can be used to correlate telemetry related to deleting conversation references in proactive scenarios."""
+
+    def __init__(self, conversation_id: str):
+        """Initializes the ProactiveDeleteConversation SpanWrapper.
+
+        :param conversation_id: The ID of the conversation being deleted, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_DELETE_CONVERSATION)
+        self._conversation_id = conversation_id
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.CONVERSATION_ID: self._conversation_id or attributes.UNKNOWN,
+        }
+
+
+class ProactiveSendActivity(SimpleSpanWrapper):
+    """Span for sending an activity in proactive scenarios, starting from when the send operation is initiated until it is completed. This span can be used to correlate telemetry related to sending activities in proactive scenarios."""
+
+    def __init__(self, conversation_id: str, activity: Activity):
+        """Initializes the ProactiveSendActivity SpanWrapper.
+
+        :param conversation_id: The ID of the conversation the activity is being sent to, used to extract attributes for the span
+        :param activity: The activity being sent, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_SEND_ACTIVITY)
+        self._conversation_id = conversation_id
+        self._activity = activity
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.CONVERSATION_ID: self._conversation_id or attributes.UNKNOWN,
+            attributes.ACTIVITY_TYPE: self._activity.type or attributes.UNKNOWN,
+            attributes.ACTIVITY_CHANNEL_ID: self._activity.channel_id
+            or attributes.UNKNOWN,
+        }
+
+
+class ProactiveContinueConversation(SimpleSpanWrapper):
+    """Span for continuing a conversation in proactive scenarios, starting from when the continue operation is initiated until it is completed. This span can be used to correlate telemetry related to continuing conversations in proactive scenarios."""
+
+    def __init__(self, conversation_id: str, activity: Activity):
+        """Initializes the ProactiveContinueConversation SpanWrapper.
+
+        :param conversation_id: The ID of the conversation being continued, used to extract attributes for the span
+        :param activity: The activity being sent, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_CONTINUE_CONVERSATION)
+        self._conversation_id = conversation_id
+        self._activity = activity
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.CONVERSATION_ID: self._conversation_id or attributes.UNKNOWN,
+            attributes.ACTIVITY_TYPE: self._activity.type or attributes.UNKNOWN,
+            attributes.ACTIVITY_CHANNEL_ID: self._activity.channel_id
+            or attributes.UNKNOWN,
+        }
+
+
+class ProactiveCreateConversation(SimpleSpanWrapper):
+    """Span for creating a conversation in proactive scenarios, starting from when the create operation is initiated until it is completed. This span can be used to correlate telemetry related to creating conversations in proactive scenarios."""
+
+    def __init__(self, options: CreateConversationOptions):
+        """Initializes the ProactiveCreateConversation SpanWrapper.
+
+        :param options: The options used to create the conversation, used to extract attributes for the span
+        """
+        super().__init__(constants.SPAN_CREATE_CONVERSATION)
+        self._channel_id = options.channel_id
+        self._members_count: str | int = (
+            len(options.parameters.members)
+            if options.parameters and options.parameters.members
+            else attributes.UNKNOWN
+        )
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.ACTIVITY_CHANNEL_ID: self._channel_id,
+            attributes.MEMBERS_COUNT: self._members_count,
+        }

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/telemetry/attributes.py

@@ -21,6 +21,7 @@
 AUTH_SUCCESS = "auth.success"
 
 CONNECTION_NAME = "auth.connection.name"
+CONVERSATION_FOUND = "proactive.conversation.found"
 CONVERSATION_ID = "activity.conversation.id"
 
 HTTP_METHOD = "http.method"
@@ -30,6 +31,8 @@
 
 KEY_COUNT = "storage.keys.count"
 
+MEMBERS_COUNT = "proactive.members.count"
+
 OPERATION = "operation"
 
 ROUTE_AUTHORIZED = "route.authorized"

tests/hosting_core/app/proactive/test_conversation.py

@@ -69,7 +69,9 @@ def test_init_empty_claims_dict(self):
 class TestConversationFromTurnContext:
     def test_from_turn_context_extracts_reference_and_identity(self):
         ref = _make_reference("ctx-conv")
-        identity = ClaimsIdentity(claims={"aud": "app-id", "tid": "t"}, is_authenticated=True)
+        identity = ClaimsIdentity(
+            claims={"aud": "app-id", "tid": "t"}, is_authenticated=True
+        )
 
         ctx = MagicMock()
         ctx.activity.get_conversation_reference.return_value = ref
@@ -195,7 +197,9 @@ def test_round_trip_preserves_conversation_id(self):
     def test_round_trip_preserves_service_url(self):
         original = Conversation(
             claims={},
-            conversation_reference=_make_reference(service_url="https://custom.service/"),
+            conversation_reference=_make_reference(
+                service_url="https://custom.service/"
+            ),
         )
         json_data = original.store_item_to_json()
         restored = Conversation.from_json_to_store_item(json_data)