feat(reminder): wire real engine into the app, drop remaining mocks

backend/src/timeflow/intelligence/realtime/agent.py

@@ -327,6 +327,9 @@ def __init__(
         self._question_id_factory = question_id_factory
         # None between replies; assigned on first use so each reply gets a fresh id.
         self._audio_id: str | None = None
+        # Survives the reset below so a late barge-in can still name the audio the
+        # phone is playing -- the model finishes generating well before playback ends.
+        self._last_audio_id: str | None = None
         self._reply_id: str | None = None
         self._spoken = ""
         self._purpose = REPLY_PURPOSE
@@ -373,6 +376,7 @@ async def audio(self, data: bytes) -> None:
         """Queue one chunk, starting the delivery on the first one."""
         if self._speaking is None:
             self._audio_id = self._audio_id_factory()
+            self._last_audio_id = self._audio_id
             self._speaking = asyncio.create_task(self._speak())
         await self._audio.put(data)
 
@@ -477,10 +481,12 @@ async def _finish_reply(self, *, canceled: bool) -> None:
             # own before the barge-in arrived -- but the model generates audio faster
             # than it plays back, so the phone can still be sounding it out. The session
             # only calls interrupted() this late when its own playable-until estimate
-            # says that's still plausible, so the client is told to stop regardless of
-            # what this turn still has queued locally. audio_id is empty because there
-            # is no reply left here to name; the client's handler does not read it.
-            await self._result_sink.deliver_canceled(AudioCanceled(audio_id=""), self._stream)
+            # says that's still plausible. Name the original audio so the client can
+            # tell this stale cancellation apart from a newer reply already starting.
+            if self._last_audio_id is not None:
+                await self._result_sink.deliver_canceled(
+                    AudioCanceled(audio_id=self._last_audio_id), self._stream
+                )
         self._spoken = ""
         self._reply_id = None
         self._audio_id = None

backend/tests/intelligence/realtime/test_realtime_agent.py

@@ -495,8 +495,8 @@ def test_a_barge_in_after_the_reply_already_finished_sending_still_tells_the_cli
     back, so a reply can finish sending -- turn_completed() already settled it -- while
     the phone is still sounding it out. The session only calls interrupted() this late
     when its own playable-until estimate says that is still plausible, so the client
-    must still be told to stop even though this turn's own bookkeeping has nothing left
-    to name; there is no reply id left here to attach to it, hence the empty audio_id.
+    must still be told to stop even though this turn's current bookkeeping has already
+    been reset. The original audio id must be preserved so a newer reply is not stopped.
     """
 
     async def scenario() -> None:
@@ -515,7 +515,29 @@ async def scenario() -> None:
         )
 
         (canceled,) = [payload for kind, payload in sink.calls if kind == "canceled"]
-        assert canceled.audio_id == ""
+        (audio_reply,) = [payload for kind, payload in sink.calls if kind == "audio_start"]
+        assert canceled.audio_id == audio_reply.audio_id
+
+    asyncio.run(scenario())
+
+
+def test_a_barge_in_before_any_reply_started_sends_no_cancellation() -> None:
+    """interrupted() with nothing ever spoken has no audio id to preserve or send.
+
+    _last_audio_id is only set once audio() queues a first chunk; a barge-in that
+    lands before the model has said anything (or produced any audio) leaves it None,
+    and there is nothing playing on the phone for the client to stop.
+    """
+
+    async def scenario() -> None:
+        sink = RecordingSink()
+        session = ScriptedSession([("interrupted", ())])
+
+        await RealtimeAgent(ScriptedFactory(session), sink).handle_audio(
+            _open_mic(), _Stream(voice_mode="continuous")
+        )
+
+        assert [kind for kind, _ in sink.calls if kind == "canceled"] == []
 
     asyncio.run(scenario())
 

frontend/src/app/AppProviders.tsx

@@ -1,7 +1,8 @@
-import { type PropsWithChildren, useEffect } from 'react';
+import { type PropsWithChildren, useCallback, useEffect } from 'react';
 
 import type { AuthController, AuthInvalidationCoordinator } from '../features/auth/application';
 import { AuthProvider, useAuth } from '../features/auth/presentation/AuthProvider';
+import { useReminderPermissionsOnLaunch } from '../features/reminder';
 import { AppServicesProvider } from './composition/AppServicesProvider';
 import type { AppServices } from './composition/createAppServices';
 
@@ -19,26 +20,37 @@ export function AppProviders({
   return (
     <AuthProvider controller={authController} invalidationCoordinator={invalidationCoordinator}>
       <AppServicesProvider services={services}>
-        <AuthenticatedRuntime runtime={services.runtime} />
+        <AuthenticatedRuntime services={services} />
         {children}
       </AppServicesProvider>
     </AuthProvider>
   );
 }
 
-function AuthenticatedRuntime({ runtime }: { readonly runtime: AppServices['runtime'] }) {
+function AuthenticatedRuntime({ services }: { readonly services: AppServices }) {
   const { viewState } = useAuth();
+  const isAuthenticated = viewState.status === 'authenticated';
+
+  const onPermissionsUpdated = useCallback(() => {
+    void services.reminder.rebuild();
+  }, [services]);
+
+  useReminderPermissionsOnLaunch(
+    isAuthenticated ? services.reminderPorts.device : null,
+    isAuthenticated ? services.alertDialog : null,
+    onPermissionsUpdated,
+  );
 
   useEffect(() => {
-    if (viewState.status !== 'authenticated') {
+    if (!isAuthenticated) {
       return;
     }
 
-    void runtime.start();
+    void services.runtime.start();
     return () => {
-      void runtime.stop();
+      void services.runtime.stop();
     };
-  }, [runtime, viewState.status]);
+  }, [isAuthenticated, services]);
 
   return null;
 }

frontend/src/app/composition/createAppServices.ts

@@ -1,30 +1,38 @@
 import { AppRuntime } from '../orchestration/AppRuntime';
 import { createAuthRuntime, type AuthRuntime, type CreateAuthRuntimeOptions } from '../authRuntime';
 import type {
+  AlertDialogPort,
   ReminderApplicationDependencies,
   ReminderApplicationPort,
 } from '../../features/reminder/application/interfaces';
 import { LocalReminderApplication } from '../../features/reminder/application';
 import {
+  LocalReminderDelivery,
   LocalReminderDispositionSync,
+  LocalReminderRecovery,
+  NoopPopup,
   SqliteLocalScheduleReader,
   SqliteReminderStateStore,
 } from '../../features/reminder/data/local';
+import { AlertReminderPresenter } from '../../features/reminder/presentation';
 import { ExpoAudioPlayback } from '../../infrastructure/audio';
 import { ExpoLocationMonitor } from '../../infrastructure/location';
 import {
-  MockPopup,
-  MockReminderRecovery,
-  MockReminderDelivery,
-  MockSystemNotification,
-  MockVibration,
+  ExpoSystemNotification,
   NativeAlarmScheduler,
   NativeDeviceCapability,
+  ReactNativeAlertDialog,
+  ReactNativeVibration,
 } from '../../infrastructure/notifications';
 import { IntervalTimeListener } from '../../infrastructure/time';
-import { MockReminderPresenter } from '../../features/reminder/presentation';
 import { ScheduleViewStore } from '../../features/schedule/presentation';
 
+export interface CreateAppServicesOptions {
+  readonly auth?: CreateAuthRuntimeOptions;
+  readonly schedules?: SqliteLocalScheduleReader;
+  readonly overrides?: Partial<ReminderApplicationDependencies>;
+}
+
 export type AppServices = {
   auth: AuthRuntime;
   protectedClient: AuthRuntime['protectedClient'];
@@ -35,33 +43,43 @@ export type AppServices = {
   scheduleView: ScheduleViewStore;
   schedules: SqliteLocalScheduleReader;
   webSocketClient: AuthRuntime['webSocketClient'];
+  alertDialog: AlertDialogPort;
 };
 
-export interface CreateAppServicesOptions {
-  readonly auth?: CreateAuthRuntimeOptions;
-}
-
 /** 应用唯一组合根：认证传输、功能服务、生命周期和账号内存清理在此接线。 */
 export function createAppServices(options: CreateAppServicesOptions = {}): AppServices {
   const auth = createAuthRuntime(options.auth);
-  const schedules = new SqliteLocalScheduleReader();
+  const alertDialog = new ReactNativeAlertDialog();
+  const schedules = options.schedules ?? new SqliteLocalScheduleReader();
   const reminderState = new SqliteReminderStateStore();
+  const presenter =
+    (options.overrides?.presenter as AlertReminderPresenter | undefined) ??
+    new AlertReminderPresenter(alertDialog);
+
+  const {
+    schedules: _ignoredSchedules,
+    presenter: _ignoredPresenter,
+    ...restOverrides
+  } = options.overrides ?? {};
+
   const reminderPorts: ReminderApplicationDependencies = {
-    schedules,
     time: new IntervalTimeListener(),
     location: new ExpoLocationMonitor(),
     alarms: new NativeAlarmScheduler(),
-    delivery: new MockReminderDelivery(),
+    delivery: new LocalReminderDelivery(),
     audio: new ExpoAudioPlayback(),
     device: new NativeDeviceCapability(),
-    presenter: new MockReminderPresenter(),
-    systemNotification: new MockSystemNotification(),
-    popup: new MockPopup(),
-    vibration: new MockVibration(),
-    recovery: new MockReminderRecovery(),
+    systemNotification: new ExpoSystemNotification(),
+    popup: new NoopPopup(),
+    vibration: new ReactNativeVibration(),
+    recovery: new LocalReminderRecovery(),
     state: reminderState,
     dispositionSync: new LocalReminderDispositionSync(),
+    ...restOverrides,
+    schedules,
+    presenter,
   };
+
   const reminder = new LocalReminderApplication(reminderPorts);
   const scheduleView = new ScheduleViewStore();
   const runtime = new AppRuntime([
@@ -84,5 +102,6 @@ export function createAppServices(options: CreateAppServicesOptions = {}): AppSe
     scheduleView,
     schedules,
     webSocketClient: auth.webSocketClient,
+    alertDialog,
   };
 }

frontend/src/features/assistant/application/AssistantContinuousConversationService.ts

@@ -57,6 +57,8 @@ export class AssistantContinuousConversationService implements AssistantApplicat
   private streamId: string | null = null;
   /** 非 null 表示当前正处于 voice.tts.start 和 voice.tts.end/canceled 之间。 */
   private currentAudioId: string | null = null;
+  /** 最近被打断的音频 id；服务端会在 canceled 后补发同 id 的 tts.end。 */
+  private canceledAudioId: string | null = null;
   private streamStartedWaiter: ((conversationId: string) => void) | null = null;
   /** 跟 streamStartedWaiter 配对；传输层报错/连接掉线时用它让等待方结束，不然会永远卡住。 */
   private streamStartRejecter: ((error: Error) => void) | null = null;
@@ -89,6 +91,8 @@ export class AssistantContinuousConversationService implements AssistantApplicat
    * startStream() 内部有一个没人等的 await（配置原生播放器），如果紧跟着的
    * 第一块音频不排在它后面，可能在原生侧还没配置完时就到达。 */
   private playbackChain: Promise<void> = Promise.resolve();
+  /** 取消时递增，让已经排队但尚未执行的旧流操作失效。 */
+  private playbackGeneration = 0;
   /** Category events can arrive before the command result creates their local row. */
   private readonly pendingCategoryUpdates = new Map<string, ScheduleCategory>();
   private disposed = false;
@@ -253,7 +257,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat
     this.replyText = null;
     this.currentAudioId = null;
     this.notifyListeners();
-    await this.deps.playback.stop().catch(() => {});
+    await this.stopPlaybackImmediately();
   }
 
   /** 用户点圆圈暂停/恢复。暂停期间空闲计时器照常跑——忘记恢复也会兜底挂断。 */
@@ -397,6 +401,8 @@ export class AssistantContinuousConversationService implements AssistantApplicat
         this.notifyListeners();
         return;
       case 'voice.tts.start':
+        this.playbackGeneration += 1;
+        this.canceledAudioId = null;
         this.currentAudioId = message.audio_id;
         this.setState({ conversationId: message.conversation_id, phase: 'speaking' });
         this.chainPlayback(() =>
@@ -407,18 +413,34 @@ export class AssistantContinuousConversationService implements AssistantApplicat
         );
         return;
       case 'voice.tts.end':
+        // canceled 后服务端仍会补发同 id 的 tts.end；它不能收尾新流，也不能把
+        // interrupted 状态提前改回 listening。
+        if (
+          (this.canceledAudioId !== null && message.audio_id === this.canceledAudioId) ||
+          (this.currentAudioId !== null && message.audio_id !== this.currentAudioId)
+        ) {
+          return;
+        }
         this.currentAudioId = null;
+        this.canceledAudioId = null;
         this.chainPlayback(() => this.deps.playback.endStream());
         this.setState({ conversationId: message.conversation_id, phase: 'listening' });
         // 播报完成后给一个全新的窗口，对应"播报完成后进入短暂等待"——不用单独
         // 再搞一个计时器，这次重置就当作那个等待。
         this.armIdleTimer();
         return;
       case 'voice.tts.canceled':
-        // 用户开口打断了正在播的回复：立刻丢掉播放端缓冲里还没放出来的音频，
-        // 而不是等 voice.tts.end（后面仍会补发，但语义已经不是"正常说完"）。
+        // 用户开口打断了正在播的回复：stop 必须绕过 playbackChain 立即执行，否则
+        // 已排队的 PCM 会先继续喂给原生播放器；旧队列随后由代次检查丢弃。
+        if (
+          this.currentAudioId !== null &&
+          (message.audio_id === '' || message.audio_id !== this.currentAudioId)
+        ) {
+          return;
+        }
+        this.canceledAudioId = message.audio_id || this.currentAudioId;
         this.currentAudioId = null;
-        this.chainPlayback(() => this.deps.playback.stop());
+        void this.stopPlaybackImmediately();
         this.setState({ conversationId: message.conversation_id, phase: 'interrupted' });
         return;
       case 'voice.session.end':
@@ -501,9 +523,25 @@ export class AssistantContinuousConversationService implements AssistantApplicat
   }
 
   /** 把一次对原生播放模块的调用接到 playbackChain 末尾，保证上一次真正执行完
-   * (不管成功与否)才轮到这一次。 */
+   * (不管成功与否)才轮到这一次；取消后旧代次的操作会被跳过。 */
   private chainPlayback(run: () => Promise<void>): void {
-    this.playbackChain = this.playbackChain.then(run).catch(() => {});
+    const generation = this.playbackGeneration;
+    this.playbackChain = this.playbackChain
+      .then(async () => {
+        if (generation !== this.playbackGeneration) {
+          return;
+        }
+        await run();
+      })
+      .catch(() => {});
+  }
+
+  /** 立即清空原生播放器，并把后续新操作排在 stop 完成之后。 */
+  private async stopPlaybackImmediately(): Promise<void> {
+    this.playbackGeneration += 1;
+    const stop = this.deps.playback.stop().catch(() => {});
+    this.playbackChain = stop;
+    await stop;
   }
 
   private handleClose(event: { code: number; reason: string }): void {

frontend/src/features/assistant/application/AssistantConversationService.ts

@@ -342,6 +342,9 @@ export class AssistantConversationService implements AssistantApplicationPort {
   }
 
   private handleClose(event: { code: number; reason: string }): void {
+    // 必须真的执行：切到连续对话时共享 WS 会因 voiceMode 不同断开重连，只置空的话
+    // 旧服务仍订阅着新连接的 TTS/PCM，同一句话会被两个服务重复送进播放器。
+    this.unsubscribeConnection?.();
     this.connection = null;
     this.unsubscribeConnection = null;
     const message = event.reason || `连接已断开（${event.code}）`;