Phase 4A：建立受控 ASR consumer routing 与语音设置

docs/design/smart-turn-v3-provider-neutral.md

@@ -28,29 +28,31 @@ also suitable for barge-in and connection lifecycle gating, but it never emits
 
 ## Consumer-neutral voice input
 
-`LLMSessionManager.bind_voice_input_consumer()` exposes the high-level final
-text boundary needed by later game/plugin integration. A binding is inert:
-MicLease remains the sole microphone-ownership authority, and the caller must
-bind before changing the lease owner to `game`.
-
-- `owner=core` keeps the existing Core transcript path.
-- `owner=game` accepts PCM only while the exact captured consumer binding is
-  still registered; otherwise the route remains fail-closed and suspended.
-- The binding receives only a route-authorized `VoiceTranscriptEvent`: a
-  provider-native logical final for streaming ASR, or a Smart Turn-sealed and
-  aggregated final for segmented ASR. It never receives PCM or partials.
-- Consumer replacement or removal is forbidden while that owner holds
-  MicLease. Lease changes invalidate queued PCM, the active turn, transcript
-  reservations, and delayed callbacks before another target can become active.
-- Consumer callback failure discards that delivery. It never falls back to
-  Core, Omni, another ASR provider, or a browser speech recognizer.
-- Provider selection remains centralized and follows the active Core/ASR route
-  policy. A game or plugin cannot select Qwen, Soniox, or another provider.
-
-Phase 3 supplies only this common binding contract and a fake-consumer
-integration test. Registering concrete games, changing game UI, and removing
-legacy browser `SpeechRecognition` are responsibilities of their own follow-up
-integration changes.
+`VoiceInputRegistry` owns the high-level transcript boundary used by ordinary
+Core chat, the active game route, and a future trusted plugin bridge. MicLease
+remains the sole microphone-ownership authority; the Registry never receives
+PCM and cannot select a provider or endpoint.
+
+- `owner=core` activates the built-in `core_chat` consumer, which accepts
+  identified partials and non-empty finals.
+- `owner=game` activates the built-in `game` consumer, which accepts only
+  non-empty finals while `is_game_route_active(lanlan_name)` is true. An
+  unavailable game route stays fail-closed and never falls back to chat.
+- Every route is pinned by its full `VoiceTurnToken`. Consumer switches,
+  unregistration, lease changes, PCM holes, aborts, and session teardown
+  terminate the old route instead of transferring it to the new consumer.
+- Final delivery consumes the route before calling business code. A duplicate,
+  late event, callback failure, or empty final cannot restore or redirect it.
+- Empty finals are terminal cleanup events, not transcripts. They reach neither
+  Core injection nor the game route.
+- `core_chat` captures the prepared session and external turn id, so
+  cancellation after a hot swap abandons the precise original session turn.
+- Plugin registration is namespace-bound and exposes no Registry, MicLease,
+  provider, PCM, process-management, or routing authority.
+
+Provider-native and Smart Turn-sealed finals therefore share one controlled
+Core-side routing contract while provider selection remains centralized below
+the independent-ASR runtime boundary.
 
 ## Audio contract
 

main_logic/asr_client/detector_runtime.py

@@ -911,6 +911,7 @@ def __init__(
         vad: SileroVad | None = None,
         gate: SileroActivityGate | None = None,
         rnnoise_onset_probability: float = 0.35,
+        resource_optimization_enabled: bool = True,
         provider_policy: AsrProviderPolicy | None = None,
         coordinator: TurnCoordinator | None = None,
         on_turn_complete: Callable[[], Awaitable[None]] | None = None,
@@ -935,6 +936,7 @@ def __init__(
         self._available = True
         self._closed = False
         self._rnnoise_onset_probability = rnnoise_onset_probability
+        self._resource_optimization_enabled = bool(resource_optimization_enabled)
         self._speech_active = False
         self._events: list[SpeechActivityEvent] = []
         self._semantic_adapter: _VoiceTurnAdapter | None = None
@@ -1383,7 +1385,8 @@ async def feed(
             if self._closed or not self._available:
                 return DetectorFeedResult((), False)
             if (
-                rnnoise_available
+                self._resource_optimization_enabled
+                and rnnoise_available
                 and speech_probability is not None
                 and not self._speech_active
                 and speech_probability < self._rnnoise_onset_probability
@@ -1479,7 +1482,8 @@ async def submit_audio(
                 None,
             )
         if (
-            rnnoise_available
+            self._resource_optimization_enabled
+            and rnnoise_available
             and speech_probability is not None
             and not self._candidate_open
             and speech_probability < self._rnnoise_onset_probability

main_logic/asr_client/runtime.py

@@ -334,6 +334,7 @@ def _init_asr_runtime_state(self) -> None:
         self._asr_audio_sequence = 0
         self._asr_audio_generation = 0
         self._asr_current_ingress_token: VoiceIngressToken | None = None
+        self._asr_partial_turn_token: VoiceTurnToken | None = None
         self._asr_accepted_final_keys: OrderedDict[FinalKey, None] = OrderedDict()
         self._asr_reserved_final_key: FinalKey | None = None
         self._asr_transcript_dispatcher = TranscriptDispatcher(
@@ -378,6 +379,8 @@ def _ensure_asr_runtime_state(self) -> None:
             self._asr_pending_detector_candidate = None
         if not hasattr(self, "_asr_overlap_onset_token"):
             self._asr_overlap_onset_token = None
+        if not hasattr(self, "_asr_partial_turn_token"):
+            self._asr_partial_turn_token = None
         if not hasattr(self, "_asr_overlap_completed_token"):
             self._asr_overlap_completed_token = None
             self._asr_overlap_completed_turns = 0
@@ -1182,6 +1185,9 @@ async def on_detector_event(event) -> None:
                     raise RuntimeError("ASR_DETECTOR_CONTROL_BACKPRESSURE")
 
             detector_ref = DetectorRuntime(
+                resource_optimization_enabled=(
+                    self._voice_input_resource_optimization_enabled
+                ),
                 provider_policy=policy,
                 on_endpointing_failure=(
                     on_detector_endpointing_failure
@@ -1274,6 +1280,7 @@ def _reset_asr_turn_state(self) -> None:
         self._asr_overlap_completed_turns = 0
         self._asr_audio_sequence = 0
         self._asr_current_ingress_token = None
+        self._asr_partial_turn_token = None
         self._asr_accepted_final_keys.clear()
         self._asr_reserved_final_key = None
         self._asr_sealed_turn_token = None
@@ -2180,6 +2187,11 @@ async def _prepare_independent_asr_turn(self, epoch: int) -> None:
                     self.display_name,
                 )
         if accepted and self._runtime_identity_matches(identity):
+            # The provider callback carries text only. Pin the source identity
+            # at the ordered prepare boundary; partial delivery later validates
+            # this exact token instead of relabeling text with whatever turn
+            # happens to be current at callback time.
+            self._asr_partial_turn_token = turn_token
             return
         transcript_dispatcher.release(final_key)
         if not self._runtime_identity_matches(identity):
@@ -2190,6 +2202,8 @@ async def _prepare_independent_asr_turn(self, epoch: int) -> None:
         ):
             self._asr_reserved_final_key = None
             self._asr_turn_prepared = False
+            if self._asr_partial_turn_token == turn_token:
+                self._asr_partial_turn_token = None
 
     async def _handle_independent_asr_endpoint(self, epoch: int) -> None:
         """Seal the current turn immediately at its semantic endpoint."""
@@ -2392,9 +2406,19 @@ async def _send_independent_asr_preview(self, text: str, epoch: int) -> None:
         if not clean or epoch != self._asr_session_epoch:
             return
         lifecycle = self._asr_lifecycle
+        turn_token = self._asr_partial_turn_token
         if (
-            lifecycle is not None
-            and not self._asr_first_partial_recorded
+            lifecycle is None
+            or turn_token is None
+            or not self._asr_turn_prepared
+            or lifecycle.snapshot.state is not VoiceLifecycleState.ACTIVE
+            or not self._ingress_token_matches(turn_token.ingress)
+            or lifecycle.snapshot.turn_id != turn_token.turn_id
+            or self._asr_audio_dispatcher.active_turn != turn_token
+        ):
+            return
+        if (
+            not self._asr_first_partial_recorded
             and self._asr_turn_audio_started_at is not None
         ):
             lifecycle.metrics.first_partial_latency_ms = int(
@@ -2403,7 +2427,7 @@ async def _send_independent_asr_preview(self, text: str, epoch: int) -> None:
             self._asr_first_partial_recorded = True
         try:
             await self._callbacks.on_partial(
-                VoicePartialEvent(text=clean, session_epoch=epoch)
+                VoicePartialEvent(turn_token=turn_token, text=clean)
             )
         except Exception:
             logger.debug(
@@ -2471,6 +2495,8 @@ async def _handle_independent_asr_final(
                 self._asr_turn_prepared = False
                 self._asr_received_audio = False
                 self._asr_sealed_turn_token = None
+                if self._asr_partial_turn_token == sealed_token.turn:
+                    self._asr_partial_turn_token = None
                 self._asr_turn_endpointed_at = None
                 self._asr_reserved_final_key = None
                 watchdog = self._asr_final_watchdog_task

main_logic/core/_shared.py

@@ -37,6 +37,7 @@
 # None collapses both into the same code path and would let recovery /
 # proactive paths accidentally bind their messages to a newer request_id.
 _REQUEST_ID_UNSET: Any = object()
+_HANDSHAKE_OVERRIDE_UNSET: Any = object()
 _MAGIC_COMMAND_IMAGE_DROP_REQUEST_MAX = 64
 _VOICE_PROACTIVE_ACK_GRACE_S = 0.05
 _TEXT_SESSION_INPUT_TYPES = frozenset({"text", "avatar_drop_image", "user_image"})