fix(dashboard): Settle stuck inline sends and report a late steer

packages/junior-dashboard/e2e/conversation-send.spec.ts

@@ -869,7 +869,7 @@ test("edits a queued message back into the composer", async ({
 
   await pending.getByRole("button", { name: "Edit queued message" }).click();
   await expect(
-    page.getByText("Junior already read that message."),
+    page.getByText("That message was already read or removed."),
   ).toBeVisible();
   await expect(composer).toHaveValue(
     `${queuedMessage.text}\nDraft in progress`,

packages/junior-dashboard/src/client/conversations/ConversationPage.tsx

@@ -704,10 +704,10 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: {
           inboundMessageIds: [inboundMessageId],
           receivedBefore,
         });
-        // Junior can read the message before the cancel lands. Then it is
-        // already in the Conversation and there is nothing to edit.
+        // Junior can read the message before the cancel lands, or someone
+        // else can remove it. Then there is nothing to edit.
         if (!result.cancelledInboundMessageIds.includes(inboundMessageId)) {
-          setQueueActionError("Junior already read that message.");
+          setQueueActionError("That message was already read or removed.");
           return;
         }
         const restored = { images, text: message.text ?? "" };
@@ -736,13 +736,24 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: {
   const onSteerMessage = useCallback((message: ConversationMailboxMessage) => {
     if (steerPendingMessagesRef.current.isPending) return;
     setQueueActionError(undefined);
+    const inboundMessageId = message.inboundMessageId;
     steerPendingMessagesRef.current.mutate(
-      { inboundMessageIds: [message.inboundMessageId] },
+      { inboundMessageIds: [inboundMessageId] },
       {
         onError: () =>
           setQueueActionError(
             "Could not send the queued message now. Try again.",
           ),
+        // No id comes back when the message is not a queued `defer` message
+        // anymore. Junior read it, someone removed it, or another tab
+        // steered it. The message names all three.
+        onSuccess: (result) => {
+          if (!result.steeredInboundMessageIds.includes(inboundMessageId)) {
+            setQueueActionError(
+              "That message was already read, steered, or removed.",
+            );
+          }
+        },
       },
     );
   }, []);

packages/junior-dashboard/src/client/conversations/conversationOutbox.ts

@@ -1,6 +1,7 @@
 import type {
   ActorIdentity,
   ConversationPendingMessage,
+  ConversationReportEvent,
   InputImage,
 } from "@sentry/junior/api/schema";
 
@@ -14,6 +15,8 @@ export type ConversationOutboxMessage = {
    * the row in place, so it does not pass through the queue first.
    */
   inline?: boolean;
+  /** Client time when the accept response arrived. */
+  acceptedAtMs?: number;
   message: string;
   images?: InputImage[];
   messageId: string;
@@ -168,6 +171,70 @@ export function isInlineOutboxMessage(
   return Boolean(message.inline) && message.status !== "failed";
 }
 
+/** Ids of history Messages and of the input that started each Turn. */
+function historyMessageIds(events: readonly ConversationReportEvent[]): {
+  messageIds: Set<string>;
+  turnInputIds: Set<string>;
+} {
+  const messageIds = new Set<string>();
+  const turnInputIds = new Set<string>();
+  for (const event of events) {
+    if (event.data.type === "message") messageIds.add(event.data.messageId);
+    if (event.data.type === "turn_lifecycle" && event.data.state === "started")
+      for (const id of event.data.inputMessageIds ?? []) turnInputIds.add(id);
+  }
+  return { messageIds, turnInputIds };
+}
+
+/**
+ * Drop local sends that one server snapshot already shows.
+ *
+ * A queued row leaves when the mailbox or history has it. An inline row stays
+ * until history has it. The server writes the mailbox before it accepts a
+ * send, and the snapshot reads the mailbox before history. So when a mailbox
+ * read that started after the accept has neither copy, the send was removed,
+ * for example by another participant.
+ */
+export function settleConversationOutboxMessages(
+  current: ConversationOutboxMessage[],
+  snapshot: {
+    events: readonly ConversationReportEvent[];
+    /** Client time when the snapshot started its mailbox read. */
+    mailboxReadAtMs?: number;
+    queued: readonly ConversationPendingMessage[];
+  },
+): ConversationOutboxMessage[] {
+  const queuedIds = new Set(
+    snapshot.queued.map((message) => message.messageId),
+  );
+  const { messageIds, turnInputIds } = historyMessageIds(snapshot.events);
+  const settle = (
+    message: ConversationOutboxMessage,
+  ): ConversationOutboxMessage | undefined => {
+    const queued = queuedIds.has(message.messageId);
+    // A Turn joins several sends into the Message of the first one, so a
+    // joined send has only its Turn input id. The Turn starts before the
+    // mailbox drops the send. Until then, keep the row, or the mailbox copy
+    // would show in the queue.
+    const read =
+      messageIds.has(message.messageId) ||
+      (!queued && turnInputIds.has(message.messageId));
+    if (read) return undefined;
+    if (!isInlineOutboxMessage(message)) return queued ? undefined : message;
+    const removed =
+      !queued &&
+      message.acceptedAtMs !== undefined &&
+      snapshot.mailboxReadAtMs !== undefined &&
+      snapshot.mailboxReadAtMs > message.acceptedAtMs;
+    return removed ? undefined : message;
+  };
+  const next = current.flatMap((message) => settle(message) ?? []);
+  const unchanged =
+    next.length === current.length &&
+    next.every((message, index) => message === current[index]);
+  return unchanged ? current : next;
+}
+
 /**
  * Split pending rows between the queue and the transcript.
  *
@@ -222,7 +289,7 @@ export function acceptConversationOutboxMessage(
 ): ConversationOutboxMessage[] {
   return (current ?? []).map((message) =>
     message.idempotencyKey === idempotencyKey
-      ? { ...message, messageId, status: "accepted" }
+      ? { ...message, acceptedAtMs: Date.now(), messageId, status: "accepted" }
       : message,
   );
 }

packages/junior-dashboard/src/client/conversations/queries.ts

@@ -43,10 +43,10 @@ import {
   conversationOutboxMessageForSubmit,
   conversationOutboxQueryKey,
   failConversationOutboxMessage,
-  isInlineOutboxMessage,
   mergeConversationMailboxMessages,
   acceptConversationOutboxMessage,
   reuseConversationMailboxMessages,
+  settleConversationOutboxMessages,
   splitInlineMailboxMessages,
   upsertConversationOutboxMessage,
   type ConversationMailboxMessage,
@@ -141,6 +141,8 @@ const FAST_POLL_WINDOW_MS = 6_000;
 
 type ConversationSnapshot = ConversationDetailReport & {
   mailbox?: ConversationPendingMessagesReport;
+  /** Client time when this snapshot started its mailbox read. */
+  mailboxReadAtMs?: number;
   detailEtag?: string;
 };
 
@@ -150,9 +152,10 @@ async function readConversationSnapshot(
   signal?: AbortSignal,
   previous?: ConversationSnapshot,
 ): Promise<ConversationSnapshot> {
+  const mailboxReadAtMs = Date.now();
   const mailbox = await readConversationPendingMessages(conversationId, signal);
   const detail = await readConversationData(conversationId, signal, previous);
-  return { ...detail, mailbox };
+  return { ...detail, mailbox, mailboxReadAtMs };
 }
 
 /** Refresh history and mailbox as one visible snapshot, including idle input. */
@@ -232,6 +235,7 @@ export function useCreateConversation() {
               inline: true,
               messageId: accepted.messageId,
             }),
+            acceptedAtMs: Date.now(),
             status: "accepted",
           }),
       );
@@ -282,6 +286,7 @@ export function useForkConversation(sourceConversationId: string) {
               inline: true,
               messageId: accepted.messageId,
             }),
+            acceptedAtMs: Date.now(),
             status: "accepted",
           }),
       );
@@ -722,30 +727,19 @@ export function useConversationData(conversationId: string | undefined) {
       ? deferredDetail
       : detail.data;
   const mailbox = detailData?.mailbox;
-  // Accept is not visibility. Keep a local row until either server resource
-  // contains its real Message id. History and mailbox arrive in one render.
+  // Accept is not visibility. Keep a local row until the server snapshot
+  // shows it. History and mailbox arrive in one render.
   useEffect(() => {
     if (!mailbox || !detailData) return;
-    const queuedIds = new Set(
-      mailbox.messages.map((message) => message.messageId),
-    );
-    const historyIds = new Set<string>();
-    for (const event of detailData.events) {
-      if (event.data.type === "message") historyIds.add(event.data.messageId);
-    }
     queryClient.setQueryData<ConversationOutboxMessage[]>(
       conversationOutboxQueryKey(conversationId),
-      (current) => {
-        // An inline row stays until history has it. The mailbox copy alone
-        // would put it back in the queue.
-        const next = current?.filter(
-          (message) =>
-            !historyIds.has(message.messageId) &&
-            (isInlineOutboxMessage(message) ||
-              !queuedIds.has(message.messageId)),
-        );
-        return next?.length === current?.length ? current : next;
-      },
+      (current) =>
+        current &&
+        settleConversationOutboxMessages(current, {
+          events: detailData.events,
+          mailboxReadAtMs: detailData.mailboxReadAtMs,
+          queued: mailbox.messages,
+        }),
     );
   }, [conversationId, detailData, mailbox, outbox.data, queryClient]);
   // The title is generated after the first message, and only the open

packages/junior-dashboard/tests/conversation-outbox.test.ts

@@ -5,6 +5,7 @@ import {
   failConversationOutboxMessage,
   mailboxMessageFromOutbox,
   mergeConversationMailboxMessages,
+  settleConversationOutboxMessages,
   splitInlineMailboxMessages,
   upsertConversationOutboxMessage,
 } from "../src/client/conversations/conversationOutbox";
@@ -190,4 +191,108 @@ describe("conversation outbox", () => {
       queued: merged,
     });
   });
+
+  it("keeps inline sends until history reads them or the queue drops them", () => {
+    const accepted = (messageId: string, message: string) => ({
+      ...conversationOutboxMessageForSubmit({
+        idempotencyKey: messageId,
+        inline: true,
+        messageId,
+        message,
+        now: "2026-01-01T00:00:00.000Z",
+      }),
+      acceptedAtMs: 1_000,
+      status: "accepted" as const,
+    });
+    const serverCopy = (message: ReturnType<typeof accepted>) => ({
+      ...mailboxMessageFromOutbox(message),
+      clientStatus: undefined,
+      idempotencyKey: undefined,
+    });
+    const inline = accepted("accepted-1", "Continue in Junior");
+    const joined = accepted("accepted-2", "And the rollback owner");
+    const queuedRow = { ...accepted("queued-1", "Queued"), inline: undefined };
+    const outbox = [inline, joined, queuedRow];
+
+    // A mailbox read from before the accept proves nothing. The queued row
+    // leaves once the mailbox lists it.
+    expect(
+      settleConversationOutboxMessages(outbox, {
+        events: [],
+        mailboxReadAtMs: 900,
+        queued: [serverCopy(queuedRow)],
+      }),
+    ).toEqual([inline, joined]);
+
+    // A Turn joined the second send into the first Message, so only its
+    // started event has the second id. The Turn starts before the mailbox
+    // drops the send, so the row stays inline and out of the queue.
+    const turnStarted = [
+      {
+        seq: 1,
+        createdAt: "2026-01-01T00:00:02.000Z",
+        data: {
+          type: "turn_lifecycle" as const,
+          turnId: "turn-1",
+          state: "started" as const,
+          inputMessageIds: ["accepted-1", "accepted-2"],
+        },
+      },
+      {
+        seq: 2,
+        createdAt: "2026-01-01T00:00:02.000Z",
+        data: {
+          type: "message" as const,
+          messageId: "accepted-1",
+          role: "user" as const,
+          text: "Continue in Junior\nAnd the rollback owner",
+        },
+      },
+    ];
+    const server = [serverCopy(inline), serverCopy(joined)];
+    const whileQueued = settleConversationOutboxMessages([inline, joined], {
+      events: turnStarted,
+      mailboxReadAtMs: 2_000,
+      queued: server,
+    });
+    expect(whileQueued).toEqual([joined]);
+    const split = splitInlineMailboxMessages(
+      mergeConversationMailboxMessages(server, whileQueued),
+      server,
+      whileQueued,
+    );
+    expect(split.inline.map((message) => message.messageId)).toEqual([
+      "accepted-2",
+    ]);
+    expect(split.queued.map((message) => message.messageId)).toEqual([
+      "accepted-1",
+    ]);
+    // History has a Message for the first send, so the queue hides it.
+
+    // After the ack, the Turn input id settles the joined send.
+    expect(
+      settleConversationOutboxMessages([joined], {
+        events: turnStarted,
+        mailboxReadAtMs: 3_000,
+        queued: [],
+      }),
+    ).toEqual([]);
+
+    // A mailbox read after the accept has neither copy: someone removed it.
+    const unchanged = [inline];
+    expect(
+      settleConversationOutboxMessages(unchanged, {
+        events: [],
+        mailboxReadAtMs: 1_000,
+        queued: [],
+      }),
+    ).toBe(unchanged);
+    expect(
+      settleConversationOutboxMessages(unchanged, {
+        events: [],
+        mailboxReadAtMs: 1_001,
+        queued: [],
+      }),
+    ).toEqual([]);
+  });
 });