feat: better attachments pre send guard

This commit is contained in:
Alex committed 2026-08-13 17:19:39 +01:00
1 parent 1430037a52
commit 49f89a67c2
6 files changed
+217 -57

No files matched your search

+17 -10
View File
@@ -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({
})}
</div>
)}
{!sendArmed && completedAttachmentCount > 0 && (
<div className="text-muted-foreground px-2 pb-1 text-xs sm:px-3">
{t('conversation.attachments.attachedCount', {
count: completedAttachmentCount,
})}
</div>
)}
{voiceError && (
<div className="px-2 pb-1 text-xs text-[#B42318] sm:px-3">
{voiceError}
+15 -1
View File
@@ -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<string | null>(null);
const [lastQueryReturnedErr, setLastQueryReturnedErr] =
useState<boolean>(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}
+1 -3
View File
@@ -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",
+69
View File
@@ -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);
});
});
+26 -5
View File
@@ -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<string>) => {
@@ -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<void> {
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();
}
});
});