Span linking in Proactive scenarios

.github/workflows/codeql.yml

@@ -97,4 +97,4 @@ jobs:
     - name: Perform CodeQL Analysis
       uses: github/codeql-action/analyze@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4.37.7
       with:
-        category: "/language:${{matrix.language}}"
+        category: "/language:${{matrix.language}}"

.github/workflows/python-package.yml

@@ -79,4 +79,4 @@ jobs:
         python -m pip install ./dist/microsoft_agents_storage_cosmos*.whl
     - name: Test with pytest
       run: |
-        pytest -W "ignore:SelectableGroups dict interface is deprecated. Use select.:DeprecationWarning"
+        pytest -W "ignore:SelectableGroups dict interface is deprecated. Use select.:DeprecationWarning"

.gitignore

@@ -130,4 +130,7 @@ cython_debug/
 bin/
 
 # Claude
-.claude/
+.claude/
+
+# Certificates
+*.pfx

dev/integration/tests/telemetry/__init__.py

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

dev/integration/tests/telemetry/test_proactive_span_linking.py

@@ -0,0 +1,193 @@
+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
+    link_context = continuation_span.links[0].context
+    assert link_context.trace_id == store_span.context.trace_id
+    assert link_context.span_id == store_span.context.span_id
+    assert link_context.trace_flags == store_span.context.trace_flags
+    assert link_context.trace_state == store_span.context.trace_state
+    assert store_span.context.is_remote is False
+    assert link_context.is_remote is True
+
+
+@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
+    link_context = continuation_span.links[0].context
+    assert link_context.trace_id == store_spans[-1].context.trace_id
+    assert link_context.span_id == store_spans[-1].context.span_id
+    assert link_context.trace_id != store_spans[0].context.trace_id
+    assert link_context.is_remote is True

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
@@ -137,6 +174,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
@@ -147,4 +185,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),
         )