fix: address voice bridge review feedback

Author: MomiJiSan

app/agent_server.py

@@ -50,6 +50,14 @@
 from utils.config_manager import get_config_manager
 from utils.tokenize import truncate_to_tokens as _tt
 from main_logic.agent_event_bus import AgentServerEventBridge
+from plugin.server.application.plugins.voice_contracts import (
+    VOICE_TRANSCRIPT_EVENT_ID,
+    VOICE_TRANSCRIPT_EVENT_TYPE,
+    arbitrate_voice_transcript_results,
+    voice_transcript_custom_event_args,
+    voice_transcript_noop,
+    voice_transcript_request_has_text,
+)
 try:
     from brain.computer_use import ComputerUseAdapter
     from brain.browser_use_adapter import BrowserUseAdapter
@@ -1651,83 +1659,11 @@ 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
+    return arbitrate_voice_transcript_results(dispatch_results)
 
 
 async def _dispatch_voice_transcript_custom_event(
@@ -1736,9 +1672,9 @@ async def _dispatch_voice_transcript_custom_event(
     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),
+        event_type=VOICE_TRANSCRIPT_EVENT_TYPE,
+        event_id=VOICE_TRANSCRIPT_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)
@@ -1747,24 +1683,19 @@ async def _dispatch_voice_transcript_custom_event(
 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")
+    result: Dict[str, Any] = voice_transcript_noop("unavailable")
 
     try:
-        if not _voice_transcript_request_has_text(event):
-            result = _voice_bridge_noop("empty_transcript")
+        if not voice_transcript_request_has_text(event):
+            result = voice_transcript_noop("empty_transcript")
         elif not Modules.analyzer_enabled:
-            result = _voice_bridge_noop("agent_disabled")
+            result = voice_transcript_noop("agent_disabled")
         elif not Modules.agent_flags.get("user_plugin_enabled", False):
-            result = _voice_bridge_noop("user_plugin_disabled")
+            result = voice_transcript_noop("user_plugin_disabled")
+        elif not bool(Modules.plugin_lifecycle_started):
+            result = voice_transcript_noop("plugin_warming_up")
         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)
+            result = await _dispatch_voice_transcript_custom_event(event)
     except asyncio.CancelledError:
         raise
     except Exception as exc:
@@ -1774,11 +1705,10 @@ async def _handle_voice_transcript_request(event: Dict[str, Any]) -> None:
             lanlan_name,
             exc,
         )
-        result = {
-            "action": "noop",
-            "reason": "dispatch_failed",
-            "error_type": type(exc).__name__,
-        }
+        result = voice_transcript_noop(
+            "dispatch_failed",
+            error_type=type(exc).__name__,
+        )
 
     await _emit_main_event(
         "voice_bridge_result",

main_logic/agent_event_bus.py

@@ -47,6 +47,10 @@ def _zmq_addr(env_key: str, default_port: int) -> str:
 _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()

main_logic/core.py

@@ -2009,16 +2009,21 @@ def _session_changed() -> bool:
         action = str(result.get("action") or "").strip()
         if action == "cancel_response":
             cancel_response = getattr(session_snapshot, "cancel_response", None)
-            if callable(cancel_response):
-                try:
-                    if _session_changed():
-                        return ""
-                    await cancel_response()
-                    logger.debug("[%s] voice bridge cancelled current response", self.lanlan_name)
-                except asyncio.CancelledError:
-                    raise
-                except Exception as exc:
-                    logger.debug("[%s] voice bridge cancel skipped/failed: %s", self.lanlan_name, exc)
+            if not callable(cancel_response):
+                logger.debug("[%s] voice bridge cancel skipped: session has no cancel_response", self.lanlan_name)
+                return ""
+            try:
+                if _session_changed():
+                    return ""
+                await cancel_response()
+                if _session_changed():
+                    return ""
+                logger.debug("[%s] voice bridge cancelled current response", self.lanlan_name)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.debug("[%s] voice bridge cancel skipped/failed: %s", self.lanlan_name, exc)
+                return ""
             return action
 
         if action == "prime_context":

plugin/plugins/study_companion/voice_contracts.py

@@ -16,6 +16,8 @@
     VOICE_TRANSCRIPT_EVENT_TYPE,
     arbitrate_voice_transcript_results,
     voice_transcript_cancel_response,
+    voice_transcript_custom_event_args,
     voice_transcript_noop,
     voice_transcript_prime_context,
+    voice_transcript_request_has_text,
 )

plugin/server/application/plugins/voice_contracts.py

@@ -90,6 +90,25 @@ def voice_transcript_prime_context(
     return payload
 
 
+def voice_transcript_request_has_text(event: Mapping[str, object] | None) -> bool:
+    """Return whether a loosely-typed voice transcript request has usable text."""
+    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]:
+    """Build canonical custom-event args for a voice transcript request."""
+    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 {},
+    }
+
+
 # ---------------------------------------------------------------------------
 # Arbitration helpers
 # ---------------------------------------------------------------------------
@@ -217,6 +236,8 @@ def arbitrate_voice_transcript_results(dispatch_results: object) -> dict[str, An
     "VOICE_TRANSCRIPT_EVENT_TYPE",
     "arbitrate_voice_transcript_results",
     "voice_transcript_cancel_response",
+    "voice_transcript_custom_event_args",
     "voice_transcript_noop",
     "voice_transcript_prime_context",
+    "voice_transcript_request_has_text",
 ]

plugin/tests/unit/plugins/test_study_companion_voice_contracts.py

@@ -4,6 +4,8 @@
 
 from plugin.plugins.study_companion.voice_contracts import (
     arbitrate_voice_transcript_results,
+    voice_transcript_custom_event_args,
+    voice_transcript_request_has_text,
 )
 
 pytestmark = pytest.mark.plugin_unit
@@ -104,3 +106,19 @@ def test_voice_transcript_arbitration_all_noop_continues_ordinary_flow() -> None
         "handlers": 1,
         "failures": 1,
     }
+
+
+def test_voice_transcript_request_helpers_normalize_event_args() -> None:
+    event = {
+        "transcript": "  Yui help  ",
+        "lanlan_name": "Lan",
+        "metadata": {"request_id": "req-1"},
+    }
+
+    assert voice_transcript_request_has_text(event) is True
+    assert voice_transcript_request_has_text({"transcript": "  "}) is False
+    assert voice_transcript_custom_event_args(event) == {
+        "transcript": "Yui help",
+        "lanlan_name": "Lan",
+        "metadata": {"request_id": "req-1"},
+    }

tests/unit/test_core_game_route_memory_contract.py

@@ -389,6 +389,82 @@ async def fake_publish(lanlan_name, transcript, *, metadata):
     assert mgr.sync_message_queue.messages == []
 
 
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_voice_transcript_bridge_cancel_without_session_handler_uses_ordinary_flow(
+    monkeypatch,
+):
+    async def fake_publish(*_args, **_kwargs):
+        return {"action": "cancel_response", "reason": "ocr_overlap"}
+
+    mgr = _make_transcript_manager()
+    mgr.session = object()
+    monkeypatch.setattr(
+        core_module,
+        "publish_voice_transcript_request_reliably",
+        fake_publish,
+    )
+
+    await core_module.LLMSessionManager.handle_input_transcript(
+        mgr,
+        "screen echo",
+        is_voice_source=True,
+    )
+
+    assert mgr._activity_tracker.voice_rms_count == 1
+    assert mgr._activity_tracker.user_messages == ["screen echo"]
+    assert mgr._session_turn_count == 1
+    mgr._publish_user_utterance_to_plugin_bus.assert_called_once_with(
+        "screen echo",
+        is_voice_source=True,
+    )
+    assert mgr.sync_message_queue.messages == [
+        {
+            "type": "user",
+            "data": {"input_type": "transcript", "data": "screen echo"},
+        }
+    ]
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_voice_transcript_bridge_cancel_failure_uses_ordinary_flow(monkeypatch):
+    class _FailingCancelSession:
+        async def cancel_response(self):
+            raise RuntimeError("cancel failed")
+
+    async def fake_publish(*_args, **_kwargs):
+        return {"action": "cancel_response", "reason": "ocr_overlap"}
+
+    mgr = _make_transcript_manager()
+    mgr.session = _FailingCancelSession()
+    monkeypatch.setattr(
+        core_module,
+        "publish_voice_transcript_request_reliably",
+        fake_publish,
+    )
+
+    await core_module.LLMSessionManager.handle_input_transcript(
+        mgr,
+        "screen echo",
+        is_voice_source=True,
+    )
+
+    assert mgr._activity_tracker.voice_rms_count == 1
+    assert mgr._activity_tracker.user_messages == ["screen echo"]
+    assert mgr._session_turn_count == 1
+    mgr._publish_user_utterance_to_plugin_bus.assert_called_once_with(
+        "screen echo",
+        is_voice_source=True,
+    )
+    assert mgr.sync_message_queue.messages == [
+        {
+            "type": "user",
+            "data": {"input_type": "transcript", "data": "screen echo"},
+        }
+    ]
+
+
 @pytest.mark.unit
 @pytest.mark.asyncio
 async def test_voice_transcript_bridge_prime_context_consumes_ordinary_flow(monkeypatch):