bug fix: Deadlock Sequence on Pubsub

Documentacao/FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md

@@ -0,0 +1,37 @@
+# Correção: deadlock/espera cross-loop na geração de sequence
+
+## Problema
+
+A API síncrona `agent_framework.observer.event()` podia ser chamada em uma worker thread sem event loop ativo. Nesse caso, a implementação anterior executava `asyncio.run(aevent(...))`, criando um novo event loop temporário. Ao mesmo tempo, `analytics/tim_sequence.py` compartilhava instâncias globais de `asyncio.Lock` (`_mongo_index_lock` e `_memory_lock`) entre chamadas que podiam vir de event loops diferentes.
+
+Na primeira operação Mongo, `_ensure_mongo_ttl_index_once()` mantinha `_mongo_index_lock` durante a criação do índice TTL. A contenção por outro loop podia deixar a segunda chamada aguardando indefinidamente.
+
+## Alterações aplicadas
+
+1. `observer.py`
+   - removido `asyncio.run()` do caminho síncrono de `event()`;
+   - adicionado um event loop dedicado e reutilizável para chamadas síncronas;
+   - submissão cross-thread feita com `asyncio.run_coroutine_threadsafe()`;
+   - encerramento best-effort do loop no shutdown do processo.
+
+2. `analytics/tim_sequence.py`
+   - `_mongo_index_lock`: `asyncio.Lock` -> `threading.Lock`;
+   - `_memory_lock`: `asyncio.Lock` -> `threading.Lock`;
+   - inicialização do índice TTL movida para uma função síncrona protegida por lock de thread e chamada via `asyncio.to_thread()`;
+   - o contador de fallback em memória usa uma seção crítica curta e thread-safe.
+
+3. Testes
+   - `tests/test_observer_cross_loop_deadlock_fix.py` valida:
+     - múltiplas worker threads usando `event()` compartilham o mesmo loop síncrono do observer;
+     - sequence em memória permanece monotônica entre event loops independentes;
+     - criação do índice TTL ocorre apenas uma vez sob contenção cross-loop.
+
+## Validação executada
+
+```bash
+PYTHONPATH=libs/agent_framework/src pytest -q tests/test_observer_cross_loop_deadlock_fix.py
+```
+
+Resultado: `3 passed`.
+
+A suíte completa do repositório possui falhas preexistentes/independentes desta alteração, incluindo conflitos de coleta de arquivos `test_long_term_memory.py`, caminhos estáticos de template e testes de checkpoint/workflow. Esses itens não foram alterados por esta correção.

libs/agent_framework/src/agent_framework/analytics/tim_sequence.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import os
+import threading
 from collections import defaultdict
 from datetime import datetime, timedelta, timezone
 from typing import Any, Literal
@@ -12,7 +13,7 @@ logger = logging.getLogger("agent_framework.analytics.tim_sequence")
 # In-process fallback. This is not cross-process/global, but keeps telemetry alive
 # when the configured shared sequence backend is unavailable, matching the
 # framework principle that observability must not break business execution.
-_memory_lock = asyncio.Lock()
+_memory_lock = threading.Lock()
 _memory_counters: dict[str, int] = defaultdict(int)
 
 SequenceProvider = Literal["auto", "redis", "mongodb", "mongo", "memory", "none"]
@@ -152,7 +153,7 @@ async def _next_sequence_redis(key: str, ttl_seconds: int) -> int | None:
 
 
 _mongo_index_checked = False
-_mongo_index_lock = asyncio.Lock()
+_mongo_index_lock = threading.Lock()
 
 
 def _next_sequence_mongodb_sync(
@@ -204,37 +205,43 @@ def _next_sequence_mongodb_sync(
         client.close()
 
 
-async def _ensure_mongo_ttl_index_once(ttl_seconds: int) -> None:
-    """Best-effort TTL index creation for Mongo sequence docs.
+def _ensure_mongo_ttl_index_once_sync(ttl_seconds: int) -> None:
+    """Best-effort TTL index initialization, safe across threads/event loops.
 
-    The sequence still works without this index. If the application user lacks
-    index privileges, we only log and continue.
+    ``asyncio.Lock`` must not be shared by independent event loops. Observer
+    compatibility calls may originate in worker threads, so this one-time
+    process-local guard deliberately uses ``threading.Lock``. The blocking
+    Mongo operation is executed by the async wrapper in a worker thread.
     """
     global _mongo_index_checked
     if _mongo_index_checked or ttl_seconds <= 0 or not _mongo_uri():
         return
 
-    async with _mongo_index_lock:
+    with _mongo_index_lock:
         if _mongo_index_checked:
             return
         try:
             from pymongo import MongoClient  # type: ignore
 
-            def _create() -> None:
             client = MongoClient(_mongo_uri())
             try:
                 collection = client[_mongo_database()][_mongo_collection()]
                 collection.create_index("expiresAt", expireAfterSeconds=0, background=True)
             finally:
                 client.close()
-
-            await asyncio.to_thread(_create)
         except Exception:
             logger.warning("tim_sequence.mongodb_ttl_index_failed", exc_info=True)
         finally:
+            # The index is an observability housekeeping concern, not a
+            # prerequisite for sequence generation. Do not retry on every
+            # event if the application user lacks index privileges.
             _mongo_index_checked = True
 
 
+async def _ensure_mongo_ttl_index_once(ttl_seconds: int) -> None:
+    await asyncio.to_thread(_ensure_mongo_ttl_index_once_sync, ttl_seconds)
+
+
 async def _next_sequence_mongodb(
     key: str,
     agent_id: str | None,
@@ -260,7 +267,9 @@ async def _next_sequence_mongodb(
 
 
 async def _next_sequence_memory(key: str) -> int:
-    async with _memory_lock:
+    # Tiny in-process critical section; a thread lock is intentional because
+    # this fallback can be reached from more than one asyncio event loop.
+    with _memory_lock:
         _memory_counters[key] += 1
         return _memory_counters[key]
 

libs/agent_framework/src/agent_framework/observer.py

@@ -14,9 +14,10 @@ rails, bridges e comandos de negócio sem quebrar o turno do cliente.
 """
 
 import asyncio
+import atexit
 import logging
 import os
-from threading import Lock
+from threading import Event, Lock, Thread
 from typing import Any
 
 from agent_framework.analytics.factory import create_analytics_publisher
@@ -29,6 +30,74 @@ _GLOBAL_CONFIG: dict[str, Any] = {}
 _LOCK = Lock()
 
 
+class _SyncEventLoopBridge:
+    """Own one reusable event loop for synchronous observer calls.
+
+    The legacy ``event()`` API is frequently invoked from worker threads that
+    do not own an asyncio loop. Creating a fresh loop with ``asyncio.run()``
+    for every such call makes the same global observer reachable from multiple
+    temporary loops. This bridge keeps those synchronous calls on one stable
+    loop and submits work through asyncio's thread-safe API.
+    """
+
+    def __init__(self) -> None:
+        self._start_lock = Lock()
+        self._ready = Event()
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._thread: Thread | None = None
+
+    def _thread_main(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        self._loop = loop
+        self._ready.set()
+        try:
+            loop.run_forever()
+        finally:
+            pending = asyncio.all_tasks(loop)
+            for task in pending:
+                task.cancel()
+            if pending:
+                loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+            loop.close()
+
+    def _ensure_started(self) -> asyncio.AbstractEventLoop:
+        loop = self._loop
+        if loop is not None and loop.is_running():
+            return loop
+        with self._start_lock:
+            loop = self._loop
+            if loop is None or not loop.is_running():
+                self._ready.clear()
+                self._thread = Thread(
+                    target=self._thread_main,
+                    name="agent-framework-observer-loop",
+                    daemon=True,
+                )
+                self._thread.start()
+                self._ready.wait()
+        assert self._loop is not None
+        return self._loop
+
+    def run(self, coro: Any) -> Any:
+        loop = self._ensure_started()
+        future = asyncio.run_coroutine_threadsafe(coro, loop)
+        return future.result()
+
+    def close(self) -> None:
+        loop = self._loop
+        thread = self._thread
+        if loop is None or not loop.is_running():
+            return
+        loop.call_soon_threadsafe(loop.stop)
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=2.0)
+
+
+_SYNC_EVENT_LOOP = _SyncEventLoopBridge()
+atexit.register(_SYNC_EVENT_LOOP.close)
+
+
 def _truthy(value: Any, default: bool = False) -> bool:
     if value is None:
         return default
@@ -202,7 +271,12 @@ def event(
     try:
         loop = asyncio.get_running_loop()
     except RuntimeError:
-        return asyncio.run(aevent(name, data=data, metadata=metadata, event_test=event_test))
+        # Do not create a temporary event loop in every worker thread. Route
+        # synchronous compatibility calls to one stable observer loop using
+        # asyncio's thread-safe submission primitive.
+        return _SYNC_EVENT_LOOP.run(
+            aevent(name, data=data, metadata=metadata, event_test=event_test)
+        )
 
     task = loop.create_task(aevent(name, data=data, metadata=metadata, event_test=event_test))
     task.add_done_callback(_log_task_exception)

tests/test_observer_cross_loop_deadlock_fix.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+from agent_framework import observer
+from agent_framework.analytics import tim_sequence
+
+
+def test_sync_event_calls_share_one_stable_asyncio_loop(monkeypatch):
+    seen_loop_ids: set[int] = set()
+    seen_lock = threading.Lock()
+
+    async def fake_aevent(name: str, **kwargs):
+        loop_id = id(asyncio.get_running_loop())
+        with seen_lock:
+            seen_loop_ids.add(loop_id)
+        await asyncio.sleep(0.01)
+        return {"eventType": name, "loop_id": loop_id}
+
+    monkeypatch.setattr(observer, "aevent", fake_aevent)
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(lambda i: observer.event(f"IC.TEST.{i}"), range(24)))
+
+    assert len(seen_loop_ids) == 1
+    assert {item["loop_id"] for item in results} == seen_loop_ids
+
+
+def test_memory_sequence_is_safe_across_independent_event_loops(monkeypatch):
+    monkeypatch.setenv("PUBSUB_SEQUENCE_ENABLED", "true")
+    monkeypatch.setenv("PUBSUB_SEQUENCE_PROVIDER", "memory")
+    tim_sequence._memory_counters.clear()
+
+    def one_call(_: int) -> int | None:
+        return asyncio.run(
+            tim_sequence.next_sequence(
+                "agent-a",
+                "session-a",
+                "transaction-cross-loop",
+            )
+        )
+
+    with ThreadPoolExecutor(max_workers=12) as pool:
+        values = list(pool.map(one_call, range(120)))
+
+    assert sorted(values) == list(range(1, 121))
+
+
+def test_mongo_ttl_index_guard_is_thread_safe_across_event_loops(monkeypatch):
+    tim_sequence._mongo_index_checked = False
+    monkeypatch.setenv("PUBSUB_SEQUENCE_MONGODB_URI", "mongodb://fake")
+
+    calls = 0
+    calls_lock = threading.Lock()
+
+    class FakeCollection:
+        def create_index(self, *args, **kwargs):
+            nonlocal calls
+            with calls_lock:
+                calls += 1
+            # Enlarge the contention window that previously exposed the
+            # cross-event-loop asyncio.Lock issue.
+            time.sleep(0.05)
+
+    class FakeDatabase:
+        def __getitem__(self, name):
+            return FakeCollection()
+
+    class FakeMongoClient:
+        def __init__(self, uri):
+            self.uri = uri
+
+        def __getitem__(self, name):
+            return FakeDatabase()
+
+        def close(self):
+            return None
+
+    monkeypatch.setitem(sys.modules, "pymongo", SimpleNamespace(MongoClient=FakeMongoClient))
+
+    def ensure_index(_: int) -> None:
+        asyncio.run(tim_sequence._ensure_mongo_ttl_index_once(60))
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(ensure_index, range(16)))
+
+    assert calls == 1
+    assert tim_sequence._mongo_index_checked is True