fix: route voice transcripts through agent bridge

Author: MomiJiSan

app/agent_server.py

@@ -13,6 +13,7 @@
 
 import mimetypes
 import json
+from collections.abc import Mapping
 mimetypes.add_type("application/javascript", ".js")
 import asyncio
 import uuid
@@ -1650,6 +1651,143 @@ async def _emit_agent_status_update(lanlan_name: Optional[str] = None) -> None:
         pass
 
 
+VOICE_TRANSCRIPT_CUSTOM_EVENT_TYPE = "voice_transcript"
+VOICE_TRANSCRIPT_CUSTOM_EVENT_ID = "handle_transcript"
+VOICE_TRANSCRIPT_CUSTOM_EVENT_TIMEOUT_SECONDS = 1.0
+
+
+def _voice_bridge_noop(reason: str, **extra: object) -> Dict[str, Any]:
+    return {
+        "action": "noop",
+        "reason": str(reason or "noop"),
+        **extra,
+    }
+
+
+def _voice_transcript_request_has_text(event: Mapping[str, object] | None) -> bool:
+    if not isinstance(event, Mapping):
+        return False
+    return bool(str(event.get("transcript") or "").strip())
+
+
+def _voice_transcript_custom_event_args(event: Mapping[str, object]) -> Dict[str, object]:
+    metadata = event.get("metadata")
+    return {
+        "transcript": str(event.get("transcript") or "").strip(),
+        "lanlan_name": str(event.get("lanlan_name") or ""),
+        "metadata": dict(metadata) if isinstance(metadata, Mapping) else {},
+    }
+
+
+def _voice_bridge_action_from_dispatch_results(dispatch_results: object) -> Dict[str, Any]:
+    if not isinstance(dispatch_results, list) or not dispatch_results:
+        return _voice_bridge_noop("no_subscribers")
+
+    from plugin.server.application.plugins.voice_contracts import (
+        arbitrate_voice_transcript_results,
+    )
+
+    arbitration_items: list[dict[str, object]] = []
+    failure_count = 0
+    for item in dispatch_results:
+        if not isinstance(item, Mapping):
+            continue
+        if not bool(item.get("success")):
+            failure_count += 1
+            continue
+        result = item.get("result")
+        if not isinstance(result, Mapping):
+            continue
+        action = str(result.get("action") or "").strip()
+        if not action:
+            continue
+        payload: Dict[str, Any] = dict(result)
+        payload["action"] = action
+        plugin_id = str(item.get("plugin_id") or "").strip()
+        if plugin_id:
+            payload.setdefault("source_plugin", plugin_id)
+        source_event_id = str(item.get("event_id") or "").strip()
+        if source_event_id:
+            payload.setdefault("source_event_id", source_event_id)
+        arbitration_items.append(
+            {
+                "plugin_id": payload.get("source_plugin") or plugin_id,
+                "event_id": payload.get("source_event_id") or source_event_id,
+                "success": True,
+                "result": payload,
+            }
+        )
+
+    if not arbitration_items:
+        return _voice_bridge_noop("no_handler_result", failures=failure_count)
+    payload = arbitrate_voice_transcript_results(arbitration_items)
+    if failure_count:
+        try:
+            existing_failures = int(payload.get("failures") or 0)
+        except (TypeError, ValueError):
+            existing_failures = 0
+        payload["failures"] = existing_failures + failure_count
+    return payload
+
+
+async def _dispatch_voice_transcript_custom_event(
+    event: Mapping[str, object],
+) -> Dict[str, Any]:
+    from plugin.server.application.plugins.dispatch_service import PluginDispatchService
+
+    dispatch_results = await PluginDispatchService().trigger_custom_event_subscribers(
+        event_type=VOICE_TRANSCRIPT_CUSTOM_EVENT_TYPE,
+        event_id=VOICE_TRANSCRIPT_CUSTOM_EVENT_ID,
+        args=_voice_transcript_custom_event_args(event),
+        timeout=VOICE_TRANSCRIPT_CUSTOM_EVENT_TIMEOUT_SECONDS,
+    )
+    return _voice_bridge_action_from_dispatch_results(dispatch_results)
+
+
+async def _handle_voice_transcript_request(event: Dict[str, Any]) -> None:
+    event_id = str((event or {}).get("event_id") or "")
+    lanlan_name = (event or {}).get("lanlan_name")
+    result: Dict[str, Any] = _voice_bridge_noop("unavailable")
+
+    try:
+        if not _voice_transcript_request_has_text(event):
+            result = _voice_bridge_noop("empty_transcript")
+        elif not Modules.analyzer_enabled:
+            result = _voice_bridge_noop("agent_disabled")
+        elif not Modules.agent_flags.get("user_plugin_enabled", False):
+            result = _voice_bridge_noop("user_plugin_disabled")
+        else:
+            lifecycle_ready = bool(Modules.plugin_lifecycle_started)
+            if not lifecycle_ready:
+                lifecycle_ready = await _ensure_plugin_lifecycle_started()
+
+            if not lifecycle_ready:
+                result = _voice_bridge_noop("plugin_lifecycle_start_failed")
+            else:
+                result = await _dispatch_voice_transcript_custom_event(event)
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        logger.debug(
+            "[VoiceBridge] plugin dispatch failed: event_id=%s lanlan=%s err=%s",
+            event_id,
+            lanlan_name,
+            exc,
+        )
+        result = {
+            "action": "noop",
+            "reason": "dispatch_failed",
+            "error_type": type(exc).__name__,
+        }
+
+    await _emit_main_event(
+        "voice_bridge_result",
+        lanlan_name if isinstance(lanlan_name, str) else None,
+        event_id=event_id,
+        result=result,
+    )
+
+
 async def _on_session_event(event: Dict[str, Any]) -> None:
     event_type = (event or {}).get("event_type")
     if event_type == "agent_intent_restore_signal":
@@ -1662,6 +1800,20 @@ async def _on_session_event(event: Dict[str, Any]) -> None:
         # has its own once-flag, so this is safe to spam.
         await _maybe_restore_agent_intent()
         return
+    if event_type == "voice_transcript_request":
+        seen_task = asyncio.create_task(
+            _emit_main_event(
+                "voice_bridge_request_seen",
+                event.get("lanlan_name") if isinstance(event.get("lanlan_name"), str) else None,
+                event_id=str(event.get("event_id") or ""),
+            )
+        )
+        Modules._background_tasks.add(seen_task)
+        seen_task.add_done_callback(Modules._background_tasks.discard)
+        task = asyncio.create_task(_handle_voice_transcript_request(event))
+        Modules._background_tasks.add(task)
+        task.add_done_callback(Modules._background_tasks.discard)
+        return
     if event_type == "analyze_request":
         messages = event.get("messages", [])
         lanlan_name = event.get("lanlan_name")

app/main_server.py

@@ -121,7 +121,13 @@ def _resolve_user_plugin_base() -> str:
     from fastapi.responses import JSONResponse, Response # noqa
     from fastapi.staticfiles import StaticFiles # noqa
     from main_logic import core as core, cross_server as cross_server # noqa
-    from main_logic.agent_event_bus import MainServerAgentBridge, notify_analyze_ack, set_main_bridge # noqa
+    from main_logic.agent_event_bus import (
+        MainServerAgentBridge,
+        notify_analyze_ack,
+        notify_voice_bridge_request_seen,
+        notify_voice_bridge_result,
+        set_main_bridge,
+    ) # noqa
     from fastapi.templating import Jinja2Templates # noqa
     from dataclasses import dataclass # noqa
     from typing import Any, Optional # noqa
@@ -608,6 +614,17 @@ async def _handle_agent_event(event: dict):
             notify_analyze_ack(str(event.get("event_id") or ""))
             return
 
+        if event_type == "voice_bridge_result":
+            notify_voice_bridge_result(
+                str(event.get("event_id") or ""),
+                event.get("result") if isinstance(event.get("result"), dict) else {},
+            )
+            return
+
+        if event_type == "voice_bridge_request_seen":
+            notify_voice_bridge_request_seen(str(event.get("event_id") or ""))
+            return
+
         # Agent status updates may be broadcast (lanlan_name omitted).
         if event_type == "agent_status_update":
             payload = {