fix(chat): bound live subscriptions and preserve unconfirmed input

7 files changed, 341 insertions, 8 deletions

diff --git a/src/features/chat/adapter/pending-inputs.ts b/src/features/chat/adapter/pending-inputs.ts
new file mode 100644
index 0000000..5798fdb
--- /dev/null
+++ b/src/features/chat/adapter/pending-inputs.ts
@@ -0,0 +1,49 @@
+import type { PendingInput, PendingInputStorage } from "../pending-inputs";
+
+export function createPendingInputStorage(
+  storage: Storage,
+  conversationId: string,
+): PendingInputStorage {
+  const prefix = `dispatch.pending-inputs:${conversationId}:`;
+  let known = new Map<string, string>();
+  return {
+    load() {
+      const inputs: PendingInput[] = [];
+      const loaded = new Map<string, string>();
+      for (let i = 0; i < storage.length; i++) {
+        const key = storage.key(i);
+        if (key === null || !key.startsWith(prefix)) continue;
+        const raw = storage.getItem(key);
+        if (raw === null) continue;
+        const input = JSON.parse(raw);
+        if (
+          (input?.request?.type !== "chat.send" && input?.request?.type !== "chat.queue") ||
+          input.request.conversationId !== conversationId ||
+          typeof input.request.requestId !== "string" ||
+          key !== prefix + input.request.requestId ||
+          (input.request.type === "chat.send"
+            ? typeof input.request.message !== "string"
+            : typeof input.request.text !== "string") ||
+          (input.rejection !== undefined && typeof input.rejection !== "string")
+        )
+          throw new Error("Invalid saved input");
+        inputs.push(input);
+        loaded.set(input.request.requestId, raw);
+      }
+      known = loaded;
+      return inputs;
+    },
+    save(inputs) {
+      // Per-request keys prevent another browser tab's send/receipt from overwriting ours.
+      // Quota/security failures reach the composer BEFORE it clears its staged input.
+      const next = new Map(inputs.map((input) => [input.request.requestId, JSON.stringify(input)]));
+      for (const [id, raw] of next) {
+        if (known.get(id) !== raw) storage.setItem(prefix + id, raw);
+      }
+      for (const id of known.keys()) {
+        if (!next.has(id)) storage.removeItem(prefix + id);
+      }
+      known = next;
+    },
+  };
+}
diff --git a/src/features/chat/delivery.test.ts b/src/features/chat/delivery.test.ts
new file mode 100644
index 0000000..99ad2d4
--- /dev/null
+++ b/src/features/chat/delivery.test.ts
@@ -0,0 +1,146 @@
+import { describe, expect, it, vi } from "vitest";
+import { createPendingInputStorage } from "./adapter/pending-inputs";
+import { createChatStore } from "./store.svelte";
+import {
+  createFakeCache,
+  createFakeHistorySync,
+  createFakeMetricsSync,
+  createFakeTransport,
+} from "./test-helpers";
+
+function storage(): Storage {
+  const values = new Map<string, string>();
+  return {
+    get length() {
+      return values.size;
+    },
+    getItem: (key) => values.get(key) ?? null,
+    setItem: (key, value) => {
+      values.set(key, value);
+    },
+    removeItem: (key) => {
+      values.delete(key);
+    },
+    clear: () => values.clear(),
+    key: (index) => [...values.keys()][index] ?? null,
+  };
+}
+
+function fixture(saved = storage(), metricsSync = createFakeMetricsSync().impl) {
+  const transport = createFakeTransport();
+  let id = 0;
+  const store = createChatStore({
+    conversationId: "c",
+    createRequestId: () => `request-${++id}`,
+    pendingInputStorage: createPendingInputStorage(saved, "c"),
+    transport: transport.impl,
+    historySync: createFakeHistorySync().impl,
+    metricsSync,
+    cache: createFakeCache().impl,
+  });
+  return { store, transport, saved };
+}
+
+const images = [{ url: "data:image/png;base64,AAAA", mimeType: "image/png" }];
+
+describe("input delivery", () => {
+  it.each([
+    "send",
+    "queueMessage",
+  ] as const)("saves %s input through refresh without automatically retransmitting", (method) => {
+    const f = fixture();
+    f.store[method]("keep me", images);
+    expect(f.store.pendingInputs).toHaveLength(1);
+    f.store.dispose();
+    const reloaded = fixture(f.saved);
+    expect(reloaded.store.pendingInputs).toMatchObject([{ request: { images } }]);
+    expect(JSON.stringify(reloaded.store.pendingInputs)).toContain("keep me");
+    expect(reloaded.transport.sent).toEqual([]);
+    expect(reloaded.transport.sentQueue).toEqual([]);
+    reloaded.store.dispose();
+  });
+
+  it("streamed user echoes and wrong receipts cannot discard an unconfirmed send", () => {
+    const f = fixture();
+    f.store.send("mine", images);
+    f.store.handleDelta({
+      type: "chat.delta",
+      event: { type: "user-message", conversationId: "c", turnId: "t", text: "mine" },
+    });
+    f.store.handleDelta({ type: "chat.accepted", conversationId: "other", requestId: "request-1" });
+    f.store.handleDelta({ type: "chat.accepted", conversationId: "c", requestId: "other" });
+    expect(f.store.pendingInputs).toHaveLength(1);
+    f.store.handleDelta({ type: "chat.accepted", conversationId: "c", requestId: "request-1" });
+    expect(f.store.pendingInputs).toEqual([]);
+    expect(createPendingInputStorage(f.saved, "c").load()).toEqual([]);
+    f.store.dispose();
+  });
+
+  it("retains explicitly rejected input and images for recovery until discarded", () => {
+    const f = fixture();
+    f.store.send("mine", images);
+    f.store.handleDelta({
+      type: "chat.error",
+      conversationId: "c",
+      requestId: "request-1",
+      message: "disk full",
+    });
+    const reloaded = fixture(f.saved);
+    expect(reloaded.store.pendingInputs).toMatchObject([
+      { rejection: "disk full", request: { message: "mine", images } },
+    ]);
+    reloaded.store.discardPendingInput("request-1");
+    expect(createPendingInputStorage(f.saved, "c").load()).toEqual([]);
+    f.store.dispose();
+    reloaded.store.dispose();
+  });
+
+  it.each([
+    "send",
+    "queueMessage",
+  ] as const)("does not clear or transmit %s input if saving fails", (method) => {
+    const saved = storage();
+    saved.setItem = () => {
+      throw new Error("quota exceeded");
+    };
+    const f = fixture(saved);
+    expect(() => f.store[method]("mine", images)).toThrow("quota exceeded");
+    expect(f.store.pendingInputs).toEqual([]);
+    expect(f.store.chunks).toEqual([]);
+    expect(f.transport.sent).toEqual([]);
+    expect(f.transport.sentQueue).toEqual([]);
+    f.store.dispose();
+  });
+
+  it("coalesces a reconnect burst and permits one metrics catch-up after completion", async () => {
+    let resolve!: (value: { turns: [] }) => void;
+    const first = {
+      promise: new Promise<{ turns: [] }>((done) => {
+        resolve = done;
+      }),
+      resolve: (value: { turns: [] }) => resolve(value),
+    };
+    const metricsSync = vi.fn().mockReturnValueOnce(first.promise).mockResolvedValue({ turns: [] });
+    const f = fixture(storage(), metricsSync);
+    for (let i = 0; i < 100; i++) f.store.resync();
+    expect(metricsSync).toHaveBeenCalledTimes(1);
+    first.resolve({ turns: [] });
+    await vi.waitFor(() => expect(metricsSync).toHaveBeenCalledTimes(2));
+    f.store.dispose();
+  });
+});
+
+it("concurrent browser tabs cannot overwrite each other's saved messages or receipts", () => {
+  const saved = storage();
+  const first = createPendingInputStorage(saved, "c");
+  const second = createPendingInputStorage(saved, "c");
+  first.load();
+  second.load();
+  const input = (requestId: string) => ({
+    request: { type: "chat.send" as const, conversationId: "c", requestId, message: requestId },
+  });
+  first.save([input("first")]);
+  second.save([input("second")]);
+  first.save([]);
+  expect(createPendingInputStorage(saved, "c").load()).toEqual([input("second")]);
+});
diff --git a/src/features/chat/index.ts b/src/features/chat/index.ts
index fec2cdb..41cb06f 100644
--- a/src/features/chat/index.ts
+++ b/src/features/chat/index.ts
@@ -6,6 +6,7 @@ export type {
 } from "../../core/chunks";
 export { groupRenderedChunks, resolveImageUrl, viewProviderRetry } from "../../core/chunks";
 export type { TurnMetricsEntry } from "../../core/metrics";
+export { createPendingInputStorage } from "./adapter/pending-inputs";
 export type { NamedAgentLike, NamedAgentSnapshotLike } from "./model-select";
 export {
   CLAUDE_SUBSCRIPTION_KEY,
@@ -22,6 +23,7 @@ export {
   openCodeNumber,
   providerKeysWithNamedAgents,
 } from "./model-select";
+export type { PendingInput, PendingInputStorage } from "./pending-inputs";
 export type { ChatTransport, HistorySync, HistoryWindow, MetricsSync } from "./ports";
 export type {
   EffortOption,
@@ -53,6 +55,7 @@ export { default as CompactionView } from "./ui/CompactionView.svelte";
 export type { ComposerStatus } from "./ui/Composer.svelte";
 export { default as Composer } from "./ui/Composer.svelte";
 export { default as ModelSelector } from "./ui/ModelSelector.svelte";
+export { default as PendingInputs } from "./ui/PendingInputs.svelte";
 export { default as ReasoningEffortSelector } from "./ui/ReasoningEffortSelector.svelte";
 
 /** Public module manifest — aggregated by the shell's "Loaded Modules" view. */
diff --git a/src/features/chat/pending-inputs.ts b/src/features/chat/pending-inputs.ts
new file mode 100644
index 0000000..d7fb0aa
--- /dev/null
+++ b/src/features/chat/pending-inputs.ts
@@ -0,0 +1,12 @@
+import type { ChatQueueMessage, ChatSendMessage } from "@dispatch/transport-contract";
+
+/** A saved copy, never automatically retransmitted: requestId is not idempotent. */
+export interface PendingInput {
+  readonly request: (ChatSendMessage | ChatQueueMessage) & { readonly requestId: string };
+  readonly rejection?: string;
+}
+
+export interface PendingInputStorage {
+  load(): readonly PendingInput[];
+  save(inputs: readonly PendingInput[]): void;
+}
diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts
index bb0b218..dacbe06 100644
--- a/src/features/chat/store.svelte.ts
+++ b/src/features/chat/store.svelte.ts
@@ -1,4 +1,5 @@
 import type {
+  ChatAcceptedMessage,
   ChatDeltaMessage,
   ChatErrorMessage,
   ChatQueueCancelMessage,
@@ -6,6 +7,7 @@ import type {
   ChatQueueMessage,
   ChatResyncRequiredMessage,
   ChatSendMessage,
+  ChatSubscriptionRejectedMessage,
 } from "@dispatch/transport-contract";
 import type { ChatMessage, ImageInput, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
 import type { RenderedChunk, TranscriptState } from "../../core/chunks";
@@ -43,9 +45,11 @@ import {
 } from "../../core/metrics";
 import type { ConversationCache } from "../conversation-cache";
 import type { NamedAgentSelectionSnapshot } from "../named-agents";
+import type { PendingInput, PendingInputStorage } from "./pending-inputs";
 import type { ChatTransport, HistorySync, MetricsSync } from "./ports";
 
 export interface ChatStoreDependencies {
+  readonly pendingInputStorage?: PendingInputStorage;
   readonly conversationId: string;
   /** Injected unique client-local correlation token for each new send. */
   readonly createRequestId: () => string;
@@ -98,6 +102,8 @@ export interface ChatStoreDependencies {
 }
 
 export interface ChatStore {
+  readonly pendingInputs: readonly PendingInput[];
+  discardPendingInput(requestId: string): void;
   readonly messages: readonly ChatMessage[];
   readonly chunks: readonly RenderedChunk[];
   readonly turnMetrics: readonly TurnMetricsEntry[];
@@ -156,7 +162,13 @@ export interface ChatStore {
    */
   noteLastActivity(at: number): void;
   handleDelta(
-    msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage,
+    msg:
+      | ChatAcceptedMessage
+      | ChatSubscriptionRejectedMessage
+      | ChatDeltaMessage
+      | ChatErrorMessage
+      | ChatQueuedMessage
+      | ChatResyncRequiredMessage,
   ): void;
   /**
    * Send a user message (start a turn via `chat.send`). Optimistically echoes
@@ -232,6 +244,14 @@ export interface ChatStore {
 }
 
 export function createChatStore(deps: ChatStoreDependencies): ChatStore {
+  let pendingInputs = $state<readonly PendingInput[]>(deps.pendingInputStorage?.load() ?? []);
+  function savePendingInputs(next: readonly PendingInput[]): void {
+    deps.pendingInputStorage?.save(next);
+    pendingInputs = next;
+  }
+  function confirmInput(requestId: string): void {
+    savePendingInputs(pendingInputs.filter((input) => input.request.requestId !== requestId));
+  }
   let transcript = $state<TranscriptState>(initialState());
   let metrics = $state<MetricsState>(initialMetricsState());
   let _pendingSync = $state(false);
@@ -507,10 +527,30 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
     }
   }
 
-  async function syncMetrics(): Promise<void> {
+  let pendingMetrics: Promise<void> | undefined;
+  let rerunMetrics = false;
+  function syncMetrics(): Promise<void> {
+    if (pendingMetrics !== undefined) {
+      rerunMetrics = true;
+      return pendingMetrics;
+    }
+    pendingMetrics = (async () => {
+      for (let attempt = 0; attempt < 2; attempt++) {
+        rerunMetrics = false;
+        await fetchMetrics();
+        if (!rerunMetrics || disposed) break;
+      }
+    })().finally(() => {
+      pendingMetrics = undefined;
+    });
+    return pendingMetrics;
+  }
+
+  async function fetchMetrics(): Promise<void> {
     if (disposed) return;
     try {
       const res = await deps.metricsSync(deps.conversationId);
+      if (disposed) return;
       metrics = applyDurableMetrics(metrics, res.turns);
     } catch (err) {
       // Metrics fetch failure must not block history sync or throw;
@@ -580,6 +620,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
   }
 
   return {
+    get pendingInputs() {
+      return pendingInputs;
+    },
+    discardPendingInput: confirmInput,
     get messages(): readonly ChatMessage[] {
       return selectMessages(transcript);
     },
@@ -626,9 +670,27 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
     },
 
     handleDelta(
-      msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage,
+      msg:
+        | ChatAcceptedMessage
+        | ChatSubscriptionRejectedMessage
+        | ChatDeltaMessage
+        | ChatErrorMessage
+        | ChatQueuedMessage
+        | ChatResyncRequiredMessage,
     ): void {
       if (disposed) return;
+      if (msg.type === "chat.accepted") {
+        if (msg.conversationId === deps.conversationId) confirmInput(msg.requestId);
+        return;
+      }
+      if (msg.type === "chat.subscription-rejected") {
+        if (msg.conversationId === deps.conversationId) {
+          _error =
+            "Live updates could not be subscribed. Message delivery is independent; saved inputs remain available below.";
+          void syncTail();
+        }
+        return;
+      }
       if (msg.type === "chat.resync-required") {
         if (msg.conversationId !== deps.conversationId) return;
         flushEvents();
@@ -656,6 +718,16 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
           return;
         }
         _error = msg.message;
+        if (msg.requestId !== undefined) {
+          savePendingInputs(
+            pendingInputs.map((input) =>
+              input.request.requestId === msg.requestId
+                ? { ...input, rejection: msg.message }
+                : input,
+            ),
+          );
+          transcript = removeLocalEcho(transcript, msg.requestId);
+        }
         return;
       }
       if (msg.type === "chat.queued") {
@@ -663,7 +735,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
         // A queued live opening may already have confirmed the echoed text.
         // Fold it first; turn-start alone must not protect still-local artifacts.
         flushEvents();
-        if (msg.requestId !== undefined) transcript = removeLocalEcho(transcript, msg.requestId);
+        if (msg.requestId !== undefined) {
+          confirmInput(msg.requestId);
+          transcript = removeLocalEcho(transcript, msg.requestId);
+        }
         return;
       }
       if (msg.event.conversationId !== deps.conversationId) {
@@ -685,9 +760,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
       // Sending ends the wait since the last turn: stamp it on that turn.
       cacheTimer = stopTurnGap(cacheTimer, now(), () => lastTurnIdOf(metrics));
       const requestId = deps.createRequestId();
-      transcript = appendUserMessage(transcript, text, images, requestId);
-      maybeTrim();
-      const msg: ChatSendMessage = {
+      const msg: ChatSendMessage & { requestId: string } = {
         type: "chat.send",
         requestId,
         conversationId: deps.conversationId,
@@ -697,19 +770,34 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
         ...(deps.workspaceId !== undefined ? { workspaceId: deps.workspaceId } : {}),
         ...(images !== undefined && images.length > 0 ? { images: [...images] } : {}),
       };
+      try {
+        savePendingInputs([...pendingInputs, { request: msg }]);
+      } catch (error) {
+        deps.onError?.("Unable to save input; the message has not been sent", error);
+        throw error;
+      }
+      transcript = appendUserMessage(transcript, text, images, requestId);
+      maybeTrim();
       deps.transport.send(msg);
     },
 
     queueMessage(text: string, images?: readonly ImageInput[]): void {
       const trimmed = text.trim();
       if (trimmed.length === 0 && !images?.length) return;
-      const msg: ChatQueueMessage = {
+      const msg: ChatQueueMessage & { requestId: string } = {
         type: "chat.queue",
+        requestId: deps.createRequestId(),
         conversationId: deps.conversationId,
         text: trimmed,
         ...(images !== undefined && images.length > 0 ? { images: [...images] } : {}),
         ...(deps.workspaceId !== undefined ? { workspaceId: deps.workspaceId } : {}),
       };
+      try {
+        savePendingInputs([...pendingInputs, { request: msg }]);
+      } catch (error) {
+        deps.onError?.("Unable to save input; the message has not been queued", error);
+        throw error;
+      }
       deps.transport.send(msg);
     },
 
diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts
index c24592d..5040e67 100644
--- a/src/features/chat/store.test.ts
+++ b/src/features/chat/store.test.ts
@@ -290,6 +290,7 @@ describe("createChatStore", () => {
       expect(transport.sentQueue).toEqual([
         {
           type: "chat.queue",
+          requestId: expect.any(String),
           conversationId: CONV_ID,
           workspaceId: "images-workspace",
           text: text.trim(),
diff --git a/src/features/chat/ui/PendingInputs.svelte b/src/features/chat/ui/PendingInputs.svelte
new file mode 100644
index 0000000..cbaf35f
--- /dev/null
+++ b/src/features/chat/ui/PendingInputs.svelte
@@ -0,0 +1,34 @@
+<script lang="ts">
+  import type { PendingInput } from "../pending-inputs";
+
+  let {
+    inputs,
+    onDiscard,
+  }: {
+    inputs: readonly PendingInput[];
+    onDiscard: (requestId: string) => void;
+  } = $props();
+</script>
+
+{#each inputs as input (input.request.requestId)}
+  <div class="m-3 rounded border border-warning p-3 text-sm" role="status">
+    <p>
+      {input.rejection
+        ? `Message rejected: ${input.rejection}`
+        : "Delivery unconfirmed. Your input is saved on this device."}
+    </p>
+    <p>Check the conversation history before sending again to avoid a duplicate.</p>
+    <textarea
+      class="textarea textarea-bordered mt-2 w-full"
+      aria-label="Saved message"
+      readonly
+      value={input.request.type === "chat.send" ? input.request.message : input.request.text}
+    ></textarea>
+    {#each input.request.images ?? [] as image}
+      <img class="my-2 max-h-32" src={image.url} alt="Saved attachment" />
+    {/each}
+    <button class="btn btn-sm mt-2" onclick={() => onDiscard(input.request.requestId)}
+      >Discard saved copy</button
+    >
+  </div>
+{/each}