diff --git a/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.md b/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.md index 288f9685f..2af86260f 100644 --- a/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.md +++ b/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.md @@ -97,3 +97,11 @@ The [Effect probes](../../../../specs/models/session-files.effect-probe.mjs) use Reproduce with a temporary effect@3.18.4 installation (`npm install --prefix --ignore-scripts --no-audit --no-fund effect@3.18.4`). Copy specs/models/session-files.effect-probe.mjs and packages/components/src/providers/store-ref-tracker.ts retaining their repository-relative paths; run `node --experimental-strip-types --test /specs/models/session-files.effect-probe.mjs`. This establishes only those boundaries, not real XHR/IPC, writer durability, cross-window coordination, or E01–E12 acceptance. Official v3 sources and current code references are in the Spec. The original design checkout had 20 broken links to uninitialized ACP submodules. The independent implementation checkout initializes the pinned submodules: document checks now have zero errors and no registered SHA-protected topics. The three action/composer suites pass 69 tests for layer 1; full repository verification and PR references are recorded with the stack status. Product draft behavior and device acceptance remain incomplete. Specs remain draft and this Note remains proposed. + +## Layer 2 implementation + +Layer 2 creates one workspace Effect resource owner for file preparation, image uploads, and send-path store borrows. React keeps Promise interfaces. Navigation does not cancel uploads; workspace disposal cancels and joins work before closing caches/transports. Noncancelable IPC must settle before dependency release. New and continuing conversations share file preparation; cancellation cannot trigger fallback upload, and multipart cleanup is awaited. + +Deterministic tests cover parallel cancellation, late store acquisition, sibling isolation, actual XHR cancellation, and progress versus successful response. Transfer still starts on addition. Persistent submission and complete draft behavior remain the next two layers. + +Layer 2 validation: `TMPDIR=/private/tmp NODE_ENV=test pnpm check` passes completely (components: 478 files, 3,661 tests). `pnpm format` and `pnpm run docs check` completed; docs have no errors. Packaged-device draft acceptance remains outstanding. diff --git a/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.zh.md b/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.zh.md index 30dcf5c43..2ce6448f5 100644 --- a/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.zh.md +++ b/.agents/notes/proposed/architecture/2026-09-14-deferred-attachment-send.zh.md @@ -94,3 +94,11 @@ public-boundary 检查及文档检查分别通过。已运行 `pnpm format` 并 复现：临时目录安装 `effect@3.18.4`（`npm install --prefix --ignore-scripts --no-audit --no-fund effect@3.18.4`）；按仓库相对路径复制 `specs/models/session-files.effect-probe.mjs` 和 `packages/components/src/providers/store-ref-tracker.ts`；运行 `node --experimental-strip-types --test /specs/models/session-files.effect-probe.mjs`。只证明这些边界，不证明真实图片 XHR、IPC、writer 持久化、多窗口协调或 E01–E12 已验收。官方 v3 资料及当前源码入口列在 Spec 末尾。 最初设计工作树有 20 个未初始化 ACP 子模块导致的断链错误。独立实施 checkout 已初始化锁定的子模块，文档检查现在为零错误，没有注册的 SHA 保护主题。第一层的三个 actions/composer 套件共 69 项通过；全仓验证与 PR 链接随栈实施状态记录。完整 draft 产品行为和设备验收仍未完成，Spec 保持 draft，本 Note 保持 proposed。 + +## Layer 2 implementation + +第二层为工作区创建一个 Effect 资源所有者；文件准备、图片上传及发送时的 store 借用都由它管理。React 仍使用 Promise 接口，切页不取消上传，工作区关闭先取消并等待任务，再关闭缓存和传输。不能取消的 IPC 必须结束后才能释放依赖。文件准备在新对话和继续对话间共用；取消不能触发备用上传，multipart 清理必须等待。 + +新增确定性测试覆盖并行取消、迟到的 store 获取、兄弟任务隔离、XHR 实际取消及上传进度与成功响应的区别。该层保持添加时上传；持久化发送和完整 draft 行为仍属于后两层。 + +第二层验证：`TMPDIR=/private/tmp NODE_ENV=test pnpm check` 全部通过（组件 478 个文件、3,661 个测试），`pnpm format` 和 `pnpm run docs check` 已完成；文档无错误。仍未声称完成真实设备上的 draft 验收。 diff --git a/locales/en.json b/locales/en.json index 6d67d0268..c72bf7d3b 100644 --- a/locales/en.json +++ b/locales/en.json @@ -63,13 +63,12 @@ "agents.authentication.authorizeAndOpen": "Authorize and open page", "agents.authentication.browserOpenFailed": "Could not open the authorization page. Check your browser settings and try again.", "agents.authentication.browserOpened": "Complete authorization in the browser window, then return to Lody.", - "agents.authentication.chooseMethod": "Choose how to sign in to {{provider}}", + "agents.authentication.chooseMethod": "Choose a sign-in method", "agents.authentication.codeCopied": "Copied", "agents.authentication.codeSubmitFailed": "Could not submit the authorization code. Please try again.", "agents.authentication.codeSubmitted": "Submitted", "agents.authentication.copyCode": "Copy code", "agents.authentication.copyCodeFailed": "Could not copy the code.", - "agents.authentication.chooseMethod": "Choose a sign-in method", "agents.authentication.enterCode": "Enter this code on the authorization page.", "agents.authentication.enterCodeWithExpiry": "Enter this code on the authorization page. It expires in {{minutes}} minutes.", "agents.authentication.finishInBrowser": "Finish signing in to {{provider}}", @@ -4149,5 +4148,6 @@ "promptShortcut.bodyOffline": "Body is not available offline. Reconnect and select again.", "promptShortcut.loadFailed": "Could not load Shortcut. Select it again to retry.", "settings.beta.promptShortcutsHelper": "Create reusable prompts and insert them with /. In development — expect rough edges.", - "settings.promptShortcuts.disabled": "Enable Prompt Shortcuts under Developer mode in Settings → About to use this feature." + "settings.promptShortcuts.disabled": "Enable Prompt Shortcuts under Developer mode in Settings → About to use this feature.", + "sessions.attachmentTransferInterrupted": "Transfer interrupted. Retry when you return." } diff --git a/locales/zh_CN.json b/locales/zh_CN.json index c4102ecb0..670a1899b 100644 --- a/locales/zh_CN.json +++ b/locales/zh_CN.json @@ -63,13 +63,12 @@ "agents.authentication.authorizeAndOpen": "授权并打开页面", "agents.authentication.browserOpenFailed": "无法打开授权页面，请检查浏览器设置后重试。", "agents.authentication.browserOpened": "请在浏览器窗口中完成授权，然后返回 Lody。", - "agents.authentication.chooseMethod": "选择登录 {{provider}} 的方式", + "agents.authentication.chooseMethod": "选择登录方式", "agents.authentication.codeCopied": "已复制", "agents.authentication.codeSubmitFailed": "无法提交授权码，请重试。", "agents.authentication.codeSubmitted": "已提交", "agents.authentication.copyCode": "复制代码", "agents.authentication.copyCodeFailed": "无法复制代码。", - "agents.authentication.chooseMethod": "选择登录方式", "agents.authentication.enterCode": "请在授权页面输入此代码。", "agents.authentication.enterCodeWithExpiry": "请在授权页面输入此代码，代码将在 {{minutes}} 分钟后过期。", "agents.authentication.finishInBrowser": "完成 {{provider}} 登录", @@ -4149,5 +4148,6 @@ "promptShortcut.bodyOffline": "离线时无法读取正文，请联网后重新选择。", "promptShortcut.loadFailed": "无法加载快捷方式，请重新选择以重试。", "settings.beta.promptShortcutsHelper": "创建可复用的提示词，通过 / 插入。功能开发中，体验可能不完善。", - "settings.promptShortcuts.disabled": "请在设置 → 关于的开发者模式下开启 Prompt Shortcuts。" + "settings.promptShortcuts.disabled": "请在设置 → 关于的开发者模式下开启 Prompt Shortcuts。", + "sessions.attachmentTransferInterrupted": "传输已中断，返回后可重试。" } diff --git a/packages/components/src/atoms/chat-landing-draft.ts b/packages/components/src/atoms/chat-landing-draft.ts index 72fa9ca4c..106339962 100644 --- a/packages/components/src/atoms/chat-landing-draft.ts +++ b/packages/components/src/atoms/chat-landing-draft.ts @@ -24,6 +24,7 @@ export type PendingImage = { progress: number; error?: string; uploaded?: SessionImagePayload; + abort?: AbortController; }; export type PendingFile = { diff --git a/packages/components/src/atoms/runtime.ts b/packages/components/src/atoms/runtime.ts index f0d6fba22..20abc6a2c 100644 --- a/packages/components/src/atoms/runtime.ts +++ b/packages/components/src/atoms/runtime.ts @@ -1,3 +1,4 @@ +import type { SessionSendResources } from '@/lib/session-send-resources'; import type { LocalFilePreviewResource } from '@lody/shared/local-file-preview'; import type { SessionData } from '@lody/shared/session-data'; import { atom } from 'jotai'; @@ -171,6 +172,7 @@ export type WorkspaceRuntime = { * The workspace id used for IndexedDB/WebSocket connections. */ readonly workspaceId: WorkspaceId; + readonly sendResources: SessionSendResources; readonly repo: LoroRepo; /** Workspace-owned, scoped LRU for owner-session file-index Flock resources. */ readonly codeCollabFileIndexCache: CodeCollabFileIndexCache; diff --git a/packages/components/src/components/sessions/session-chat-input-area.tsx b/packages/components/src/components/sessions/session-chat-input-area.tsx index a576a82f9..1caf3d11d 100644 --- a/packages/components/src/components/sessions/session-chat-input-area.tsx +++ b/packages/components/src/components/sessions/session-chat-input-area.tsx @@ -1,3 +1,7 @@ +import { + prepareSessionFile, + SessionFilePreparationAuthError, +} from '@/lib/session-file-preparation'; import { useState, useCallback, @@ -107,15 +111,11 @@ import { isImeComposingKeyboardEvent } from '@/lib/ime'; import { toast } from 'sonner'; import { uploadSessionImage, validateSessionImageFile } from '@/lib/session-image-upload'; import { - computeSha256Hex, - computeTextPreviewable, isUploadAbortedError, isSessionFileTransferPhase, - uploadSessionFile, SESSION_FILE_MAX_SIZE_MB, validateSessionFile, type SessionFileTransferPhase, - type SessionFileUploadProgress, } from '@/lib/session-file-upload'; import { formatFileSize } from '@/lib/session-file-presentation'; import { SESSION_FILE_MAX_COUNT, SESSION_IMAGE_MAX_SIZE_BYTES } from '@lody/shared'; @@ -160,6 +160,7 @@ type PendingImage = { progress: number; error?: string; uploaded?: SessionImagePayload; + abort?: AbortController; }; type PendingFile = { @@ -287,8 +288,11 @@ const setSessionPastedTextDrafts = ( return next; }; -const revokeImagePreviewUrls = (images: readonly Pick[]): void => { +const revokeImagePreviewUrls = ( + images: readonly Pick[] +): void => { for (const image of images) { + image.abort?.abort(); URL.revokeObjectURL(image.previewUrl); } }; @@ -968,7 +972,12 @@ export const SessionChatInputArea = memo( const startUpload = useCallback( async (targetSessionId: SessionId, localId: string, file: File) => { - if (!workspaceId || !authToken) { + if ( + !workspaceId || + !authToken || + !workspaceRuntime || + workspaceRuntime.workspaceId !== workspaceId + ) { capturePostHogEvent(postHog, 'session/image_upload_failed', { channel: 'web', entrypoint: 'session_chat', @@ -992,9 +1001,11 @@ export const SessionChatInputArea = memo( return; } + const abort = new AbortController(); updatePendingImage(targetSessionId, localId, (image) => ({ ...image, status: 'uploading', + abort, progress: 0, error: undefined, })); @@ -1014,20 +1025,26 @@ export const SessionChatInputArea = memo( const uploadStartedAtMs = getPerformanceNowMs(); try { - const uploaded = await uploadSessionImage({ - workspaceId, - sessionId: targetSessionId, - token: authToken, - file, - onProgress: (progress) => { - updatePendingImage(targetSessionId, localId, (image) => ({ ...image, progress })); - }, - }); + const uploaded = await workspaceRuntime.sendResources.run( + (signal) => + uploadSessionImage({ + signal, + workspaceId, + sessionId: targetSessionId, + token: authToken, + file, + onProgress: (progress) => { + updatePendingImage(targetSessionId, localId, (image) => ({ ...image, progress })); + }, + }), + abort.signal + ); updatePendingImage(targetSessionId, localId, (image) => ({ ...image, status: 'uploaded', progress: 100, uploaded, + abort: undefined, error: undefined, })); capturePostHogEvent(postHog, 'session/image_upload_succeeded', { @@ -1044,6 +1061,15 @@ export const SessionChatInputArea = memo( upload_duration_ms: getDurationSinceMs(uploadStartedAtMs), }); } catch (error) { + if (isUploadAbortedError(error)) { + updatePendingImage(targetSessionId, localId, (image) => ({ + ...image, + status: 'failed', + error: t('sessions.attachmentTransferInterrupted'), + abort: undefined, + })); + return; + } const errorMessage = error instanceof Error ? error.message : imageUploadFailedLabel; const reasonCode = toImageUploadReason(classifyImageUploadReason(error)); if ( @@ -1052,12 +1078,17 @@ export const SessionChatInputArea = memo( getSessionFileDrafts(targetSessionId).length < SESSION_FILE_MAX_COUNT ) { try { - const outcome = await sendSessionFileToLocalRuntime({ - workspaceId, - sessionId: targetSessionId, - machineId: session.machineId, - file, - }); + const outcome = await workspaceRuntime.sendResources.run( + (signal) => + sendSessionFileToLocalRuntime({ + signal, + workspaceId, + sessionId: targetSessionId, + machineId: session.machineId, + file, + }), + abort.signal + ); const localFile = outcome?.ok ? outcome.files[0] : undefined; if (localFile) { updatePendingImagesForSession(targetSessionId, (prev) => { @@ -1144,52 +1175,13 @@ export const SessionChatInputArea = memo( updatePendingImagesForSession, t, workspaceId, + workspaceRuntime, ] ); const startFileUpload = useCallback( async (targetSessionId: SessionId, localId: string, file: File) => { - if (!workspaceId) { - updatePendingFile(targetSessionId, localId, (entry) => ({ - ...entry, - status: 'failed', - progress: 0, - error: fileUploadMissingAuthLabel, - })); - return; - } - - // Desktop local-transport fast path: hand bytes straight to the local CLI - // (zero relay round trip). The CLI stores the blob and returns a - // transport:'local' block, which we drop into `uploaded` exactly like a - // cloud upload — the block then rides the outgoing message via - // toFileInputBlock. No progress bar: the handoff completes in one step. - // On any failure we fall through to the cloud path below. - if (canSendFileLocally && session.machineId) { - try { - const outcome = await sendSessionFileToLocalRuntime({ - workspaceId, - sessionId: targetSessionId, - machineId: session.machineId, - file, - }); - if (outcome?.ok && outcome.files[0]) { - updatePendingFile(targetSessionId, localId, (entry) => ({ - ...entry, - status: 'uploaded', - progress: 100, - uploaded: outcome.files[0], - error: undefined, - abort: undefined, - })); - return; - } - } catch { - // Local handoff threw; fall back to the cloud upload path. - } - } - - if (!authToken) { + if (!workspaceId || !workspaceRuntime || workspaceRuntime.workspaceId !== workspaceId) { updatePendingFile(targetSessionId, localId, (entry) => ({ ...entry, status: 'failed', @@ -1209,30 +1201,15 @@ export const SessionChatInputArea = memo( })); try { - // Compute the integrity hash + text-previewability once before upload; - // both ride along to the server and the latter pre-fills the block. - const [sha256, textPreview] = await Promise.all([ - computeSha256Hex(file, { - signal: abort.signal, - onProgress: (progress) => { - updatePendingFile(targetSessionId, localId, (entry) => ({ - ...entry, - status: progress.phase, - progress: progress.percent, - })); - }, - }), - computeTextPreviewable(file), - ]); - const uploaded = await uploadSessionFile({ + const uploaded = await prepareSessionFile(workspaceRuntime.sendResources, { workspaceId, sessionId: targetSessionId, + machineId: session.machineId ?? null, + canSendLocally: canSendFileLocally, token: authToken, file, - sha256, - textPreview, signal: abort.signal, - onProgress: (progress: SessionFileUploadProgress) => { + onProgress: (progress) => { updatePendingFile(targetSessionId, localId, (entry) => ({ ...entry, status: progress.phase, @@ -1250,11 +1227,20 @@ export const SessionChatInputArea = memo( })); } catch (error) { if (isUploadAbortedError(error)) { - // Removal/clearing aborts in-flight uploads; the entry is already - // gone, so leave state untouched. + updatePendingFile(targetSessionId, localId, (entry) => ({ + ...entry, + status: 'failed', + error: t('sessions.attachmentTransferInterrupted'), + abort: undefined, + })); return; } - const errorMessage = error instanceof Error ? error.message : fileUploadFailedLabel; + const errorMessage = + error instanceof SessionFilePreparationAuthError + ? fileUploadMissingAuthLabel + : error instanceof Error + ? error.message + : fileUploadFailedLabel; updatePendingFile(targetSessionId, localId, (entry) => ({ ...entry, status: 'failed', @@ -1272,6 +1258,8 @@ export const SessionChatInputArea = memo( session.machineId, updatePendingFile, workspaceId, + workspaceRuntime, + t, ] ); diff --git a/packages/components/src/hooks/use-chat-landing-file-draft.ts b/packages/components/src/hooks/use-chat-landing-file-draft.ts index cf602435b..2a36f06b3 100644 --- a/packages/components/src/hooks/use-chat-landing-file-draft.ts +++ b/packages/components/src/hooks/use-chat-landing-file-draft.ts @@ -1,3 +1,8 @@ +import { + prepareSessionFile, + SessionFilePreparationAuthError, +} from '@/lib/session-file-preparation'; +import { activeWorkspaceRuntimeAtom } from '@/atoms/runtime'; import { useCallback, useMemo } from 'react'; import { SESSION_FILE_MAX_COUNT, @@ -13,20 +18,13 @@ import { useTranslation } from 'react-i18next'; import { chatLandingPendingFilesAtomFamily, type PendingFile } from '@/atoms/chat-landing-draft'; import { localMachineIdAtom } from '@/atoms/local-probe'; import { formatFileSize } from '@/lib/session-file-presentation'; -import { - canUseElectronLocalFileSend, - sendSessionFileToLocalRuntime, -} from '@/lib/electron-session-file-sender'; +import { canUseElectronLocalFileSend } from '@/lib/electron-session-file-sender'; import { SESSION_FILE_MAX_SIZE_MB, - computeSha256Hex, - computeTextPreviewable, isUploadAbortedError, isSessionFileTransferPhase, - uploadSessionFile, validateSessionFile, type SessionFileTransferPhase, - type SessionFileUploadProgress, } from '@/lib/session-file-upload'; export type ChatLandingFileDraftItem = { @@ -92,6 +90,7 @@ export function useChatLandingFileDraft(args: { ensureSessionId, } = args; const localMachineId = useAtomValue(localMachineIdAtom); + const workspaceRuntime = useAtomValue(activeWorkspaceRuntimeAtom); const [pendingFiles, setPendingFiles] = useAtom(chatLandingPendingFilesAtomFamily(draftKey)); // Desktop local-transport fast path: available only when the selected machine @@ -135,45 +134,7 @@ export function useChatLandingFileDraft(args: { const startUpload = useCallback( async (localId: string, file: File, sessionId: SessionId) => { - if (!workspaceId) { - updatePendingFile(localId, (entry) => ({ - ...entry, - status: 'failed', - progress: 0, - error: fileUploadMissingAuthLabel, - })); - return; - } - - // Desktop local-transport fast path: hand bytes straight to the local CLI - // (zero relay round trip). The CLI returns a transport:'local' block that - // drops into `uploaded` exactly like a cloud upload. On any failure we fall - // through to the cloud path below. - if (canSendFileLocally && machineId) { - try { - const outcome = await sendSessionFileToLocalRuntime({ - workspaceId, - sessionId, - machineId, - file, - }); - if (outcome?.ok && outcome.files[0]) { - updatePendingFile(localId, (entry) => ({ - ...entry, - status: 'uploaded', - progress: 100, - uploaded: outcome.files[0], - error: undefined, - abort: undefined, - })); - return; - } - } catch { - // Local handoff threw; fall back to the cloud upload path. - } - } - - if (!authToken) { + if (!workspaceId || !workspaceRuntime || workspaceRuntime.workspaceId !== workspaceId) { updatePendingFile(localId, (entry) => ({ ...entry, status: 'failed', @@ -193,30 +154,15 @@ export function useChatLandingFileDraft(args: { })); try { - // Compute the integrity hash + text-previewability once before upload; - // both ride along to the server and the latter pre-fills the block. - const [sha256, textPreview] = await Promise.all([ - computeSha256Hex(file, { - signal: abort.signal, - onProgress: (progress) => { - updatePendingFile(localId, (entry) => ({ - ...entry, - status: progress.phase, - progress: progress.percent, - })); - }, - }), - computeTextPreviewable(file), - ]); - const uploaded = await uploadSessionFile({ + const uploaded = await prepareSessionFile(workspaceRuntime.sendResources, { workspaceId, - sessionId, + sessionId: sessionId, + machineId: machineId ?? null, + canSendLocally: canSendFileLocally, token: authToken, file, - sha256, - textPreview, signal: abort.signal, - onProgress: (progress: SessionFileUploadProgress) => { + onProgress: (progress) => { updatePendingFile(localId, (entry) => ({ ...entry, status: progress.phase, @@ -234,11 +180,20 @@ export function useChatLandingFileDraft(args: { })); } catch (error) { if (isUploadAbortedError(error)) { - // Removal/clearing aborts in-flight uploads; the entry is already - // gone, so leave state untouched. + updatePendingFile(localId, (entry) => ({ + ...entry, + status: 'failed', + error: t('sessions.attachmentTransferInterrupted'), + abort: undefined, + })); return; } - const errorMessage = error instanceof Error ? error.message : fileUploadFailedLabel; + const errorMessage = + error instanceof SessionFilePreparationAuthError + ? fileUploadMissingAuthLabel + : error instanceof Error + ? error.message + : fileUploadFailedLabel; updatePendingFile(localId, (entry) => ({ ...entry, status: 'failed', @@ -256,6 +211,8 @@ export function useChatLandingFileDraft(args: { machineId, updatePendingFile, workspaceId, + workspaceRuntime, + t, ] ); diff --git a/packages/components/src/hooks/use-chat-landing-image-draft.ts b/packages/components/src/hooks/use-chat-landing-image-draft.ts index 8976d0c56..43396b53e 100644 --- a/packages/components/src/hooks/use-chat-landing-image-draft.ts +++ b/packages/components/src/hooks/use-chat-landing-image-draft.ts @@ -7,7 +7,9 @@ import { type SessionInputBlock, type WorkspaceId, } from '@lody/shared'; -import { useAtom } from 'jotai'; +import { useAtom, useAtomValue } from 'jotai'; +import { activeWorkspaceRuntimeAtom } from '@/atoms/runtime'; +import { isUploadAbortedError } from '@/lib/session-file-upload'; import { toast } from 'sonner'; import { useTranslation } from 'react-i18next'; import { usePostHog } from '@posthog/react'; @@ -62,6 +64,7 @@ export function useChatLandingImageDraft(args: { ensureSessionId, } = args; const postHog = usePostHog(); + const workspaceRuntime = useAtomValue(activeWorkspaceRuntimeAtom); const [pendingImages, setPendingImages] = useAtom(chatLandingPendingImagesAtomFamily(draftKey)); const imageUploadFailedLabel = t('sessions.imageUploadFailed', 'Image upload failed'); const imageUploadMissingAuthLabel = t( @@ -103,6 +106,7 @@ export function useChatLandingImageDraft(args: { const clearPendingImages = useCallback(() => { setPendingImages((prev) => { for (const image of prev) { + image.abort?.abort(); URL.revokeObjectURL(image.previewUrl); } return []; @@ -125,7 +129,12 @@ export function useChatLandingImageDraft(args: { const startUpload = useCallback( async (localId: string, file: File, sessionId: SessionId) => { - if (!workspaceId || !authToken) { + if ( + !workspaceId || + !authToken || + !workspaceRuntime || + workspaceRuntime.workspaceId !== workspaceId + ) { capturePostHogEvent(postHog, 'session/image_upload_failed', { channel: 'web', entrypoint: 'chat_landing', @@ -146,9 +155,11 @@ export function useChatLandingImageDraft(args: { return; } + const abort = new AbortController(); updatePendingImage(localId, (image) => ({ ...image, status: 'uploading', + abort, progress: 0, error: undefined, })); @@ -164,20 +175,26 @@ export function useChatLandingImageDraft(args: { }); try { - const uploaded = await uploadSessionImage({ - workspaceId, - sessionId, - token: authToken, - file, - onProgress: (progress) => { - updatePendingImage(localId, (image) => ({ ...image, progress })); - }, - }); + const uploaded = await workspaceRuntime.sendResources.run( + (signal) => + uploadSessionImage({ + signal, + workspaceId, + sessionId, + token: authToken, + file, + onProgress: (progress) => { + updatePendingImage(localId, (image) => ({ ...image, progress })); + }, + }), + abort.signal + ); updatePendingImage(localId, (image) => ({ ...image, status: 'uploaded', progress: 100, uploaded, + abort: undefined, error: undefined, })); capturePostHogEvent(postHog, 'session/image_upload_succeeded', { @@ -192,6 +209,15 @@ export function useChatLandingImageDraft(args: { mime_type: uploaded.mimeType, }); } catch (error) { + if (isUploadAbortedError(error)) { + updatePendingImage(localId, (image) => ({ + ...image, + status: 'failed', + error: t('sessions.attachmentTransferInterrupted'), + abort: undefined, + })); + return; + } const errorMessage = error instanceof Error ? error.message : imageUploadFailedLabel; updatePendingImage(localId, (image) => ({ ...image, @@ -221,6 +247,8 @@ export function useChatLandingImageDraft(args: { projectKind, updatePendingImage, workspaceId, + workspaceRuntime, + t, ] ); @@ -307,6 +335,7 @@ export function useChatLandingImageDraft(args: { setPendingImages((prev) => { const target = prev.find((item) => item.localId === localId); if (target) { + target.abort?.abort(); URL.revokeObjectURL(target.previewUrl); } return prev.filter((item) => item.localId !== localId); diff --git a/packages/components/src/lib/AGENTS.md b/packages/components/src/lib/AGENTS.md index 7c3ceb116..076db96e4 100644 --- a/packages/components/src/lib/AGENTS.md +++ b/packages/components/src/lib/AGENTS.md @@ -117,3 +117,5 @@ Rationale: [components](../../../../.agents/docs/components-package.md) and `formatAgentRateLimitWindowLabel`, even when duration/utilization/reset match. - Before creating top-level or child sessions, call `filterAcpSessionConfigOptionValues()` so cached values outside the current selector schema are neither dispatched nor persisted. + +Attachment transfer lifecycle changes follow [workspace ownership](../providers/AGENTS.md#attachment-transfer-ownership). diff --git a/packages/components/src/lib/electron-session-file-sender.ts b/packages/components/src/lib/electron-session-file-sender.ts index 7c896ab63..eebf2d0cc 100644 --- a/packages/components/src/lib/electron-session-file-sender.ts +++ b/packages/components/src/lib/electron-session-file-sender.ts @@ -1,3 +1,4 @@ +import { throwIfSendAborted } from './session-send-resources'; import type { SessionFilePayload } from '@lody/shared'; import { isElectronRenderer } from './electron'; import { getIpcServices } from './electron-ipc-client'; @@ -38,17 +39,21 @@ export const sendSessionFileToLocalRuntime = async (args: { sessionId: string; machineId: string; file: File; + signal?: AbortSignal; }): Promise => { + throwIfSendAborted(args.signal); if (!getIpcServices()) { return null; } const bytes = await fileToArrayBuffer(args.file); + throwIfSendAborted(args.signal); const result = await getIpcServices()!.localProjects.sendSessionFileLocal({ workspaceId: args.workspaceId, sessionId: args.sessionId, machineId: args.machineId, files: [{ fileName: args.file.name, bytes }], }); + throwIfSendAborted(args.signal); if (!result.ok) { return { ok: false, error: result.error }; } diff --git a/packages/components/src/lib/multipart-upload.ts b/packages/components/src/lib/multipart-upload.ts index 1957f3c8d..b5b991df0 100644 --- a/packages/components/src/lib/multipart-upload.ts +++ b/packages/components/src/lib/multipart-upload.ts @@ -6,6 +6,7 @@ type PostMultipartArgs = { onProgress?: (percent: number) => void; /** Prefix for network/abort error messages, e.g. "Image upload". */ errorLabel: string; + signal?: AbortSignal; }; /** @@ -23,9 +24,20 @@ export function postMultipartWithProgress({ formData, onProgress, errorLabel, + signal, }: PostMultipartArgs): Promise { return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new DOMException(`${errorLabel} aborted`, 'AbortError')); + return; + } const request = new XMLHttpRequest(); + const abort = () => request.abort(); + const cleanup = () => { + signal?.removeEventListener('abort', abort); + request.upload.onprogress = null; + request.onerror = request.onabort = request.onload = null; + }; request.open('POST', url); request.responseType = 'json'; request.setRequestHeader('Authorization', `Bearer ${token}`); @@ -40,10 +52,17 @@ export function postMultipartWithProgress({ }; } - request.onerror = () => reject(new Error(`${errorLabel} failed`)); - request.onabort = () => reject(new Error(`${errorLabel} aborted`)); + request.onerror = () => { + cleanup(); + reject(new Error(`${errorLabel} failed`)); + }; + request.onabort = () => { + cleanup(); + reject(new DOMException(`${errorLabel} aborted`, 'AbortError')); + }; request.onload = () => { + cleanup(); let responseBody: unknown = null; try { responseBody = @@ -68,6 +87,17 @@ export function postMultipartWithProgress({ resolve(responseBody); }; - request.send(formData); + try { + signal?.addEventListener('abort', abort, { once: true }); + if (signal?.aborted) { + cleanup(); + reject(new DOMException(`${errorLabel} aborted`, 'AbortError')); + return; + } + request.send(formData); + } catch (error) { + cleanup(); + reject(error); + } }); } diff --git a/packages/components/src/lib/session-file-preparation.ts b/packages/components/src/lib/session-file-preparation.ts new file mode 100644 index 000000000..98c31ffa9 --- /dev/null +++ b/packages/components/src/lib/session-file-preparation.ts @@ -0,0 +1,72 @@ +import type { MachineId, SessionFilePayload, SessionId, WorkspaceId } from '@lody/shared'; +import { sendSessionFileToLocalRuntime } from './electron-session-file-sender'; +import { + computeSha256Hex, + computeTextPreviewable, + uploadSessionFile, + type SessionFileUploadProgress, +} from './session-file-upload'; +import { throwIfSendAborted, type SessionSendResources } from './session-send-resources'; + +export class SessionFilePreparationAuthError extends Error { + constructor() { + super('Missing workspace or auth token'); + this.name = 'SessionFilePreparationAuthError'; + } +} + +/** One transfer path for landing and continuation; UI only observes its result. */ +export function prepareSessionFile( + resources: SessionSendResources, + args: { + workspaceId: WorkspaceId; + sessionId: SessionId; + machineId: MachineId | null; + canSendLocally: boolean; + token: string | null; + file: File; + signal?: AbortSignal; + onProgress?: (progress: SessionFileUploadProgress) => void; + } +): Promise { + return resources.run(async (signal) => { + if (args.canSendLocally && args.machineId) { + try { + const result = await sendSessionFileToLocalRuntime({ + ...args, + machineId: args.machineId, + signal, + }); + throwIfSendAborted(signal); + if (result?.ok && result.files[0]) return result.files[0]; + } catch (error) { + // Preserve the existing transport fallback, but cancellation never + // authorizes a second transfer after a late local handoff. + throwIfSendAborted(signal); + if (error instanceof DOMException && error.name === 'AbortError') throw error; + } + } + throwIfSendAborted(signal); + if (!args.token) throw new SessionFilePreparationAuthError(); + const progress = (value: SessionFileUploadProgress) => { + if (!signal.aborted) args.onProgress?.(value); + }; + const [hash, preview] = await Promise.allSettled([ + computeSha256Hex(args.file, { signal, onProgress: progress }), + computeTextPreviewable(args.file), + ]); + throwIfSendAborted(signal); + if (hash.status === 'rejected') throw hash.reason; + if (preview.status === 'rejected') throw preview.reason; + return await uploadSessionFile({ + workspaceId: args.workspaceId, + sessionId: args.sessionId, + token: args.token, + file: args.file, + sha256: hash.value, + textPreview: preview.value, + signal, + onProgress: progress, + }); + }, args.signal); +} diff --git a/packages/components/src/lib/session-file-upload.ts b/packages/components/src/lib/session-file-upload.ts index 151452270..f38a05f50 100644 --- a/packages/components/src/lib/session-file-upload.ts +++ b/packages/components/src/lib/session-file-upload.ts @@ -592,15 +592,16 @@ const uploadMultipart = async (args: UploadSessionFileArgs): Promise void; + signal?: AbortSignal; }; export const uploadSessionImage = async ({ @@ -95,6 +96,7 @@ export const uploadSessionImage = async ({ token, file, onProgress, + signal, }: UploadSessionImageArgs): Promise => { const validationError = validateSessionImageFile(file); if (validationError) { @@ -111,6 +113,7 @@ export const uploadSessionImage = async ({ formData, onProgress, errorLabel: 'Image upload', + signal, }); const imageValue = diff --git a/packages/components/src/lib/session-send-resources.ts b/packages/components/src/lib/session-send-resources.ts new file mode 100644 index 000000000..de3335612 --- /dev/null +++ b/packages/components/src/lib/session-send-resources.ts @@ -0,0 +1,111 @@ +import { + Cause, + Context, + Effect, + ExecutionStrategy, + Exit, + Fiber, + Layer, + ManagedRuntime, + Scope, +} from 'effect'; +import type { SessionId } from '@lody/shared'; +import type { SessionDocStore } from '@/atoms/runtime'; + +class SendScope extends Context.Tag('lody/SessionSendScope')() {} + +export const throwIfSendAborted = (signal?: AbortSignal): void => { + if (signal?.aborted) throw new DOMException('Attachment operation aborted', 'AbortError'); +}; + +/** Interruption stops cooperative I/O and joins raw work before releasing its owner. */ +function ownedPromise(work: (signal: AbortSignal) => Promise): Effect.Effect { + return Effect.async((resume) => { + const controller = new AbortController(); + const raw = Promise.resolve().then(() => { + throwIfSendAborted(controller.signal); + return work(controller.signal); + }); + void raw.then( + (value) => resume(Effect.succeed(value)), + (error: unknown) => resume(Effect.fail(error)) + ); + return Effect.promise(async () => { + controller.abort(); + // IPC and some acquisitions cannot be interrupted. Their existing finally + // blocks must finish before the workspace destroys their dependencies. + await raw.catch(() => undefined); + }); + }); +} + +export type SessionSendResources = ReturnType; + +export function createSessionSendResources(stores: { + acquire: (sessionId: SessionId) => Promise; + releaseRef: (sessionId: SessionId) => void; +}) { + const managed = ManagedRuntime.make( + Layer.scoped( + SendScope, + Effect.acquireRelease(Scope.make(ExecutionStrategy.parallel), (scope) => + Scope.close(scope, Exit.void) + ) + ) + ); + let closing: Promise | undefined; + + const run = async ( + work: (signal: AbortSignal) => Promise, + signal?: AbortSignal + ): Promise => { + throwIfSendAborted(signal); + if (closing) throw new DOMException('Workspace send resources closed', 'AbortError'); + const fiber = await managed.runPromise( + Effect.gen(function* () { + const scope = yield* SendScope; + return yield* Effect.forkIn(ownedPromise(work), scope); + }) + ); + const interrupt = () => { + // The fiber belongs to SendScope; this only requests its cancellation. + void Effect.runPromise(Fiber.interrupt(fiber)); + }; + signal?.addEventListener('abort', interrupt, { once: true }); + if (signal?.aborted) interrupt(); + try { + const exit = await Effect.runPromise(Fiber.await(fiber)); + throwIfSendAborted(signal); + if (Exit.isSuccess(exit)) return exit.value; + if (Cause.isInterrupted(exit.cause)) { + throw new DOMException('Attachment operation aborted', 'AbortError'); + } + throw Cause.squash(exit.cause); + } finally { + signal?.removeEventListener('abort', interrupt); + } + }; + + return { + run, + withSessionStore: ( + sessionId: SessionId, + use: (store: SessionDocStore, signal: AbortSignal) => Promise | A, + signal?: AbortSignal + ): Promise => + run(async (ownedSignal) => { + const store = await stores.acquire(sessionId); + try { + throwIfSendAborted(ownedSignal); + return await use(store, ownedSignal); + } finally { + // Cache alone owns dispose/unload; only return this operation's borrow. + stores.releaseRef(sessionId); + } + }, signal), + dispose: (): Promise => { + closing ??= Promise.resolve().then(() => managed.dispose()); + return closing; + }, + }; +} diff --git a/packages/components/src/lib/session-submission.ts b/packages/components/src/lib/session-submission.ts index 07f9e436f..0b6edc313 100644 --- a/packages/components/src/lib/session-submission.ts +++ b/packages/components/src/lib/session-submission.ts @@ -309,7 +309,7 @@ export function createSessionSubmission(ports: SessionSubmissionPorts) { if (!runtime) { throw new Error('Runtime not ready'); } - const entry = await runtime.withSessionStore(sessionId, async (sessionStore) => { + const entry = await runtime.sendResources.withSessionStore(sessionId, async (sessionStore) => { const read = await sessionStore.sessionData.history.readTurn(userTurnId); return read.state === 'ready' && read.turn.role === 'user' ? read.turn : undefined; }); @@ -331,8 +331,8 @@ export function createSessionSubmission(ports: SessionSubmissionPorts) { // Local history writes are the accept boundary. Remote document sync is a // sibling of dispatch signaling, never a blocker for clearing the composer. // Hold a store ref for the flush so eviction cannot unload the doc mid-flush. - void runtime - .withSessionStore(sessionId, (sessionStore) => sessionStore.waitUntilSynced()) + void runtime.sendResources + .withSessionStore(sessionId, (sessionStore, signal) => sessionStore.waitUntilSynced(signal)) .catch((error: unknown) => { console.warn('Failed to sync session doc after dispatch request', { sessionId, @@ -380,7 +380,7 @@ export function createSessionSubmission(ports: SessionSubmissionPorts) { if (!runtime) { throw new Error('Runtime not ready'); } - const entry = await runtime.withSessionStore(sessionId, async (sessionStore) => { + const entry = await runtime.sendResources.withSessionStore(sessionId, async (sessionStore) => { const read = await sessionStore.sessionData.history.readTurn(userTurnId); return read.state === 'ready' && read.turn.role === 'user' ? read.turn : undefined; }); @@ -434,26 +434,29 @@ export function createSessionSubmission(ports: SessionSubmissionPorts) { // pending_apply because replay could deliver the input twice. // Re-acquire the store for the write: the steer RPC above can run long, // and we must not hold a store ref across it. - const promoted = await runtime.withSessionStore(sessionId, async (sessionStore) => { - const changed = - ( + const promoted = await runtime.sendResources.withSessionStore( + sessionId, + async (sessionStore) => { + const changed = + ( await sessionStore.sessionData.commands.applyHistoryAction({ kind: 'user-status', turnId: userTurnId, status: 'pending', onlyPendingApply: true, }) - ).matched ?? false; - if (changed) return true; - // CLI promotion can write history before its activation pointer - // fails. Auto-seen may also have observed that pending entry. - const read = await sessionStore.sessionData.history.readTurn(userTurnId); - return ( - read.state === 'ready' && - read.turn.role === 'user' && - (read.turn.status === 'pending' || read.turn.status === 'seen') - ); - }); + ).matched ?? false; + if (changed) return true; + // CLI promotion can write history before its activation pointer + // fails. Auto-seen may also have observed that pending entry. + const read = await sessionStore.sessionData.history.readTurn(userTurnId); + return ( + read.state === 'ready' && + read.turn.role === 'user' && + (read.turn.status === 'pending' || read.turn.status === 'seen') + ); + } + ); // Pending promotion is repairable; a started, terminal, or removed turn is not. if (!promoted) { return false; diff --git a/packages/components/src/providers/AGENTS.md b/packages/components/src/providers/AGENTS.md index 9aaa61d70..1cc39de03 100644 --- a/packages/components/src/providers/AGENTS.md +++ b/packages/components/src/providers/AGENTS.md @@ -84,3 +84,14 @@ and update only decision fields through HistoryWriter; never replace a rendered - Doc-metadata bootstrap and the live repo watch overlap by design: merge per field with live winning (`mergeBootstrapMetaCache`), never letting the snapshot undo an archive already applied live. + +## Attachment transfer ownership + +These rules also bind attachment helpers and UI callers. + +- Workspace `sendResources` owns attachment preparation, upload cancellation, and + send-path session-store borrows. React unmount does not dispose this owner. +- Workspace disposal awaits its cleanup before destroying transports or caches. + Noncancelable IPC must settle before release; only the cache disposes stores. +- Cancellation reaches underlying I/O and fences late completion; it never + authorizes a fallback upload. Await multipart cleanup before returning failure. diff --git a/packages/components/src/providers/create-workspace-runtime.ts b/packages/components/src/providers/create-workspace-runtime.ts index 572b623fb..2a23f6711 100644 --- a/packages/components/src/providers/create-workspace-runtime.ts +++ b/packages/components/src/providers/create-workspace-runtime.ts @@ -1,3 +1,4 @@ +import { createSessionSendResources } from '@/lib/session-send-resources'; import { jotaiStore } from '@/lib/utils'; import { desktopWindowId } from '@/lib/desktop-window'; import { navigationSidebarHiddenAtom } from '@/atoms/layout-state'; @@ -3901,7 +3902,8 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise { - await transportReady.promise; + if (signal) await waitForPromiseOrAbort(transportReady.promise, signal); + else await transportReady.promise; if (signal?.aborted) { return; } @@ -4429,6 +4431,8 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise { + // Cancel and join send I/O while its cache, transport and repo still exist. + await sendResources.dispose(); cancelDelayedBackgroundSyncStart?.(); cancelDelayedBackgroundSyncStart = null; cancelDelayedStartupAcpCapabilitiesRefresh?.(); @@ -4610,11 +4614,16 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise targetRouter.prepareSessionTarget(sessionId, machineId), diff --git a/packages/components/tests/chat-landing-draft-persistence.test.tsx b/packages/components/tests/chat-landing-draft-persistence.test.tsx index 814d961d2..5173b891c 100644 --- a/packages/components/tests/chat-landing-draft-persistence.test.tsx +++ b/packages/components/tests/chat-landing-draft-persistence.test.tsx @@ -6,6 +6,11 @@ import { Provider, createStore, useAtom } from 'jotai'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { SessionId, WorkspaceId } from '@lody/shared'; +import { runtimeAtom, type WorkspaceRuntime } from '../src/atoms/runtime'; +import { createSessionSendResources } from '../src/lib/session-send-resources'; + +globalThis.IS_REACT_ACT_ENVIRONMENT = true; + import { buildChatLandingDraftKey } from '../src/atoms/chat-landing-draft'; import { chatLandingSessionStateAtomFamily } from '../src/atoms/local-storage-cache'; import { useChatLandingDraftSession } from '../src/hooks/use-chat-landing-draft-session'; @@ -18,21 +23,29 @@ import { type ChatLandingFileDraftItem, } from '../src/hooks/use-chat-landing-file-draft'; -type Deferred = { promise: Promise; resolve: (value: T) => void }; +type Deferred = { + promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +}; function deferred(): Deferred { let resolve!: (value: T) => void; - const promise = new Promise((done) => { + let reject!: (error: unknown) => void; + const promise = new Promise((done, fail) => { + reject = fail; resolve = done; }); - return { promise, resolve }; + return { promise, resolve, reject }; } const uploadMocks = vi.hoisted(() => ({ + imageUploadStarted: null as Deferred | null, imageUpload: null as Deferred | null, fileUpload: null as Deferred | null, /** Resolves the moment the hook reaches `uploadSessionFile`, so the test * waits on that call rather than on a guessed number of microtasks. */ + fileUploadAborted: null as Deferred | null, fileUploadStarted: null as Deferred | null, fileUploadSignals: [] as (AbortSignal | undefined)[], })); @@ -52,8 +65,15 @@ vi.mock('../src/lib/posthog-analytics', () => ({ capturePostHogEvent: vi.fn() }) vi.mock('../src/lib/session-image-upload', () => ({ validateSessionImageFile: () => null, - uploadSessionImage: () => { + uploadSessionImage: ({ signal }: { signal?: AbortSignal }) => { uploadMocks.imageUpload = deferred(); + const upload = uploadMocks.imageUpload; + signal?.addEventListener( + 'abort', + () => upload.reject(new DOMException('Aborted', 'AbortError')), + { once: true } + ); + uploadMocks.imageUploadStarted?.resolve(); return uploadMocks.imageUpload.promise; }, })); @@ -63,11 +83,21 @@ vi.mock('../src/lib/session-file-upload', () => ({ validateSessionFile: () => null, computeSha256Hex: async () => 'sha256', computeTextPreviewable: async () => undefined, - isUploadAbortedError: () => false, + isUploadAbortedError: (error: unknown) => + error instanceof DOMException && error.name === 'AbortError', isSessionFileTransferPhase: (status: string) => status === 'preparing' || status === 'uploading', uploadSessionFile: ({ signal }: { signal?: AbortSignal }) => { uploadMocks.fileUploadSignals.push(signal); uploadMocks.fileUpload = deferred(); + const upload = uploadMocks.fileUpload; + signal?.addEventListener( + 'abort', + () => { + upload.reject(new DOMException('Aborted', 'AbortError')); + uploadMocks.fileUploadAborted?.resolve(); + }, + { once: true } + ); uploadMocks.fileUploadStarted?.resolve(); return uploadMocks.fileUpload.promise; }, @@ -135,6 +165,7 @@ function DraftHarness({ draftKey }: { draftKey: string }) { return null; } +let resources: ReturnType; let store = createStore(); let root: Root | null = null; let container: HTMLDivElement | null = null; @@ -142,6 +173,10 @@ let objectUrlSeq = 0; let revokedUrls: string[] = []; function mountLanding(draftKey: string): void { + store.set(runtimeAtom, { + workspaceId: 'workspace-a', + sendResources: resources, + } as WorkspaceRuntime); container = document.createElement('div'); document.body.appendChild(container); root = createRoot(container); @@ -173,6 +208,13 @@ function textFile(name: string): File { } beforeEach(() => { + resources = createSessionSendResources({ + acquire: async () => { + throw new Error('Unexpected store acquisition'); + }, + releaseRef: () => {}, + }); + uploadMocks.imageUploadStarted = deferred(); localStorage.clear(); sessionStorage.clear(); Object.defineProperty(window, '__LODY_ELECTRON__', { @@ -187,6 +229,7 @@ beforeEach(() => { uploadMocks.imageUpload = null; uploadMocks.fileUpload = null; uploadMocks.fileUploadStarted = deferred(); + uploadMocks.fileUploadAborted = deferred(); uploadMocks.fileUploadSignals = []; URL.createObjectURL = () => `blob:preview/${(objectUrlSeq += 1)}`; URL.revokeObjectURL = (url: string) => { @@ -194,8 +237,9 @@ beforeEach(() => { }; }); -afterEach(() => { +afterEach(async () => { if (root) unmountLanding(); + await resources.dispose(); }); describe('chat landing draft persistence', () => { @@ -287,8 +331,9 @@ describe('chat landing draft persistence', () => { it('lets an image upload that was in flight at unmount finish into the restored draft', async () => { mountLanding(WORKSPACE_A_KEY); - act(() => { + await act(async () => { readHarness().addImages([pngFile('shot.png')]); + await uploadMocks.imageUploadStarted?.promise; }); expect(readHarness().imageItems[0]!.status).toBe('uploading'); @@ -348,8 +393,9 @@ describe('chat landing draft persistence', () => { }); const [image] = readHarness().imageItems; - act(() => { + await act(async () => { readHarness().clearDraft(); + await uploadMocks.fileUploadAborted?.promise; }); expect(revokedUrls).toEqual([image!.previewUrl]); diff --git a/packages/components/tests/session-file-upload.test.ts b/packages/components/tests/session-file-upload.test.ts index 2ebe31962..8b4cdb528 100644 --- a/packages/components/tests/session-file-upload.test.ts +++ b/packages/components/tests/session-file-upload.test.ts @@ -216,3 +216,93 @@ describe('multipart session file progress', () => { expect(Math.max(...loadedBytes)).toBe(file.size); }); }); + +describe('multipart response and cancellation', () => { + it('waits for the response after 100% progress and detaches cancellation after success', async () => { + const { postMultipartWithProgress } = await import('../src/lib/multipart-upload'); + let request!: ControlledRequest; + class ControlledRequest { + upload = { onprogress: null as ((event: ProgressEvent) => void) | null }; + onload: (() => void) | null = null; + onerror: (() => void) | null = null; + onabort: (() => void) | null = null; + responseType = ''; + response = { imageId: 'uploaded' }; + status = 200; + aborted = false; + constructor() { + request = this; + } + open() {} + setRequestHeader() {} + send() {} + abort() { + this.aborted = true; + this.onabort?.(); + } + } + vi.stubGlobal('XMLHttpRequest', ControlledRequest); + const controller = new AbortController(); + const progress: number[] = []; + let settled = false; + const upload = postMultipartWithProgress({ + url: '/upload', + token: 'test', + formData: new FormData(), + errorLabel: 'Image', + signal: controller.signal, + onProgress: (value) => progress.push(value), + }); + void upload.then(() => { + settled = true; + }); + request.upload.onprogress?.({ lengthComputable: true, loaded: 10, total: 10 } as ProgressEvent); + expect(progress).toEqual([100]); + expect(settled).toBe(false); + request.onload?.(); + expect(await upload).toEqual({ imageId: 'uploaded' }); + controller.abort(); + expect(request.aborted).toBe(false); + expect(request.upload.onprogress).toBeNull(); + }); + + it('aborts the actual request and rejects without retaining callbacks', async () => { + const { postMultipartWithProgress } = await import('../src/lib/multipart-upload'); + let request!: ControlledRequest; + class ControlledRequest { + upload = { onprogress: null }; + onload: (() => void) | null = null; + onerror: (() => void) | null = null; + onabort: (() => void) | null = null; + aborted = false; + constructor() { + request = this; + } + open() {} + setRequestHeader() {} + send() {} + abort() { + this.aborted = true; + this.onabort?.(); + } + } + vi.stubGlobal('XMLHttpRequest', ControlledRequest); + const controller = new AbortController(); + const upload = postMultipartWithProgress({ + url: '/upload', + token: 'test', + formData: new FormData(), + errorLabel: 'Image', + signal: controller.signal, + }); + controller.abort(); + await expect(upload).rejects.toMatchObject({ name: 'AbortError' }); + expect(request.aborted).toBe(true); + expect([request.onload, request.onerror, request.onabort, request.upload.onprogress]).toEqual([ + null, + null, + null, + null, + ]); + }); +}); diff --git a/packages/components/tests/session-send-resources.test.ts b/packages/components/tests/session-send-resources.test.ts new file mode 100644 index 000000000..b8c87cb05 --- /dev/null +++ b/packages/components/tests/session-send-resources.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from 'vitest'; +import type { SessionId } from '@lody/shared'; +import type { SessionDocStore } from '../src/atoms/runtime'; +import { createSessionSendResources } from '../src/lib/session-send-resources'; + +function gate() { + let resolve!: (value: T) => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +const unusedStores = { + acquire: async (): Promise => { + throw new Error('Unexpected acquisition'); + }, + releaseRef: () => {}, +}; + +describe('workspace send resources', () => { + it('aborts all children and waits for non-cooperative work before closing', async () => { + const resources = createSessionSendResources(unusedStores); + const started = gate(); + const rawFinished = gate(); + const cooperativeStarted = gate(); + const aborted = gate(); + const raw = resources.run(async () => { + started.resolve(); + await rawFinished.promise; + }); + const cooperative = resources.run(async (signal) => { + cooperativeStarted.resolve(); + await new Promise((_, reject) => { + signal.addEventListener( + 'abort', + () => { + aborted.resolve(); + reject(new DOMException('Aborted', 'AbortError')); + }, + { once: true } + ); + }); + }); + const outcomes = Promise.allSettled([raw, cooperative]); + await Promise.all([started.promise, cooperativeStarted.promise]); + let closed = false; + const closing = resources.dispose(); + expect(resources.dispose()).toBe(closing); + void closing.then(() => { + closed = true; + }); + await aborted.promise; + expect(closed).toBe(false); + await expect(resources.run(async () => 'late')).rejects.toMatchObject({ name: 'AbortError' }); + rawFinished.resolve(); + await closing; + for (const outcome of await outcomes) { + expect(outcome).toMatchObject({ status: 'rejected', reason: { name: 'AbortError' } }); + } + }); + + it('returns a late acquisition without exposing it to cancelled work', async () => { + const acquired = gate(); + const acquiring = gate(); + const store = { id: 'session' } as unknown as SessionDocStore; + const borrowed = new Set(); + const resources = createSessionSendResources({ + acquire: async (id) => { + acquiring.resolve(); + const value = await acquired.promise; + borrowed.add(id); + return value; + }, + releaseRef: (id) => { + borrowed.delete(id); + }, + }); + const controller = new AbortController(); + let used = false; + const work = resources.withSessionStore( + 'session' as SessionId, + () => { + used = true; + }, + controller.signal + ); + const result = work.catch((error: unknown) => error); + await acquiring.promise; + controller.abort(); + acquired.resolve(store); + expect(await result).toMatchObject({ name: 'AbortError' }); + expect(used).toBe(false); + expect(borrowed.size).toBe(0); + await resources.dispose(); + }); + + it('preserves failures without cancelling sibling operations', async () => { + const resources = createSessionSendResources(unusedStores); + const siblingReady = gate(); + const siblingFinish = gate(); + const sibling = resources.run(async () => { + siblingReady.resolve(); + return siblingFinish.promise; + }); + await siblingReady.promise; + const error = new Error('Upload rejected'); + await expect( + resources.run(async () => { + throw error; + }) + ).rejects.toBe(error); + siblingFinish.resolve('completed'); + expect(await sibling).toBe('completed'); + await resources.dispose(); + }); +}); diff --git a/packages/components/tests/use-session-actions.test.ts b/packages/components/tests/use-session-actions.test.ts index b450507e3..89cc8f1cb 100644 --- a/packages/components/tests/use-session-actions.test.ts +++ b/packages/components/tests/use-session-actions.test.ts @@ -1,3 +1,4 @@ +import { createSessionSendResources, type SessionSendResources } from '../src/lib/session-send-resources'; import { applyHistoryAction } from '../../shared/src/session-data/history-actions'; import type { HistoryAction, SessionEntry } from '@lody/shared/session-data'; // @vitest-environment jsdom @@ -139,6 +140,8 @@ const sessionDataOver = (history: unknown[]) => ({ }, }); +const sendResourceOwners = new Set(); + const createRuntime = ( overrides: Partial< Pick @@ -198,7 +201,7 @@ const createRuntime = ( reorderSessionMessages: vi.fn(async () => undefined), } as unknown as WorkspaceRuntime['writer']); - return { + const runtime = { workspaceSlug: overrides.workspaceSlug ?? 'workspace-slug', workspaceId: overrides.workspaceId ?? ('workspace-1' as WorkspaceId), repo, @@ -216,6 +219,13 @@ const createRuntime = ( }) ), } as unknown as WorkspaceRuntime; + const resources = createSessionSendResources({ + acquire: (sessionId) => runtime.withSessionStore(sessionId, (store) => store), + releaseRef: () => {}, + }); + sendResourceOwners.add(resources); + Object.defineProperty(runtime, 'sendResources', { value: resources }); + return runtime; }; const createSessionPayload = (sessionId: SessionId): SessionToCreate => @@ -315,6 +325,8 @@ describe('useSessionActions', () => { }); afterEach(async () => { + await Promise.all([...sendResourceOwners].map((resources) => resources.dispose())); + sendResourceOwners.clear(); if (root) { await act(async () => { root?.unmount();