agent_event_bus.py

"""
用于 main_server <-> agent_server 通信的 ZeroMQ 事件总线。

重要说明：这里使用 **同步** 的 zmq.Context + zmq.Socket，并通过后台
守护线程执行 recv。原因是 zmq.asyncio.Socket.recv 依赖事件循环的
fd 轮询（add_reader），而该机制在 Windows ProactorEventLoop 上不可用。
发送侧使用 zmq.NOBLOCK，并在 asyncio 线程内调用（本地 TCP 延迟很低）。
"""

import asyncio
import os
import threading
import time
import uuid
from typing import Any, Awaitable, Callable, Dict, Optional

import orjson

from utils.logger_config import get_module_logger

try:
    import zmq
except Exception:  # pragma: no cover - optional dependency at runtime
    zmq = None

logger = get_module_logger(__name__, "Main")

# ZMQ 地址：支持环境变量覆盖，便于 launcher 在默认端口落入
# Hyper-V 保留区时进行迁移。
def _zmq_addr(env_key: str, default_port: int) -> str:
    raw = os.getenv(env_key, "").strip()
    if raw:
        try:
            val = int(raw)
            if 1 <= val <= 65535:
                return f"tcp://127.0.0.1:{val}"
        except (ValueError, TypeError):
            pass
    return f"tcp://127.0.0.1:{default_port}"

SESSION_PUB_ADDR  = _zmq_addr("NEKO_ZMQ_SESSION_PUB_PORT", 48961)   # main -> agent（PUB/SUB）
AGENT_PUSH_ADDR   = _zmq_addr("NEKO_ZMQ_AGENT_PUSH_PORT", 48962)    # agent -> main（PUSH/PULL）
ANALYZE_PUSH_ADDR = _zmq_addr("NEKO_ZMQ_ANALYZE_PUSH_PORT", 48963)  # main -> agent（PUSH/PULL，可靠分析队列）

_main_bridge_ref: Optional["MainServerAgentBridge"] = None
_ack_waiters: dict[str, asyncio.Future] = {}
_ack_waiters_lock = threading.Lock()
_voice_bridge_waiters: dict[str, asyncio.Future] = {}
_voice_bridge_request_seen_waiters: dict[str, asyncio.Future] = {}
# Voice bridge waiter state is protected by _voice_bridge_waiters_lock.
# queued means a result has been enqueued on the main loop but resolution has
# not started; resolving means notify_voice_bridge_result is actively resolving
# it. Normal flow is queued -> resolving -> removed.
_voice_bridge_waiters_queued: set[str] = set()
_voice_bridge_waiters_resolving: set[str] = set()
_voice_bridge_waiters_lock = threading.Lock()
VOICE_BRIDGE_AGENT_SEEN_TIMEOUT_SECONDS = 0.08


# ---------------------------------------------------------------------------
#  main_server 侧桥接器
# ---------------------------------------------------------------------------

class MainServerAgentBridge:
    """运行于 main_server 进程内，绑定 PUB、PUSH(analyze)、PULL(agent→main)。"""

    def __init__(self, on_agent_event: Callable[[Dict[str, Any]], Awaitable[None]]) -> None:
        self.on_agent_event = on_agent_event
        self.ctx: Any = None
        self.pub: Any = None
        self.analyze_push: Any = None
        self.pull: Any = None
        self._recv_thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self.owner_loop: Optional[asyncio.AbstractEventLoop] = None
        self.owner_thread_id: Optional[int] = None
        self.ready = False

    async def start(self) -> None:
        if zmq is None:
            logger.warning("pyzmq not installed, event bus disabled on main_server")
            return

        self.ctx = zmq.Context()

        self.pub = self.ctx.socket(zmq.PUB)
        self.pub.setsockopt(zmq.LINGER, 1000)
        self.pub.bind(SESSION_PUB_ADDR)

        self.analyze_push = self.ctx.socket(zmq.PUSH)
        self.analyze_push.setsockopt(zmq.LINGER, 1000)
        self.analyze_push.bind(ANALYZE_PUSH_ADDR)

        self.pull = self.ctx.socket(zmq.PULL)
        self.pull.setsockopt(zmq.LINGER, 1000)
        self.pull.setsockopt(zmq.RCVTIMEO, 1000)
        self.pull.bind(AGENT_PUSH_ADDR)

        self.owner_loop = asyncio.get_running_loop()
        self.owner_thread_id = threading.get_ident()
        self.ready = True

        self._recv_thread = threading.Thread(
            target=self._recv_thread_fn, name="zmq-main-recv", daemon=True,
        )
        self._recv_thread.start()
        logger.info("[EventBus] Main bridge started (pid=%s)", os.getpid())

    # -- 后台接收（agent → main） -------------------------------------------

    def _recv_thread_fn(self) -> None:
        while not self._stop.is_set():
            try:
                msg = orjson.loads(self.pull.recv())
                if isinstance(msg, dict) and self.owner_loop is not None:
                    if msg.get("event_type") == "voice_bridge_result":
                        mark_voice_bridge_result_queued(str(msg.get("event_id") or ""))
                    asyncio.run_coroutine_threadsafe(
                        self.on_agent_event(msg), self.owner_loop,
                    )
            except zmq.Again:
                continue
            except Exception as e:
                if not self._stop.is_set():
                    logger.debug("[EventBus] main recv thread error: %s", e)
                    time.sleep(0.05)

    # -- 发送辅助函数（在 asyncio 线程中调用） -------------------------------

    async def publish_session_event(self, event: Dict[str, Any]) -> bool:
        if not self.ready or self.pub is None:
            return False
        try:
            self.pub.send(orjson.dumps(event), zmq.NOBLOCK)
            return True
        except Exception:
            return False

    async def publish_analyze_request(self, event: Dict[str, Any]) -> bool:
        if not self.ready or self.analyze_push is None:
            return False
        try:
            self.analyze_push.send(orjson.dumps(event), zmq.NOBLOCK)
            return True
        except Exception:
            return False

    async def stop(self) -> None:
        """Shut down ZMQ resources and background thread."""
        self._stop.set()
        self.ready = False
        if self._recv_thread is not None:
            await asyncio.to_thread(self._recv_thread.join, 2.0)
        for sock in (self.pull, self.analyze_push, self.pub):
            if sock is not None:
                try:
                    sock.close(linger=0)
                except Exception:
                    pass
        if self.ctx is not None:
            _ctx = self.ctx
            self.ctx = None
            try:
                await asyncio.wait_for(asyncio.to_thread(_ctx.term), timeout=3.0)
            except (asyncio.TimeoutError, Exception):
                logger.debug("[EventBus] Main bridge ctx.term timed out or failed, skipping")
        logger.debug("[EventBus] Main bridge stopped")

    async def publish_session_event_threadsafe(self, event: Dict[str, Any]) -> bool:
        if self.owner_loop is None:
            return False
        if threading.get_ident() == self.owner_thread_id:
            return await self.publish_session_event(event)
        try:
            cf = asyncio.run_coroutine_threadsafe(
                self.publish_session_event(event), self.owner_loop,
            )
            return await asyncio.wrap_future(cf)
        except Exception:
            return False


# ---------------------------------------------------------------------------
#  agent_server 侧桥接器
# ---------------------------------------------------------------------------

class AgentServerEventBridge:
    """运行于 agent_server 进程内，连接 SUB、PULL(analyze)、PUSH(agent→main)。"""

    def __init__(self, on_session_event: Callable[[Dict[str, Any]], Awaitable[None]]) -> None:
        self.on_session_event = on_session_event
        self.ctx: Any = None
        self.sub: Any = None
        self.analyze_pull: Any = None
        self.push: Any = None
        self._recv_thread: Optional[threading.Thread] = None
        self._analyze_recv_thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self._owner_loop: Optional[asyncio.AbstractEventLoop] = None
        self.ready = False

    async def start(self) -> None:
        if zmq is None:
            logger.warning("pyzmq not installed, event bus disabled on agent_server")
            return

        self.ctx = zmq.Context()

        self.sub = self.ctx.socket(zmq.SUB)
        self.sub.setsockopt(zmq.LINGER, 1000)
        self.sub.setsockopt(zmq.RCVTIMEO, 1000)
        self.sub.connect(SESSION_PUB_ADDR)
        self.sub.setsockopt_string(zmq.SUBSCRIBE, "")

        self.analyze_pull = self.ctx.socket(zmq.PULL)
        self.analyze_pull.setsockopt(zmq.LINGER, 1000)
        self.analyze_pull.setsockopt(zmq.RCVTIMEO, 1000)
        self.analyze_pull.connect(ANALYZE_PUSH_ADDR)

        self.push = self.ctx.socket(zmq.PUSH)
        self.push.setsockopt(zmq.LINGER, 1000)
        self.push.connect(AGENT_PUSH_ADDR)

        self._owner_loop = asyncio.get_running_loop()
        self.ready = True

        self._recv_thread = threading.Thread(
            target=self._recv_sub_fn, name="zmq-agent-sub", daemon=True,
        )
        self._recv_thread.start()

        self._analyze_recv_thread = threading.Thread(
            target=self._recv_analyze_fn, name="zmq-agent-analyze", daemon=True,
        )
        self._analyze_recv_thread.start()
        logger.info("[EventBus] Agent bridge started (pid=%s)", os.getpid())

    # -- 后台接收线程 -------------------------------------------------------

    def _recv_sub_fn(self) -> None:
        while not self._stop.is_set():
            try:
                msg = orjson.loads(self.sub.recv())
                if isinstance(msg, dict) and self._owner_loop is not None:
                    asyncio.run_coroutine_threadsafe(
                        self.on_session_event(msg), self._owner_loop,
                    )
            except zmq.Again:
                continue
            except Exception as e:
                if not self._stop.is_set():
                    logger.debug("[EventBus] agent sub recv thread error: %s", e)
                    time.sleep(0.05)

    def _recv_analyze_fn(self) -> None:
        while not self._stop.is_set():
            try:
                msg = orjson.loads(self.analyze_pull.recv())
                if isinstance(msg, dict):
                    if msg.get("event_type") == "analyze_request":
                        logger.info(
                            "[EventBus] analyze_request dequeued on agent: event_id=%s lanlan=%s trigger=%s",
                            msg.get("event_id"),
                            msg.get("lanlan_name"),
                            msg.get("trigger"),
                        )
                    if self._owner_loop is not None:
                        asyncio.run_coroutine_threadsafe(
                            self.on_session_event(msg), self._owner_loop,
                        )
            except zmq.Again:
                continue
            except Exception as e:
                if not self._stop.is_set():
                    logger.debug("[EventBus] agent analyze recv thread error: %s", e)
                    time.sleep(0.05)

    # -- 发送辅助函数（在 asyncio 线程中调用） -------------------------------

    async def emit_to_main(self, event: Dict[str, Any]) -> bool:
        if not self.ready or self.push is None:
            return False
        try:
            self.push.send(orjson.dumps(event), zmq.NOBLOCK)
            return True
        except Exception:
            return False


# ---------------------------------------------------------------------------
#  模块级辅助函数（API 保持不变）
# ---------------------------------------------------------------------------

def set_main_bridge(bridge: Optional[MainServerAgentBridge]) -> None:
    global _main_bridge_ref
    _main_bridge_ref = bridge


async def publish_session_event(event: Dict[str, Any]) -> bool:
    if _main_bridge_ref is None:
        return False
    return await _main_bridge_ref.publish_session_event(event)


async def publish_session_event_threadsafe(event: Dict[str, Any]) -> bool:
    if _main_bridge_ref is None:
        return False
    bridge = _main_bridge_ref
    if hasattr(bridge, "publish_session_event_threadsafe"):
        return await bridge.publish_session_event_threadsafe(event)
    return await bridge.publish_session_event(event)


def notify_analyze_ack(event_id: str) -> None:
    if not event_id:
        return
    waiter = None
    with _ack_waiters_lock:
        waiter = _ack_waiters.pop(event_id, None)
    if waiter is None or waiter.done():
        return
    loop = waiter.get_loop()

    def _resolve() -> None:
        if not waiter.done():
            waiter.set_result(True)

    loop.call_soon_threadsafe(_resolve)


def _resolve_voice_bridge_seen_waiter(event_id: str, waiter: asyncio.Future | None) -> None:
    if waiter is None or waiter.done():
        return
    loop = waiter.get_loop()

    def _resolve() -> None:
        if not waiter.done():
            waiter.set_result(True)

    try:
        loop.call_soon_threadsafe(_resolve)
    except RuntimeError:
        with _voice_bridge_waiters_lock:
            _voice_bridge_request_seen_waiters.pop(event_id, None)
        if not waiter.done():
            waiter.cancel()


def notify_voice_bridge_request_seen(event_id: str) -> None:
    """Mark that agent_server received a voice-bridge request."""
    if not event_id:
        return
    with _voice_bridge_waiters_lock:
        waiter = _voice_bridge_request_seen_waiters.get(event_id)
    _resolve_voice_bridge_seen_waiter(event_id, waiter)


def mark_voice_bridge_result_queued(event_id: str) -> None:
    """Mark that a voice result is queued on the main loop but not resolved yet."""
    if not event_id:
        return
    with _voice_bridge_waiters_lock:
        if event_id in _voice_bridge_waiters:
            _voice_bridge_waiters_queued.add(event_id)
            waiter = _voice_bridge_request_seen_waiters.get(event_id)
        else:
            waiter = None
    _resolve_voice_bridge_seen_waiter(event_id, waiter)


def notify_voice_bridge_result(event_id: str, result: Dict[str, Any]) -> None:
    """Resolve a pending voice-bridge request sent from main_server to agent_server."""
    if not event_id:
        return
    payload = result if isinstance(result, dict) else {}
    with _voice_bridge_waiters_lock:
        waiter = _voice_bridge_waiters.get(event_id)
        if waiter is not None and not waiter.done():
            _voice_bridge_waiters_resolving.add(event_id)
            _voice_bridge_waiters_queued.discard(event_id)
            seen_waiter = _voice_bridge_request_seen_waiters.get(event_id)
        else:
            _voice_bridge_waiters_resolving.discard(event_id)
            _voice_bridge_waiters_queued.discard(event_id)
            seen_waiter = None
    _resolve_voice_bridge_seen_waiter(event_id, seen_waiter)
    if waiter is None or waiter.done():
        return
    loop = waiter.get_loop()

    def _resolve() -> None:
        with _voice_bridge_waiters_lock:
            if _voice_bridge_waiters.get(event_id) is not waiter:
                _voice_bridge_waiters_resolving.discard(event_id)
                _voice_bridge_waiters_queued.discard(event_id)
                return
            _voice_bridge_waiters.pop(event_id, None)
            _voice_bridge_request_seen_waiters.pop(event_id, None)
            _voice_bridge_waiters_resolving.discard(event_id)
            _voice_bridge_waiters_queued.discard(event_id)
        if not waiter.done():
            waiter.set_result(payload)

    try:
        loop.call_soon_threadsafe(_resolve)
    except RuntimeError:
        with _voice_bridge_waiters_lock:
            if _voice_bridge_waiters.get(event_id) is waiter:
                if not waiter.done():
                    waiter.cancel()
                _voice_bridge_waiters.pop(event_id, None)
            _voice_bridge_request_seen_waiters.pop(event_id, None)
            _voice_bridge_waiters_resolving.discard(event_id)
            _voice_bridge_waiters_queued.discard(event_id)


# ---------------------------------------------------------------------------
#  Layering-inversion sinks: lower layers (main_logic) emit, higher layers
#  (plugin / main_routers) register.
#
#  Why this exists
#  ---------------
#  ``main_logic.core`` used to ``import`` from ``plugin.core.state`` and
#  ``main_routers.system_router`` to publish user utterances and to consult
#  the mini-game-invite keyword matcher. Both are layering inversions
#  (main_logic L2 → plugin L4 / main_routers L3) — banned by
#  ``scripts/check_module_layering.py``.
#
#  Now main_logic emits via ``dispatch_*`` and the higher layers attach via
#  ``register_*``. The CONSUMERS self-register at module-import time
#  (plugin/core/state.py and main_routers/system_router.py), so any context
#  that loads those modules — even directly, without going through the
#  ``app`` entrypoint — gets its sink wired automatically. This preserves
#  the side-effect that direct ``main_logic.core`` consumers (testbench /
#  ad-hoc scripts) used to enjoy via the previous chained import. The
#  registries dedupe on identity, so ``app/runtime_bindings.py`` calling
#  ``register_*`` again after the consumer module is loaded is a no-op.
#
#  If nothing is registered (e.g. memory_server entrypoint doesn't ship
#  plugin runtime), the dispatchers silently no-op.
# ---------------------------------------------------------------------------

# Fire-and-forget user-utterance sink. Plugin's user-context bus subscribes
# here. Multiple subscribers are allowed; per-sink errors are swallowed so
# one misbehaving consumer cannot break the chat pipeline.
_user_utterance_sinks: list[Callable[[str, Dict[str, Any]], None]] = []


def register_user_utterance_sink(
    fn: Callable[[str, Dict[str, Any]], None],
) -> None:
    """Subscribe to user-utterance events: ``fn(bucket: str, event: dict)``.

    Dedupes on identity — re-registering the same callable is a no-op.
    Important because both ``plugin.core.state`` (self-register on import)
    and ``app.runtime_bindings`` (explicit wiring) call this for the same
    function; without dedup, every utterance would fire twice.
    """
    if fn in _user_utterance_sinks:
        return
    _user_utterance_sinks.append(fn)


def dispatch_user_utterance(bucket: str, event: Dict[str, Any]) -> None:
    """Fan a user utterance out to every registered sink."""
    for fn in _user_utterance_sinks:
        try:
            fn(bucket, event)
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("[EventBus] user_utterance sink raised: %s", exc)


# First-hit-wins text-message hook with optional return value.
# main_routers' mini-game-invite keyword matcher is the canonical consumer.
_text_user_message_hooks: list[
    Callable[[str, str], Optional[Dict[str, Any]]]
] = []


def register_text_user_message_hook(
    fn: Callable[[str, str], Optional[Dict[str, Any]]],
) -> None:
    """Subscribe to text user-message events: ``fn(lanlan_name, text) -> dict?``.

    First hook returning a truthy value wins; later hooks are skipped.
    Dedupes on identity (see ``register_user_utterance_sink`` rationale).
    """
    if fn in _text_user_message_hooks:
        return
    _text_user_message_hooks.append(fn)


def dispatch_text_user_message(
    lanlan_name: str, text: str,
) -> Optional[Dict[str, Any]]:
    """Run hooks in registration order; return the first truthy result."""
    for fn in _text_user_message_hooks:
        try:
            result = fn(lanlan_name, text)
            if result:
                return result
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("[EventBus] text_user_message hook raised: %s", exc)
    return None


async def publish_analyze_request_reliably(
    lanlan_name: str,
    trigger: str,
    messages: list[dict],
    *,
    ack_timeout_s: float = 0.8,
    retries: int = 1,
    conversation_id: Optional[str] = None,
) -> bool:
    """可靠发布 analyze_request：携带 event_id + ack，并支持短重试。"""
    event_id = uuid.uuid4().hex
    sent_at = time.perf_counter()

    for attempt in range(max(retries, 0) + 1):
        event = {
            "event_type": "analyze_request",
            "event_id": event_id,
            "trigger": trigger,
            "lanlan_name": lanlan_name,
            "messages": messages,
        }
        if conversation_id:
            event["conversation_id"] = conversation_id

        loop = asyncio.get_running_loop()
        waiter: asyncio.Future = loop.create_future()
        with _ack_waiters_lock:
            _ack_waiters[event_id] = waiter

        bridge = _main_bridge_ref
        if bridge is None:
            with _ack_waiters_lock:
                _ack_waiters.pop(event_id, None)
            return False
        if bridge.owner_loop is None:
            with _ack_waiters_lock:
                _ack_waiters.pop(event_id, None)
            return False

        if threading.get_ident() == bridge.owner_thread_id:
            sent = await bridge.publish_analyze_request(event)
        else:
            try:
                if bridge.owner_loop.is_closed():
                    logger.debug("[EventBus] owner_loop closed, skipping publish")
                    sent = False
                else:
                    coro = bridge.publish_analyze_request(event)
                    try:
                        cf = asyncio.run_coroutine_threadsafe(coro, bridge.owner_loop)
                        sent = await asyncio.wrap_future(cf)
                    except Exception as e:
                        coro.close()
                        logger.debug("[EventBus] publish_analyze_request threadsafe failed: %s", e)
                        sent = False
            except Exception as e:
                logger.debug("[EventBus] publish_analyze_request threadsafe failed: %s", e)
                sent = False

        if not sent:
            with _ack_waiters_lock:
                _ack_waiters.pop(event_id, None)
            continue

        try:
            await asyncio.wait_for(waiter, timeout=ack_timeout_s)
            logger.info(
                "[EventBus] analyze_request acked: event_id=%s lanlan=%s trigger=%s latency_ms=%.1f",
                event_id,
                lanlan_name,
                trigger,
                (time.perf_counter() - sent_at) * 1000.0,
            )
            return True
        except asyncio.TimeoutError:
            with _ack_waiters_lock:
                _ack_waiters.pop(event_id, None)
            logger.info(
                "[EventBus] analyze_request ack timeout (attempt %d): event_id=%s lanlan=%s trigger=%s",
                attempt + 1,
                event_id,
                lanlan_name,
                trigger,
            )

    return False


async def publish_voice_transcript_request_reliably(
    lanlan_name: str,
    transcript: str,
    *,
    timeout_s: float = 1.2,
    agent_seen_timeout_s: float = VOICE_BRIDGE_AGENT_SEEN_TIMEOUT_SECONDS,
    retries: int = 0,
    metadata: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
    """Ask agent_server/plugins how a realtime voice transcript should be handled.

    Returns a plugin-produced action payload, or ``None`` when the bridge is not
    running or no response arrives within the bounded timeout.
    """
    text = str(transcript or "").strip()
    if not text:
        return None
    event_id = uuid.uuid4().hex
    loop = asyncio.get_running_loop()
    waiter: asyncio.Future = loop.create_future()
    seen_waiter: asyncio.Future = loop.create_future()
    with _voice_bridge_waiters_lock:
        _voice_bridge_waiters[event_id] = waiter
        _voice_bridge_request_seen_waiters[event_id] = seen_waiter

    try:
        for attempt in range(max(retries, 0) + 1):
            event = {
                "event_type": "voice_transcript_request",
                "event_id": event_id,
                "lanlan_name": lanlan_name,
                "transcript": text,
                "metadata": dict(metadata or {}),
                "attempt": attempt,
            }
            sent = await publish_session_event(event)
            if not sent:
                logger.debug(
                    "[EventBus] voice_transcript_request not sent: no main bridge lanlan=%s attempt=%s",
                    lanlan_name,
                    attempt,
                )
                continue
            await asyncio.wait(
                {asyncio.shield(waiter), asyncio.shield(seen_waiter)},
                timeout=max(0.0, float(agent_seen_timeout_s)),
                return_when=asyncio.FIRST_COMPLETED,
            )
            if waiter.done():
                result = waiter.result()
                return result if isinstance(result, dict) else {}
            if not seen_waiter.done():
                logger.debug(
                    "[EventBus] voice_transcript_request not acknowledged by agent: event_id=%s lanlan=%s attempt=%s",
                    event_id,
                    lanlan_name,
                    attempt,
                )
                continue
            try:
                result = await asyncio.wait_for(asyncio.shield(waiter), timeout=timeout_s)
                return result if isinstance(result, dict) else {}
            except asyncio.TimeoutError:
                with _voice_bridge_waiters_lock:
                    result_is_queued = (
                        _voice_bridge_waiters.get(event_id) is waiter
                        and (
                            event_id in _voice_bridge_waiters_resolving
                            or event_id in _voice_bridge_waiters_queued
                        )
                    )
                if result_is_queued:
                    try:
                        result = await asyncio.wait_for(
                            asyncio.shield(waiter),
                            timeout=0.05,
                        )
                        return result if isinstance(result, dict) else {}
                    except asyncio.TimeoutError:
                        pass
                logger.debug(
                    "[EventBus] voice_transcript_request timed out: event_id=%s lanlan=%s attempt=%s",
                    event_id,
                    lanlan_name,
                    attempt,
                )
                continue
        return None
    finally:
        should_cancel = False
        with _voice_bridge_waiters_lock:
            if _voice_bridge_waiters.get(event_id) is waiter:
                _voice_bridge_waiters.pop(event_id, None)
                should_cancel = True
            _voice_bridge_request_seen_waiters.pop(event_id, None)
            _voice_bridge_waiters_queued.discard(event_id)
            _voice_bridge_waiters_resolving.discard(event_id)
        if should_cancel and not waiter.done():
            waiter.cancel()
        if not seen_waiter.done():
            seen_waiter.cancel()