fix(otel): use asyncio.Lock per loop in LiveOTelExporter to avoid event-loop deadlock

The rollout stage runs all OTelTracedSession instances in a single
event loop (`asyncio.run` -> `rollout.execute`). Holding a sync
`threading.Lock` across the inner `await invoke_callable(...)` in
`OTelTracedSession.run_turn` blocked the entire loop: when
`rollout.concurrency > 1`, the second coroutine could not acquire the
lock because the first was awaiting an HTTP call on the same loop, and
the first could never resume because the loop thread was parked on the
sync lock.

Reproducer (pre-fix): `rollout.concurrency: 4` with the LangGraph
travel-planner callable target deadlocked indefinitely after the first
batch of concurrent rollouts. Heartbeat froze and no transcripts were
produced.

Fix:
- `LiveOTelExporter._lock` becomes an `asyncio.Lock` created lazily
  per running event loop via `LiveOTelExporter.get_lock()`. This
  serializes the clear-invoke-export cycle within one loop while
  yielding the loop to other coroutines while waiting.
- `OTelTracedSession.run_turn` switches from `lock.acquire()` /
  `finally lock.release()` to `async with lock_ctx`, using
  `contextlib.nullcontext()` when `live_otel` is False.
- A class-level `asyncio.Lock()` would bind to the first loop on use
  and raise on subsequent `asyncio.run` calls; the per-loop cache
  keeps repeated runs and tests safe.

Verification:
- `uv run pytest -q` -> 505 passed, 14 skipped.
- LangGraph travel-planner config patched to `rollout.concurrency: 4`
  with 20 seeds end-to-end: PASS in 375s, 20 transcripts, 20 scores.

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

Author: changliu2

p2m/core/otel.py

@@ -15,8 +15,8 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
-import threading
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Protocol, runtime_checkable
@@ -451,7 +451,27 @@ class LiveOTelExporter:
     _instance: "LiveOTelExporter | None" = None
     _setup_done: bool = False
     _sdk_exporter: Any = None
-    _lock: threading.Lock = threading.Lock()
+    # Per-event-loop async lock, created lazily on first use. A class-level
+    # ``asyncio.Lock()`` would bind to whichever loop happened to be running
+    # at first use and then raise in any subsequent ``asyncio.run()``. We
+    # cache (loop, lock) and recreate when the loop changes so:
+    #   - within one rollout (one event loop), all concurrent sessions share
+    #     the same lock and serialize the clear-invoke-export cycle;
+    #   - tests / repeated runs that create new event loops still work.
+    # NOTE: ``threading.Lock`` MUST NOT be used here — it would block the
+    # entire event loop across the inner ``await`` and deadlock when
+    # ``rollout.concurrency > 1``.
+    _lock: asyncio.Lock | None = None
+    _lock_loop: asyncio.AbstractEventLoop | None = None
+
+    @classmethod
+    def get_lock(cls) -> asyncio.Lock:
+        """Return an ``asyncio.Lock`` bound to the current running loop."""
+        loop = asyncio.get_running_loop()
+        if cls._lock is None or cls._lock_loop is not loop:
+            cls._lock = asyncio.Lock()
+            cls._lock_loop = loop
+        return cls._lock
 
     def __new__(cls) -> "LiveOTelExporter":
         if cls._instance is None:

p2m/core/otel_session.py

@@ -19,8 +19,8 @@
 
 import importlib
 import inspect
-import threading
 import uuid
+from contextlib import nullcontext
 from typing import Any
 
 from p2m.core.async_utils import invoke_callable
@@ -141,16 +141,16 @@ async def run_turn(self, messages: list[Message]) -> TurnResult:
         - ``span.id``: Each span → its own event (maximum granularity)
 
         When using ``LiveOTelExporter`` (singleton, shared span buffer), the
-        clear-invoke-export cycle is serialized via a lock to prevent
-        concurrent sessions from contaminating each other's span data.
+        clear-invoke-export cycle is serialized via an ``asyncio.Lock`` to
+        prevent concurrent sessions from contaminating each other's span
+        data. The lock MUST be ``asyncio.Lock`` (not ``threading.Lock``):
+        the rollout stage runs all sessions in one event loop, so holding a
+        sync threading lock across the inner ``await`` would block the loop
+        and deadlock when ``rollout.concurrency > 1``.
         """
-        # Acquire the live exporter lock for the entire clear-invoke-export
-        # cycle so concurrent OTelTracedSessions don't mix spans.
         from p2m.core.otel import LiveOTelExporter
-        lock = LiveOTelExporter._lock if self._live_otel else None
-        if lock:
-            lock.acquire()
-        try:
+        lock_ctx = LiveOTelExporter.get_lock() if self._live_otel else nullcontext()
+        async with lock_ctx:
             # Clear spans from previous turn so we only capture this turn's execution
             if self._live_exporter is not None:
                 self._live_exporter.clear()
@@ -188,9 +188,6 @@ async def run_turn(self, messages: list[Message]) -> TurnResult:
 
             # Collect traces for this turn
             turn_spans = self._exporter.export_session(turn_id)
-        finally:
-            if lock:
-                lock.release()
 
         validation = validate_spans(turn_spans)