fix(chat): isolate pending flows and cancel stale requests

src/stores/chat/__tests__/chat-actor-cancellation.test.ts

@@ -0,0 +1,100 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createActor } from "xstate";
+
+const repository = vi.hoisted(() => ({
+  prefetchMediaForSendResponse: vi.fn(async () => ({
+    success: true as const,
+    data: undefined,
+  })),
+  sendMessage: vi.fn(),
+  unlockHistory: vi.fn(),
+}));
+
+const historySync = vi.hoisted(() => ({
+  readAndSyncHistory: vi.fn(),
+  readLocalHistorySnapshot: vi.fn(async () => ({
+    messages: [],
+    localCount: 0,
+  })),
+  resolveHistoryCacheIdentity: vi.fn(async () => null),
+  syncNetworkHistory: vi.fn(),
+}));
+
+vi.mock("@/data/repositories/chat_repository_loader", () => ({
+  loadChatRepository: vi.fn(async () => repository),
+}));
+
+vi.mock("@/data/repositories/chat_cache_identity", () => ({
+  resolveChatConversationKey: vi.fn(async () => ({
+    success: true,
+    data: "user:test::character:character_elio",
+  })),
+}));
+
+vi.mock("@/stores/chat/chat-history-sync", () => historySync);
+
+import { loadHistoryActor } from "@/stores/chat/machine/actors/history";
+import { sendMessageHttpActor } from "@/stores/chat/machine/actors/send";
+import { unlockHistoryActor } from "@/stores/chat/machine/actors/unlock";
+
+describe("chat actor request cancellation", () => {
+  beforeEach(() => {
+    repository.sendMessage.mockReset();
+    repository.unlockHistory.mockReset();
+    historySync.syncNetworkHistory.mockReset();
+  });
+
+  it("aborts an in-flight send request when its actor stops", async () => {
+    repository.sendMessage.mockImplementation(() => new Promise(() => undefined));
+    const actor = createActor(sendMessageHttpActor, {
+      input: { characterId: "character_elio", content: "hello" },
+    }).start();
+    await vi.waitFor(() => expect(repository.sendMessage).toHaveBeenCalled());
+
+    const signal = getSignal(repository.sendMessage.mock.calls[0]?.[2]);
+    expect(signal.aborted).toBe(false);
+    actor.stop();
+    expect(signal.aborted).toBe(true);
+  });
+
+  it("aborts an in-flight history request when its callback actor stops", async () => {
+    historySync.syncNetworkHistory.mockImplementation(
+      () => new Promise(() => undefined),
+    );
+    const actor = createActor(loadHistoryActor, {
+      input: { characterId: "character_elio" },
+    }).start();
+    await vi.waitFor(() =>
+      expect(historySync.syncNetworkHistory).toHaveBeenCalled(),
+    );
+
+    const signal = getSignal(
+      historySync.syncNetworkHistory.mock.calls[0]?.[3],
+    );
+    expect(signal.aborted).toBe(false);
+    actor.stop();
+    expect(signal.aborted).toBe(true);
+  });
+
+  it("aborts an in-flight unlock request when its actor stops", async () => {
+    repository.unlockHistory.mockImplementation(
+      () => new Promise(() => undefined),
+    );
+    const actor = createActor(unlockHistoryActor, {
+      input: { characterId: "character_elio" },
+    }).start();
+    await vi.waitFor(() => expect(repository.unlockHistory).toHaveBeenCalled());
+
+    const signal = getSignal(repository.unlockHistory.mock.calls[0]?.[1]);
+    expect(signal.aborted).toBe(false);
+    actor.stop();
+    expect(signal.aborted).toBe(true);
+  });
+});
+
+function getSignal(options: unknown): AbortSignal {
+  if (options instanceof AbortSignal) return options;
+  const signal = (options as { signal?: AbortSignal } | undefined)?.signal;
+  if (!signal) throw new Error("Missing AbortSignal");
+  return signal;
+}

src/stores/chat/__tests__/chat-promotion.test.ts

@@ -9,6 +9,7 @@ import {
 } from "@/stores/chat/helper/promotion";
 
 const promotion: PendingChatPromotion = {
+  characterId: "character_elio",
   promotionType: "image",
   lockType: "image_paywall",
   clientLockId: "promotion-1",

src/stores/chat/__tests__/chat-session-flow.test.ts

@@ -80,6 +80,7 @@ describe("chat session flow", () => {
     actor.send({
       type: "ChatPromotionInjected",
       promotion: {
+        characterId: "character_elio",
         promotionType: "voice",
         lockType: "voice_message",
         clientLockId: "promotion-1",

src/stores/chat/chat-history-sync.ts

@@ -4,6 +4,7 @@ import { loadChatRepository } from "@/data/repositories/chat_repository_loader";
 import { Logger } from "@/utils/logger";
 import { Result } from "@/utils/result";
 import { todayString } from "@/utils/date";
+import { isAbortError } from "@/utils/abort";
 
 import { CHAT_HISTORY_LIMIT } from "./helper/history";
 import { localMessagesToUi } from "./helper/message-mappers";
@@ -81,6 +82,7 @@ export async function syncNetworkHistory(
   characterId: string,
   localCount: number,
   cacheIdentity: string | null,
+  signal?: AbortSignal,
 ): Promise<NetworkHistorySyncOutput | null> {
   const chatRepo = await loadChatRepository();
   const greetingMessage = createGreetingMessage();
@@ -89,13 +91,16 @@ export async function syncNetworkHistory(
     characterId,
     CHAT_HISTORY_LIMIT,
     0,
+    { signal },
   );
   if (Result.isErr(networkResult)) {
+    if (isAbortError(networkResult.error)) throw networkResult.error;
     log.error("[chat-machine] loadHistory NETWORK FAILED", {
       error: networkResult.error,
     });
     return null;
   }
+  signal?.throwIfAborted();
 
   const networkUi = localMessagesToUi(networkResult.data.messages);
   log.debug("[chat-machine] loadHistory NETWORK DONE", {
@@ -144,6 +149,7 @@ export async function syncNetworkHistory(
  */
 export async function readAndSyncHistory(
   characterId: string,
+  signal?: AbortSignal,
 ): Promise<ReadAndSyncHistoryOutput> {
   const cacheIdentity = await resolveHistoryCacheIdentity(characterId);
   const localSnapshot = await readLocalHistorySnapshot(cacheIdentity);
@@ -151,6 +157,7 @@ export async function readAndSyncHistory(
     characterId,
     localSnapshot.localCount,
     cacheIdentity,
+    signal,
   );
   if (networkSnapshot) return networkSnapshot;
 

src/stores/chat/machine/actors/history.ts

@@ -4,6 +4,7 @@ import type { UiMessage } from "@/stores/chat/ui-message";
 import { loadChatRepository } from "@/data/repositories/chat_repository_loader";
 import { Logger } from "@/utils/logger";
 import { Result } from "@/utils/result";
+import { isAbortError } from "@/utils/abort";
 
 import {
   readLocalHistorySnapshot,
@@ -38,6 +39,7 @@ export const loadHistoryActor = fromCallback<
   ChatHistoryActorInput
 >(({ input, sendBack }) => {
   let cancelled = false;
+  const controller = new AbortController();
 
   void (async () => {
     const cacheIdentity = await resolveHistoryCacheIdentity(input.characterId);
@@ -52,6 +54,7 @@ export const loadHistoryActor = fromCallback<
       input.characterId,
       localSnapshot.localCount,
       cacheIdentity,
+      controller.signal,
     );
     if (cancelled || !networkSnapshot) return;
     sendBack({
@@ -59,13 +62,14 @@ export const loadHistoryActor = fromCallback<
       output: networkSnapshot,
     });
   })().catch((error: unknown) => {
-    if (cancelled) return;
+    if (cancelled || isAbortError(error)) return;
     log.error("[chat-machine] loadHistoryActor failed", { error });
     sendBack({ type: "ChatHistoryLoadFailed", error });
   });
 
   return () => {
     cancelled = true;
+    controller.abort();
   };
 });
 
@@ -74,10 +78,13 @@ export const loadMoreHistoryActor =
     ({ input, receive, sendBack }) => {
       let cancelled = false;
       let running = false;
+      let activeController: AbortController | null = null;
 
       receive((event) => {
         if (running || cancelled) return;
         running = true;
+        const controller = new AbortController();
+        activeController = controller;
 
         void (async () => {
           const chatRepo = await loadChatRepository();
@@ -85,6 +92,7 @@ export const loadMoreHistoryActor =
             input.characterId,
             event.limit,
             event.offset,
+            { signal: controller.signal },
           );
           if (Result.isErr(historyResult)) throw historyResult.error;
           if (cancelled) return;
@@ -111,17 +119,19 @@ export const loadMoreHistoryActor =
           );
         })()
           .catch((error: unknown) => {
-            if (cancelled) return;
+            if (cancelled || isAbortError(error)) return;
             log.warn("[chat-machine] loadMoreHistoryActor failed", { error });
             sendBack({ type: "ChatOlderHistoryLoadFailed", error });
           })
           .finally(() => {
+            if (activeController === controller) activeController = null;
             running = false;
           });
       });
 
       return () => {
         cancelled = true;
+        activeController?.abort();
       };
     },
   );

src/stores/chat/machine/actors/send.ts

@@ -7,6 +7,7 @@ import { loadChatRepository } from "@/data/repositories/chat_repository_loader";
 import { resolveChatConversationKey } from "@/data/repositories/chat_cache_identity";
 import { Logger } from "@/utils/logger";
 import { Result } from "@/utils/result";
+import { isAbortError } from "@/utils/abort";
 
 import type { ChatEvent } from "../../chat-events";
 import { sendResponseToUiMessage } from "../../helper/message-mappers";
@@ -18,8 +19,8 @@ type UiMessage = import("@/stores/chat/ui-message").UiMessage;
 export const sendMessageHttpActor = fromPromise<
   { response: ChatSendResponse; reply: UiMessage | null },
   { characterId: string; content: string }
->(async ({ input }) => {
-  return sendMessageViaHttp(input.characterId, input.content);
+>(async ({ input, signal }) => {
+  return sendMessageViaHttp(input.characterId, input.content, signal);
 });
 
 export const httpMessageQueueActor = fromCallback<
@@ -37,13 +38,21 @@ function createMessageQueueActor(
   receive: (listener: (event: ChatEvent) => void) => void,
 ): () => void {
   const queue = new MessageQueue();
+  let activeController: AbortController | null = null;
 
   queue.setConsumer(async (content) => {
+    const controller = new AbortController();
+    activeController = controller;
     sendBack({ type: "ChatQueuedSendStarted" });
     try {
-      const output = await sendMessageViaHttp(characterId, content);
+      const output = await sendMessageViaHttp(
+        characterId,
+        content,
+        controller.signal,
+      );
       sendBack({ type: "ChatQueuedHttpDone", output });
     } catch (error) {
+      if (isAbortError(error)) return;
       const errorMessage = ExceptionHandler.message(error);
       log.error("[chat-machine] message queue send failed", {
         error,
@@ -53,6 +62,8 @@ function createMessageQueueActor(
         content,
         errorMessage,
       });
+    } finally {
+      if (activeController === controller) activeController = null;
     }
   });
 
@@ -63,22 +74,31 @@ function createMessageQueueActor(
     queue.enqueue(content);
   });
 
-  return () => queue.dispose();
+  return () => {
+    activeController?.abort();
+    queue.dispose();
+  };
 }
 
-async function sendMessageViaHttp(characterId: string, content: string): Promise<{
+async function sendMessageViaHttp(
+  characterId: string,
+  content: string,
+  signal?: AbortSignal,
+): Promise<{
   response: ChatSendResponse;
   reply: UiMessage | null;
 }> {
   const chatRepo = await loadChatRepository();
   const cacheIdentityResult = await resolveChatConversationKey(characterId);
-  const result = await chatRepo.sendMessage(characterId, content);
+  const result = await chatRepo.sendMessage(characterId, content, { signal });
   if (Result.isErr(result)) {
+    if (isAbortError(result.error)) throw result.error;
     log.error("[chat-machine] sendMessageHttpActor failed", {
       error: result.error,
     });
     throw result.error;
   }
+  signal?.throwIfAborted();
   if (Result.isOk(cacheIdentityResult)) {
     void chatRepo.prefetchMediaForSendResponse(
       result.data,

src/stores/chat/machine/actors/unlock.ts

@@ -4,6 +4,7 @@ import { loadChatRepository } from "@/data/repositories/chat_repository_loader";
 import { resolveChatConversationKey } from "@/data/repositories/chat_cache_identity";
 import { Logger } from "@/utils/logger";
 import { Result } from "@/utils/result";
+import { isAbortError } from "@/utils/abort";
 
 import { readAndSyncHistory } from "../../chat-history-sync";
 import {
@@ -25,17 +26,21 @@ export interface UnlockHistoryOutput {
 export const unlockHistoryActor = fromPromise<
   UnlockHistoryOutput,
   { characterId: string }
->(async ({ input }) => {
+>(async ({ input, signal }) => {
   const chatRepo = await loadChatRepository();
-  const unlockResult = await chatRepo.unlockHistory(input.characterId);
+  const unlockResult = await chatRepo.unlockHistory(input.characterId, {
+    signal,
+  });
   if (Result.isErr(unlockResult)) {
+    if (isAbortError(unlockResult.error)) throw unlockResult.error;
     log.error("[chat-machine] unlockHistoryActor failed", {
       error: unlockResult.error,
     });
     throw unlockResult.error;
   }
 
-  const history = await readAndSyncHistory(input.characterId);
+  signal.throwIfAborted();
+  const history = await readAndSyncHistory(input.characterId, signal);
   return {
     unlocked: unlockResult.data.unlocked,
     reason: unlockResult.data.reason,
@@ -47,18 +52,22 @@ export const unlockHistoryActor = fromPromise<
 export const unlockMessageActor = fromPromise<
   UnlockMessageOutput,
   UnlockMessageRequest & { characterId: string }
->(async ({ input }) => {
+>(async ({ input, signal }) => {
   const chatRepo = await loadChatRepository();
   const cacheIdentityResult = await resolveChatConversationKey(
     input.characterId,
   );
-  const unlockResult = await chatRepo.unlockPrivateMessage({
-    characterId: input.characterId,
-    ...(input.messageId ? { messageId: input.messageId } : {}),
-    ...(input.lockType ? { lockType: input.lockType } : {}),
-    ...(input.clientLockId ? { clientLockId: input.clientLockId } : {}),
-  });
+  const unlockResult = await chatRepo.unlockPrivateMessage(
+    {
+      characterId: input.characterId,
+      ...(input.messageId ? { messageId: input.messageId } : {}),
+      ...(input.lockType ? { lockType: input.lockType } : {}),
+      ...(input.clientLockId ? { clientLockId: input.clientLockId } : {}),
+    },
+    { signal },
+  );
   if (Result.isErr(unlockResult)) {
+    if (isAbortError(unlockResult.error)) throw unlockResult.error;
     log.error("[chat-machine] unlockMessageActor failed", {
       messageId: input.messageId,
       clientLockId: input.clientLockId,
@@ -66,6 +75,7 @@ export const unlockMessageActor = fromPromise<
     });
     throw unlockResult.error;
   }
+  signal.throwIfAborted();
 
   if (
     unlockResult.data.unlocked &&