fix(meeting-agent): serialize live-transcript metadata read-modify-write

Recall webhooks are dispatched one-thread-each via asyncio.to_thread, so
rapid-fire transcript.data events ran _upsert_live_transcript concurrently.
Each did an unsynchronized read (get_meeting) -> append -> UPDATE of the
persisted meeting transcript, so two concurrent captions could read the same
base and one write would clobber the other, dropping transcript lines.

Add a shared reentrant lock (pipeline._metadata_lock) around every metadata
read-modify-write in merge_meeting_metadata, and hold it across the read in
_upsert_live_transcript so the whole read-append-write is atomic. In-memory
fan-out (live.py) already had a lock; this closes the same gap for the
persisted transcript. Example-only, no package params/exports/deps added.

examples/python/meeting-agent/integrations/recall/processor.py

@@ -23,29 +23,34 @@ def _upsert_live_transcript(meeting_id: str, line: str, *, partial: bool = False
     """Merge one live caption line; partial updates replace the last same-speaker line."""
     from praisonai_tools.tools.meeting_tools import get_meeting
     from integrations.recall.live import publish_live_chunk
+    from pipeline import _metadata_lock
 
-    record = get_meeting.__wrapped__(meeting_id)
-    meta = record.get("metadata") or {} if isinstance(record, dict) else {}
-    existing = str(meta.get("transcript") or "")
-    lines = existing.splitlines()
-    prefix = _speaker_prefix(line)
+    # Hold the merge lock across the read too: streaming transcript events are
+    # dispatched one-thread-each, so an unsynchronized read-append-write would
+    # let concurrent captions clobber each other and drop lines.
+    with _metadata_lock:
+        record = get_meeting.__wrapped__(meeting_id)
+        meta = record.get("metadata") or {} if isinstance(record, dict) else {}
+        existing = str(meta.get("transcript") or "")
+        lines = existing.splitlines()
+        prefix = _speaker_prefix(line)
 
-    if lines and lines[-1] == line:
-        full = existing
-    elif partial and lines and prefix and lines[-1].startswith(prefix):
-        lines[-1] = line
-        full = "\n".join(lines)
-    elif not partial and lines and prefix and lines[-1].startswith(prefix):
-        # Final utterance replaces the in-progress partial for this speaker.
-        lines[-1] = line
-        full = "\n".join(lines)
-    else:
-        full = f"{existing}\n{line}".strip() if existing else line
+        if lines and lines[-1] == line:
+            full = existing
+        elif partial and lines and prefix and lines[-1].startswith(prefix):
+            lines[-1] = line
+            full = "\n".join(lines)
+        elif not partial and lines and prefix and lines[-1].startswith(prefix):
+            # Final utterance replaces the in-progress partial for this speaker.
+            lines[-1] = line
+            full = "\n".join(lines)
+        else:
+            full = f"{existing}\n{line}".strip() if existing else line
 
-    _merge_metadata(
-        meeting_id,
-        {"transcript": full, "live_status": "live", "status": "live"},
-    )
+        _merge_metadata(
+            meeting_id,
+            {"transcript": full, "live_status": "live", "status": "live"},
+        )
     publish_live_chunk(meeting_id, line=line, full_text=full, live_status="live")
     return full
 

examples/python/meeting-agent/pipeline.py

@@ -10,12 +10,19 @@ from __future__ import annotations
 import json
 import logging
 import re
+import threading
 from contextlib import closing
 from pathlib import Path
 from typing import Any
 
 logger = logging.getLogger(__name__)
 
+# Serializes metadata read-modify-write. Webhooks are dispatched one-thread-each
+# (asyncio.to_thread), so concurrent transcript events would otherwise clobber
+# each other's appends. Reentrant so callers can hold it across their own
+# read + merge (see processor._upsert_live_transcript).
+_metadata_lock = threading.RLock()
+
 AUDIO_SUFFIXES = {".mp3", ".mp4", ".m4a", ".wav", ".webm", ".ogg", ".mpeg", ".mpga"}
 _MEETING_ID_RE = re.compile(
     r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
@@ -27,19 +34,20 @@ def merge_meeting_metadata(meeting_id: str, patch: dict[str, Any]) -> dict[str,
     """Merge keys into a meeting's metadata blob."""
     from praisonai_tools.tools import meeting_tools as mt
 
-    record = mt.get_meeting.__wrapped__(meeting_id)
-    if "error" in record:
-        return record
-    meta = {**(record.get("metadata") or {}), **patch}
-    try:
-        with closing(mt._connect()) as conn, conn:
-            conn.execute(
-                "UPDATE meetings SET metadata = ? WHERE meeting_id = ?",
-                (json.dumps(meta, sort_keys=True), meeting_id),
-            )
-    except Exception as exc:  # noqa: BLE001
-        logger.error("merge_meeting_metadata failed: %s", exc)
-        return {"error": str(exc)}
+    with _metadata_lock:
+        record = mt.get_meeting.__wrapped__(meeting_id)
+        if "error" in record:
+            return record
+        meta = {**(record.get("metadata") or {}), **patch}
+        try:
+            with closing(mt._connect()) as conn, conn:
+                conn.execute(
+                    "UPDATE meetings SET metadata = ? WHERE meeting_id = ?",
+                    (json.dumps(meta, sort_keys=True), meeting_id),
+                )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("merge_meeting_metadata failed: %s", exc)
+            return {"error": str(exc)}
     return {"meeting_id": meeting_id, "metadata": meta}