Start canonical Voice speech before settlement

Amp-Thread-ID: https://ampcode.com/threads/T-01a06d67-bcd6-77dc-b3c4-4b51537a64c8
Co-authored-by: Amp <amp@ampcode.com>

Author: kostandinang

apps/petrinaut-website/src/main/app/local-storage-demo/brunch-panel-transport.test.ts

@@ -25,12 +25,18 @@ test("delegates one typed message to the supplied Flue conversation", async () =
       turnId: "turn-1",
       position: { batch: 1, index: 0 },
     });
+    await options?.onEvent?.({
+      type: "message-completed",
+      conversationId: "conversation-stable",
+      messageId: "assistant-1",
+      position: { batch: 1, index: 1 },
+    });
     await options?.onEvent?.({
       type: "submission-settled",
       conversationId: "conversation-stable",
       submissionId: admission.submissionId,
       outcome: "completed",
-      position: { batch: 1, index: 1 },
+      position: { batch: 1, index: 2 },
     });
   });
   const client = {
@@ -43,6 +49,10 @@ test("delegates one typed message to the supplied Flue conversation", async () =
     { kind: "user", messageId: "user-1" },
     admissionListener,
   );
+  const responseCompletedListener = vi.fn();
+  tracker.subscribeToResponseMessageCompleted(responseCompletedListener);
+  const responseStartedListener = vi.fn();
+  tracker.subscribeToResponseMessageStarted(responseStartedListener);
   const onAdmission = vi.fn();
   const transport = createBrunchPanelTransport(
     Promise.resolve(client),
@@ -80,6 +90,18 @@ test("delegates one typed message to the supplied Flue conversation", async () =
   expect(tracker.submissionsForResponse("assistant-1")).toEqual([
     "submission-1",
   ]);
+  expect(responseStartedListener).toHaveBeenCalledOnce();
+  expect(responseStartedListener).toHaveBeenCalledWith({
+    messageId: "assistant-1",
+    position: { batch: 1, index: 0 },
+    submissionId: "submission-1",
+  });
+  expect(responseCompletedListener).toHaveBeenCalledOnce();
+  expect(responseCompletedListener).toHaveBeenCalledWith({
+    messageId: "assistant-1",
+    position: { batch: 1, index: 1 },
+    submissionId: "submission-1",
+  });
   expect(onAdmission).toHaveBeenCalledOnce();
   expect(onAdmission).toHaveBeenCalledWith(admission);
 });
@@ -141,15 +163,66 @@ test("matches client-tool admissions once and supports unsubscribe", () => {
 
 test("records every submission that wrote a resumed assistant message", () => {
   const tracker = new BrunchPanelConversationTracker();
-  tracker.recordResponse("assistant-1", "submission-1");
-  tracker.recordResponse("assistant-1", "submission-continuation");
-  tracker.recordResponse("assistant-1", "submission-continuation");
+  const responseStartedListener = vi.fn();
+  tracker.subscribeToResponseMessageStarted(responseStartedListener);
+  tracker.recordResponse({
+    messageId: "assistant-1",
+    position: { batch: 1, index: 0 },
+    submissionId: "submission-1",
+  });
+  tracker.recordResponse({
+    messageId: "assistant-1",
+    position: { batch: 2, index: 0 },
+    submissionId: "submission-continuation",
+  });
+  tracker.recordResponse({
+    messageId: "assistant-1",
+    position: { batch: 2, index: 0 },
+    submissionId: "submission-continuation",
+  });
 
   expect(tracker.submissionsForResponse("assistant-1")).toEqual([
     "submission-1",
     "submission-continuation",
   ]);
   expect(tracker.submissionsForResponse("assistant-2")).toBeUndefined();
+  expect(responseStartedListener.mock.calls).toEqual([
+    [
+      {
+        messageId: "assistant-1",
+        position: { batch: 1, index: 0 },
+        submissionId: "submission-1",
+      },
+    ],
+    [
+      {
+        messageId: "assistant-1",
+        position: { batch: 2, index: 0 },
+        submissionId: "submission-continuation",
+      },
+    ],
+    [
+      {
+        messageId: "assistant-1",
+        position: { batch: 2, index: 0 },
+        submissionId: "submission-continuation",
+      },
+    ],
+  ]);
+});
+
+test("publishes Stop immediately and supports unsubscribe", () => {
+  const tracker = new BrunchPanelConversationTracker();
+  const listener = vi.fn();
+  const unsubscribedListener = vi.fn();
+  tracker.subscribeToStopRequested(listener);
+  const unsubscribe = tracker.subscribeToStopRequested(unsubscribedListener);
+  unsubscribe();
+
+  tracker.recordStopRequested();
+
+  expect(listener).toHaveBeenCalledOnce();
+  expect(unsubscribedListener).not.toHaveBeenCalled();
 });
 
 test("settles in-flight submissions before a durable abort can target them", async () => {

apps/petrinaut-website/src/main/app/local-storage-demo/brunch-panel-transport.ts

@@ -14,7 +14,11 @@ import type {
   SweepCompletionReport,
 } from "../brunch-sweep-output";
 import type { AgentSendResult, FlueClient } from "@flue/sdk";
-import type { FlueChatTransportOptions } from "@hashintel/brunch-agent-transport-aisdk";
+import type {
+  FlueChatResponseMessageCompletedEvent,
+  FlueChatResponseMessageStartedEvent,
+  FlueChatTransportOptions,
+} from "@hashintel/brunch-agent-transport-aisdk";
 import type { PetrinautAiChatTransport } from "@hashintel/petrinaut/ui";
 import type { UIMessageChunk } from "ai";
 
@@ -44,6 +48,13 @@ export class BrunchPanelConversationTracker {
     string,
     AgentSendResult["submissionId"][]
   >();
+  readonly #responseMessageStartedListeners = new Set<
+    (event: FlueChatResponseMessageStartedEvent) => void
+  >();
+  readonly #responseMessageCompletedListeners = new Set<
+    (event: FlueChatResponseMessageCompletedEvent) => void
+  >();
+  readonly #stopRequestedListeners = new Set<() => void>();
 
   public recordAdmission(admission: BrunchPanelAdmission): void {
     if (admission.kind === "user") {
@@ -69,15 +80,29 @@ export class BrunchPanelConversationTracker {
    * all: Voice correlates a reply by membership, whichever side admitted the
    * continuation.
    */
-  public recordResponse(
-    messageId: string,
-    submissionId: AgentSendResult["submissionId"],
-  ): void {
-    const recorded = this.#responseSubmissions.get(messageId);
+  public recordResponse(event: FlueChatResponseMessageStartedEvent): void {
+    const recorded = this.#responseSubmissions.get(event.messageId);
     if (recorded === undefined) {
-      this.#responseSubmissions.set(messageId, [submissionId]);
-    } else if (!recorded.includes(submissionId)) {
-      recorded.push(submissionId);
+      this.#responseSubmissions.set(event.messageId, [event.submissionId]);
+    } else if (!recorded.includes(event.submissionId)) {
+      recorded.push(event.submissionId);
+    }
+    for (const listener of this.#responseMessageStartedListeners) {
+      listener(event);
+    }
+  }
+
+  public recordResponseMessageCompleted(
+    event: FlueChatResponseMessageCompletedEvent,
+  ): void {
+    for (const listener of this.#responseMessageCompletedListeners) {
+      listener(event);
+    }
+  }
+
+  public recordStopRequested(): void {
+    for (const listener of this.#stopRequestedListeners) {
+      listener();
     }
   }
 
@@ -143,6 +168,25 @@ export class BrunchPanelConversationTracker {
     this.#admissionFailureSubscriptions.add(subscription);
     return () => this.#admissionFailureSubscriptions.delete(subscription);
   }
+
+  public subscribeToResponseMessageCompleted(
+    listener: (event: FlueChatResponseMessageCompletedEvent) => void,
+  ): () => void {
+    this.#responseMessageCompletedListeners.add(listener);
+    return () => this.#responseMessageCompletedListeners.delete(listener);
+  }
+
+  public subscribeToResponseMessageStarted(
+    listener: (event: FlueChatResponseMessageStartedEvent) => void,
+  ): () => void {
+    this.#responseMessageStartedListeners.add(listener);
+    return () => this.#responseMessageStartedListeners.delete(listener);
+  }
+
+  public subscribeToStopRequested(listener: () => void): () => void {
+    this.#stopRequestedListeners.add(listener);
+    return () => this.#stopRequestedListeners.delete(listener);
+  }
 }
 
 const formatFailure = (failure: SweepCompletionFailure): string => {
@@ -282,8 +326,9 @@ export const createBrunchPanelTransport = (
             tracker.recordAdmission(event);
             hooks?.onAdmission?.(event.admission);
           },
-          onResponseMessage: ({ messageId, submissionId }) =>
-            tracker.recordResponse(messageId, submissionId),
+          onResponseMessage: (event) => tracker.recordResponse(event),
+          onResponseMessageCompleted: (event) =>
+            tracker.recordResponseMessageCompleted(event),
         });
         try {
           return decorateBrunchStream(

apps/petrinaut-website/src/main/app/local-storage-demo/local-storage-demo-app.test.tsx

@@ -106,26 +106,61 @@ describe("local storage demo Brunch voice integration", () => {
       throw new Error("Expected the configured composer control to render.");
     }
     const failureListener = vi.fn();
+    const responseCompletedListener = vi.fn();
+    const responseStartedListener = vi.fn();
+    const stopListener = vi.fn();
     const target = { kind: "user" as const, messageId: "voice-turn-1" };
     const controlProps = control.props as {
       config: typeof config;
       subscribeToAdmissionFailure: (
         admissionTarget: typeof target,
         listener: (error: FlueChatAdmissionError) => void,
       ) => () => void;
+      subscribeToResponseMessageCompleted: (
+        listener: typeof responseCompletedListener,
+      ) => () => void;
+      subscribeToResponseMessageStarted: (
+        listener: typeof responseStartedListener,
+      ) => () => void;
+      subscribeToStopRequested: (listener: () => void) => () => void;
     };
     expect(control.type).toBe(VoiceInterviewControl);
     expect(controlProps.config).toBe(config);
     const unsubscribe = controlProps.subscribeToAdmissionFailure(
       target,
       failureListener,
     );
+    const unsubscribeFromStop =
+      controlProps.subscribeToStopRequested(stopListener);
+    const unsubscribeFromResponseCompleted =
+      controlProps.subscribeToResponseMessageCompleted(
+        responseCompletedListener,
+      );
+    const unsubscribeFromResponseStarted =
+      controlProps.subscribeToResponseMessageStarted(responseStartedListener);
     const admissionError = new FlueChatAdmissionError({ kind: "ambiguous" });
 
     tracker.recordAdmissionFailure(target, admissionError);
+    tracker.recordResponse({
+      messageId: "assistant-1",
+      position: { batch: 1, index: 0 },
+      submissionId: "submission-1",
+    });
+    tracker.recordResponseMessageCompleted({
+      messageId: "assistant-1",
+      position: { batch: 1, index: 1 },
+      submissionId: "submission-1",
+    });
+    tracker.recordStopRequested();
 
     expect(failureListener).toHaveBeenCalledWith(admissionError);
+    expect(responseStartedListener).toHaveBeenCalledOnce();
+    expect(responseCompletedListener).toHaveBeenCalledOnce();
+    expect(stopListener).toHaveBeenCalledOnce();
     unsubscribe();
+    unsubscribeFromResponseCompleted();
+    unsubscribeFromResponseStarted();
+    unsubscribeFromStop();
   });
 
   test("registers no brunch_ask tool in the production Brunch preview", async () => {
@@ -271,6 +306,8 @@ describe("local storage demo Brunch voice integration", () => {
     const abort = vi.fn<FlueClient["abort"]>(async () => ({ aborted: true }));
     const client = { abort } as Pick<FlueClient, "abort"> as FlueClient;
     const tracker = new BrunchPanelConversationTracker();
+    const stopListener = vi.fn();
+    tracker.subscribeToStopRequested(stopListener);
     let admit: (() => void) | undefined;
     void tracker.trackSubmission(
       new Promise<void>((resolve) => {
@@ -279,6 +316,7 @@ describe("local storage demo Brunch voice integration", () => {
     );
 
     const stop = requestFlueStop(Promise.resolve(client), tracker);
+    expect(stopListener).toHaveBeenCalledOnce();
     await Promise.resolve();
     await Promise.resolve();
     expect(abort).not.toHaveBeenCalled();

apps/petrinaut-website/src/main/app/local-storage-demo/local-storage-demo-app.tsx

@@ -130,6 +130,23 @@ export const getBrunchVoiceMode = (
           resolveResponseSubmission={(messageId) =>
             tracker?.submissionsForResponse(messageId)
           }
+          subscribeToResponseMessageCompleted={
+            tracker === undefined
+              ? undefined
+              : (listener) =>
+                  tracker.subscribeToResponseMessageCompleted(listener)
+          }
+          subscribeToResponseMessageStarted={
+            tracker === undefined
+              ? undefined
+              : (listener) =>
+                  tracker.subscribeToResponseMessageStarted(listener)
+          }
+          subscribeToStopRequested={
+            tracker === undefined
+              ? undefined
+              : (listener) => tracker.subscribeToStopRequested(listener)
+          }
           subscribeToAdmission={
             tracker === undefined
               ? undefined
@@ -188,6 +205,7 @@ export const requestFlueStop = async (
   clientPromise: Promise<ReturnType<typeof createFlueClient>>,
   tracker: BrunchPanelConversationTracker,
 ): Promise<PetrinautAiStopResult> => {
+  tracker.recordStopRequested();
   const client = await clientPromise;
   await tracker.settleInFlightSubmissions();
   const result = await client.abort();