diff --git a/packages/junior-dashboard/e2e/conversation-send.spec.ts b/packages/junior-dashboard/e2e/conversation-send.spec.ts index d594fe47a9..f35d22cbe0 100644 --- a/packages/junior-dashboard/e2e/conversation-send.spec.ts +++ b/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`, diff --git a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx b/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx index 07b2a0e335..996bda0555 100644 --- a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx +++ b/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.", + ); + } + }, }, ); }, []); diff --git a/packages/junior-dashboard/src/client/conversations/conversationOutbox.ts b/packages/junior-dashboard/src/client/conversations/conversationOutbox.ts index a23f51f7f0..d75c9db934 100644 --- a/packages/junior-dashboard/src/client/conversations/conversationOutbox.ts +++ b/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; + turnInputIds: Set; +} { + const messageIds = new Set(); + const turnInputIds = new Set(); + 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, ); } diff --git a/packages/junior-dashboard/src/client/conversations/queries.ts b/packages/junior-dashboard/src/client/conversations/queries.ts index ddf47067b9..ae8ab4cad7 100644 --- a/packages/junior-dashboard/src/client/conversations/queries.ts +++ b/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 { + 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(); - for (const event of detailData.events) { - if (event.data.type === "message") historyIds.add(event.data.messageId); - } queryClient.setQueryData( 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 diff --git a/packages/junior-dashboard/tests/conversation-outbox.test.ts b/packages/junior-dashboard/tests/conversation-outbox.test.ts index 8b5a37a6ec..37f22e492b 100644 --- a/packages/junior-dashboard/tests/conversation-outbox.test.ts +++ b/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) => ({ + ...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([]); + }); });