fix(observability): cancel resolver futures when export can't await them

When export runs inside an active event loop, the exporter can't await an
async token resolver and fails that export. It closed bare coroutines but left
a resolver-returned asyncio.Future or Task running, where it could still
mutate a token cache or raise an unobserved exception. Cancel those too.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 5cbf5f6b-cc40-4b7e-a591-65848db73a12

Author: DheerajPannala

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter.py

@@ -299,6 +299,8 @@ def _resolve_token(self, agent_id: str, tenant_id: str) -> str | None:
                 return asyncio.run(_await_token(cast(Awaitable[str | None], token)))
             if inspect.iscoroutine(token):
                 token.close()
+            elif isinstance(token, asyncio.Future):
+                token.cancel()
             raise RuntimeError(
                 "Agent365Exporter cannot await an async token_resolver while running "
                 "inside an active event loop; use a synchronous cached resolver or refresh "

tests/observability/core/test_agent365_exporter.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import asyncio
+import inspect
 import json
 import os
 import unittest
@@ -352,6 +354,34 @@ async def resolver(agent_id, tenant_id):
             self.assertEqual(result, SpanExportResult.SUCCESS)
             mock_post.assert_called_once()
 
+    def test_awaitable_resolver_in_active_event_loop_fails_and_is_released(self):
+        """Inside a running loop, export fails and the resolver's coroutine, future or task is released."""
+
+        async def pending_token():
+            await asyncio.sleep(3600)
+            return "never-used"
+
+        async def run_in_loop():
+            coroutine = pending_token()
+            future = asyncio.get_running_loop().create_future()
+            task = asyncio.ensure_future(pending_token())
+            for awaitable in (coroutine, future, task):
+                exporter = _Agent365Exporter(
+                    token_resolver=lambda _agent_id, _tenant_id, value=awaitable: value,
+                    cluster_category="test",
+                )
+                spans = [self._create_mock_span("active_loop_span")]
+                with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                    self.assertEqual(exporter.export(spans), SpanExportResult.FAILURE)
+                    mock_post.assert_not_called()
+
+            await asyncio.wait([task], timeout=1)
+            self.assertEqual(inspect.getcoroutinestate(coroutine), inspect.CORO_CLOSED)
+            self.assertTrue(future.cancelled())
+            self.assertTrue(task.cancelled())
+
+        asyncio.run(run_in_loop())
+
     def test_auth_not_found_errors_do_not_fallback_to_delegated_route(self):
         """401/403/404 failures do not trigger a delegated route fallback."""
         for status_code in (401, 403, 404):