fix(web): stop a sent image message reporting "attachment not found"

apps/web/src/components/ChatView.tsx

@@ -36,6 +36,7 @@ import {
 } from "@t3tools/client-runtime/state/thread-settled";
 import {
   ANONYMOUS_OUTBOX_IDENTITY,
+  selectThreadOutboxReplay,
   shouldRetryThreadOutboxDelivery,
   threadOutboxRetryDelayMs,
 } from "@t3tools/client-runtime/outbox";
@@ -2696,15 +2697,19 @@ function ChatViewContent(props: ChatViewProps) {
     if (isLocalSendBusy || outboxEnvironmentConnectionPhase !== "connected") {
       return;
     }
-    // Oldest first: IndexedDB iterates by messageId (a random UUID), so an
-    // unsorted pick would flush the queue in arbitrary order.
-    const queued = durableOutboxItems
-      .filter(
-        (item) =>
-          item.deliveryState !== "failed" && !outboxReplayInFlightRef.current.has(item.messageId),
-      )
-      .sort((left, right) => left.createdAt.localeCompare(right.createdAt))
-      .at(0);
+    // T3-CUSTOM(expbkt3): the send gate above clears on the server's projection
+    // acknowledgement, which lands before the RPC reply removes the queue row.
+    // Replaying then puts a second copy of a turn that already started on the
+    // wire, and since upstream #8048 that copy fails outright — its uploaded
+    // attachment was released the moment the first copy was acknowledged.
+    const { stale, next: queued } = selectThreadOutboxReplay({
+      items: durableOutboxItems,
+      identityKey: outboxIdentityKey,
+      replayingMessageIds: outboxReplayInFlightRef.current,
+    });
+    for (const item of stale) {
+      void discardDurableOutbox(item);
+    }
     if (queued === undefined) return;
     outboxReplayInFlightRef.current.add(queued.messageId);
     void startThreadTurn({
@@ -2764,6 +2769,7 @@ function ChatViewContent(props: ChatViewProps) {
       }, threadOutboxRetryDelayMs(attempt));
     });
   }, [
+    discardDurableOutbox,
     durableOutboxItems,
     isLocalSendBusy,
     outboxEnvironmentConnectionPhase,

packages/client-runtime/src/outbox/delivery.test.ts

@@ -0,0 +1,180 @@
+import { beforeEach, describe, expect, it } from "@effect/vitest";
+
+import {
+  beginThreadOutboxDelivery,
+  isThreadOutboxDeliveryDelivered,
+  isThreadOutboxDeliveryInFlight,
+  resetThreadOutboxDeliveries,
+  selectThreadOutboxReplay,
+  settleThreadOutboxDelivery,
+  threadOutboxDeliveryKey,
+} from "./delivery.ts";
+
+const ref = (messageId: string, identityKey: string | undefined = "user-1") => ({
+  environmentId: "env-1",
+  identityKey,
+  messageId,
+});
+
+describe("thread outbox delivery registry", () => {
+  beforeEach(() => {
+    resetThreadOutboxDeliveries();
+  });
+
+  it("reports a dispatch as in flight until it settles", () => {
+    const key = threadOutboxDeliveryKey(ref("message-1"));
+    beginThreadOutboxDelivery(key);
+
+    expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(true);
+    expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(false);
+
+    settleThreadOutboxDelivery(key, true);
+
+    expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(false);
+    expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(true);
+  });
+
+  it("forgets a rejected dispatch so retry and reconnect replay still work", () => {
+    const key = threadOutboxDeliveryKey(ref("message-1"));
+    beginThreadOutboxDelivery(key);
+    settleThreadOutboxDelivery(key, false);
+
+    expect(isThreadOutboxDeliveryInFlight(ref("message-1"))).toBe(false);
+    expect(isThreadOutboxDeliveryDelivered(ref("message-1"))).toBe(false);
+  });
+
+  it("keeps environments and accounts apart", () => {
+    settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1")), true);
+
+    expect(isThreadOutboxDeliveryDelivered(ref("message-1", "user-2"))).toBe(false);
+    expect(
+      isThreadOutboxDeliveryDelivered({
+        environmentId: "env-2",
+        identityKey: "user-1",
+        messageId: "message-1",
+      }),
+    ).toBe(false);
+  });
+
+  it("bounds the delivered history so a long-lived tab cannot grow it forever", () => {
+    for (let index = 0; index < 300; index += 1) {
+      settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref(`message-${index}`)), true);
+    }
+
+    expect(isThreadOutboxDeliveryDelivered(ref("message-0"))).toBe(false);
+    expect(isThreadOutboxDeliveryDelivered(ref("message-299"))).toBe(true);
+  });
+});
+
+const row = (
+  messageId: string,
+  overrides: {
+    readonly deliveryState?: "pending" | "failed";
+    readonly createdAt?: string;
+  } = {},
+) => ({
+  environmentId: "env-1",
+  identityKey: "user-1",
+  messageId,
+  createdAt: overrides.createdAt ?? "2026-09-01T11:43:00.000Z",
+  ...(overrides.deliveryState === undefined ? {} : { deliveryState: overrides.deliveryState }),
+});
+
+describe("thread outbox replay selection", () => {
+  beforeEach(() => {
+    resetThreadOutboxDeliveries();
+  });
+
+  it("sends the oldest queued row", () => {
+    const older = row("message-1", { createdAt: "2026-09-01T11:40:00.000Z" });
+    const newer = row("message-2", { createdAt: "2026-09-01T11:43:00.000Z" });
+
+    const selection = selectThreadOutboxReplay({
+      items: [newer, older],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(),
+    });
+
+    expect(selection.next).toBe(older);
+    expect(selection.stale).toEqual([]);
+  });
+
+  it("leaves a row alone while its own dispatch is on the wire", () => {
+    beginThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1")));
+
+    const selection = selectThreadOutboxReplay({
+      items: [row("message-1")],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(),
+    });
+
+    expect(selection.next).toBeUndefined();
+    expect(selection.stale).toEqual([]);
+  });
+
+  it("discards a row whose turn was already delivered instead of resending it", () => {
+    settleThreadOutboxDelivery(threadOutboxDeliveryKey(ref("message-1")), true);
+    const delivered = row("message-1");
+
+    const selection = selectThreadOutboxReplay({
+      items: [delivered],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(),
+    });
+
+    expect(selection.next).toBeUndefined();
+    expect(selection.stale).toEqual([delivered]);
+  });
+
+  it("still sends a row whose dispatch was rejected", () => {
+    const key = threadOutboxDeliveryKey(ref("message-1"));
+    beginThreadOutboxDelivery(key);
+    settleThreadOutboxDelivery(key, false);
+    const rejected = row("message-1");
+
+    const selection = selectThreadOutboxReplay({
+      items: [rejected],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(),
+    });
+
+    expect(selection.next).toBe(rejected);
+    expect(selection.stale).toEqual([]);
+  });
+
+  it("skips rows the replay loop already has on the wire and rows marked failed", () => {
+    const selection = selectThreadOutboxReplay({
+      items: [row("message-1"), row("message-2", { deliveryState: "failed" })],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(["message-1"]),
+    });
+
+    expect(selection.next).toBeUndefined();
+    expect(selection.stale).toEqual([]);
+  });
+
+  it("falls back to the current identity for legacy rows that omit one", () => {
+    settleThreadOutboxDelivery(
+      threadOutboxDeliveryKey({
+        environmentId: "env-1",
+        identityKey: "user-1",
+        messageId: "message-1",
+      }),
+      true,
+    );
+    const legacy = {
+      environmentId: "env-1",
+      messageId: "message-1",
+      createdAt: "2026-09-01T11:43:00.000Z",
+    };
+
+    const selection = selectThreadOutboxReplay({
+      items: [legacy],
+      identityKey: "user-1",
+      replayingMessageIds: new Set(),
+    });
+
+    expect(selection.next).toBeUndefined();
+    expect(selection.stale).toEqual([legacy]);
+  });
+});

packages/client-runtime/src/outbox/delivery.ts

@@ -0,0 +1,118 @@
+// T3-CUSTOM(expbkt3): in-process record of which queued turns this client has
+// already put on the wire.
+//
+// `dispatchPersistedOutboxItem` was written when a duplicate dispatch was
+// harmless: the server deduplicates by commandId, so a queue row that outlived
+// its acknowledgement cost nothing to send twice. Since upstream #8048 that is
+// no longer true. The web client uploads an image before sending and references
+// it by a `pending-…` id, then releases that upload as soon as the turn is
+// acknowledged. A second dispatch is normalized before any commandId dedup,
+// finds the pending upload gone, and fails with "attachment not found (removed
+// or expired)" — marking a turn that actually ran as failed.
+//
+// A replay loop cannot tell the difference on its own: its send gate clears on
+// the server's projection acknowledgement, which races ahead of both the RPC
+// reply and the queue-row removal. This registry is the missing signal.
+
+/** Bounded so a long-lived tab cannot grow the delivered set without limit. */
+const DELIVERED_HISTORY_LIMIT = 256;
+
+const inFlightDeliveries = new Set<string>();
+const deliveredDeliveries = new Set<string>();
+
+export interface ThreadOutboxDeliveryRef {
+  readonly environmentId: string;
+  readonly identityKey: string | undefined;
+  readonly messageId: string;
+}
+
+export function threadOutboxDeliveryKey(ref: ThreadOutboxDeliveryRef): string {
+  return `${ref.identityKey ?? ""} ${ref.environmentId} ${ref.messageId}`;
+}
+
+export function beginThreadOutboxDelivery(key: string): void {
+  inFlightDeliveries.add(key);
+}
+
+/**
+ * Records the outcome of a dispatch. A delivered turn stays remembered so a
+ * queue row that outlived its removal is recognised as stale; a failed one is
+ * forgotten so retry and reconnect replay still work.
+ */
+export function settleThreadOutboxDelivery(key: string, delivered: boolean): void {
+  inFlightDeliveries.delete(key);
+  if (!delivered) {
+    return;
+  }
+  deliveredDeliveries.add(key);
+  while (deliveredDeliveries.size > DELIVERED_HISTORY_LIMIT) {
+    const oldest = deliveredDeliveries.values().next();
+    if (oldest.done === true) {
+      return;
+    }
+    deliveredDeliveries.delete(oldest.value);
+  }
+}
+
+/** This turn is on the wire right now — leave its queue row to the dispatch that owns it. */
+export function isThreadOutboxDeliveryInFlight(ref: ThreadOutboxDeliveryRef): boolean {
+  return inFlightDeliveries.has(threadOutboxDeliveryKey(ref));
+}
+
+/** This turn was acknowledged, so its queue row is stale and must be discarded, not replayed. */
+export function isThreadOutboxDeliveryDelivered(ref: ThreadOutboxDeliveryRef): boolean {
+  return deliveredDeliveries.has(threadOutboxDeliveryKey(ref));
+}
+
+/** Test seam: the registry is module state shared by every caller in the app. */
+export function resetThreadOutboxDeliveries(): void {
+  inFlightDeliveries.clear();
+  deliveredDeliveries.clear();
+}
+
+interface ReplayCandidate {
+  readonly environmentId: string;
+  readonly identityKey?: string | undefined;
+  readonly messageId: string;
+  readonly deliveryState?: string | undefined;
+  readonly createdAt: string;
+}
+
+/**
+ * Splits the queue into the rows a replay loop should drop and the single row it
+ * should send next. Stale rows are discarded rather than skipped: a row left in
+ * place keeps the composer latched as busy for as long as it survives.
+ */
+export function selectThreadOutboxReplay<Item extends ReplayCandidate>(input: {
+  readonly items: ReadonlyArray<Item>;
+  readonly identityKey: string;
+  /** Rows the replay loop itself already has on the wire. */
+  readonly replayingMessageIds: ReadonlySet<string>;
+}): { readonly stale: ReadonlyArray<Item>; readonly next: Item | undefined } {
+  const stale: Item[] = [];
+  const sendable: Item[] = [];
+  for (const item of input.items) {
+    if (item.deliveryState === "failed") {
+      continue;
+    }
+    const ref = {
+      environmentId: item.environmentId,
+      identityKey: item.identityKey ?? input.identityKey,
+      messageId: item.messageId,
+    };
+    if (isThreadOutboxDeliveryDelivered(ref)) {
+      stale.push(item);
+      continue;
+    }
+    if (input.replayingMessageIds.has(item.messageId) || isThreadOutboxDeliveryInFlight(ref)) {
+      continue;
+    }
+    sendable.push(item);
+  }
+  // Oldest first: IndexedDB iterates by messageId (a random UUID), so an
+  // unsorted pick would flush the queue in arbitrary order.
+  return {
+    stale,
+    next: sendable.sort((left, right) => left.createdAt.localeCompare(right.createdAt)).at(0),
+  };
+}

packages/client-runtime/src/outbox/index.ts

@@ -1,3 +1,4 @@
+export * from "./delivery.ts";
 export * from "./dispatch.ts";
 export * from "./metrics.ts";
 export * from "./model.ts";

packages/client-runtime/src/state/threadCommands.ts

@@ -85,8 +85,11 @@ import type { EnvironmentRegistry } from "../connection/registry.ts";
 // T3-CUSTOM(expbkt3): every client persists the exact turn before dispatch.
 import {
   ANONYMOUS_OUTBOX_IDENTITY,
+  beginThreadOutboxDelivery,
   recordThreadOutboxFailure,
+  settleThreadOutboxDelivery,
   shouldRetryThreadOutboxDelivery,
+  threadOutboxDeliveryKey,
   ThreadOutboxPersistenceError,
   type QueuedThreadMessage,
 } from "../outbox/index.ts";
@@ -171,7 +174,25 @@ const startThreadTurnDurably = Effect.fn("ThreadCommands.startThreadTurnDurably"
       }),
     ),
   );
-  const dispatched = yield* Effect.exit(dispatch);
+  // T3-CUSTOM(expbkt3): a replay loop must not send this turn again while it is
+  // on the wire, or after it lands — an uploaded attachment is released on the
+  // first acknowledgement and the duplicate would fail as "attachment not
+  // found". Settling inside `ensuring` also covers interruption.
+  const deliveryKey = threadOutboxDeliveryKey({
+    environmentId,
+    identityKey: outboxIdentityKey,
+    messageId: queuedMessage.messageId,
+  });
+  beginThreadOutboxDelivery(deliveryKey);
+  let deliveredExit = false;
+  const dispatched = yield* Effect.exit(dispatch).pipe(
+    Effect.tap((exit) =>
+      Effect.sync(() => {
+        deliveredExit = Exit.isSuccess(exit);
+      }),
+    ),
+    Effect.ensuring(Effect.sync(() => settleThreadOutboxDelivery(deliveryKey, deliveredExit))),
+  );
   if (Exit.isFailure(dispatched)) {
     const error = Cause.squash(dispatched.cause);
     const retrying = shouldRetryThreadOutboxDelivery(error);