Span linking in Proactive scenarios

changelog.md

@@ -4,6 +4,7 @@
 - Added support for Workload Identity
 - **Entra JWT Issuer Validation**: Added tenant ID cross-checking for Entra issuer claims and support for configuring issuer lists through environment variables (#515)
 - **AgentApplication Adaptive Card Routing**: Added `AgentApplication.adaptive_card` with decorator-based handlers for Adaptive Card `Action.Submit`, `Action.Execute`, and `Data.Query` dynamic search with verb and dataset matching selection.
+- Distributed tracing across Proactive operations.
 
 ## New Models & APIs
 - **Regionalized UserTokenClient Support**: Added optional argument to `CloudAdapter` to configure Token Service endpoint used by `RestChannelServiceClientFactory` when creating `UserTokenClient` instances.
@@ -13,6 +14,9 @@
 
 ## Developer Experience
 - Building packages with `py.typed` files for improved typing support
+- Support for linking with OpenTelemetry span creation throught the `SimpleSpanWrapper` constructor.
+
+---
 
 # Microsoft 365 Agents SDK for Python - Release Notes v1.3.0
 

dev/integration/tests/telemetry/__init__.py

@@ -0,0 +1 @@
+"""Telemetry integration tests."""

dev/integration/tests/telemetry/test_proactive_span_linking.py

@@ -0,0 +1,184 @@
+import pytest
+
+from microsoft_agents.activity import Activity, ActivityTypes
+from microsoft_agents.hosting.aiohttp import CloudAdapter
+from microsoft_agents.hosting.core import (
+    AgentApplication,
+    AgentAuthConfiguration,
+    ApplicationOptions,
+    Authorization,
+    MemoryStorage,
+    TurnContext,
+    TurnState,
+)
+from microsoft_agents.hosting.core.app.proactive import ProactiveOptions
+from microsoft_agents.hosting.core.app.proactive.telemetry import constants
+from microsoft_agents.hosting.core.authorization import ClaimsIdentity
+from microsoft_agents.testing import AgentEnvironment, AiohttpScenario
+
+from ..utils.telemetry_fixtures import (  # noqa: F401
+    test_exporter,
+    test_telemetry,
+)
+
+
+class _FakeTokenProvider:
+    def __init__(self) -> None:
+        self._configuration = AgentAuthConfiguration()
+
+    @property
+    def configuration(self) -> AgentAuthConfiguration:
+        return self._configuration
+
+    async def get_access_token(
+        self,
+        resource_url: str,
+        scopes: list[str],
+        force_refresh: bool = False,
+    ) -> str:
+        return "test-access-token"
+
+
+class _FakeConnections:
+    def __init__(self) -> None:
+        self._provider = _FakeTokenProvider()
+
+    def get_connection(self, connection_name: str):
+        return self._provider
+
+    def get_default_connection(self):
+        return self._provider
+
+    def get_token_provider(
+        self,
+        claims_identity: ClaimsIdentity,
+        service_url: str,
+    ):
+        return self._provider
+
+    def get_token_provider_from_activity(
+        self,
+        claims_identity: ClaimsIdentity,
+        activity: Activity,
+    ):
+        return self._provider
+
+    def get_default_connection_configuration(self) -> AgentAuthConfiguration:
+        return self._provider.configuration
+
+
+def _create_scenario() -> AiohttpScenario:
+    connections = _FakeConnections()
+    storage = MemoryStorage()
+    adapter = CloudAdapter(connection_manager=connections)
+    authorization = Authorization(storage, connections)
+    app = AgentApplication[TurnState](
+        options=ApplicationOptions(
+            storage=storage,
+            adapter=adapter,
+            proactive=ProactiveOptions(),
+        ),
+        authorization=authorization,
+    )
+
+    @app.activity(ActivityTypes.message)
+    async def store_conversation(context: TurnContext, state: TurnState) -> None:
+        await app.proactive.store_conversation(context)
+
+    environment = AgentEnvironment(
+        config={},
+        agent_application=app,
+        authorization=authorization,
+        adapter=adapter,
+        storage=storage,
+        connections=connections,
+    )
+    return AiohttpScenario(environment, use_jwt_middleware=False)
+
+
+_SCENARIO = _create_scenario()
+
+
+def _get_span(spans, name):
+    return next(span for span in spans if span.name == name)
+
+
+@pytest.mark.asyncio
+@pytest.mark.agent_test(_SCENARIO)
+async def test_continue_conversation_links_to_stored_context(
+    test_exporter,
+    agent_client,
+    agent_application,
+    adapter,
+):
+    activity = agent_client.template.create(
+        {
+            "type": ActivityTypes.message,
+            "id": "proactive-linking-activity",
+        }
+    )
+    await agent_client.send(activity)
+
+    async def continue_handler(context: TurnContext, state: TurnState) -> None:
+        pass
+
+    await agent_application.proactive.continue_conversation(
+        adapter,
+        activity.conversation.id,
+        continue_handler,
+    )
+
+    spans = test_exporter.get_finished_spans()
+    store_span = _get_span(spans, constants.SPAN_STORE_CONVERSATION)
+    continuation_span = _get_span(spans, constants.SPAN_CONTINUE_CONVERSATION)
+
+    assert len(continuation_span.links) == 1
+    assert continuation_span.links[0].context == store_span.context
+
+
+@pytest.mark.asyncio
+@pytest.mark.agent_test(_SCENARIO)
+async def test_overwriting_conversation_links_to_latest_store_span(
+    test_exporter,
+    agent_client,
+    agent_application,
+    adapter,
+):
+    conversation_id = "proactive-overwrite-conversation"
+    first_activity = agent_client.template.create(
+        {
+            "type": ActivityTypes.message,
+            "id": "first-store-activity",
+            "conversation": {"id": conversation_id},
+        }
+    )
+    second_activity = agent_client.template.create(
+        {
+            "type": ActivityTypes.message,
+            "id": "second-store-activity",
+            "conversation": {"id": conversation_id},
+        }
+    )
+
+    await agent_client.send(first_activity)
+    await agent_client.send(second_activity)
+
+    async def continue_handler(context: TurnContext, state: TurnState) -> None:
+        pass
+
+    await agent_application.proactive.continue_conversation(
+        adapter,
+        conversation_id,
+        continue_handler,
+    )
+
+    spans = test_exporter.get_finished_spans()
+    store_spans = [
+        span for span in spans if span.name == constants.SPAN_STORE_CONVERSATION
+    ]
+    continuation_span = _get_span(spans, constants.SPAN_CONTINUE_CONVERSATION)
+
+    assert len(store_spans) == 2
+    assert len(continuation_span.links) == 1
+    assert continuation_span.links[0].context == store_spans[-1].context
+    assert continuation_span.links[0].context != store_spans[0].context

dev/integration/tests/test_telemetry.py → dev/integration/tests/telemetry/test_telemetry.py

@@ -10,14 +10,14 @@
 from microsoft_agents.hosting.core.connector.telemetry import constants as connector_constants
 from microsoft_agents.hosting.core.storage.telemetry import constants as storage_constants
 
-from .scenarios import load_scenario
+from ..scenarios import load_scenario
 
-from .utils.telemetry_fixtures import ( # unused imports are needed for fixtures
+from ..utils.telemetry_fixtures import ( # unused imports are needed for fixtures
     test_telemetry,
     test_exporter,
     test_metric_reader,
 )
-from .utils.telemetry_utils import (
+from ..utils.telemetry_utils import (
     sum_counter,
     sum_hist_count,
     find_metric

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

@@ -1,19 +1,20 @@
-"""
-Copyright (c) Microsoft Corporation. All rights reserved.
-Licensed under the MIT License.
-"""
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
 
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
+from opentelemetry.trace import SpanContext
+
 from microsoft_agents.activity import ConversationReference
 from microsoft_agents.hosting.core.authorization import ClaimsIdentity
 from microsoft_agents.hosting.core.storage.store_item import StoreItem
 
 if TYPE_CHECKING:
     from microsoft_agents.hosting.core.turn_context import TurnContext
-    from microsoft_agents.hosting.core.channel_adapter import ChannelAdapter
+
+from .telemetry._utils import _deserialize_span_context, _dump_span_context
 
 # JWT claim keys that are persisted alongside a ConversationReference.
 _PERSISTED_CLAIM_KEYS = frozenset({"aud", "azp", "appid", "idtyp", "ver", "iss", "tid"})
@@ -41,14 +42,50 @@ def __init__(
         self,
         claims: dict[str, str] | ClaimsIdentity,
         conversation_reference: ConversationReference,
+        *,
+        _span_context: dict | None = None,
     ) -> None:
+        """Creates a new :class:`~microsoft_agents.hosting.core.app.proactive.Conversation` instance.
+
+        :param claims: Filtered JWT claims (``aud``, ``azp``, ``appid``, ``idtyp``,
+            ``ver``, ``iss``, ``tid``).  May be a raw ``dict`` or a
+            :class:`~microsoft_agents.hosting.core.authorization.ClaimsIdentity`.
+        :type claims: dict[str, str] or ClaimsIdentity
+        :param conversation_reference: The conversation reference.
+        :type conversation_reference: :class:`~microsoft_agents.activity.ConversationReference`
+        :param _span_context: Optional serialized span context for telemetry linking. For internal use only; this is not part of the public API.
+        :type _span_context: dict or None
+        """
         if isinstance(claims, ClaimsIdentity):
             self.claims: dict[str, str] = Conversation.claims_from_identity(claims)
         else:
             self.claims = {
                 k: v for k, v in claims.items() if k in _PERSISTED_CLAIM_KEYS
             }
         self.conversation_reference: ConversationReference = conversation_reference
+        self._span_context_dict: dict | None = _span_context
+
+    def _set_span_context(self, span_context: SpanContext) -> None:
+        """Sets the span context for this conversation, serializing it to a dictionary for storage.
+
+        For internal use only; this is not part of the public API.
+
+        :param span_context: The SpanContext to set.
+        :type span_context: SpanContext
+        """
+        self._span_context_dict = _dump_span_context(span_context)
+
+    def _get_span_context(self) -> SpanContext | None:
+        """Gets the span context for this conversation, deserializing it from a dictionary.
+
+        For internal use only; this is not part of the public API.
+
+        :return: The SpanContext, or None if not set.
+        :rtype: SpanContext or None
+        """
+        if self._span_context_dict is None:
+            return None
+        return _deserialize_span_context(self._span_context_dict)
 
     # ------------------------------------------------------------------
     # Factory helpers
@@ -135,6 +172,7 @@ def store_item_to_json(self) -> dict:
             "conversation_reference": self.conversation_reference.model_dump(
                 mode="json", by_alias=True, exclude_unset=True
             ),
+            "_span_context": self._span_context_dict,
         }
 
     @staticmethod
@@ -145,4 +183,5 @@ def from_json_to_store_item(json_data: dict) -> Conversation:
         return Conversation(
             claims=json_data.get("claims", {}),
             conversation_reference=reference,
+            _span_context=json_data.get("_span_context", None),
         )