From 49f89a67c2786bd1d5fd6073544a9290c1a4c97f Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 13 Aug 2026 17:19:39 +0100 Subject: [PATCH] feat: better attachments pre send guard --- frontend/src/components/MessageInput.tsx | 27 ++-- frontend/src/conversation/Conversation.tsx | 16 ++- frontend/src/locale/en.json | 4 +- frontend/src/upload/uploadSlice.test.ts | 69 ++++++++++ frontend/src/upload/uploadSlice.ts | 31 ++++- .../specs/tier-a/composer-resilience.spec.ts | 127 ++++++++++++------ 6 files changed, 217 insertions(+), 57 deletions(-) diff --git a/frontend/src/components/MessageInput.tsx b/frontend/src/components/MessageInput.tsx index a35aa9a1..2dc6d364 100644 --- a/frontend/src/components/MessageInput.tsx +++ b/frontend/src/components/MessageInput.tsx @@ -304,6 +304,11 @@ type MessageInputProps = { // attachments (used by doc-driven workflow runs). Normal chat leaves this // unset, preserving the text-required behavior. allowSendWithoutText?: boolean; + // A question queued by a send path outside the composer (hero + // suggestion cards) while attachments were still pending: seeds the + // input and arms the send so the standard waiting banner takes over. + queuedQuestion?: string | null; + onQueuedQuestionConsumed?: () => void; }; export default function MessageInput({ @@ -313,6 +318,8 @@ export default function MessageInput({ showToolButton = true, autoFocus = true, allowSendWithoutText = false, + queuedQuestion = null, + onQueuedQuestionConsumed, }: MessageInputProps) { const { t } = useTranslation(); const [value, setValue] = useState(''); @@ -1486,9 +1493,6 @@ export default function MessageInput({ // When ``allowSendWithoutText`` is set, an attachment-only submit is // permitted as long as at least one attachment exists; a still-pending // one arms the send instead of submitting (see handleSubmit). - const completedAttachmentCount = attachments.filter( - (attachment) => attachment.status === 'completed', - ).length; const hasSubmittableContent = Boolean(value.trim()) || (allowSendWithoutText && attachments.length > 0); const canSubmit = @@ -1518,6 +1522,16 @@ export default function MessageInput({ cancel: cancelArmedSend, } = useArmedSend({ attachments, onFlush: submitNow }); + // Adopt a question queued outside the composer: seed the input, arm, + // and hand the wait to the standard banner. If the attachments already + // resolved by the time this runs, the armed-send effect flushes at once. + useEffect(() => { + if (queuedQuestion == null || queuedQuestion === '') return; + setValue(queuedQuestion); + armSend(); + onQueuedQuestionConsumed?.(); + }, [queuedQuestion]); + const handleSubmit = () => { if (!canSubmit) return; // Attachments still uploading/parsing (or failed) must never be @@ -1619,13 +1633,6 @@ export default function MessageInput({ })} )} - {!sendArmed && completedAttachmentCount > 0 && ( -
- {t('conversation.attachments.attachedCount', { - count: completedAttachmentCount, - })} -
- )} {voiceError && (
{voiceError} diff --git a/frontend/src/conversation/Conversation.tsx b/frontend/src/conversation/Conversation.tsx index 9d2604c2..1cb2487e 100644 --- a/frontend/src/conversation/Conversation.tsx +++ b/frontend/src/conversation/Conversation.tsx @@ -32,8 +32,10 @@ import { submitToolActions, updateQuery, } from './conversationSlice'; +import { getSendReadiness } from '../components/message-input/armedSend'; import { clearAttachments, + selectAttachments, selectCompletedAttachments, } from '../upload/uploadSlice'; @@ -58,6 +60,10 @@ export default function Conversation() { const conversationId = useSelector(selectConversationId); const selectedAgent = useSelector(selectSelectedAgent); const completedAttachments = useSelector(selectCompletedAttachments); + const attachments = useSelector(selectAttachments); + // A direct send (hero card) that must wait for pending attachments is + // parked here; MessageInput consumes it into an armed composer send. + const [queuedQuestion, setQueuedQuestion] = useState(null); const [lastQueryReturnedErr, setLastQueryReturnedErr] = useState(false); @@ -214,6 +220,12 @@ export default function Conversation() { index, attachmentIds: rowAttachmentIds, }); + } else if (getSendReadiness(attachments).state !== 'ready') { + // Direct new sends (hero suggestion cards) bypass MessageInput's + // submit gate. With files still uploading/parsing (or failed), + // sending now would silently drop them — route the question into + // the composer instead, where the armed-send banner takes over. + setQueuedQuestion(trimmedQuestion); } else { const filesAttached = completedAttachments .filter((a) => a.id) @@ -236,7 +248,7 @@ export default function Conversation() { if (filesAttached.length > 0) dispatch(clearAttachments()); } }, - [dispatch, handleFetchAnswer, completedAttachments, queries], + [dispatch, handleFetchAnswer, completedAttachments, attachments, queries], ); const handleFeedback = (query: Query, feedback: FEEDBACK, index: number) => { @@ -423,6 +435,8 @@ export default function Conversation() { onSubmit={(text) => { handleQuestionSubmission(text); }} + queuedQuestion={queuedQuestion} + onQueuedQuestionConsumed={() => setQueuedQuestion(null)} loading={status === 'loading'} showSourceButton={selectedAgent ? false : true} showToolButton={selectedAgent ? false : true} diff --git a/frontend/src/locale/en.json b/frontend/src/locale/en.json index dd70bb88..52adfe35 100644 --- a/frontend/src/locale/en.json +++ b/frontend/src/locale/en.json @@ -1098,9 +1098,7 @@ "waitingToSend_one": "Will send when {{count}} file finishes processing…", "waitingToSend_other": "Will send when {{count}} files finish processing…", "cancelQueuedSend": "Cancel", - "sendBlockedByFailed": "{{names}} couldn't be processed — remove to send", - "attachedCount_one": "Attached: {{count}} file", - "attachedCount_other": "Attached: {{count}} files" + "sendBlockedByFailed": "{{names}} couldn't be processed — remove to send" }, "retry": "Retry", "reasoning": "Reasoning", diff --git a/frontend/src/upload/uploadSlice.test.ts b/frontend/src/upload/uploadSlice.test.ts index 86826041..365aa727 100644 --- a/frontend/src/upload/uploadSlice.test.ts +++ b/frontend/src/upload/uploadSlice.test.ts @@ -476,3 +476,72 @@ describe('attachment race recovery', () => { expect(uploadState.attachments[0].status).toBe('failed'); }); }); + +describe('upload race — SSE auto-created duplicate absorbed on sourceId bind', () => { + it('merges the orphan row when the upload response binds the sourceId', () => { + // Client row exists without a sourceId (upload XHR still in flight)... + let state = stateWithTask( + makeTask({ + id: 'client-1', + sourceId: undefined, + status: 'uploading', + fileName: 'doc4.pdf', + }), + ); + // ...and the worker's queued event lands first — the slice cannot + // match it, so it auto-creates an orphan row keyed by sourceId. + state = reducer( + state, + ingest('source.ingest.queued', { filename: 'doc4.pdf' }), + ); + expect(state.tasks).toHaveLength(2); + + // The XHR response now binds the sourceId to the client row: the + // orphan must be absorbed, not left behind as a stuck duplicate. + state = reducer( + state, + updateUploadTask({ + id: 'client-1', + updates: { + taskId: 'task-9', + sourceId: SOURCE_ID, + status: 'training', + progress: 0, + }, + }), + ); + + expect(state.tasks).toHaveLength(1); + expect(state.tasks[0].id).toBe('client-1'); + expect(state.tasks[0].sourceId).toBe(SOURCE_ID); + }); + + it('keeps the orphan progress and terminal state when it advanced further', () => { + let state = stateWithTask( + makeTask({ id: 'client-1', sourceId: undefined, status: 'uploading' }), + ); + state = reducer( + state, + ingest('source.ingest.queued', { filename: 'doc.pdf' }), + ); + state = reducer( + state, + ingest('source.ingest.progress', { current: 80, stage: 'embedding' }), + ); + state = reducer(state, ingest('source.ingest.completed', {})); + + state = reducer( + state, + updateUploadTask({ + id: 'client-1', + updates: { sourceId: SOURCE_ID, status: 'training', progress: 0 }, + }), + ); + + expect(state.tasks).toHaveLength(1); + const task = state.tasks[0]; + expect(task.id).toBe('client-1'); + expect(task.status).toBe('completed'); + expect(task.progress).toBe(100); + }); +}); diff --git a/frontend/src/upload/uploadSlice.ts b/frontend/src/upload/uploadSlice.ts index c0ecf35a..468a8353 100644 --- a/frontend/src/upload/uploadSlice.ts +++ b/frontend/src/upload/uploadSlice.ts @@ -49,11 +49,7 @@ export interface Attachment { } export type UploadTaskStatus = - | 'preparing' - | 'uploading' - | 'training' - | 'completed' - | 'failed'; + 'preparing' | 'uploading' | 'training' | 'completed' | 'failed'; export interface UploadTask { id: string; @@ -193,6 +189,31 @@ export const uploadSlice = createSlice({ ...updates, }; } + + // Absorb an SSE auto-created duplicate. When ``source.ingest.*`` + // events arrive before this update binds the sourceId (worker + // beats the upload XHR's onload), the SSE reducer cannot match + // the client row and auto-creates a second row keyed by + // sourceId — the "two docs at once" toast. Merge whatever state + // the orphan accumulated during the race window and drop it. + if (incomingSourceId) { + const dupIndex = state.tasks.findIndex( + (t, i) => i !== index && t.sourceId === incomingSourceId, + ); + if (dupIndex !== -1) { + const dup = state.tasks[dupIndex]; + const row = state.tasks[index]; + if (dup.progress > row.progress) row.progress = dup.progress; + if (dup.stage) row.stage = dup.stage; + if (dup.status === 'completed' || dup.status === 'failed') { + row.status = dup.status; + row.errorMessage = dup.errorMessage; + row.tokenLimitReached = dup.tokenLimitReached; + if (!sourceWasDismissed) row.dismissed = false; + } + state.tasks.splice(dupIndex, 1); + } + } } }, dismissUploadTask: (state, action: PayloadAction) => { diff --git a/tests/e2e/specs/tier-a/composer-resilience.spec.ts b/tests/e2e/specs/tier-a/composer-resilience.spec.ts index c3f343f5..5bdc9f39 100644 --- a/tests/e2e/specs/tier-a/composer-resilience.spec.ts +++ b/tests/e2e/specs/tier-a/composer-resilience.spec.ts @@ -19,33 +19,33 @@ * here `setInputFiles` on the dropzone input is the point. */ -import { dirname, resolve } from 'node:path'; -import { fileURLToPath } from 'node:url'; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; -import * as playwright from '@playwright/test'; +import * as playwright from "@playwright/test"; const { expect, test } = playwright; -import type { Page } from '@playwright/test'; +import type { Page } from "@playwright/test"; -import { newUserContext } from '../../helpers/auth.js'; -import { pg } from '../../helpers/db.js'; -import { resetDb } from '../../helpers/reset.js'; +import { newUserContext } from "../../helpers/auth.js"; +import { pg } from "../../helpers/db.js"; +import { resetDb } from "../../helpers/reset.js"; const HERE = dirname(fileURLToPath(import.meta.url)); -const FIXTURES_DIR = resolve(HERE, '..', '..', 'fixtures', 'docs'); -const SMALL_FIXTURE_PATH = resolve(FIXTURES_DIR, 'notes.txt'); +const FIXTURES_DIR = resolve(HERE, "..", "..", "fixtures", "docs"); +const SMALL_FIXTURE_PATH = resolve(FIXTURES_DIR, "notes.txt"); async function openComposer(page: Page): Promise { - await page.goto('/'); - await expect(page.locator('#message-input')).toBeVisible(); + await page.goto("/"); + await expect(page.locator("#message-input")).toBeVisible(); } -test.describe('tier-a · composer resilience', () => { +test.describe("tier-a · composer resilience", () => { test.beforeEach(async () => { await resetDb(); }); - test('failed /stream surfaces a visible error and Retry re-sends successfully', async ({ + test("failed /stream surfaces a visible error and Retry re-sends successfully", async ({ browser, }) => { const { context, sub } = await newUserContext(browser); @@ -55,31 +55,31 @@ test.describe('tier-a · composer resilience', () => { // Force the stream POST to die at the network layer — the shape of // an outage the frontend cannot see an HTTP status for. - await page.route('**/stream', (route) => route.abort('failed')); + await page.route("**/stream", (route) => route.abort("failed")); - const textarea = page.locator('#message-input'); - await textarea.fill('resilience probe — e2e-composer-retry'); - await textarea.press('Enter'); + const textarea = page.locator("#message-input"); + await textarea.fill("resilience probe — e2e-composer-retry"); + await textarea.press("Enter"); // Visible failure, not a silent drop: the error bubble and a // usable Retry control. - await expect(page.getByText('Something went wrong')).toBeVisible({ + await expect(page.getByText("Something went wrong")).toBeVisible({ timeout: 15_000, }); - const retryButton = page.getByRole('button', { name: /retry/i }); + const retryButton = page.getByRole("button", { name: /retry/i }); await expect(retryButton).toBeEnabled(); // Network heals; Retry must complete the SAME turn. - await page.unroute('**/stream'); + await page.unroute("**/stream"); const streamDone = page.waitForResponse( - (r) => r.url().includes('/stream') && r.request().method() === 'POST', + (r) => r.url().includes("/stream") && r.request().method() === "POST", { timeout: 45_000 }, ); await retryButton.click(); const streamRes = await streamDone; expect(streamRes.status()).toBe(200); - await expect(page.getByText('Something went wrong')).toBeHidden({ + await expect(page.getByText("Something went wrong")).toBeHidden({ timeout: 15_000, }); @@ -95,14 +95,14 @@ test.describe('tier-a · composer resilience', () => { [sub], ); expect(rows).toHaveLength(1); - expect(rows[0].prompt).toBe('resilience probe — e2e-composer-retry'); + expect(rows[0].prompt).toBe("resilience probe — e2e-composer-retry"); expect(rows[0].response).toBeTruthy(); } finally { await context.close(); } }); - test('send while a file is still processing arms, waits, and flushes with the file bound to the turn', async ({ + test("send while a file is still processing arms, waits, and flushes with the file bound to the turn", async ({ browser, }) => { const { context, sub } = await newUserContext(browser); @@ -112,15 +112,15 @@ test.describe('tier-a · composer resilience', () => { // Track /stream POSTs so we can prove the send was HELD. const streamPosts: string[] = []; - page.on('request', (req) => { - if (req.url().includes('/stream') && req.method() === 'POST') { - streamPosts.push(req.postData() ?? ''); + page.on("request", (req) => { + if (req.url().includes("/stream") && req.method() === "POST") { + streamPosts.push(req.postData() ?? ""); } }); // Hold the upload response long enough to press Enter while the // attachment is still visibly 'uploading'. - await page.route('**/api/store_attachment', async (route) => { + await page.route("**/api/store_attachment", async (route) => { await new Promise((r) => setTimeout(r, 2_000)); await route.continue(); }); @@ -132,9 +132,9 @@ test.describe('tier-a · composer resilience', () => { .first() .setInputFiles(SMALL_FIXTURE_PATH); - const textarea = page.locator('#message-input'); - await textarea.fill('what is in the attached file? e2e-armed-send'); - await textarea.press('Enter'); + const textarea = page.locator("#message-input"); + await textarea.fill("what is in the attached file? e2e-armed-send"); + await textarea.press("Enter"); // The send is armed, not fired: banner visible, no /stream yet, // and the prompt is still in the composer. @@ -143,13 +143,13 @@ test.describe('tier-a · composer resilience', () => { ).toBeVisible({ timeout: 5_000 }); expect(streamPosts).toHaveLength(0); await expect(textarea).toHaveValue( - 'what is in the attached file? e2e-armed-send', + "what is in the attached file? e2e-armed-send", ); // Upload completes → Celery parses → attachment.completed lands via // SSE → the armed send auto-flushes. notes.txt parses in seconds. const streamDone = page.waitForResponse( - (r) => r.url().includes('/stream') && r.request().method() === 'POST', + (r) => r.url().includes("/stream") && r.request().method() === "POST", { timeout: 60_000 }, ); const streamRes = await streamDone; @@ -163,10 +163,10 @@ test.describe('tier-a · composer resilience', () => { expect(payload.attachments).toHaveLength(1); // Banner cleared, composer emptied by the flush. - await expect( - page.getByText(/will send when/i), - ).toBeHidden({ timeout: 15_000 }); - await expect(textarea).toHaveValue(''); + await expect(page.getByText(/will send when/i)).toBeHidden({ + timeout: 15_000, + }); + await expect(textarea).toHaveValue(""); // The invariant that motivated all of this: the file is BOUND to // the turn — conversation_messages.attachments[] holds one PG PK @@ -187,7 +187,7 @@ test.describe('tier-a · composer resilience', () => { ); expect(rows).toHaveLength(1); expect(rows[0].prompt).toBe( - 'what is in the attached file? e2e-armed-send', + "what is in the attached file? e2e-armed-send", ); expect(Number(rows[0].attachment_count)).toBe(1); expect(Number(rows[0].resolved)).toBe(1); @@ -195,4 +195,55 @@ test.describe('tier-a · composer resilience', () => { await context.close(); } }); + + test("hero suggestion card while a file is processing routes into the armed composer instead of sending without it", async ({ + browser, + }) => { + const { context } = await newUserContext(browser); + try { + const page = await context.newPage(); + await openComposer(page); + + const streamPosts: string[] = []; + page.on("request", (req) => { + if (req.url().includes("/stream") && req.method() === "POST") { + streamPosts.push(req.postData() ?? ""); + } + }); + + await page.route("**/api/store_attachment", async (route) => { + await new Promise((r) => setTimeout(r, 2_000)); + await route.continue(); + }); + await page + .locator('input[type="file"]') + .first() + .setInputFiles(SMALL_FIXTURE_PATH); + + // The hero cards call handleQuestion directly — the historical + // bypass around the composer's arming gate. + await page.getByText("Learn about DocsGPT").click(); + + // Not sent: the question landed in the composer, armed. + await expect( + page.getByText(/will send when 1 file finishes processing/i), + ).toBeVisible({ timeout: 5_000 }); + expect(streamPosts).toHaveLength(0); + await expect(page.locator("#message-input")).toHaveValue( + /what is docsgpt/i, + ); + + // Once the file resolves, the flush carries it on the turn. + const streamRes = await page.waitForResponse( + (r) => r.url().includes("/stream") && r.request().method() === "POST", + { timeout: 60_000 }, + ); + expect(streamRes.status()).toBe(200); + expect(streamPosts).toHaveLength(1); + const payload = JSON.parse(streamPosts[0]) as { attachments?: string[] }; + expect(payload.attachments).toHaveLength(1); + } finally { + await context.close(); + } + }); });