diff --git a/apps/mobile/src/features/threads/ComposerQueuedEdit.tsx b/apps/mobile/src/features/threads/ComposerQueuedEdit.tsx index 3263a3af35602..028129d1e3180 100644 --- a/apps/mobile/src/features/threads/ComposerQueuedEdit.tsx +++ b/apps/mobile/src/features/threads/ComposerQueuedEdit.tsx @@ -1,12 +1,8 @@ import { Button } from "../../components/ui/Button"; -import type { ChatAttachment, EnvironmentId } from "@t3tools/contracts"; -import { Image } from "expo-image"; -import { Pressable, ScrollView, View } from "react-native"; +import { View } from "react-native"; import { AppText as Text } from "../../components/ui/AppText"; import { SymbolView } from "../../components/ui/AppSymbol"; -import { useAssetUrl } from "../../state/assets"; -import { assetImageSource } from "../../lib/assetImageSource"; export function ComposerQueuedEditBanner(props: { readonly saving: boolean; @@ -28,84 +24,3 @@ export function ComposerQueuedEditBanner(props: { ); } - -export function ComposerQueuedEditAttachments(props: { - readonly environmentId: EnvironmentId; - readonly attachments: ReadonlyArray; - readonly disabled: boolean; - readonly onRemove: (attachmentId: string) => void; -}) { - if (props.attachments.length === 0) return null; - return ( - - {props.attachments.map((attachment) => ( - - ))} - - ); -} - -function QueuedEditAttachmentChip(props: { - readonly environmentId: EnvironmentId; - readonly attachment: ChatAttachment; - readonly disabled: boolean; - readonly onRemove: (attachmentId: string) => void; -}) { - const { attachment } = props; - const isImage = attachment.mimeType.startsWith("image/"); - const resource = { - _tag: "attachment", - attachmentId: attachment.id, - fileName: attachment.name, - mimeType: attachment.mimeType, - disposition: "inline", - } as const; - const url = useAssetUrl(props.environmentId, isImage ? resource : null); - - return ( - - {isImage && url !== null ? ( - - ) : ( - - - - )} - - {attachment.name} - - props.onRemove(attachment.id)} - className="active:opacity-70 disabled:opacity-40" - > - - - - ); -} diff --git a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx index 2bc26707324e2..cb8e76a307a43 100644 --- a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx +++ b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx @@ -997,12 +997,14 @@ export function NewTaskDraftScreen(props: { activeThreadBusy={false} canStopThread={false} queuedRunEdit={null} + savedQueuedEditDrafts={[]} + recoveryDraftKey={null} + onSelectRecoveryDraft={() => {}} composerDraftKey={flow.draftKey} followUpBehavior="queue" canSteerActiveTurn={false} isSavingQueuedEdit={false} onCancelQueuedRunEdit={() => {}} - onRemoveQueuedEditAttachment={() => {}} projectWorkspaceRoot={selectedProject.workspaceRoot} threadCwd={composerWorkspaceCwd} selectedThreadQueueCount={0} diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index e36900736e8e2..582e9b9fcb5e3 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -17,7 +17,6 @@ import { pastedTextDisposition, replaceTextSelection } from "@t3tools/client-run import { PROVIDER_SEND_TURN_MAX_ATTACHMENTS, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, - type ChatAttachment, type EnvironmentId, type MessageId, type ModelSelection, @@ -100,7 +99,6 @@ import { import { compactEditorInsets } from "./compactEditorInsets"; import { ComposerCommandPopover } from "./ComposerCommandPopover"; import { ComposerPopoverAnchor } from "./ComposerPopoverAnchor"; -import { ComposerQueuedEditAttachments } from "./ComposerQueuedEdit"; import { useComposerCommandMenu } from "./use-composer-command-menu"; import type { ComposerEnvironmentCandidate } from "@t3tools/client-runtime/composerThreadItems"; import { offersLinkedEnvironments } from "@t3tools/client-runtime/link-request"; @@ -228,9 +226,7 @@ export interface ThreadComposerProps { readonly interactionLocked?: boolean; readonly draftKey?: string; readonly queuedEdit?: { - readonly existingAttachments: ReadonlyArray; readonly saving: boolean; - readonly onRemoveExistingAttachment: (attachmentId: string) => void; } | null; readonly followUpBehavior: FollowUpBehavior; readonly canSteerActiveTurn: boolean; @@ -452,10 +448,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps const composerDraftKey = props.draftKey ?? composerOwnerKey; const { text: draftMessage, attachments: draftAttachments } = useComposerDraft(composerDraftKey); const queuedEdit = props.queuedEdit ?? null; - const hasContent = - draftMessage.trim().length > 0 || - draftAttachments.length > 0 || - (queuedEdit?.existingAttachments.length ?? 0) > 0; + const hasContent = draftMessage.trim().length > 0 || draftAttachments.length > 0; const stripAttachments = composerStripAttachments(draftAttachments); const showStopAction = !hasContent && props.canStopThread && queuedEdit === null; const onResumeQueue = props.onResumeQueue; @@ -599,9 +592,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps const compactEditor = useCompactComposerEditor(bodyText, draftMessage); const showsCompactDictation = isVoiceInputPresented; const isToolbarVisible = isVoiceInputPresented; - const hasEditingAttachments = - (queuedEdit !== null && queuedEdit.existingAttachments.length > 0) || - stripAttachments.length > 0; + const hasEditingAttachments = stripAttachments.length > 0; const attachmentBlockReason = composerAttachmentUploadBlockReason({ environmentId: props.environmentId, attachments: draftAttachments, @@ -753,20 +744,6 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps // Removable attachments: inside the card, or above the focused one-row composer. const editingAttachments = hasEditingAttachments ? ( <> - {queuedEdit !== null && queuedEdit.existingAttachments.length > 0 ? ( - - - - ) : null} {stripAttachments.length > 0 ? ( ; + readonly recoveryDraftKey: string | null; + readonly onSelectRecoveryDraft: (key: string | null) => void; readonly composerDraftKey: string | null; readonly followUpBehavior: FollowUpBehavior; readonly canSteerActiveTurn: boolean; readonly isSavingQueuedEdit: boolean; readonly onCancelQueuedRunEdit: () => void; - readonly onRemoveQueuedEditAttachment: (attachmentId: string) => void; readonly environmentId: EnvironmentId; readonly projectWorkspaceRoot: string | null; readonly threadCwd: string | null; @@ -1588,6 +1591,29 @@ function ThreadDetailBody( pushes the resting content floor up by the same amount. */} + {props.savedQueuedEditDrafts.length > 0 ? ( + + {props.recoveryDraftKey !== null ? ( + + ) : null} + {queuedRunEdit.savedEditDrafts.map((target, index) => ( + + ))} + + ) : null}
@@ -3714,10 +3740,7 @@ export default function ChatView(props: ChatViewProps) { setThreadError={setThreadError} onExpandImage={onExpandTimelineImageStable} onFileOpen={openFileAttachmentStable} - editingQueuedAttachments={queuedRunEdit.editingAttachments} - onRemoveEditingQueuedAttachment={ - queuedRunEdit.removeEditingAttachment - } + isEditingQueuedMessage={queuedRunEdit.isEditing} /> )}
diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index cba7c434b1bab..6b45aa6bf5d0d 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -20,7 +20,6 @@ import { import type { KeybindingCommand, AssistantCitation, - ChatAttachment as ContractChatAttachment, ChatFileAttachment, EnvironmentId, ModelSelection, @@ -203,7 +202,6 @@ import { ComposerAttachmentFileList, ComposerAttachmentMediaRow, ComposerFilePreviewDialog, - ComposerQueuedAttachmentRow, ComposerRestingImagePreviews, } from "./ComposerAttachmentStrip"; import { ComposerPrimaryActions } from "./ComposerPrimaryActions"; @@ -442,11 +440,7 @@ export interface ChatComposerProps { onPageScrollKeyUp: (key: string) => void; onPageScrollRelease: () => void; - editingQueuedAttachments: ReadonlyArray<{ - readonly attachment: ContractChatAttachment; - readonly url: string | null; - }> | null; - onRemoveEditingQueuedAttachment: (attachmentId: string) => void; + isEditingQueuedMessage: boolean; onCompactContext: () => void; onSend: ( @@ -573,8 +567,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) setThreadError, onExpandImage, onFileOpen, - editingQueuedAttachments, - onRemoveEditingQueuedAttachment, + isEditingQueuedMessage, } = props; const primaryEnvironmentId = usePrimaryEnvironmentId(); const composerDraftTargetKey = composerTargetKey(composerDraftTarget); @@ -604,7 +597,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const activePendingProgress = holdingUserInput ? null : incomingPendingProgress; const activePendingIsResponding = !holdingUserInput && incomingPendingIsResponding; const isLiteralPendingAnswer = activePendingProgress?.activeQuestion?.initialAnswer !== undefined; - const isEditingQueuedMessage = editingQueuedAttachments !== null; const applyThreadModelPick = useApplyThreadModelPick(); const composerDraft = useComposerThreadDraft(composerDraftTarget); const composerDraftTargetKeyRef = useRef(""); @@ -1076,7 +1068,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const composerSendState = deriveComposerSendState({ prompt, imageCount: composerImages.length + composerFiles.length, - existingAttachmentCount: editingQueuedAttachments?.length ?? 0, terminalContexts: composerTerminalContexts, elementContextCount: composerPreviewAnnotations.length + composerReviewComments.length, }); @@ -3421,16 +3412,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) )} - {!isComposerCollapsedMobile && - pendingUserInputs.length === 0 && - editingQueuedAttachments !== null && - editingQueuedAttachments.length > 0 && ( - - )} - {composerAttachmentMediaRow} {composerAttachmentFileList} diff --git a/apps/web/src/components/chat/ComposerAttachmentStrip.tsx b/apps/web/src/components/chat/ComposerAttachmentStrip.tsx index ca98b4d2f6c79..2177506fd8bed 100644 --- a/apps/web/src/components/chat/ComposerAttachmentStrip.tsx +++ b/apps/web/src/components/chat/ComposerAttachmentStrip.tsx @@ -1,5 +1,4 @@ import type { - ChatAttachment, ChatFileAttachment, EnvironmentId, ScopedThreadRef, @@ -659,46 +658,6 @@ export function ComposerRestingImagePreviews({ ); } -/** Attachments of the queued message being edited; they are already uploaded. */ -export function ComposerQueuedAttachmentRow({ - attachments, - onRemove, -}: { - readonly attachments: ReadonlyArray<{ - readonly attachment: ChatAttachment; - readonly url: string | null; - }>; - readonly onRemove: (attachmentId: string) => void; -}) { - return ( -
- {attachments.map(({ attachment, url }) => ( -
- {attachment.type === "image" && url ? ( - {attachment.name} - ) : ( -
- {attachment.name} -
- )} - -
- ))} -
- ); -} - export function ComposerAttachButton({ addComposerAttachments, focusComposer, diff --git a/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx b/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx index 7e25e84135ba5..bc5b409d17f42 100644 --- a/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx +++ b/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx @@ -65,6 +65,30 @@ async function renderQueue(props: Partial); } +it("keeps held messages editable without offering Send now", async () => { + setup(); + state.projection = { projection: { messages: [] } }; + state.workflow = { + activeRun: null, + canSendNow: true, + queuedRuns: [ + { + run: { id: "run:held", userMessageId: "message:held", queueEditId: "edit:other-device" }, + text: "Wait for my edit", + attachments: [], + }, + ], + }; + const onEditQueuedRun = vi.fn(); + await using _screen = await renderQueue({ onEditQueuedRun }); + await expect.element(page.getByText("Editing elsewhere")).toBeVisible(); + await expect.element(page.getByRole("button", { name: "Send now" })).not.toBeInTheDocument(); + await page.getByRole("button", { name: "Edit queued message", exact: true }).click(); + expect(onEditQueuedRun).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ runId: "run:held" }), + ); +}); + it("previews assistant quotes as readable text and preserves the citation when editing", async () => { setup(); const quotedText = "Cache the parsed config per worktree."; diff --git a/apps/web/src/components/chat/QueuedRunsControl.tsx b/apps/web/src/components/chat/QueuedRunsControl.tsx index 2b2cfa78dce2b..a1c52b7702d36 100644 --- a/apps/web/src/components/chat/QueuedRunsControl.tsx +++ b/apps/web/src/components/chat/QueuedRunsControl.tsx @@ -147,6 +147,7 @@ export function QueuedRunsControl({ url: queuedImageUrlById.get(attachment.id) ?? null, })), pending: false, + held: run.queueEditId != null, })), ...optimisticQueued.map((message) => ({ key: message.id, @@ -166,14 +167,19 @@ export function QueuedRunsControl({ url: attachment.previewUrl ?? null, })), pending: true, + held: false, })), ]; const sendQueuedMessage = async (queuedRunId: RunId) => { - const freshContext = queued.some( - ({ run }) => run.id === queuedRunId && run.freshContext === true, - ); - if (!canSendRun(freshContext) || steerInFlightRef.current) return; + const item = queued.find(({ run }) => run.id === queuedRunId); + if ( + !item || + item.run.queueEditId != null || + !canSendRun(item.run.freshContext === true) || + steerInFlightRef.current + ) + return; steerInFlightRef.current = true; setBusyRunId(queuedRunId); if (workflow?.canSendNow) { @@ -194,7 +200,8 @@ export function QueuedRunsControl({ useImperativeHandle(ref, () => ({ steerNext(repeat) { const next = queued[0]; - if (!next || !canSendRun(next.run.freshContext === true)) return false; + if (!next || next.run.queueEditId != null || !canSendRun(next.run.freshContext === true)) + return false; if (!repeat && busyRunId === null) void sendQueuedMessage(next.run.id); return true; }, @@ -275,7 +282,7 @@ export function QueuedRunsControl({ timestampFormat={props.timestampFormat} onOpenThread={props.onOpenThread} > - {canSendRun(item.freshContext) && !isEditing ? ( + {canSendRun(item.freshContext) && !isEditing && !item.held ? ( <> }> @@ -305,6 +312,9 @@ export function QueuedRunsControl({ Saving ) : null} + {item.held && !isEditing ? ( + Editing elsewhere + ) : null} {isEditing ? ( <> Editing in the composer diff --git a/apps/web/src/components/chat/queuedMessageEdit.test.ts b/apps/web/src/components/chat/queuedMessageEdit.test.ts index f7aed151f2fcc..cbea9e83bdd31 100644 --- a/apps/web/src/components/chat/queuedMessageEdit.test.ts +++ b/apps/web/src/components/chat/queuedMessageEdit.test.ts @@ -1,20 +1,57 @@ -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { + collectComposerContextReferences, + formatComposerContextReference, +} from "@t3tools/shared/composerContextReferences"; +import { + ChatAttachmentId, + CommandId, + ComposerContextId, + EnvironmentId, + MessageId, + RunId, + ThreadId, +} from "@t3tools/contracts"; import { scopeThreadRef } from "@t3tools/client-runtime/environment"; -import { beforeEach, describe, expect, it } from "vite-plus/test"; +import { afterAll, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +vi.hoisted(() => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + vi.stubGlobal("localStorage", storage); + vi.stubGlobal("window", { localStorage: storage }); +}); +afterAll(() => vi.unstubAllGlobals()); import { DraftId, useComposerDraftStore, + partializeComposerDraftStoreState, + hydrateImagesFromPersisted, type ComposerFileAttachment, type ComposerImageAttachment, } from "../../composerDraftStore"; -import { prepareQueuedEditAttachments, recoverQueuedMessageEdit } from "./queuedMessageEdit"; +import { + prepareQueuedEditPayload, + prepareQueuedMessageEditDraft, + recoverQueuedMessageEdit, + releaseQueuedMessageEditDraft, +} from "./queuedMessageEdit"; const environmentId = EnvironmentId.make("remote-environment"); const threadTarget = scopeThreadRef(environmentId, ThreadId.make("thread:edit")); const editTarget = DraftId.make("queued-edit:test"); +const edit = { + editId: CommandId.make("edit-1"), + sequence: 1, + runId: RunId.make("run-1"), + messageId: MessageId.make("message-1"), +}; + const file: ComposerFileAttachment = { type: "file", - id: "file:report", + id: ChatAttachmentId.make("file-report"), name: "report.pdf", mimeType: "application/pdf", sizeBytes: 6, @@ -22,22 +59,60 @@ const file: ComposerFileAttachment = { }; const image: ComposerImageAttachment = { type: "image", - id: "image:screen", + id: ChatAttachmentId.make("image-screen"), name: "screen.png", mimeType: "image/png", sizeBytes: 5, file: new File(["image"], "screen.png", { type: "image/png" }), previewUrl: "blob:screen", }; +const emptyContent = { + images: [], + files: [], + terminalContexts: [], + reviewComments: [], + previewAnnotations: [], + threadContexts: [], + environmentContexts: [], +}; +const payloadEdit = { ...edit, sourceText: "", originalText: "", originalAttachmentIds: [] }; const uploadedFile = { type: "file" as const, - id: "upload:report", + id: ChatAttachmentId.make("upload-report"), name: file.name, mimeType: file.mimeType, sizeBytes: file.sizeBytes, }; describe("queued message file edits", () => { + it("records stashed attachments as intentional removals", async () => { + const store = useComposerDraftStore.getState(); + await prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original message", + attachments: [uploadedFile], + readImage: async () => { + throw new Error("Unexpected image"); + }, + }); + store.clearComposerPromptAndImages(editTarget); + store.setPrompt(editTarget, "Replacement text"); + const draft = store.getComposerDraft(editTarget)!; + expect(draft.queuedEdit?.removedAttachmentIds).toEqual([uploadedFile.id]); + await expect( + prepareQueuedEditPayload({ + edit: draft.queuedEdit!, + content: draft, + readImage: async () => { + throw new Error("Unexpected image"); + }, + uploadFiles: async () => [], + }), + ).resolves.toMatchObject({ attachments: [] }); + }); + beforeEach(() => { useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); }); @@ -46,17 +121,16 @@ describe("queued message file edits", () => { { images: [], label: "file-only" }, { images: [image], label: "mixed" }, ])("preserves a generic file in a $label save", async ({ images }) => { - const attachments = await prepareQueuedEditAttachments({ - existingAttachments: [], - images, - files: [file], + const attachments = await prepareQueuedEditPayload({ + edit: payloadEdit, + content: { ...emptyContent, images, files: [file] }, uploadFiles: async () => [uploadedFile], readImage: async () => "data:image/png;base64,aW1hZ2U=", }); - expect(attachments.at(-1)).toEqual(uploadedFile); - expect(attachments.length).toBe(images.length + 1); + expect(attachments.attachments.at(-1)).toEqual(uploadedFile); + expect(attachments.attachments.length).toBe(images.length + 1); if (images.length > 0) - expect(attachments[0]).toMatchObject({ + expect(attachments.attachments[0]).toMatchObject({ type: "image", dataUrl: "data:image/png;base64,aW1hZ2U=", }); @@ -64,10 +138,9 @@ describe("queued message file edits", () => { it("fails a save instead of dropping a file whose upload is missing", async () => { await expect( - prepareQueuedEditAttachments({ - existingAttachments: [], - images: [image], - files: [file], + prepareQueuedEditPayload({ + edit: payloadEdit, + content: { ...emptyContent, images: [image], files: [file] }, uploadFiles: async () => [], readImage: async () => "image", }), @@ -86,39 +159,445 @@ describe("queued message file edits", () => { expect(store.getComposerDraft(editTarget)).toBeNull(); }); - it("reports a removed queued attachment as an edit when the run leaves", () => { + it("reports a removed queued attachment as an edit when the run leaves", async () => { const store = useComposerDraftStore.getState(); - store.setPrompt(editTarget, "Original message"); + await prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original message", + attachments: [uploadedFile], + readImage: async () => { + throw new Error("Unexpected image"); + }, + }); + store.removeFile(editTarget, uploadedFile.id); expect( recoverQueuedMessageEdit({ editTarget, threadTarget, originalText: "Original message", - removedExistingAttachment: true, + originalAttachmentIds: [uploadedFile.id], }), ).toBe("kept"); expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Original message"); - store.clearComposerContent(threadTarget); - // An attachment-only message leaves nothing to put back in the composer. + }); + + it("does not overwrite a separate draft when the queued run leaves", () => { + const store = useComposerDraftStore.getState(); + store.setPrompt(threadTarget, "Separate draft"); + store.addFiles(editTarget, [file]); + expect(recoverQueuedMessageEdit({ editTarget, threadTarget, originalText: "" })).toBe( + "retained", + ); + expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Separate draft"); + expect(store.getComposerDraft(editTarget)?.files).toEqual([file]); + }); + + it("restores ownership and dirty content through draft hydration, then releases locally", async () => { + const store = useComposerDraftStore.getState(); + await prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original", + attachments: [], + readImage: async () => { + throw new Error("unused"); + }, + }); + store.setPrompt(editTarget, ""); + store.setPrompt(threadTarget, "Separate draft"); + const persisted = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), + ); + useComposerDraftStore.setState({ draftsByThreadKey: {} }); + const merge = useComposerDraftStore.persist.getOptions().merge!; + useComposerDraftStore.setState(merge(persisted, useComposerDraftStore.getState())); + expect(store.getComposerDraft(editTarget)?.queuedEdit?.ownership).toEqual({ + editId: edit.editId, + sequence: 1, + }); + expect(store.getComposerDraft(editTarget)?.prompt).toBe(""); + expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Separate draft"); + releaseQueuedMessageEditDraft(editTarget); + expect(store.getComposerDraft(editTarget)).toBeNull(); + expect( + partializeComposerDraftStoreState(useComposerDraftStore.getState()).draftsByThreadKey[ + editTarget + ], + ).toBeUndefined(); + }); + + it("retains the acquired hold through attachment failure and reload until Cancel succeeds", async () => { + const store = useComposerDraftStore.getState(); + await expect( + prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original", + attachments: [ + { + type: "image", + id: image.id, + name: image.name, + mimeType: image.mimeType, + sizeBytes: image.sizeBytes, + }, + ], + readImage: async () => { + throw new Error("offline"); + }, + }), + ).rejects.toThrow("offline"); + // Cleanup cannot reach the server while offline, so ownership must survive reload. + const persisted = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), + ); + useComposerDraftStore.setState({ draftsByThreadKey: {} }); + useComposerDraftStore.setState( + useComposerDraftStore.persist.getOptions().merge!( + persisted, + useComposerDraftStore.getState(), + ), + ); + const draft = store.getComposerDraft(editTarget)!; + expect(draft.queuedEdit?.ownership).toEqual({ editId: edit.editId, sequence: edit.sequence }); + await expect( + prepareQueuedEditPayload({ + edit: draft.queuedEdit!, + content: draft, + readImage: async () => { + throw new Error("unused"); + }, + uploadFiles: async () => [], + }), + ).rejects.toThrow("attachment could not be restored"); + // The normal Cancel path clears the draft only after the token's release succeeds. + store.clearComposerContent(editTarget); + expect(store.getComposerDraft(editTarget)).toBeNull(); + }); + + it("recovers text typed while failed image preparation was pending after release", async () => { + const store = useComposerDraftStore.getState(); + let rejectFetch!: (error: Error) => void; + const fetch = new Promise((_resolve, reject) => { + rejectFetch = reject; + }); + const preparation = prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original", + attachments: [image], + readImage: () => fetch, + }); + const preparedEdit = store.getComposerDraft(editTarget)!.queuedEdit!; + store.setPrompt(editTarget, "Keep what I typed during loading"); + rejectFetch(new Error("offline")); + await expect(preparation).rejects.toThrow("offline"); expect( recoverQueuedMessageEdit({ editTarget, threadTarget, - originalText: "", - removedExistingAttachment: true, + originalText: preparedEdit.originalText, + originalAttachmentIds: preparedEdit.originalAttachmentIds, + expectedEdit: preparedEdit, }), - ).toBe("discarded"); + ).toBe("kept"); + expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Keep what I typed during loading"); + expect(store.getComposerDraft(editTarget)).toBeNull(); + }); + + it("leaves a replacement fallback edit intact after an old preparation fails", async () => { + const store = useComposerDraftStore.getState(); + let rejectFetch!: (error: Error) => void; + const fetch = new Promise((_resolve, reject) => { + rejectFetch = reject; + }); + const preparation = prepareQueuedMessageEditDraft({ + editTarget, + edit: { ...edit, editId: null }, + environmentId, + text: "Original", + attachments: [image], + readImage: () => fetch, + }); + const preparedEdit = store.getComposerDraft(editTarget)!.queuedEdit!; + store.clearComposerContent(editTarget); + await prepareQueuedMessageEditDraft({ + editTarget, + edit: { ...edit, editId: null }, + environmentId, + text: "Original", + attachments: [], + readImage: async () => { + throw new Error("unused"); + }, + }); + store.setPrompt(editTarget, "Replacement edit"); + const replacement = store.getComposerDraft(editTarget); + rejectFetch(new Error("offline")); + await expect(preparation).rejects.toThrow("offline"); + recoverQueuedMessageEdit({ + editTarget, + threadTarget, + originalText: preparedEdit.originalText, + originalAttachmentIds: preparedEdit.originalAttachmentIds, + expectedEdit: preparedEdit, + }); + expect(store.getComposerDraft(editTarget)).toBe(replacement); expect(store.getComposerDraft(threadTarget)).toBeNull(); }); - it("does not overwrite a separate draft when the queued run leaves", () => { + it.each(["text", "attachments"])( + "retains an older dirty edit separately when server %s changes", + async (change) => { + const store = useComposerDraftStore.getState(); + const prepare = (text: string, attachments: (typeof uploadedFile)[]) => + prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text, + attachments, + readImage: async () => { + throw new Error("unused"); + }, + }); + await prepare("Original", []); + store.setPrompt(editTarget, "My older edit"); + releaseQueuedMessageEditDraft(editTarget); + await prepare("Original", []); + expect(store.getComposerDraft(editTarget)?.prompt).toBe("My older edit"); + releaseQueuedMessageEditDraft(editTarget); + await prepare( + change === "text" ? "New server text" : "Original", + change === "attachments" ? [uploadedFile] : [], + ); + const fresh = store.getComposerDraft(editTarget)!; + expect(fresh.prompt).toContain(change === "text" ? "New server text" : "Original"); + expect(fresh.files.map((file) => file.id)).toEqual( + change === "attachments" ? [uploadedFile.id] : [], + ); + const retained = Object.entries(useComposerDraftStore.getState().draftsByThreadKey).find( + ([key]) => key.startsWith(`${editTarget}:saved:`), + )?.[1]; + expect(retained?.prompt).toBe("My older edit"); + expect(retained?.queuedEdit?.ownership).toBeUndefined(); + const retainedKey = Object.keys(useComposerDraftStore.getState().draftsByThreadKey).find( + (key) => key.startsWith(editTarget + ":saved:"), + )!; + store.setPrompt(DraftId.make(retainedKey), ""); + expect(store.getComposerDraft(DraftId.make(retainedKey))).toBeNull(); + }, + ); + + it.each(["attachment removal", "text edit"])( + "recovers the selected server attachments after %s and restart", + async (change) => { + const store = useComposerDraftStore.getState(); + const first = { + type: "image" as const, + id: ChatAttachmentId.make("saved-first"), + name: "first.png", + mimeType: "image/png", + sizeBytes: 5, + }; + const second = { ...first, id: "saved:second", name: "second.png" }; + const savedFile = { ...uploadedFile, id: "saved:file" }; + const attachments = [first, second, savedFile]; + await prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original message", + attachments, + readImage: async (attachment) => ({ + file: new File(["image"], attachment.name, { type: attachment.mimeType }), + dataUrl: "data:image/png;base64,aW1hZ2U=", + }), + }); + const originalText = store.getComposerDraft(editTarget)!.prompt; + store.setPrompt(threadTarget, "Unrelated main draft"); + if (change === "attachment removal") { + store.removeImage(editTarget, first.id); + expect(store.getComposerDraft(editTarget)?.prompt).toBe(originalText); + } else store.setPrompt(editTarget, `${originalText} edited`); + expect( + recoverQueuedMessageEdit({ + editTarget, + threadTarget, + originalText, + originalAttachmentIds: attachments.map((attachment) => attachment.id), + }), + ).toBe("retained"); + const persisted = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), + ); + expect( + persisted.draftsByThreadKey["queued-edit:test"].attachments.map( + (attachment: { id: string }) => attachment.id, + ), + ).toEqual(change === "attachment removal" ? [second.id] : [first.id, second.id]); + expect( + hydrateImagesFromPersisted(persisted.draftsByThreadKey["queued-edit:test"].attachments).map( + (attachment) => attachment.id, + ), + ).toEqual(change === "attachment removal" ? [second.id] : [first.id, second.id]); + expect(persisted.draftsByThreadKey["queued-edit:test"].files).toEqual([ + expect.objectContaining({ attachmentId: savedFile.id, environmentId }), + ]); + expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Unrelated main draft"); + }, + ); + it("saves bound existing image and file references, then removes them without orphaning text", async () => { const store = useComposerDraftStore.getState(); - store.setPrompt(threadTarget, "Separate draft"); - store.addFiles(editTarget, [file]); - expect(recoverQueuedMessageEdit({ editTarget, threadTarget, originalText: "" })).toBe( - "discarded", + const savedImage = { + type: "image" as const, + id: ChatAttachmentId.make("server-image"), + name: "screen.png", + mimeType: "image/png", + sizeBytes: 5, + source: { + kind: "snap-shot" as const, + capturedAt: "2026-10-19T00:00:00.000Z", + appName: "Browser", + windowTitle: "Test", + }, + }; + const savedFile = { ...uploadedFile, id: "server-file" }; + const attachments = [savedFile, savedImage]; + const records = attachments.map((attachment, index) => ({ + version: 1 as const, + ...(attachment.type === "file" ? { kind: "file" as const } : { kind: "image" as const }), + contextId: ComposerContextId.make(`local-context-${index}`), + label: attachment.name, + attachmentId: attachment.id, + name: attachment.name, + mimeType: attachment.mimeType, + sizeBytes: attachment.sizeBytes, + })); + const text = records.map(formatComposerContextReference).join(" "); + await prepareQueuedMessageEditDraft({ + editTarget, + edit: { ...edit, context: { version: 1, records } }, + environmentId, + text, + attachments, + readImage: async () => ({ file: image.file, dataUrl: "data:image/png;base64,aW1hZ2U=" }), + }); + const readImage = vi.fn(async () => "data:image/png;base64,aW1hZ2U="); + const save = () => { + const draft = store.getComposerDraft(editTarget)!; + return prepareQueuedEditPayload({ + edit: draft.queuedEdit!, + content: draft, + readImage, + uploadFiles: async (files) => + files.map((file) => ({ ...savedFile, id: file.uploadedAttachmentId! })), + }); + }; + store.setPrompt(editTarget, store.getComposerDraft(editTarget)!.prompt + " edited"); + const kept = await save(); + expect(readImage).not.toHaveBeenCalled(); + expect(kept.attachments).toEqual(attachments); + expect(kept.attachments.map((attachment) => attachment.id)).toEqual([ + savedFile.id, + savedImage.id, + ]); + const refs = collectComposerContextReferences(store.getComposerDraft(editTarget)!.prompt); + expect(refs).toHaveLength(2); + for (const ref of refs) { + expect( + kept.context.records.find((record) => record.contextId === ref.contextId), + ).toMatchObject({ + attachmentId: ref.kind === "image" ? savedImage.id : savedFile.id, + }); + } + const savedText = store.getComposerDraft(editTarget)!.prompt; + store.clearComposerContent(editTarget); + await prepareQueuedMessageEditDraft({ + editTarget, + edit: { ...edit, context: kept.context }, + environmentId, + text: savedText, + attachments: kept.attachments, + readImage: async () => ({ file: image.file, dataUrl: "data:image/png;base64,aW1hZ2U=" }), + }); + store.removeImage(editTarget, savedImage.id); + store.removeFile(editTarget, savedFile.id); + const persisted = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), ); - expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Separate draft"); + const merge = useComposerDraftStore.persist.getOptions().merge!; + useComposerDraftStore.setState(merge(persisted, useComposerDraftStore.getState())); + const removed = await save(); + expect(removed.attachments).toEqual([]); + expect(removed.context.records).toEqual([]); + expect(collectComposerContextReferences(store.getComposerDraft(editTarget)!.prompt)).toEqual( + [], + ); + }); + + it("rejects an image lost during persistence but saves an intentional removal after hydration", async () => { + const store = useComposerDraftStore.getState(); + const saved = { + type: "image" as const, + id: ChatAttachmentId.make("server-image"), + name: image.name, + mimeType: image.mimeType, + sizeBytes: image.sizeBytes, + }; + const prepare = () => + prepareQueuedMessageEditDraft({ + editTarget, + edit, + environmentId, + text: "Original", + attachments: [saved], + readImage: async () => ({ file: image.file, dataUrl: "data:image/png;base64,aW1hZ2U=" }), + }); + await prepare(); + store.setPrompt(threadTarget, "Occupied draft"); + // Simulate the quota verifier omitting the image while retaining owned edit metadata. + const persisted = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), + ); + persisted.draftsByThreadKey[editTarget].attachments = []; + const merge = useComposerDraftStore.persist.getOptions().merge!; + useComposerDraftStore.setState(merge(persisted, useComposerDraftStore.getState())); + const save = () => { + const draft = store.getComposerDraft(editTarget)!; + return prepareQueuedEditPayload({ + edit: draft.queuedEdit!, + content: draft, + readImage: async () => "unused", + uploadFiles: async () => [], + }); + }; + await expect(save()).rejects.toThrow("attachment could not be restored"); + expect( + recoverQueuedMessageEdit({ + editTarget, + threadTarget, + originalText: "Original", + originalAttachmentIds: [saved.id], + }), + ).toBe("clean"); expect(store.getComposerDraft(editTarget)).toBeNull(); + expect(store.getComposerDraft(threadTarget)?.prompt).toBe("Occupied draft"); + // Missing content cannot be reused as an intentional selection when reopening. + await prepare(); + expect(store.getComposerDraft(editTarget)?.images.map((image) => image.id)).toEqual([saved.id]); + store.removeImage(editTarget, saved.id); + const removed = JSON.parse( + JSON.stringify(partializeComposerDraftStoreState(useComposerDraftStore.getState())), + ); + useComposerDraftStore.setState(merge(removed, useComposerDraftStore.getState())); + expect((await save()).attachments).toEqual([]); + await prepare(); + expect(store.getComposerDraft(editTarget)?.images).toEqual([]); }); }); diff --git a/apps/web/src/components/chat/queuedMessageEdit.ts b/apps/web/src/components/chat/queuedMessageEdit.ts index 40a222a9f383a..f408224f1cced 100644 --- a/apps/web/src/components/chat/queuedMessageEdit.ts +++ b/apps/web/src/components/chat/queuedMessageEdit.ts @@ -1,61 +1,297 @@ -import { resolveQueuedEditExit } from "@t3tools/client-runtime/queued-run-edit"; -import type { ChatAttachment } from "@t3tools/contracts"; +import { + stripAttachmentOnlyBootstrapPrompt, + replaceComposerContextReferences, + formatComposerContextReference, + toKindScopedComposerContextId, +} from "@t3tools/shared/composerContextReferences"; +import { buildMessageContext } from "../../lib/composerContextRecords"; +import { randomUUID } from "../../lib/utils"; +import { + queuedEditDraftMatchesSource, + queuedEditDraftHasMissingAttachments, + type QueuedRunEditDraft, +} from "@t3tools/client-runtime/state/queued-run-edit-draft"; +import type { + ChatAttachment, + ChatImageAttachment, + ChatFileAttachment, + EnvironmentId, +} from "@t3tools/contracts"; import { composerDraftHasUserContent, + createEmptyThreadDraft, + resolveComposerDraftKey, useComposerDraftStore, type ComposerFileAttachment, - type ComposerImageAttachment, type ComposerThreadTarget, + type ComposerThreadDraftState, } from "../../composerDraftStore"; +export async function prepareQueuedMessageEditDraft(input: { + readonly editTarget: ComposerThreadTarget; + readonly edit: Pick & + NonNullable; + readonly environmentId: EnvironmentId; + readonly text: string; + readonly attachments: ReadonlyArray; + readonly readImage: (attachment: ChatAttachment) => Promise<{ file: File; dataUrl: string }>; +}) { + const store = useComposerDraftStore.getState(); + const existing = store.getComposerDraft(input.editTarget); + const key = resolveComposerDraftKey(store, input.editTarget); + if (key === null) throw new Error("Queued edit draft unavailable."); + const contextIds = new Map( + (input.edit.context?.records ?? []).flatMap((record) => + "attachmentId" in record && (record.kind === "image" || record.kind === "file") + ? [ + [ + record.contextId, + toKindScopedComposerContextId(record.kind, record.attachmentId), + ] as const, + ] + : [], + ), + ); + const text = replaceComposerContextReferences( + stripAttachmentOnlyBootstrapPrompt(input.text), + (ref) => { + const contextId = contextIds.get(ref.contextId); + return contextId ? formatComposerContextReference({ ...ref, contextId }) : ref.source; + }, + ); + const context = input.edit.context && { + version: 1 as const, + records: input.edit.context.records.map((record) => ({ + ...record, + contextId: contextIds.get(record.contextId) ?? record.contextId, + ...("screenshotContextId" in record && record.screenshotContextId + ? { + screenshotContextId: + contextIds.get(record.screenshotContextId) ?? record.screenshotContextId, + } + : {}), + })), + }; + const matches = + queuedEditDraftMatchesSource(existing?.queuedEdit, input.text, input.attachments) && + existing?.queuedEdit !== undefined && + !queuedEditDraftHasMissingAttachments(existing.queuedEdit, [ + ...existing.images, + ...existing.files, + ]) && + existing.queuedEdit.context?.records.every( + (record) => + !("attachmentId" in record) || + (record.kind !== "image" && record.kind !== "file") || + record.contextId === toKindScopedComposerContextId(record.kind, record.attachmentId), + ) !== false; + if (existing !== null && !matches) { + const { queuedEdit, ...content } = existing; + const retained = queuedEdit + ? { ...content, queuedEdit: { ...queuedEdit, ownership: undefined } } + : content; + const next = { ...store.draftsByThreadKey }; + if (composerDraftHasUserContent(retained)) next[`${key}:saved:${randomUUID()}`] = retained; + delete next[key]; + useComposerDraftStore.setState({ draftsByThreadKey: next }); + } + if (matches && existing?.queuedEdit) { + const queuedEdit = existing.queuedEdit; + useComposerDraftStore.setState((state) => ({ + draftsByThreadKey: { + ...state.draftsByThreadKey, + [key]: { + ...existing, + queuedEdit: { + ...queuedEdit, + ownership: { editId: input.edit.editId, sequence: input.edit.sequence }, + }, + }, + }, + })); + return; + } + useComposerDraftStore.setState((state) => { + const draft = state.draftsByThreadKey[key] ?? createEmptyThreadDraft(); + return { + draftsByThreadKey: { + ...state.draftsByThreadKey, + [key]: { + ...draft, + prompt: text, + queuedEdit: { + runId: input.edit.runId, + messageId: input.edit.messageId, + sourceText: input.text, + originalText: text, + originalAttachmentIds: input.attachments.map((attachment) => attachment.id), + ...(context ? { context } : {}), + ownership: { editId: input.edit.editId, sequence: input.edit.sequence }, + }, + }, + }, + }; + }); + const ownership = store.getComposerDraft(input.editTarget)?.queuedEdit?.ownership; + const ownsDraft = () => + store.getComposerDraft(input.editTarget)?.queuedEdit?.ownership === ownership; + if ( + input.attachments.some( + (attachment) => attachment.type !== "image" && attachment.type !== "file", + ) + ) + throw new Error("This attachment cannot be edited."); + const images = await Promise.all( + input.attachments + .filter((attachment): attachment is ChatImageAttachment => attachment.type === "image") + .map(async (image) => { + const { file, dataUrl } = await input.readImage(image); + return { ...image, file, previewUrl: dataUrl, dataUrl }; + }), + ); + if (!ownsDraft()) return; + store.addImages(input.editTarget, images, { allowDuplicates: true }); + await store.syncPersistedAttachments(input.editTarget, images); + if (!ownsDraft()) return; + store.addFiles( + input.editTarget, + input.attachments + .filter((attachment): attachment is ChatFileAttachment => attachment.type === "file") + .map((file) => ({ + ...file, + file: null, + uploadedAttachmentId: file.id, + uploadEnvironmentId: input.environmentId, + })), + { allowDuplicates: true }, + ); +} + +export function releaseQueuedMessageEditDraft(editTarget: ComposerThreadTarget) { + const store = useComposerDraftStore.getState(); + const key = resolveComposerDraftKey(store, editTarget); + if (key === null) return; + const draft = store.draftsByThreadKey[key]; + if (!draft?.queuedEdit) return; + useComposerDraftStore.setState({ + draftsByThreadKey: { + ...store.draftsByThreadKey, + [key]: { ...draft, queuedEdit: { ...draft.queuedEdit, ownership: undefined } }, + }, + }); + store.setPrompt(editTarget, draft.prompt); +} + export function recoverQueuedMessageEdit(input: { readonly editTarget: ComposerThreadTarget; readonly threadTarget: ComposerThreadTarget; readonly originalText: string; - readonly removedExistingAttachment?: boolean; -}): "kept" | "discarded" | "clean" { + readonly originalAttachmentIds?: ReadonlyArray; + readonly expectedEdit?: Pick; +}): "kept" | "retained" | "clean" { const store = useComposerDraftStore.getState(); + const currentEdit = store.getComposerDraft(input.editTarget)?.queuedEdit; + if ( + input.expectedEdit && + (currentEdit?.runId !== input.expectedEdit.runId || + currentEdit?.ownership !== input.expectedEdit.ownership) + ) + return "clean"; + releaseQueuedMessageEditDraft(input.editTarget); const draft = store.getComposerDraft(input.editTarget); - const recovery = resolveQueuedEditExit({ - editChanged: - input.removedExistingAttachment === true || - (draft !== null && - (draft.prompt !== input.originalText || draft.images.length > 0 || draft.files.length > 0)), - // A cleared edit has nothing to put back, so it can only be discarded. - threadDraftEmpty: - composerDraftHasUserContent(draft) && - !composerDraftHasUserContent(store.getComposerDraft(input.threadTarget)), - }); - if (recovery === "kept") { - store.moveComposerPromptAndImages(input.editTarget, input.threadTarget); - } else { + const dirty = + draft !== null && + (draft.prompt !== input.originalText || + (draft.queuedEdit?.removedAttachmentIds?.length ?? 0) > 0 || + [...draft.images, ...draft.files].some( + (attachment) => !input.originalAttachmentIds?.includes(attachment.id), + ) || + draft.terminalContexts.length > 0 || + draft.reviewComments.length > 0 || + draft.previewAnnotations.length > 0 || + draft.threadContexts.length > 0); + if (!dirty || draft === null) { store.clearComposerContent(input.editTarget); + return "clean"; } - return recovery; + const fromKey = resolveComposerDraftKey(store, input.editTarget); + const toKey = resolveComposerDraftKey(store, input.threadTarget); + if (fromKey === null || toKey === null || fromKey === toKey) return "kept"; + const destination = store.getComposerDraft(input.threadTarget); + if (composerDraftHasUserContent(destination)) return "retained"; + const { queuedEdit: _queuedEdit, ...recovered } = draft; + const next = { ...store.draftsByThreadKey, [toKey]: recovered }; + delete next[fromKey]; + useComposerDraftStore.setState({ draftsByThreadKey: next }); + return "kept"; } -export async function prepareQueuedEditAttachments(input: { - readonly existingAttachments: ReadonlyArray; - readonly images: ReadonlyArray; - readonly files: ReadonlyArray; +export async function prepareQueuedEditPayload(input: { + readonly edit: QueuedRunEditDraft; + readonly content: { + readonly [ + Key in + | "images" + | "files" + | "terminalContexts" + | "reviewComments" + | "previewAnnotations" + | "threadContexts" + | "environmentContexts" + ]: Readonly; + }; readonly uploadFiles: ( files: ReadonlyArray, ) => Promise>; readonly readImage: (file: File) => Promise; }) { - const files = input.files.length === 0 ? [] : await input.uploadFiles(input.files); - if (files.length !== input.files.length) + const { edit, content } = input; + const drafts = [...content.images, ...content.files]; + if (queuedEditDraftHasMissingAttachments(edit, drafts)) + throw new Error( + "An attachment could not be restored. Cancel and reopen the edit to restore it before saving.", + ); + const files = content.files.length === 0 ? [] : await input.uploadFiles(content.files); + if (files.length !== content.files.length) throw new Error("Retry or remove failed uploads before saving."); const images = await Promise.all( - input.images.map(async (image) => ({ + content.images.map(async (image) => ({ type: "image" as const, id: image.id, name: image.name, mimeType: image.mimeType, sizeBytes: image.sizeBytes, - dataUrl: await input.readImage(image.file), + ...(!edit.originalAttachmentIds.includes(image.id) + ? { dataUrl: await input.readImage(image.file) } + : {}), ...(image.source ? { source: image.source } : {}), })), ); - return [...input.existingAttachments, ...images, ...files]; + const uploaded = [...images, ...files]; + const originalOrder = new Map(edit.originalAttachmentIds.map((id, index) => [id, index])); + const attachments = [...uploaded].sort( + (left, right) => + (originalOrder.get(left.id) ?? originalOrder.size) - + (originalOrder.get(right.id) ?? originalOrder.size), + ); + const records = [ + ...(edit.context?.records ?? []).filter((record) => !("attachmentId" in record)), + ...(buildMessageContext({ + ...content, + terminalContexts: content.terminalContexts.filter( + (context) => context.text.trim().length > 0, + ), + attachments: drafts.map((attachment, index) => ({ + attachment, + attachmentId: uploaded[index]!.id, + })), + })?.records ?? []), + ]; + return { + attachments, + context: { + version: 1 as const, + records: [...new Map(records.map((record) => [record.contextId, record])).values()], + }, + }; } diff --git a/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx b/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx index 4a58d25e48757..df88b54cd85d6 100644 --- a/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx +++ b/apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx @@ -6,6 +6,7 @@ import { RunId, ThreadId, type ScopedThreadRef, + type ChatAttachment, } from "@t3tools/contracts"; import { useImperativeHandle, type RefObject } from "react"; import { describe, expect, it, vi } from "vite-plus/test"; @@ -27,6 +28,9 @@ interface RunStatus { const mocks = await vi.hoisted(async () => { const { Atom } = await import("effect/reactivity"); return { + acquireEdit: vi.fn(async () => ({ _tag: "Success", value: { editId: null, sequence: 0 } })), + releaseEdit: vi.fn(async () => ({ _tag: "Success" })), + setThreadError: vi.fn(), editQueuedRun: vi.fn<(value: unknown) => Promise<{ readonly _tag: "Success" | "Failure" }>>(), // The projection's runs per thread, written by each test to stand in for the server. threadRuns: Atom.family((_key: string) => @@ -36,22 +40,58 @@ const mocks = await vi.hoisted(async () => { }); vi.mock("../../state/use-atom-command", () => ({ - useAtomCommand: () => mocks.editQueuedRun, + useAtomCommand: (command: unknown) => + command === "begin" + ? mocks.acquireEdit + : command === "cancel" + ? mocks.releaseEdit + : mocks.editQueuedRun, })); vi.mock("../../hooks/useSettings", () => ({ useEnvironmentIdentificationMode: () => "none", })); +vi.mock("./composerAttachmentFiles", async (importOriginal) => ({ + ...(await importOriginal()), + fileAttachmentCapabilityBlockReason: () => null, +})); + +vi.mock("../../lib/attachmentUploadQueue", async (importOriginal) => ({ + ...(await importOriginal()), + startAttachmentUpload: () => {}, + awaitAttachmentUploads: async () => {}, + getUploadedAttachments: ({ images }: { images: ReadonlyArray }) => + images.map(({ type, id, name, mimeType, sizeBytes }) => ({ + type, + id, + name, + mimeType, + sizeBytes, + })), +})); + const runsKey = (ref: ScopedThreadRef) => `${ref.environmentId}:${ref.threadId}`; vi.mock("../../state/threads", async () => { const { Atom } = await import("effect/reactivity"); + const { useAtomValue } = await import("@effect/atom-react"); + const Option = await import("effect/Option"); const key = (ref: ScopedThreadRef) => `${ref.environmentId}:${ref.threadId}`; return { + useEnvironmentThread: (environmentId: string, threadId: string) => { + const runs = useAtomValue(mocks.threadRuns(`${environmentId}:${threadId}`)); + return { + status: "live", + sequence: 0, + data: runs === null ? Option.none() : Option.some({ runs }), + }; + }, threadEnvironment: { snapshotAtom: () => Atom.make(null), editQueuedRun: {}, + beginQueuedRunEdit: "begin", + cancelQueuedRunEdit: "cancel", }, environmentThreadDetails: { statusAtom: () => Atom.make("ready"), @@ -101,7 +141,6 @@ function Harness(props: { const sendState = deriveComposerSendState({ prompt: draft.prompt, imageCount: draft.images.length + draft.files.length, - existingAttachmentCount: result.editingAttachments?.length ?? 0, terminalContexts: draft.terminalContexts, }); return result.isEditing ? ( @@ -117,7 +156,12 @@ function Harness(props: { isPreparingWorktree={false} hasSendableContent={sendState.hasSendableContent} isEditingQueuedMessage - onSubmitMessage={() => void result.saveEdit(savedContent(draft.prompt), () => {})} + onSubmitMessage={() => + void result.saveEdit( + { ...savedContent(draft.prompt), images: draft.images, files: draft.files }, + mocks.setThreadError, + ) + } onPreviousPendingQuestion={() => {}} onInterrupt={() => {}} /> @@ -134,6 +178,7 @@ function view(threadRef: ScopedThreadRef, runStatus: string, hook: { current: Ho function setup() { mocks.editQueuedRun.mockReset(); + mocks.setThreadError.mockReset(); useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); setRuns(THREAD_A, "queued"); setRuns(THREAD_B, "queued"); @@ -183,10 +228,10 @@ describe("useQueuedRunEdit", () => { using toast = vi.spyOn(toastManager, "add"); const attachments = [ { - type: "image" as const, + type: "file" as const, id: ChatAttachmentId.make("kept"), - name: "kept.png", - mimeType: "image/png", + name: "kept.pdf", + mimeType: "application/pdf", sizeBytes: 10, }, ]; @@ -201,9 +246,9 @@ describe("useQueuedRunEdit", () => { useComposerDraftStore.getState().getComposerDraft(requireHook(hook).draftTarget)?.prompt ?? "", ).toBe(prefix.trimEnd()); - expect(requireHook(hook).editingAttachments?.map(({ attachment }) => attachment)).toEqual( - attachments, - ); + expect( + useComposerDraftStore.getState().getComposerDraft(requireHook(hook).draftTarget)?.files, + ).toEqual([expect.objectContaining(attachments[0])]); await act(() => useComposerDraftStore .getState() @@ -217,6 +262,9 @@ describe("useQueuedRunEdit", () => { const update = screen.getByRole("button", { name: "Update queued message" }); await expect.element(update).toBeEnabled(); await update.click(); + await vi.waitFor(() => + expect(mocks.setThreadError).toHaveBeenCalledWith(THREAD_A.threadId, null), + ); await vi.waitFor(() => expect(mocks.editQueuedRun).toHaveBeenCalledOnce()); expect(mocks.editQueuedRun).toHaveBeenCalledWith( expect.objectContaining({ @@ -240,16 +288,18 @@ describe("useQueuedRunEdit", () => { text: `Ultrathink:\n${ATTACHMENT_ONLY_BOOTSTRAP_PROMPT}`, attachments: [ { - type: "image", + type: "file", id: ChatAttachmentId.make("removed"), - name: "removed.png", - mimeType: "image/png", + name: "removed.pdf", + mimeType: "application/pdf", sizeBytes: 10, }, ], }), ); - await act(() => requireHook(hook).removeEditingAttachment("removed")); + await act(() => + useComposerDraftStore.getState().removeFile(requireHook(hook).draftTarget, "removed"), + ); await screen.getByRole("button", { name: "Update queued message" }).click(); await vi.waitFor(() => expect(mocks.editQueuedRun).toHaveBeenCalledOnce()); expect(mocks.editQueuedRun).toHaveBeenCalledWith( diff --git a/apps/web/src/components/chat/useQueuedRunEdit.ts b/apps/web/src/components/chat/useQueuedRunEdit.ts index 8feb302dc6aae..25e70389e4b61 100644 --- a/apps/web/src/components/chat/useQueuedRunEdit.ts +++ b/apps/web/src/components/chat/useQueuedRunEdit.ts @@ -1,30 +1,31 @@ +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; import { useAtomValue } from "@effect/atom-react"; import { makeQueuedRunEditStore, makeQueuedRunsAtom, } from "@t3tools/client-runtime/queued-run-edit"; -import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; +import { Atom } from "effect/reactivity"; import { PROVIDER_SEND_TURN_MAX_ATTACHMENTS, - type ChatAttachment, + type CommandId, type EnvironmentContextRecord, type EnvironmentId, - type MessageId, - type OrchestrationMessageContext, type PreviewAnnotationPayload, - type RunId, type ScopedThreadRef, type ThreadContextRecord, + type RunId, type ThreadId, } from "@t3tools/contracts"; +import { queuedRunEditOwnership } from "@t3tools/client-runtime/state/threads"; +import { readLocalApi } from "../../localApi"; import { scopedThreadKey } from "@t3tools/client-runtime/environment"; -import { - restoreAttachmentOnlyBootstrapPrompt, - stripAttachmentOnlyBootstrapPrompt, -} from "@t3tools/shared/composerContextReferences"; -import { Atom } from "effect/reactivity"; -import type { RefObject } from "react"; -import { useAssetUrls } from "../../assets/assetUrls"; +import { useShallow } from "zustand/react/shallow"; +import { useEffect, useRef, useState, type RefObject } from "react"; +import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets"; +import { executeAtomQuery, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import { assetEnvironment } from "../../state/assets"; +import { readPreparedConnection } from "../../state/session"; +import { restoreAttachmentOnlyBootstrapPrompt } from "@t3tools/shared/composerContextReferences"; import { DraftId, useComposerDraftStore, @@ -39,7 +40,6 @@ import { releaseDraftAttachments, startAttachmentUpload, } from "../../lib/attachmentUploadQueue"; -import { buildMessageContext } from "../../lib/composerContextRecords"; import { readFileAsDataUrl } from "../../lib/imageCompression"; import type { TerminalContextDraft } from "../../lib/terminalContext"; import type { ReviewCommentContext } from "../../reviewCommentContext"; @@ -47,34 +47,64 @@ import { appAtomRegistry } from "../../rpc/atomRegistry"; import { environmentServerConfigsAtom } from "../../state/server"; import { environmentThreadDetails, - environmentThreadShells, threadEnvironment, + useEnvironmentThread, } from "../../state/threads"; import { useAtomCommand } from "../../state/use-atom-command"; import { stackedThreadToast, toastManager } from "../ui/toast"; import type { ChatComposerHandle } from "./ChatComposer"; import { fileAttachmentCapabilityBlockReason } from "./composerAttachmentFiles"; -import { prepareQueuedEditAttachments, recoverQueuedMessageEdit } from "./queuedMessageEdit"; +import { + prepareQueuedEditPayload, + prepareQueuedMessageEditDraft, + recoverQueuedMessageEdit, +} from "./queuedMessageEdit"; import type { EditQueuedRunRequest } from "./QueuedRunsControl"; -interface EditingQueuedRun { +import type { QueuedRunEditDraft } from "@t3tools/client-runtime/state/queued-run-edit-draft"; + +interface EditingQueuedRun extends QueuedRunEditDraft { + readonly editId: CommandId | null; + readonly sequence: number; readonly threadId: ThreadId; - readonly runId: RunId; - readonly messageId: MessageId; - readonly originalText: string; - readonly existingAttachments: ReadonlyArray; - readonly context?: OrchestrationMessageContext | undefined; - /** Removing a server attachment changes the message even when the draft looks untouched. */ - readonly removedExistingAttachment?: true; } -// Keyed by the route thread, so an edit survives navigating away and back. const queuedRunEdits = makeQueuedRunEditStore({ registry: appAtomRegistry, label: "web-queued-run-edits", }); -interface QueuedRunEditContent { +function watchQueuedEdit( + routeThreadRef: ScopedThreadRef, + threadTarget: ComposerThreadTarget, + edit: EditingQueuedRun, +) { + const key = scopedThreadKey(routeThreadRef); + if (queuedRunEdits.get(key)?.ownership === edit.ownership) return; + queuedRunEdits.begin(key, edit, { + runs: makeQueuedRunsAtom({ + threadId: edit.threadId, + snapshot: threadEnvironment.snapshotAtom(routeThreadRef.environmentId), + detailStatus: environmentThreadDetails.statusAtom(routeThreadRef), + runs: Atom.make( + (get) => + get(environmentThreadDetails.quietThreadAtom(routeThreadRef))?.projection.runs ?? null, + ), + }), + onRunLeftQueue: (_, endedEdit) => { + recoverQueuedMessageEdit({ + editTarget: queuedEditDraftTarget(routeThreadRef, endedEdit.runId), + threadTarget, + originalText: endedEdit.originalText, + originalAttachmentIds: endedEdit.originalAttachmentIds, + expectedEdit: endedEdit, + }); + }, + }); +} + +/** What the composer holds when the user saves a queued-message edit. */ +export interface QueuedRunEditContent { readonly prompt: string; readonly images: ReadonlyArray; readonly files: ReadonlyArray; @@ -85,10 +115,12 @@ interface QueuedRunEditContent { readonly environmentContexts: ReadonlyArray; } +/** Draft id for editing a queued run of the route thread. */ function queuedEditDraftTarget(routeThreadRef: ScopedThreadRef, runId: RunId) { return DraftId.make(`queued-edit:${scopedThreadKey(routeThreadRef)}:${runId}`); } +/** Uploads the edit's new attachments and saves it. Returns the error to show, or null. */ async function submitQueuedRunEdit(input: { readonly environmentId: EnvironmentId; readonly draftTarget: ComposerThreadTarget; @@ -98,7 +130,6 @@ async function submitQueuedRunEdit(input: { }): Promise { const { environmentId, draftTarget, editing, content } = input; const editText = content.prompt.trim(); - const newAttachments = [...content.images, ...content.files]; const uploadFiles = async (files: ReadonlyArray) => { const validateFiles = () => { const config = appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId) ?? null; @@ -120,51 +151,22 @@ async function submitQueuedRunEdit(input: { return uploaded; }; try { - const uploads = await prepareQueuedEditAttachments({ - existingAttachments: editing.existingAttachments, - images: content.images, - files: content.files, + const payload = await prepareQueuedEditPayload({ + edit: editing, + content, readImage: readFileAsDataUrl, uploadFiles, }); - const keptRecords = (editing.context?.records ?? []).filter( - (record) => - !("attachmentId" in record) || - editing.existingAttachments.some((attachment) => attachment.id === record.attachmentId), - ); - const newRecords = - buildMessageContext({ - terminalContexts: content.terminalContexts.filter( - (context) => context.text.trim().length > 0, - ), - reviewComments: content.reviewComments, - previewAnnotations: content.previewAnnotations, - threadContexts: content.threadContexts, - environmentContexts: content.environmentContexts, - attachments: newAttachments.map((attachment, index) => ({ - attachment, - attachmentId: uploads[editing.existingAttachments.length + index]?.id ?? attachment.id, - })), - })?.records ?? []; const result = await input.editQueuedRun({ environmentId, input: { threadId: editing.threadId, runId: editing.runId, - text: restoreAttachmentOnlyBootstrapPrompt(editText, uploads.length > 0), + editId: editing.editId, + text: restoreAttachmentOnlyBootstrapPrompt(editText, payload.attachments.length > 0), edit: { messageId: editing.messageId, - attachments: uploads, - context: { - version: 1, - records: [ - ...new Map( - [...keptRecords, ...newRecords].map( - (record) => [record.contextId, record] as const, - ), - ).values(), - ], - }, + ...payload, }, }, }); @@ -178,11 +180,17 @@ function useEditQueuedRunCommand() { return useAtomCommand(threadEnvironment.editQueuedRun, { reportFailure: false }); } -/** Edits preserve the thread draft; attachment removals stay local until save. */ +/** + * Queued-message edit mode. While active, the composer is retargeted to a + * per-run edit draft so the user's in-progress draft on the thread survives + * untouched. Stored attachments join that draft so its current selection can + * be recovered through the same paths as newly attached files. + */ export function useQueuedRunEdit(input: { readonly environmentId: EnvironmentId; readonly routeThreadRef: ScopedThreadRef; readonly threadDraftTarget: ComposerThreadTarget; + // Null while the route has no thread yet; an edit needs one to target. readonly activeThreadId: ThreadId | null; readonly serverProjection: ThreadStructure | null; readonly promptRef: { current: string }; @@ -198,27 +206,45 @@ export function useQueuedRunEdit(input: { composerRef, } = input; const editQueuedRunCommand = useEditQueuedRunCommand(); + const acquireEdit = useAtomCommand(threadEnvironment.beginQueuedRunEdit); + const releaseEdit = useAtomCommand(threadEnvironment.cancelQueuedRunEdit); + const releaseFailedPreparation = useAtomCommand(threadEnvironment.cancelQueuedRunEdit, { + reportFailure: false, + }); + const threadState = useEnvironmentThread(environmentId, activeThreadId); + const beginInFlight = useRef(false); + const recoveryPrefix = `queued-edit:${scopedThreadKey(routeThreadRef)}:`; + const activeEdit = useComposerDraftStore( + (state) => + Object.entries(state.draftsByThreadKey).find( + ([key, draft]) => + key.startsWith(recoveryPrefix) && draft.queuedEdit?.ownership !== undefined, + )?.[1].queuedEdit ?? null, + ); + const editing: EditingQueuedRun | null = + activeEdit?.ownership && activeThreadId !== null + ? { + ...activeEdit, + ...activeEdit.ownership, + threadId: activeThreadId, + } + : null; const editKey = scopedThreadKey(routeThreadRef); - const storedEdit = useAtomValue(queuedRunEdits.atom)[editKey] ?? null; - const editing = activeThreadId === null ? null : storedEdit; const isSaving = useAtomValue(queuedRunEdits.savingAtom)[editKey] === true; - const draftTarget = - editing === null ? threadDraftTarget : queuedEditDraftTarget(routeThreadRef, editing.runId); - - const imageResources = (editing?.existingAttachments ?? []) - .filter((attachment) => attachment.type === "image") - .map((attachment) => ({ _tag: "attachment" as const, attachmentId: attachment.id })); - const imageUrls = useAssetUrls(environmentId, imageResources); - const urlByAttachmentId = new Map( - imageResources.map((resource, index) => [resource.attachmentId, imageUrls[index] ?? null]), + const savedEditDrafts = useComposerDraftStore( + useShallow((state) => + Object.keys(state.draftsByThreadKey).filter((key) => key.startsWith(recoveryPrefix)), + ), ); - const editingAttachments = - editing === null - ? null - : editing.existingAttachments.map((attachment) => ({ - attachment, - url: urlByAttachmentId.get(attachment.id) ?? null, - })); + const [recoveryTarget, setRecoveryTarget] = useState(null); + const selectedRecoveryTarget = + recoveryTarget !== null && savedEditDrafts.includes(recoveryTarget) + ? DraftId.make(recoveryTarget) + : null; + const draftTarget = + editing !== null + ? queuedEditDraftTarget(routeThreadRef, editing.runId) + : (selectedRecoveryTarget ?? threadDraftTarget); const focusComposerSoon = () => { window.requestAnimationFrame(() => composerRef.current?.focusAtEnd()); @@ -228,109 +254,192 @@ export function useQueuedRunEdit(input: { .getState() .clearComposerContent(queuedEditDraftTarget(routeThreadRef, runId)); - const beginEdit = useStableCallback((request: EditQueuedRunRequest) => { - if (activeThreadId === null) return; - if (editing !== null && editing.runId !== request.runId) clearEditDraft(editing.runId); - const target = queuedEditDraftTarget(routeThreadRef, request.runId); - const text = stripAttachmentOnlyBootstrapPrompt(request.text); - const store = useComposerDraftStore.getState(); - store.clearComposerContent(target); - store.setPrompt(target, text); - const threadRef = { environmentId, threadId: activeThreadId }; - queuedRunEdits.begin( - editKey, - { - threadId: activeThreadId, - runId: request.runId, - messageId: request.messageId, - originalText: text, - existingAttachments: request.attachments, - context: serverProjection?.messages.find((message) => message.id === request.messageId) - ?.context, - }, - { - runs: makeQueuedRunsAtom({ - threadId: activeThreadId, - snapshot: threadEnvironment.snapshotAtom(environmentId), - detailStatus: environmentThreadDetails.statusAtom(threadRef), - runs: Atom.make( - (get) => - get(environmentThreadDetails.quietThreadAtom(threadRef))?.projection.runs ?? null, - ), - }), - // The run started or was cancelled from another client. A changed - // edit moves into the thread's own draft when that draft is empty; - // otherwise it is dropped. Edits outlive navigation, so the toast is - // unscoped and names the thread: the user may be anywhere by now. - onRunLeftQueue: (_, edit) => { - const recovery = recoverQueuedMessageEdit({ - editTarget: queuedEditDraftTarget(routeThreadRef, edit.runId), + const beginEdit = useStableCallback(async (request: EditQueuedRunRequest) => { + if ( + activeThreadId === null || + beginInFlight.current || + appAtomRegistry.get(queuedRunEdits.savingAtom)[editKey] + ) + return; + beginInFlight.current = true; + const openEdit = async () => { + try { + const run = serverProjection?.runs.find((run) => run.id === request.runId); + const previousEditId = run?.queueEditId ?? null; + if (editing?.runId === request.runId && editing.editId === previousEditId) { + focusComposerSoon(); + return; + } + if (previousEditId !== null) { + const confirmed = await readLocalApi()?.dialogs.confirm( + "This message is being edited elsewhere. Take over editing? The other editor will keep their unsaved changes.", + ); + if (!confirmed) return; + } + if (editing !== null) { + const released = await releaseEdit({ + environmentId, + input: { threadId: editing.threadId, runId: editing.runId, editId: editing.editId }, + }); + if (released._tag !== "Success") return; + recoverQueuedMessageEdit({ + editTarget: queuedEditDraftTarget(routeThreadRef, editing.runId), threadTarget: threadDraftTarget, - originalText: edit.originalText, - removedExistingAttachment: edit.removedExistingAttachment === true, + originalText: editing.originalText, + originalAttachmentIds: editing.originalAttachmentIds, }); - if (recovery === "clean") return; - const title = appAtomRegistry.get( - environmentThreadShells.threadShellAtom(threadRef), - )?.title; - const where = title === undefined ? "" : ` in “${title}”`; - toastManager.add( - stackedThreadToast({ - type: recovery === "kept" ? "info" : "warning", - title: "Queued message is no longer queued", - description: - recovery === "kept" - ? `Your unsaved edit${where} was kept in the composer.` - : `Your unsaved edit${where} was discarded.`, - }), - ); - }, - }, - ); - focusComposerSoon(); + } + const acquired = await acquireEdit({ + environmentId, + input: { threadId: activeThreadId, runId: request.runId, previousEditId }, + }); + if (acquired._tag !== "Success") return; + const target = queuedEditDraftTarget(routeThreadRef, request.runId); + const preparation = prepareQueuedMessageEditDraft({ + editTarget: target, + edit: { + ...acquired.value, + runId: request.runId, + messageId: request.messageId, + context: serverProjection?.messages.find((message) => message.id === request.messageId) + ?.context, + }, + environmentId, + text: request.text, + attachments: request.attachments, + readImage: async (attachment) => { + const connection = readPreparedConnection(environmentId); + if (!connection) throw new Error("Reconnect before editing attachments."); + const result = await executeAtomQuery( + appAtomRegistry, + assetEnvironment.createUrl({ + environmentId, + input: { resource: { _tag: "attachment", attachmentId: attachment.id } }, + }), + { refresh: true, reportFailure: false }, + ); + if (result._tag === "Failure") throw squashAtomCommandFailure(result); + const url = resolveAssetUrl(connection.httpBaseUrl, result.value.relativeUrl); + if (url === null) throw new Error("Attachment URL unavailable."); + const response = await fetch(url, { signal: AbortSignal.timeout(30_000) }); + if (!response.ok) throw new Error(`Could not restore attachment: ${attachment.name}`); + const file = new File([await response.blob()], attachment.name, { + type: attachment.mimeType, + }); + return { file, dataUrl: await readFileAsDataUrl(file) }; + }, + }); + const preparedEdit = useComposerDraftStore.getState().getComposerDraft(target)?.queuedEdit; + try { + await preparation; + } catch (error) { + const released = await releaseFailedPreparation({ + environmentId, + input: { + threadId: activeThreadId, + runId: request.runId, + editId: acquired.value.editId, + }, + }); + if (released._tag === "Success" && preparedEdit?.ownership) + recoverQueuedMessageEdit({ + editTarget: target, + threadTarget: threadDraftTarget, + originalText: preparedEdit.originalText, + originalAttachmentIds: preparedEdit.originalAttachmentIds, + expectedEdit: preparedEdit, + }); + toastManager.add({ + type: "error", + title: "Could not open queued edit", + description: error instanceof Error ? error.message : "Could not restore attachments.", + }); + return; + } + focusComposerSoon(); + } catch (error) { + toastManager.add({ + type: "error", + title: "Could not open queued edit", + description: error instanceof Error ? error.message : "Please try again.", + }); + } + }; + await openEdit().finally(() => { + beginInFlight.current = false; + }); }); - const cancelEdit = useStableCallback(() => { + const cancelEdit = useStableCallback(async () => { if (editing === null) return; + watchQueuedEdit(routeThreadRef, threadDraftTarget, editing); + const finishCancel = queuedRunEdits.startSave(editKey); + if (finishCancel === null) return; + const cancelledDraft = useComposerDraftStore.getState().getComposerDraft(draftTarget); + const released = await releaseEdit({ + environmentId, + input: { threadId: editing.threadId, runId: editing.runId, editId: editing.editId }, + }); + const current = finishCancel(released._tag === "Success"); + if ( + released._tag !== "Success" || + !current || + useComposerDraftStore.getState().getComposerDraft(draftTarget) !== cancelledDraft + ) + return; clearEditDraft(editing.runId); - queuedRunEdits.end(editKey); focusComposerSoon(); }); - const removeEditingAttachment = useStableCallback((attachmentId: string) => { - queuedRunEdits.update(editKey, (current) => ({ - ...current, - removedExistingAttachment: true, - existingAttachments: current.existingAttachments.filter( - (attachment) => attachment.id !== attachmentId, - ), - })); - }); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- keeps the edit projection watch and recovery toast in sync with server ownership after navigation + useEffect(() => { + if (editing === null || activeThreadId !== editing.threadId || isSaving) return; + watchQueuedEdit(routeThreadRef, threadDraftTarget, editing); + if (queuedRunEditOwnership(threadState, editing) !== "lost") return; + queuedRunEdits.end(editKey); + const recovery = recoverQueuedMessageEdit({ + editTarget: queuedEditDraftTarget(routeThreadRef, editing.runId), + threadTarget: threadDraftTarget, + originalText: editing.originalText, + originalAttachmentIds: editing.originalAttachmentIds, + }); + toastManager.add( + stackedThreadToast({ + type: "info", + title: "Queued message edit ended", + description: + recovery === "kept" + ? "Your unsaved changes are in the composer." + : recovery === "retained" + ? "Open Saved queued edit above the composer to recover your changes." + : "The message left the queue or another editor took over.", + }), + ); + }, [activeThreadId, editing, routeThreadRef, threadState, threadDraftTarget, isSaving, editKey]); + /** + * Saves the composer content into the queued message being edited. Reports + * failures through `setThreadError`, which callers own because the error + * banner lives with the thread. + */ const saveEdit = async ( content: QueuedRunEditContent, setThreadError: (threadId: ThreadId, error: string | null) => void, ) => { if (editing === null || isSaving) return; const newAttachments = [...content.images, ...content.files]; - if ( - editing.existingAttachments.length + newAttachments.length > - PROVIDER_SEND_TURN_MAX_ATTACHMENTS - ) { + if (newAttachments.length > PROVIDER_SEND_TURN_MAX_ATTACHMENTS) { setThreadError( editing.threadId, `A message can have at most ${PROVIDER_SEND_TURN_MAX_ATTACHMENTS} attachments.`, ); return; } - if ( - content.prompt.trim().length === 0 && - editing.existingAttachments.length === 0 && - newAttachments.length === 0 - ) { + if (content.prompt.trim().length === 0 && newAttachments.length === 0) { return; } - // The save's result decides the edit's fate, even if the run starts meanwhile. + watchQueuedEdit(routeThreadRef, threadDraftTarget, editing); const finishSave = queuedRunEdits.startSave(editKey); if (finishSave === null) return; + const savedDraft = useComposerDraftStore.getState().getComposerDraft(draftTarget); const error = await submitQueuedRunEdit({ environmentId, draftTarget, @@ -340,10 +449,13 @@ export function useQueuedRunEdit(input: { }); setThreadError(editing.threadId, error); const current = finishSave(error === null); - if (error !== null) return; + if ( + error !== null || + !current || + useComposerDraftStore.getState().getComposerDraft(draftTarget) !== savedDraft + ) + return; releaseDraftAttachments(newAttachments); - // The edit was cancelled or begun again while this save ran; its drafts are the newer edit's. - if (!current) return; promptRef.current = ""; clearEditDraft(editing.runId); composerRef.current?.resetCursorState(); @@ -351,14 +463,15 @@ export function useQueuedRunEdit(input: { }; return { + savedEditDrafts: editing === null ? savedEditDrafts : [], + recoveryTarget: selectedRecoveryTarget, + selectRecoveryTarget: setRecoveryTarget, editingRunId: editing?.runId ?? null, isEditing: editing !== null, isSaving, draftTarget, - editingAttachments, beginEdit, cancelEdit, - removeEditingAttachment, saveEdit, }; } diff --git a/apps/web/src/composerDraftStore.ts b/apps/web/src/composerDraftStore.ts index dde304da3d03e..c4490793cf5a9 100644 --- a/apps/web/src/composerDraftStore.ts +++ b/apps/web/src/composerDraftStore.ts @@ -1,3 +1,4 @@ +import { QueuedRunEditDraft } from "@t3tools/client-runtime/state/queued-run-edit-draft"; import { stripInlineContextReferences } from "./lib/composerContextReferences"; import { composerFileAttachmentMimeType } from "./components/chat/composerAttachmentFiles"; import { elementContextToPreviewAnnotation } from "./lib/elementContext"; @@ -101,6 +102,7 @@ import { import { UnifiedSettings } from "@t3tools/contracts/settings"; const isElementContextDetails = Schema.is(ElementContextDetails); +const isQueuedRunEditDraft = Schema.is(QueuedRunEditDraft); import { ReviewCommentContextSchema, type ReviewCommentContext } from "./reviewCommentContext"; const isRuntimeMode = Schema.is(RuntimeMode); const isProviderDriverKind = Schema.is(ProviderDriverKind); @@ -240,6 +242,7 @@ const PersistedTerminalContextDraft = Schema.Struct({ type PersistedTerminalContextDraft = typeof PersistedTerminalContextDraft.Type; const PersistedComposerThreadDraftState = Schema.Struct({ + queuedEdit: Schema.optionalKey(QueuedRunEditDraft), prompt: Schema.String, attachments: Schema.Array(PersistedComposerImageAttachment), files: Schema.optionalKey(Schema.Array(PersistedComposerDraftFileAttachment)), @@ -362,6 +365,7 @@ export type ComposerContextInsertionHandler = ( const contextInsertionHandlers = new Map(); export interface ComposerThreadDraftState { + queuedEdit?: QueuedRunEditDraft | undefined; prompt: string; images: ComposerImageAttachment[]; files: ComposerFileAttachment[]; @@ -763,7 +767,7 @@ const EMPTY_THREAD_DRAFT = Object.freeze({ interactionMode: null, }); -function createEmptyThreadDraft(): ComposerThreadDraftState { +export function createEmptyThreadDraft(): ComposerThreadDraftState { return { prompt: "", images: [], @@ -884,7 +888,8 @@ function shouldRemoveDraft(draft: ComposerThreadDraftState): boolean { Object.keys(draft.modelSelectionByProvider).length === 0 && draft.activeProvider === null && draft.runtimeMode === null && - draft.interactionMode === null + draft.interactionMode === null && + draft.queuedEdit?.ownership === undefined ); } @@ -1985,7 +1990,9 @@ function normalizePersistedDraftsByThreadId( environmentContexts.length === 0 && !hasModelData && !runtimeMode && - !interactionMode + !interactionMode && + (!isQueuedRunEditDraft(draftCandidate.queuedEdit) || + draftCandidate.queuedEdit.ownership === undefined) ) { continue; } @@ -2002,6 +2009,13 @@ function normalizePersistedDraftsByThreadId( : threadKeyOrId; })(); nextDraftsByThreadKey[normalizedThreadKey] = { + ...(isQueuedRunEditDraft(draftCandidate.queuedEdit) + ? { + queuedEdit: structuredClone( + draftCandidate.queuedEdit, + ) as DeepMutable, + } + : {}), prompt, attachments, ...(files.length > 0 ? { files } : {}), @@ -2116,11 +2130,15 @@ export function partializeComposerDraftStoreState( draft.environmentContexts.length === 0 && !hasModelData && draft.runtimeMode === null && - draft.interactionMode === null + draft.interactionMode === null && + draft.queuedEdit?.ownership === undefined ) { continue; } const persistedDraft: DeepMutable = { + ...(draft.queuedEdit + ? { queuedEdit: structuredClone(draft.queuedEdit) as DeepMutable } + : {}), prompt: draft.prompt, attachments: draft.persistedAttachments, ...(draft.files.length > 0 @@ -2426,6 +2444,7 @@ function toHydratedThreadDraft( })) ?? []; return { + ...(persistedDraft.queuedEdit ? { queuedEdit: persistedDraft.queuedEdit } : {}), prompt: ensureInlineContextReferences(persistedDraft.prompt, [ ...(persistedDraft.reviewComments ?? []).map(reviewCommentContextReference), ...(persistedDraft.previewAnnotations ?? []).map(previewAnnotationContextReference), @@ -3322,6 +3341,16 @@ const composerDraftStore = create()( } const nextDraft: ComposerThreadDraftState = { ...current, + ...(current.queuedEdit && current.queuedEdit.originalAttachmentIds.includes(imageId) + ? { + queuedEdit: { + ...current.queuedEdit, + removedAttachmentIds: [ + ...new Set([...(current.queuedEdit.removedAttachmentIds ?? []), imageId]), + ], + }, + } + : {}), prompt: removeInlineContextReference( current.prompt, toKindScopedComposerContextId("image", imageId), @@ -3434,6 +3463,16 @@ const composerDraftStore = create()( } const nextDraft = { ...current, + ...(current.queuedEdit && current.queuedEdit.originalAttachmentIds.includes(fileId) + ? { + queuedEdit: { + ...current.queuedEdit, + removedAttachmentIds: [ + ...new Set([...(current.queuedEdit.removedAttachmentIds ?? []), fileId]), + ], + }, + } + : {}), prompt: removeInlineContextReference( current.prompt, toKindScopedComposerContextId("file", fileId), @@ -4055,6 +4094,7 @@ const composerDraftStore = create()( } const nextDraft: ComposerThreadDraftState = { ...current, + queuedEdit: undefined, prompt: "", images: [], files: [], @@ -4097,6 +4137,23 @@ const composerDraftStore = create()( ...current.environmentContexts.map(environmentContextReference), ...current.previewAnnotations.map(previewAnnotationContextReference), ]), + ...(current.queuedEdit + ? { + queuedEdit: { + ...current.queuedEdit, + removedAttachmentIds: [ + ...new Set([ + ...(current.queuedEdit.removedAttachmentIds ?? []), + ...[...current.images, ...current.files] + .filter((attachment) => + current.queuedEdit?.originalAttachmentIds.includes(attachment.id), + ) + .map((attachment) => attachment.id), + ]), + ], + }, + } + : {}), images: [], files: [], nonPersistedImageIds: [], diff --git a/docs/user/composer.md b/docs/user/composer.md index df880e49c6ca5..d87f4c8f8cccf 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -83,6 +83,18 @@ Press `Option+Up` on macOS or `Alt+Up` on Windows and Linux with the cursor at t start of the composer to edit the most recently queued message. Change `thread.editQueuedMessage` to use another shortcut. +Editing holds that message and the messages behind it until you Save or Cancel; +the active turn continues. The hold survives disconnects, restarts, and resuming +the queue. Return to the thread after navigating away or restarting the client to +continue your edit. To edit a message held on another device, choose Edit and +confirm takeover. +If another editor takes over or the message is removed, unsaved changes stay in +the composer or remain available as a saved queued edit above it. If another device +saves a newer version, reopening Edit starts from that version and keeps the older +draft under Saved queued edit. Failure and workspace pauses remain separate: finish +the edit, then resume a paused queue. Older servers allow local editing but cannot +hold the queue while you edit. + Mobile has the same choice under **Settings → Chat**. While a turn is running the send button shows which action it will take. Long-press it to use the other action for a single message. From a hardware keyboard, `Cmd+Return` @@ -255,7 +267,8 @@ shortcut described above. On web and desktop, queued messages appear at the end of the thread as dashed messages that have not been sent yet. On mobile they appear above the composer. Each shows its text and any attached images. **Send now** sends a message right away, into the running turn if there is one. You can -also edit or remove it. +also edit or remove it. Finish an open edit with Save or Cancel before using **Send now**; +trying to send an edit-held message does not release a Stop pause. If you stop the agent or the server restarts, queued messages keep their order and are held. Press **Resume** in an empty composer to continue. **Resume** also continues a stopped @@ -266,9 +279,9 @@ The pencil on a queued message opens it in the composer for editing. The origina stays in the queue until you save, and it is highlighted while you edit. The message's attachments appear above the text with a remove control, and new images can be added the usual way. The checkmark saves the queued message in place; **Cancel** next to it leaves it unchanged. Whatever -you had typed in the composer before starting the edit is restored afterwards. If the queued -message starts or is removed while you are editing, the edit ends: changed content moves into the -composer when it is empty, and is discarded otherwise. +you had typed in the composer before starting the edit is restored afterwards. If another editor +takes over or the message leaves the queue, your changes return to an empty composer or remain +available under **Saved queued edit**. They do not overwrite an existing draft. ## Commands and skills diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index c9e1260b956f7..ef950f28b1ff9 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -323,6 +323,10 @@ "types": "./src/state/terminal.ts", "default": "./src/state/terminal.ts" }, + "./state/queued-run-edit-draft": { + "types": "./src/state/queuedRunEditDraft.ts", + "default": "./src/state/queuedRunEditDraft.ts" + }, "./state/threads": { "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 821a92721d357..9a4124d432bbc 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -37,6 +37,9 @@ import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { v2Now, v2Projection, v2ThreadId } from "../state/orchestrationV2TestFixtures.ts"; import { archiveThread, + editQueuedRun, + beginQueuedRunEdit, + cancelQueuedRunEdit, interruptThreadTurn, revertThreadCheckpoint, resumeThreadQueue, @@ -61,6 +64,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly launches?: OrchestrationV2ThreadLaunchInput[]; readonly projection?: OrchestrationV2ThreadProjection; readonly projectionRequests?: ThreadId[]; + readonly queuedRunEditHold?: boolean; readonly advertiseThreadStop?: boolean; }) { const client = { @@ -109,6 +113,9 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct environment: { capabilities: { repositoryIdentity: true, + ...(input.queuedRunEditHold === undefined + ? {} + : { queuedRunEditHold: input.queuedRunEditHold }), ...(input.advertiseThreadStop === true ? { threadStop: true } : {}), }, }, @@ -131,6 +138,41 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct }); describe("V2 environment commands", () => { + it.effect.each([false, true])( + "negotiates queued edit holds with advertised capability %s", + (supported) => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + ...(supported ? { queuedRunEditHold: true } : {}), + }); + const provide = Effect.provideService( + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ); + const target = { threadId: v2ThreadId, runId: RunId.make("queued-run") }; + const acquired = yield* beginQueuedRunEdit({ + ...target, + commandId: CommandId.make("acquire"), + previousEditId: null, + }).pipe(provide); + expect(acquired).toEqual( + supported ? { editId: "acquire", sequence: 1 } : { editId: null, sequence: 0 }, + ); + yield* editQueuedRun({ ...target, editId: acquired.editId, text: "Saved" }).pipe(provide); + yield* cancelQueuedRunEdit({ ...target, editId: acquired.editId }).pipe(provide); + expect(commands.map((command) => command.type)).toEqual( + supported + ? ["queued-run.edit.begin", "queued-run.edit", "queued-run.edit.cancel"] + : ["queued-run.edit"], + ); + const saved = commands.find((command) => command.type === "queued-run.edit"); + if (supported) expect(saved).toHaveProperty("editId", acquired.editId); + else expect(saved).not.toHaveProperty("editId"); + }).pipe(Effect.provide(layerTestRandom)), + ); it.effect("distinguishes sending a selected queued message from resuming queue order", () => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index f813ece95dc45..c8ce87bbdd639 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -254,7 +254,18 @@ export interface RetryWorkspacePreparationInput extends ThreadCommandInput { readonly runId: RunId; } +export interface BeginQueuedRunEditInput extends ThreadCommandInput { + readonly runId: RunId; + readonly previousEditId: CommandId | null; +} + +export interface CancelQueuedRunEditInput extends ThreadCommandInput { + readonly runId: RunId; + readonly editId: CommandId | null; +} + export interface EditQueuedRunInput extends ThreadCommandInput { + readonly editId?: CommandId | null; readonly runId: RunId; readonly text: string; /** @@ -1072,6 +1083,37 @@ export const retryWorkspacePreparation = Effect.fn("EnvironmentCommands.retryWor }, ); +export const beginQueuedRunEdit = Effect.fn("EnvironmentCommands.beginQueuedRunEdit")(function* ( + input: BeginQueuedRunEditInput, +) { + const config = yield* getInitialServerConfig(); + if (config.environment.capabilities.queuedRunEditHold !== true) { + return { editId: null, sequence: 0 }; + } + const editId = yield* allocateCommandId(input); + const receipt = yield* dispatch({ + type: "queued-run.edit.begin", + commandId: editId, + threadId: input.threadId, + runId: input.runId, + previousEditId: input.previousEditId, + }); + return { editId, sequence: receipt.sequence }; +}); + +export const cancelQueuedRunEdit = Effect.fn("EnvironmentCommands.cancelQueuedRunEdit")(function* ( + input: CancelQueuedRunEditInput, +) { + if (input.editId === null) return { sequence: 0 }; + return yield* dispatch({ + type: "queued-run.edit.cancel", + commandId: yield* allocateCommandId(input), + threadId: input.threadId, + runId: input.runId, + editId: input.editId, + }); +}); + export const editQueuedRun = Effect.fn("EnvironmentCommands.editQueuedRun")(function* ( input: EditQueuedRunInput, ) { @@ -1084,6 +1126,7 @@ export const editQueuedRun = Effect.fn("EnvironmentCommands.editQueuedRun")(func commandId: yield* allocateCommandId(input), threadId: input.threadId, runId: input.runId, + ...(input.editId == null ? {} : { editId: input.editId }), text: input.text, ...(attachments === undefined ? {} : { attachments }), ...(input.edit?.context && attachments diff --git a/packages/client-runtime/src/state/queuedRunEditDraft.ts b/packages/client-runtime/src/state/queuedRunEditDraft.ts new file mode 100644 index 0000000000000..e770d7726c668 --- /dev/null +++ b/packages/client-runtime/src/state/queuedRunEditDraft.ts @@ -0,0 +1,44 @@ +import { CommandId, MessageId, OrchestrationMessageContext, RunId } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +export const QueuedRunEditDraft = Schema.Struct({ + runId: RunId, + messageId: MessageId, + // File references can normalize composer text; compare newer saves against the server text. + sourceText: Schema.String, + originalText: Schema.String, + originalAttachmentIds: Schema.Array(Schema.String), + removedAttachmentIds: Schema.optional(Schema.Array(Schema.String)), + context: Schema.optional(OrchestrationMessageContext), + ownership: Schema.optional( + Schema.Struct({ + editId: Schema.NullOr(CommandId), + sequence: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + }), + ), +}); +export type QueuedRunEditDraft = typeof QueuedRunEditDraft.Type; + +export function queuedEditDraftHasMissingAttachments( + draft: QueuedRunEditDraft, + attachments: ReadonlyArray<{ readonly id: string }>, +): boolean { + return draft.originalAttachmentIds.some( + (id) => + !draft.removedAttachmentIds?.includes(id) && + !attachments.some((attachment) => attachment.id === id), + ); +} + +export function queuedEditDraftMatchesSource( + draft: QueuedRunEditDraft | undefined, + text: string, + attachments: ReadonlyArray<{ readonly id: string }>, +): boolean { + return ( + draft !== undefined && + draft.sourceText === text && + draft.originalAttachmentIds.length === attachments.length && + draft.originalAttachmentIds.every((id, index) => id === attachments[index]?.id) + ); +} diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 3761d0fa6e715..7ece513cf50de 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -18,6 +18,8 @@ import { createEnvironmentRpcCommand, } from "./runtime.ts"; import { + type BeginQueuedRunEditInput, + type CancelQueuedRunEditInput, type InterruptThreadTurnInput, type UpdateThreadMetadataInput, archiveThread, @@ -27,6 +29,8 @@ import { createThread, deleteThread, editQueuedRun, + beginQueuedRunEdit, + cancelQueuedRunEdit, interruptThreadTurn, forkSideChat, forkThreadFromRun, @@ -359,6 +363,18 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + beginQueuedRunEdit: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:begin-queued-run-edit", + execute: (input: BeginQueuedRunEditInput) => beginQueuedRunEdit(input), + scheduler, + concurrency, + }), + cancelQueuedRunEdit: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:cancel-queued-run-edit", + execute: (input: CancelQueuedRunEditInput) => cancelQueuedRunEdit(input), + scheduler, + concurrency, + }), editQueuedRun: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:edit-queued-run", execute: editQueuedRun, diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index fd9f1c4de8cfa..ee2f13d762eed 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -1,4 +1,5 @@ import { + CommandId, TurnItemId, NodeId, MessageId, @@ -59,6 +60,30 @@ function run(id: string, ordinal: number, status: OrchestrationV2RunStatus) { } describe("thread execution presentation", () => { + it("shows a completed thread as idle behind an edit hold while preserving active work", () => { + const completed = { ...run("done", 1, "completed"), completedAt: now }; + const edited = { ...run("edited", 2, "queued"), queueEditId: CommandId.make("edit") }; + const later = run("later", 3, "queued"); + const projection = { ...v2Projection, runs: [completed, edited, later], updatedAt: now }; + expect(deriveLatestThreadRun(projection)?.runId).toBe(completed.id); + expect(deriveThreadRuntime(projection)).toMatchObject({ + status: "completed", + activeRunId: null, + }); + expect( + deriveThreadRuntime({ + ...projection, + runs: [{ ...completed, status: "running" }, edited, later], + }), + ).toMatchObject({ status: "running", activeRunId: completed.id }); + expect( + deriveThreadRuntime({ + ...projection, + runs: [completed, { ...edited, queuePosition: 4 }, later], + }), + ).toMatchObject({ status: "queued" }); + }); + it("shows extension statuses only from the selected provider's current live session", () => { const instanceId = v2Projection.thread.modelSelection.instanceId; const providerThreadId = ProviderThreadId.make("native-status-thread"); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 131d277c5e6fc..c2025297ef1d5 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -1,4 +1,6 @@ import { + CommandId, + RunId, EnvironmentAuthorizationError, EnvironmentId, EventId, @@ -54,6 +56,7 @@ import * as ThreadHistoryController from "./threadHistoryController.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, makeEnvironmentThreadState, + queuedRunEditOwnership, type EnvironmentThreadState, type ThreadSnapshotLoadResult, } from "./threads.ts"; @@ -404,6 +407,81 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { + it.effect( + "keeps edit ownership pending until the receipt sequence and detects a later takeover", + () => + Effect.gen(function* () { + const run = { + id: RunId.make("queued-edit-run"), + threadId: THREAD_ID, + ordinal: 1, + providerInstanceId: BASE_PROJECTION.thread.providerInstanceId, + modelSelection: BASE_PROJECTION.thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("queued-edit-message"), + rootNodeId: null, + activeAttemptId: null, + status: "queued" as const, + queueEditId: CommandId.make("previous-editor"), + requestedAt: BASE_PROJECTION.updatedAt, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const h = yield* makeHarness({ cached: { ...BASE_PROJECTION, runs: [run] } }); + const before = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + const edit = { runId: run.id, editId: CommandId.make("accepted-editor"), sequence: 10 }; + expect(queuedRunEditOwnership(before, edit)).toBe("pending"); + + yield* Queue.offer(h.inputs, titleUpdated("Buffered before acquisition", 9)); + const buffered = yield* awaitThreadState(h.observed, (value) => value.sequence === 9); + expect(queuedRunEditOwnership(buffered, edit)).toBe("pending"); + + yield* Queue.offer(h.inputs, { + kind: "event", + sequence: edit.sequence, + event: { + id: EventId.make("edit-acquired"), + type: "run.updated", + threadId: THREAD_ID, + occurredAt: BASE_PROJECTION.updatedAt, + payload: { ...run, queueEditId: edit.editId }, + }, + }); + const acquired = yield* awaitThreadState(h.observed, (value) => value.sequence === 10); + expect(queuedRunEditOwnership(acquired, edit)).toBe("owned"); + expect(queuedRunEditOwnership({ ...acquired, status: "cached" }, edit)).toBe("pending"); + const legacy = { ...edit, editId: null, sequence: 0 }; + expect( + queuedRunEditOwnership( + { + ...acquired, + data: Option.some({ ...BASE_PROJECTION, runs: [{ ...run, queueEditId: undefined }] }), + }, + legacy, + ), + ).toBe("owned"); + expect(queuedRunEditOwnership(acquired, legacy)).toBe("lost"); + + const reopened = { ...edit, editId: CommandId.make("reopened-editor"), sequence: 11 }; + expect(queuedRunEditOwnership(acquired, reopened)).toBe("pending"); + yield* Queue.offer( + h.inputs, + snapshot( + { + ...BASE_PROJECTION, + runs: [{ ...run, queueEditId: CommandId.make("another-device") }], + }, + 12, + ), + ); + const superseded = yield* awaitThreadState(h.observed, (value) => value.sequence === 12); + expect(queuedRunEditOwnership(superseded, edit)).toBe("lost"); + expect(queuedRunEditOwnership(superseded, reopened)).toBe("lost"); + }), + ); + it.effect.each([ ["disk", false], ["retained", false], diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 3e8f3709a915e..efc703f05b9ab 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,6 +1,8 @@ import { isOrchestrationV2WorkActive, ORCHESTRATION_V2_WS_METHODS, + type CommandId, + type RunId, type EnvironmentId as EnvironmentIdType, type OrchestrationV2ThreadDetailSnapshot, type OrchestrationV2ThreadProjection, @@ -55,6 +57,19 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +export function queuedRunEditOwnership( + state: EnvironmentThreadState, + edit: { readonly runId: RunId; readonly editId: CommandId | null; readonly sequence: number }, +): "pending" | "owned" | "lost" { + if ( + (state.status !== "live" && state.status !== "deleted") || + (state.sequence ?? 0) < edit.sequence + ) + return "pending"; + const run = Option.getOrNull(state.data)?.runs.find((run) => run.id === edit.runId); + return run?.status === "queued" && (run.queueEditId ?? null) === edit.editId ? "owned" : "lost"; +} + function statusWithoutLiveData( data: Option.Option, ): EnvironmentThreadStatus { @@ -524,7 +539,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } if (sequence === appliedSequence) return; yield* SubscriptionRef.set(lastSequence, sequence); - if (fresh.length === 0) return; + if (fresh.length === 0) { + yield* SubscriptionRef.update(state, (current) => ({ ...current, sequence })); + return; + } const waiting = yield* Ref.get(awaitingCompletion); // Apply against the latest projection/history in one update so a concurrent @@ -581,6 +599,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const updated: EnvironmentThreadState = { ...current, + sequence, data: Option.some(next), status: waiting ? "synchronizing" : "live", error: Option.none(), diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 5e9cca70752b6..f9c28a7ca7e23 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -92,6 +92,7 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Every protocol-2 server sends this; older mobile builds still read it. */ inlineMessageContext: Schema.optionalKey(Schema.Boolean), requiredWorktreeBootstrap: Schema.optionalKey(Schema.Boolean), + queuedRunEditHold: Schema.optionalKey(Schema.Boolean), threadSettlement: Schema.optionalKey(Schema.Boolean), threadAutoSettlement: Schema.optionalKey(Schema.Boolean), storageCleanup: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b1fd02b3d3a07..d00cf052347c0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -694,6 +694,7 @@ export const OrchestrationV2Run = Schema.Struct({ status: OrchestrationV2RunStatus, queuePosition: Schema.optional(Schema.NullOr(PositiveInt)), queueHeld: Schema.optional(Schema.Boolean), + queueEditId: Schema.optional(Schema.NullOr(CommandId)), requestedAt: Schema.DateTimeUtc, startedAt: Schema.NullOr(Schema.DateTimeUtc), completedAt: Schema.NullOr(Schema.DateTimeUtc), @@ -3304,8 +3305,23 @@ export const OrchestrationV2Command = Schema.Union([ threadId: ThreadId, runId: RunId, }), + Schema.Struct({ + type: Schema.Literal("queued-run.edit.begin"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + previousEditId: Schema.NullOr(CommandId), + }), + Schema.Struct({ + type: Schema.Literal("queued-run.edit.cancel"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + editId: CommandId, + }), Schema.Struct({ type: Schema.Literal("queued-run.edit"), + editId: Schema.optional(CommandId), context: Schema.optional(OrchestrationMessageContext), commandId: CommandId, threadId: ThreadId, diff --git a/packages/shared/src/orchestrationV2ThreadError.ts b/packages/shared/src/orchestrationV2ThreadError.ts index 5b9940729e2e1..a66b74f30c303 100644 --- a/packages/shared/src/orchestrationV2ThreadError.ts +++ b/packages/shared/src/orchestrationV2ThreadError.ts @@ -101,8 +101,19 @@ export function latestUnheldRun( runs: ReadonlyArray, ): OrchestrationV2Run | null { let latest: OrchestrationV2Run | null = null; + const editPosition = runs.reduce( + (position, run) => + run.status === "queued" && run.queueEditId != null + ? Math.min(position, run.queuePosition ?? run.ordinal) + : position, + Number.POSITIVE_INFINITY, + ); for (const run of runs) { - if (run.status === "queued" && run.queueHeld === true) continue; + if ( + run.status === "queued" && + (run.queueHeld === true || (run.queuePosition ?? run.ordinal) >= editPosition) + ) + continue; if (latest === null || run.ordinal > latest.ordinal) latest = run; } return latest;