diff --git a/apps/web/src/components/ChatView.browser.test.tsx b/apps/web/src/components/ChatView.browser.test.tsx index 5bb8af3ae46d0..a0458f9e80280 100644 --- a/apps/web/src/components/ChatView.browser.test.tsx +++ b/apps/web/src/components/ChatView.browser.test.tsx @@ -60,6 +60,10 @@ function acceptAttachmentUploads(): Disposable { const mint = answerRpc(WS_METHODS.attachmentsCreateUploadUrl, () => Effect.succeed({ attachmentId: "upload-1", relativeUrl: "/upload", expiresAt: 0 }), ); + // Sending verifies each ready upload still exists on the server. + const verify = answerRpc(WS_METHODS.assetsCreateUrl, () => + Effect.succeed({ relativeUrl: "/upload-1", expiresAt: 0 }), + ); const status = vi.spyOn(XMLHttpRequest.prototype, "status", "get").mockReturnValue(200); const send = vi.spyOn(XMLHttpRequest.prototype, "send").mockImplementation(function ( this: XMLHttpRequest, @@ -71,6 +75,7 @@ function acceptAttachmentUploads(): Disposable { send.mockRestore(); status.mockRestore(); mint[Symbol.dispose](); + verify[Symbol.dispose](); }, }; } diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts index c5c65e7d6e783..58f12549d15f3 100644 --- a/apps/web/src/components/chat/chatViewSend.ts +++ b/apps/web/src/components/chat/chatViewSend.ts @@ -16,7 +16,10 @@ import { type AtomCommandResult, } from "@t3tools/client-runtime/state/runtime"; import { readLocalApi } from "../../localApi"; -import { awaitAttachmentUploads, getUploadedAttachments } from "../../lib/attachmentUploadQueue"; +import { + getUploadedAttachments, + refreshExpiredAttachmentUploads, +} from "../../lib/attachmentUploadQueue"; import { findRevertCheckpoint } from "../../session-logic"; import { type ChatMessage } from "../../types"; import { randomUUID } from "~/lib/utils"; @@ -25,6 +28,7 @@ import { newThreadId } from "~/lib/utils"; import { type ComposerFileAttachment, type ComposerImageAttachment, + type ComposerThreadTarget, useComposerDraftStore, DraftId, } from "../../composerDraftStore"; @@ -328,12 +332,17 @@ export function prepareTurnAttachments(input: { readonly environmentId: EnvironmentId; readonly attachments: ReadonlyArray; readonly usesAttachmentUploads: boolean; + readonly draftTarget: ComposerThreadTarget; }) { - const { environmentId, usesAttachmentUploads } = input; + const { environmentId, usesAttachmentUploads, draftTarget } = input; return Promise.all( input.attachments.map(async (attachment) => { if (usesAttachmentUploads) { - await awaitAttachmentUploads([attachment.id]); + await refreshExpiredAttachmentUploads({ + environmentId, + images: [attachment], + draftTarget, + }); const uploaded = getUploadedAttachments({ environmentId, images: [attachment] })?.[0]; if (!uploaded) { throw new Error(`Attachment '${attachment.name}' failed to upload.`); diff --git a/apps/web/src/components/chat/composerSend.logic.test.ts b/apps/web/src/components/chat/composerSend.logic.test.ts index 13e36ea5659c9..dea00844d6a50 100644 --- a/apps/web/src/components/chat/composerSend.logic.test.ts +++ b/apps/web/src/components/chat/composerSend.logic.test.ts @@ -1,11 +1,13 @@ import { EnvironmentId, ThreadId } from "@t3tools/contracts"; import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import { describe, expect, it } from "vite-plus/test"; +import type { ComposerThreadDraftState } from "../../composerDraftStore"; import { formatInlineContextReference } from "../../lib/composerContextReferences"; import { threadContextReference } from "../../lib/composerContextRecords"; import { threadContextRecord } from "@t3tools/shared/composerContextReferences"; import { approvalRedirectNotice, + composerDraftContentChanged, loadBalancingSendWarning, multipleModelSendBlock, referencedThreadContexts, @@ -16,6 +18,46 @@ const remote = EnvironmentId.make("remote"); const localThread = threadContextRecord(scopeThreadRef(local, ThreadId.make("a")), "Local"); const remoteThread = threadContextRecord(scopeThreadRef(remote, ThreadId.make("b")), "Remote"); +it("detects draft edits during upload preparation without treating upload metadata as an edit", () => { + const draft: ComposerThreadDraftState = { + prompt: "before", + images: [], + files: [ + { + type: "file", + id: "file", + name: "a.pdf", + mimeType: "application/pdf", + sizeBytes: 1, + file: null, + }, + ], + nonPersistedImageIds: [], + persistedAttachments: [], + terminalContexts: [], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + environmentContexts: [], + replyTo: null, + modelSelectionByProvider: {}, + activeProvider: null, + runtimeMode: null, + interactionMode: null, + }; + expect( + composerDraftContentChanged(draft, { + ...draft, + files: [{ ...draft.files[0]!, uploadedAttachmentId: "upload", uploadEnvironmentId: local }], + }), + ).toBe(false); + expect(composerDraftContentChanged(draft, { ...draft, prompt: "after" })).toBe(true); + expect(composerDraftContentChanged(draft, { ...draft, files: [] })).toBe(true); + expect(composerDraftContentChanged(draft, { ...draft, threadContexts: [localThread] })).toBe( + true, + ); +}); + describe("referencedThreadContexts", () => { it("keeps only the attached threads the prompt still links", () => { const prompt = `see ${formatInlineContextReference(threadContextReference(remoteThread))}`; diff --git a/apps/web/src/components/chat/composerSend.logic.ts b/apps/web/src/components/chat/composerSend.logic.ts index e19512e6c90d4..5b7a5ac3a09c7 100644 --- a/apps/web/src/components/chat/composerSend.logic.ts +++ b/apps/web/src/components/chat/composerSend.logic.ts @@ -1,5 +1,28 @@ import type { EnvironmentId, ThreadContextRecord } from "@t3tools/contracts"; import { collectInlineContextIds } from "../../lib/composerContextReferences"; +import type { ComposerThreadDraftState } from "../../composerDraftStore"; + +/** Compare immutable user content without copying image bytes or upload metadata. */ +export function composerDraftContentChanged( + before: ComposerThreadDraftState | null, + after: ComposerThreadDraftState | null, +): boolean { + if (before === after) return false; + if (!before || !after) return true; + const sameIds = (left: ReadonlyArray<{ id: string }>, right: ReadonlyArray<{ id: string }>) => + left.length === right.length && left.every((item, index) => item.id === right[index]?.id); + return ( + before.prompt !== after.prompt || + !sameIds(before.images, after.images) || + !sameIds(before.files, after.files) || + before.terminalContexts !== after.terminalContexts || + before.previewAnnotations !== after.previewAnnotations || + before.reviewComments !== after.reviewComments || + before.threadContexts !== after.threadContexts || + before.environmentContexts !== after.environmentContexts || + before.replyTo !== after.replyTo + ); +} /** Attached threads the prompt still links to; replacing linked text leaves stale records. */ export function referencedThreadContexts(draft: { diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 80555864284a6..1335bbdde2d14 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -99,6 +99,7 @@ import { useStableCallback } from "../../hooks/useStableCallback"; import { environmentPresentations } from "../../state/presentation"; import { approvalRedirectNotice, + composerDraftContentChanged, loadBalancingSendWarning, multipleModelSendBlock, referencedThreadContexts, @@ -106,8 +107,8 @@ import { import { useMultipleModelSend } from "./useMultipleModelSend"; import { type ComposerDispatchMode } from "@t3tools/client-runtime/state/composer-dispatch"; import { - awaitAttachmentUploads, getUploadedAttachments, + refreshExpiredAttachmentUploads, readAttachmentUpload, releaseDraftAttachments, retryAttachmentUpload, @@ -872,6 +873,14 @@ export function useComposerSend(input: ComposerSendInput) { multipleTargets.push(resolved); } + const draftBeforePreparation = useComposerDraftStore + .getState() + .getComposerDraft(composerDraftTarget); + const draftChangedDuringPreparation = () => { + if (composerRef.current?.hasPendingAttachments() === true) return true; + const draft = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); + return composerDraftContentChanged(draftBeforePreparation, draft); + }; sendInFlightRef.current = true; const sendGeneration = ++composerSendGenerationRef.current; const attachmentCapabilitiesBeforeUpload = readLiveAttachmentCapabilities(); @@ -919,6 +928,17 @@ export function useComposerSend(input: ComposerSendInput) { } const attachmentCapabilitiesBeforeDispatch = readLiveAttachmentCapabilities(); + if (draftChangedDuringPreparation()) { + sendInFlightRef.current = false; + setThreadError( + threadIdForSend, + "Draft changed while preparing this message. Review it and send again.", + ); + setDockedDraftHeroThreadKey((currentThreadKey) => + currentThreadKey === activeThreadKey ? null : currentThreadKey, + ); + return; + } if ( !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || attachmentCapabilitiesBeforeDispatch.fileBlockReason !== null @@ -940,7 +960,22 @@ export function useComposerSend(input: ComposerSendInput) { composerAttachmentsSnapshot.length > 0 && (sendCtx?.shownPendingApprovalIds.length ?? 0) > 0 ) { - await awaitAttachmentUploads(composerAttachmentsSnapshot.map((attachment) => attachment.id)); + await refreshExpiredAttachmentUploads({ + environmentId, + images: composerAttachmentsSnapshot, + draftTarget: composerDraftTarget, + }); + if (draftChangedDuringPreparation()) { + sendInFlightRef.current = false; + setThreadError( + threadIdForSend, + "Draft changed while preparing this message. Review it and send again.", + ); + setDockedDraftHeroThreadKey((currentThreadKey) => + currentThreadKey === activeThreadKey ? null : currentThreadKey, + ); + return; + } const uploadBlockReason = readLiveAttachmentCapabilities().fileBlockReason ?? (getUploadedAttachments({ environmentId, images: composerAttachmentsSnapshot }) === null @@ -1103,6 +1138,7 @@ export function useComposerSend(input: ComposerSendInput) { environmentId, attachments: composerAttachmentsSnapshot, usesAttachmentUploads: turnUsesAttachmentUploads, + draftTarget: composerDraftTarget, }); if (multipleModelSelections !== null) { await sendToMultipleModels({ @@ -1118,6 +1154,7 @@ export function useComposerSend(input: ComposerSendInput) { readFileBlockReason: () => readLiveAttachmentCapabilities().fileBlockReason, buildOutgoingMessageContext, isCurrentSend: () => composerSendGenerationRef.current === sendGeneration, + draftChangedDuringPreparation, }); return; } diff --git a/apps/web/src/components/chat/useMultipleModelSend.ts b/apps/web/src/components/chat/useMultipleModelSend.ts index ccea2378968f5..ccf3b32621ce2 100644 --- a/apps/web/src/components/chat/useMultipleModelSend.ts +++ b/apps/web/src/components/chat/useMultipleModelSend.ts @@ -97,6 +97,7 @@ export function useMultipleModelSend(input: MultipleModelSendInput) { readFileBlockReason, buildOutgoingMessageContext, isCurrentSend, + draftChangedDuringPreparation, }: { readonly threadId: ThreadId; readonly project: EnvironmentProject; @@ -113,6 +114,7 @@ export function useMultipleModelSend(input: MultipleModelSendInput) { ) => ReturnType; /** False once a newer send from this composer started. */ readonly isCurrentSend: () => boolean; + readonly draftChangedDuringPreparation: () => boolean; }) => { const { attachments: attachmentsSnapshot, messageText } = sendSnapshot; const failedSelections: ModelSelection[] = []; @@ -124,6 +126,8 @@ export function useMultipleModelSend(input: MultipleModelSendInput) { // fails every selection; the draft restore below still runs. const sendResult = await settlePromise(async () => { const attachments = await attachmentsPromise; + if (draftChangedDuringPreparation()) + throw new Error("Draft changed while preparing this message. Review it and send again."); const fileBlockReason = readFileBlockReason(); if (fileBlockReason !== null) throw new Error(fileBlockReason); const context = buildOutgoingMessageContext( diff --git a/apps/web/src/components/chat/usePendingThreadRequests.ts b/apps/web/src/components/chat/usePendingThreadRequests.ts index 14c7242821076..8c32b07a29271 100644 --- a/apps/web/src/components/chat/usePendingThreadRequests.ts +++ b/apps/web/src/components/chat/usePendingThreadRequests.ts @@ -25,7 +25,12 @@ import { useComposerDraftStore, } from "../../composerDraftStore"; import { useStableCallback } from "../../hooks/useStableCallback"; -import { getUploadedAttachments, useAttachmentUploadStore } from "../../lib/attachmentUploadQueue"; +import { + getUploadedAttachments, + refreshExpiredAttachmentUploads, + useAttachmentUploadStore, +} from "../../lib/attachmentUploadQueue"; +import { composerDraftContentChanged } from "./composerSend.logic"; import { buildPendingUserInputAnswers, setPendingUserInputCustomAnswer, @@ -390,60 +395,102 @@ export function usePendingThreadRequests(input: { if (!pendingInput || pendingInput.responseCapability === "not_resumable") return; const responseKey = JSON.stringify([environmentId, activeThreadId, requestId]); if (userInputResponsesInFlight.current.has(responseKey)) return; - const attachmentsByQuestionId = new Map(); - const providerAnswers = { ...answers }; - const threadDraft = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); - for (const question of pendingInput.questions) { - const target = questionAttachmentDraftId( - environmentId, - activeThreadId, - requestId, - question.id, - ); - if ((useQuestionAttachmentPreparation.getState().counts[target] ?? 0) > 0) return; - const draft = useComposerDraftStore.getState().getComposerDraft(target); - providerAnswers[question.id] = projectPendingUserInputAnswer( - answers[question.id], - answerContextRecords(draft, threadDraft), - ); - const attachments = draft ? [...draft.images, ...draft.files] : []; - if (attachments.length === 0) continue; - const uploaded = getUploadedAttachments({ environmentId, images: attachments }); - if (!uploaded) { + userInputResponsesInFlight.current.add(responseKey); + const answerDraftKey = userInputDraftKey(environmentId, activeThreadId, requestId); + const answerDraftBeforePreparation = usePendingUserInputAnswers.getState()[answerDraftKey]; + setRespondingUserInputRequestIds((existing) => + existing.includes(requestId) ? existing : [...existing, requestId], + ); + const prepareAndRespond = async () => { + const attachmentsByQuestionId = new Map(); + const providerAnswers = { ...answers }; + const threadDraft = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); + const questionDrafts = pendingInput.questions.map((question) => { + const target = questionAttachmentDraftId( + environmentId, + activeThreadId, + requestId, + question.id, + ); + const draft = useComposerDraftStore.getState().getComposerDraft(target); + return { question, target, draft }; + }); + for (const { question, target, draft } of questionDrafts) { + if ((useQuestionAttachmentPreparation.getState().counts[target] ?? 0) > 0) return; + providerAnswers[question.id] = projectPendingUserInputAnswer( + answers[question.id], + answerContextRecords(draft, threadDraft), + ); + const attachments = draft ? [...draft.images, ...draft.files] : []; + if (attachments.length === 0) continue; + await refreshExpiredAttachmentUploads({ + environmentId, + images: attachments, + draftTarget: target, + }); + if ( + composerDraftContentChanged( + draft, + useComposerDraftStore.getState().getComposerDraft(target), + ) + ) { + setThreadError( + activeThreadId, + "Answer changed while preparing attachments. Review it and send again.", + ); + return; + } + const uploaded = getUploadedAttachments({ environmentId, images: attachments }); + if (!uploaded) { + setThreadError( + activeThreadId, + "Wait for attachments to finish uploading, or remove failed uploads.", + ); + return; + } + attachmentsByQuestionId.set(question.id, uploaded as UserInputAttachments[string]); + } + if ( + usePendingUserInputAnswers.getState()[answerDraftKey] !== answerDraftBeforePreparation || + questionDrafts.some( + ({ target, draft }) => + (useQuestionAttachmentPreparation.getState().counts[target] ?? 0) > 0 || + composerDraftContentChanged( + draft, + useComposerDraftStore.getState().getComposerDraft(target), + ), + ) + ) { setThreadError( activeThreadId, - "Wait for attachments to finish uploading, or remove failed uploads.", + "Answer changed while preparing attachments. Review it and send again.", ); return; } - attachmentsByQuestionId.set(question.id, uploaded as UserInputAttachments[string]); - } - userInputResponsesInFlight.current.add(responseKey); - - setRespondingUserInputRequestIds((existing) => - existing.includes(requestId) ? existing : [...existing, requestId], - ); - const result = await respondToThreadUserInput({ - environmentId, - input: { - threadId: activeThreadId, - requestId, - answers: providerAnswers, - ...(attachmentsByQuestionId.size > 0 - ? { attachmentsByQuestionId: Object.fromEntries(attachmentsByQuestionId) } - : {}), - }, + const result = await respondToThreadUserInput({ + environmentId, + input: { + threadId: activeThreadId, + requestId, + answers: providerAnswers, + ...(attachmentsByQuestionId.size > 0 + ? { attachmentsByQuestionId: Object.fromEntries(attachmentsByQuestionId) } + : {}), + }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + activeThreadId, + error instanceof Error ? error.message : "Failed to submit user input.", + ); + } + return result; + }; + return prepareAndRespond().finally(() => { + userInputResponsesInFlight.current.delete(responseKey); + setRespondingUserInputRequestIds((existing) => existing.filter((id) => id !== requestId)); }); - if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { - const error = squashAtomCommandFailure(result); - setThreadError( - activeThreadId, - error instanceof Error ? error.message : "Failed to submit user input.", - ); - } - userInputResponsesInFlight.current.delete(responseKey); - setRespondingUserInputRequestIds((existing) => existing.filter((id) => id !== requestId)); - return result; }; const onDismissUserInput = async (requestId: RuntimeRequestId) => { diff --git a/apps/web/src/components/chat/useQueuedRunEdit.ts b/apps/web/src/components/chat/useQueuedRunEdit.ts index e7f321bec9dc2..d5cb8505a013e 100644 --- a/apps/web/src/components/chat/useQueuedRunEdit.ts +++ b/apps/web/src/components/chat/useQueuedRunEdit.ts @@ -38,6 +38,7 @@ import { useStableCallback } from "../../hooks/useStableCallback"; import { awaitAttachmentUploads, getUploadedAttachments, + refreshExpiredAttachmentUploads, releaseDraftAttachments, startAttachmentUpload, } from "../../lib/attachmentUploadQueue"; @@ -60,6 +61,7 @@ import { useAtomCommand } from "../../state/use-atom-command"; import { stackedThreadToast, toastManager } from "../ui/toast"; import type { ChatComposerHandle } from "./ChatComposer"; import { fileAttachmentCapabilityBlockReason } from "./composerAttachmentFiles"; +import { composerDraftContentChanged } from "./composerSend.logic"; import { discardQueuedEditDraft, prepareQueuedEditAttachments, @@ -105,6 +107,7 @@ async function submitQueuedRunEdit(input: { readonly editing: EditingQueuedRun; readonly content: QueuedRunEditContent; readonly editQueuedRun: ReturnType; + readonly draftChanged: () => boolean; }): Promise { const { environmentId, draftTarget, editing, content } = input; const editText = content.prompt.trim(); @@ -124,6 +127,7 @@ async function submitQueuedRunEdit(input: { validateFiles(); for (const file of files) startAttachmentUpload({ environmentId, image: file, draftTarget }); await awaitAttachmentUploads(files.map((file) => file.id)); + await refreshExpiredAttachmentUploads({ environmentId, images: files, draftTarget }); validateFiles(); const uploaded = getUploadedAttachments({ environmentId, images: files }); if (uploaded === null) throw new Error("Retry or remove failed uploads before saving."); @@ -137,6 +141,8 @@ async function submitQueuedRunEdit(input: { readImage: readFileAsDataUrl, uploadFiles, }); + if (input.draftChanged()) + return "Draft changed while preparing this message. Review it and save again."; // The edit draft owns thread and machine mentions, so a chip removed while editing is dropped. const keptRecords = (editing.context?.records ?? []).filter( (record) => @@ -363,15 +369,29 @@ export function useQueuedRunEdit(input: { // The save's result decides the edit's fate, even if the run starts meanwhile. const finishSave = queuedRunEdits.startSave(editKey); if (finishSave === null) return; + const draftBeforeSave = useComposerDraftStore.getState().getComposerDraft(draftTarget); + const draftChanged = () => + appAtomRegistry.get(queuedRunEdits.atom)[editKey] !== editing || + composerDraftContentChanged( + draftBeforeSave, + useComposerDraftStore.getState().getComposerDraft(draftTarget), + ); const error = await submitQueuedRunEdit({ environmentId, draftTarget, editing, content, editQueuedRun: editQueuedRunCommand, + draftChanged, }); - setThreadError(editing.threadId, error); - const current = finishSave(error === null); + const changedAfterSave = error === null && draftChanged(); + setThreadError( + editing.threadId, + changedAfterSave + ? "The earlier edit was saved. Review your newer changes and save again." + : error, + ); + const current = finishSave(error === null && !changedAfterSave); if (!current) { // A failed save can recover into the thread draft; a newer edit can also retain resources. const drafts = Object.values(useComposerDraftStore.getState().draftsByThreadKey); @@ -394,6 +414,7 @@ export function useQueuedRunEdit(input: { return; } if (error !== null) return; + if (changedAfterSave) return; releaseDraftAttachments(newAttachments); promptRef.current = ""; clearEditDraft(editing.runId); diff --git a/apps/web/src/lib/attachmentUploadQueue.test.ts b/apps/web/src/lib/attachmentUploadQueue.test.ts index 1b9902336e027..a9483e1755e35 100644 --- a/apps/web/src/lib/attachmentUploadQueue.test.ts +++ b/apps/web/src/lib/attachmentUploadQueue.test.ts @@ -57,6 +57,7 @@ import { awaitAttachmentUploads, getUploadedAttachments, readAttachmentUpload, + refreshExpiredAttachmentUploads, releaseAttachmentUpload, releaseDraftAttachment, releaseDraftAttachments, @@ -210,6 +211,180 @@ describe("attachmentUploadQueue", () => { vi.unstubAllGlobals(); }); + it("reuploads an expired image while retaining other ready images", async () => { + const images = [makeImage("first"), makeImage("second")].map((image) => ({ + ...image, + name: "image.png", + })); + let minted = 0; + mocks.runAtomCommand.mockImplementation(async (_registry, command) => { + if (command !== mocks.createUploadUrl) return { _tag: "Success", value: undefined }; + const attachmentId = `pending-test-${++minted}`; + return { + _tag: "Success", + value: { + attachmentId, + relativeUrl: `/api/attachments/upload/${attachmentId}`, + expiresAt: 1, + }, + }; + }); + for (const image of images) startAttachmentUpload({ environmentId: firstEnvironment, image }); + await Promise.resolve(); + const initial = awaitAttachmentUploads(images.map((image) => image.id)); + TestXmlHttpRequest.requests.forEach((request) => request.complete()); + await initial; + mocks.executeAtomQuery.mockImplementation( + async (_registry, target: { input: { resource: { attachmentId: string } } }) => + target.input.resource.attachmentId === "pending-test-1" + ? { _tag: "Failure", error: { _tag: "AssetAttachmentNotFoundError" } } + : { _tag: "Success", value: {} }, + ); + const refreshing = refreshExpiredAttachmentUploads({ environmentId: firstEnvironment, images }); + await vi.waitFor(() => expect(TestXmlHttpRequest.requests).toHaveLength(3)); + TestXmlHttpRequest.requests[2]!.complete(); + await refreshing; + expect( + getUploadedAttachments({ environmentId: firstEnvironment, images })?.map((image) => image.id), + ).toEqual(["pending-test-3", "pending-test-2"]); + expect(TestXmlHttpRequest.requests).toHaveLength(3); + }); + + it("reuploads local file bytes after its persisted upload is known to be missing", async () => { + const file = { + ...makeFile("expired-saved-file"), + uploadedAttachmentId: "pending-expired", + uploadEnvironmentId: firstEnvironment, + }; + useAttachmentUploadStore.setState({ + uploadsByImageId: { + [file.id]: { + status: "ready", + environmentId: firstEnvironment, + attachmentId: "pending-expired", + }, + }, + }); + mocks.executeAtomQuery + .mockResolvedValueOnce({ + _tag: "Failure", + error: { _tag: "AssetAttachmentNotFoundError" }, + }) + .mockResolvedValue({ _tag: "Failure", error: new Error("Connection interrupted") }); + let uploadStarted!: () => void; + const started = new Promise((resolve) => { + uploadStarted = resolve; + }); + const runCommand = mocks.runAtomCommand.getMockImplementation()!; + mocks.runAtomCommand.mockImplementation(async (...args) => { + const result = await runCommand(...args); + if (args[1] === mocks.createUploadUrl) uploadStarted(); + return result; + }); + const refresh = refreshExpiredAttachmentUploads({ + environmentId: firstEnvironment, + images: [file], + }); + expect(await Promise.race([started.then(() => "upload"), refresh.then(() => "settled")])).toBe( + "upload", + ); + expect(TestXmlHttpRequest.requests).toHaveLength(1); + TestXmlHttpRequest.requests[0]!.complete(); + await refresh; + expect(mocks.executeAtomQuery).toHaveBeenCalledTimes(1); + expect( + getUploadedAttachments({ environmentId: firstEnvironment, images: [file] }), + ).toMatchObject([{ id: "pending-environment-1-expired-saved-file.pdf" }]); + }); + + it.each([false, true])( + "retains cleanup identity after verification fails (retry: %s)", + async (retry) => { + const image = makeImage("connection-failure"); + useAttachmentUploadStore.setState({ + uploadsByImageId: { + [image.id]: { + status: "ready", + environmentId: firstEnvironment, + attachmentId: "pending-original", + }, + }, + }); + mocks.executeAtomQuery.mockRejectedValueOnce(new Error("offline")); + await refreshExpiredAttachmentUploads({ environmentId: firstEnvironment, images: [image] }); + expect(readAttachmentUpload(image.id)).toMatchObject({ status: "failed" }); + expect( + getUploadedAttachments({ environmentId: firstEnvironment, images: [image] }), + ).toBeNull(); + expect(TestXmlHttpRequest.requests).toHaveLength(0); + if (retry) retryAttachmentUpload({ environmentId: firstEnvironment, image }); + releaseAttachmentUpload(image.id); + expect(readAttachmentUpload(image.id)).toBeUndefined(); + expect(mocks.runAtomCommand).toHaveBeenCalledExactlyOnceWith( + expect.anything(), + mocks.removeUpload, + { environmentId: firstEnvironment, input: { attachmentId: "pending-original" } }, + expect.anything(), + ); + }, + ); + + it("reuploads local bytes for a connection without read permission", async () => { + const image = makeImage("operate-only"); + useAttachmentUploadStore.setState({ + uploadsByImageId: { + [image.id]: { + status: "ready", + environmentId: firstEnvironment, + attachmentId: "pending-original", + }, + }, + }); + mocks.readEnvironmentScope.mockImplementation( + (_environmentId, scope) => scope !== "orchestration:read", + ); + const refresh = refreshExpiredAttachmentUploads({ + environmentId: firstEnvironment, + images: [image], + }); + await vi.waitFor(() => expect(TestXmlHttpRequest.requests).toHaveLength(1)); + TestXmlHttpRequest.requests[0]!.complete(); + await refresh; + expect(readAttachmentUpload(image.id)).toMatchObject({ + status: "ready", + attachmentId: "pending-environment-1-operate-only.png", + }); + expect(mocks.executeAtomQuery).not.toHaveBeenCalled(); + }); + + it("does not resurrect an image removed while its upload is being checked", async () => { + const image = makeImage("removed-during-verification"); + useAttachmentUploadStore.setState({ + uploadsByImageId: { + [image.id]: { + status: "ready", + environmentId: firstEnvironment, + attachmentId: "pending-original", + }, + }, + }); + let finish!: (value: unknown) => void; + mocks.executeAtomQuery.mockReturnValueOnce( + new Promise((resolve) => { + finish = resolve; + }), + ); + const refresh = refreshExpiredAttachmentUploads({ + environmentId: firstEnvironment, + images: [image], + }); + releaseAttachmentUpload(image.id); + finish({ _tag: "Failure", error: { _tag: "AssetAttachmentNotFoundError" } }); + await refresh; + expect(readAttachmentUpload(image.id)).toBeUndefined(); + expect(TestXmlHttpRequest.requests).toHaveLength(0); + }); + it("keeps an attachment local until its own environment grants write access", async () => { const image = makeImage("permission-gain"); mocks.readEnvironmentScope.mockImplementation( diff --git a/apps/web/src/lib/attachmentUploadQueue.ts b/apps/web/src/lib/attachmentUploadQueue.ts index 73458849164cd..cf9bbedeb853a 100644 --- a/apps/web/src/lib/attachmentUploadQueue.ts +++ b/apps/web/src/lib/attachmentUploadQueue.ts @@ -1,5 +1,6 @@ import { AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES, type ChatAttachment, type EnvironmentId, @@ -395,6 +396,7 @@ export function startAttachmentUpload(input: { readonly environmentId: EnvironmentId; readonly image: ComposerImageAttachment | ComposerFileAttachment; readonly draftTarget?: ComposerThreadTarget; + readonly force?: boolean; }): void { if (!readEnvironmentScope(input.environmentId, AuthOrchestrationOperateScope)) return; const existingJob = jobsByImageId.get(input.image.id); @@ -403,7 +405,11 @@ export function startAttachmentUpload(input: { } const existing = readAttachmentUpload(input.image.id); - if (existing?.status === "ready" && existing.environmentId === input.environmentId) { + if ( + !input.force && + existing?.status === "ready" && + existing.environmentId === input.environmentId + ) { // A file restored into a draft after its upload finished (stash, failed send) missed the // completion stamp; stamping here keeps it from re-uploading after a reload. stampDraftFileUpload(input, existing.attachmentId); @@ -440,7 +446,8 @@ export function startAttachmentUpload(input: { environmentId: input.environmentId, ...(input.draftTarget !== undefined ? { draftTarget: input.draftTarget } : {}), ...(previous ? { previous } : {}), - ...(input.image.type === "file" && + ...(!input.force && + input.image.type === "file" && input.image.uploadEnvironmentId === input.environmentId && input.image.uploadedAttachmentId ? { persistedAttachmentId: input.image.uploadedAttachmentId } @@ -582,6 +589,66 @@ export async function awaitAttachmentUploads(imageIds: ReadonlyArray): P await Promise.all(imageIds.map((imageId) => jobsByImageId.get(imageId)?.settled)); } +/** Revalidate ready uploads at send time; a long-lived draft can outlive server cleanup. */ +export async function refreshExpiredAttachmentUploads(input: { + readonly environmentId: EnvironmentId; + readonly images: ReadonlyArray; + readonly draftTarget?: ComposerThreadTarget; +}): Promise { + const canVerify = readEnvironmentScope(input.environmentId, AuthOrchestrationReadScope); + await Promise.all( + input.images.map(async (image) => { + const upload = readAttachmentUpload(image.id); + if (upload?.status !== "ready" || upload.environmentId !== input.environmentId) return; + if (!canVerify) { + if (image.file) { + startAttachmentUpload({ + environmentId: input.environmentId, + image, + force: true, + ...(input.draftTarget !== undefined ? { draftTarget: input.draftTarget } : {}), + }); + } else { + setUploadState(image.id, { + status: "failed", + environmentId: input.environmentId, + previous: upload, + reason: "Uploaded attachment could not be verified. Attach the file again.", + }); + } + return; + } + const verification = await verifyStashedAttachmentUpload({ + environmentId: input.environmentId, + attachmentId: upload.attachmentId, + }).catch((error: unknown): PersistedAttachmentVerification => ({ status: "failed", error })); + // The user may have removed or replaced the attachment while verification was in flight. + if (readAttachmentUpload(image.id) !== upload) return; + if (verification.status === "verified") return; + if (verification.status === "missing" && image.file) { + clearUploadState(image.id); + startAttachmentUpload({ + environmentId: input.environmentId, + image, + force: true, + ...(input.draftTarget !== undefined ? { draftTarget: input.draftTarget } : {}), + }); + return; + } + setUploadState(image.id, { + status: "failed", + environmentId: input.environmentId, + previous: upload, + reason: + verification.status === "missing" + ? "Uploaded attachment expired. Remove it and attach it again." + : "Uploaded attachment could not be verified. Retry when the server reconnects.", + }); + }), + ); + await awaitAttachmentUploads(input.images.map((image) => image.id)); +} + export function getUploadedAttachments(input: { readonly environmentId: EnvironmentId; readonly images: ReadonlyArray;