diff --git a/packages/app/e2e/browser/chat-file-attachment-types.spec.ts b/packages/app/e2e/browser/chat-file-attachment-types.spec.ts new file mode 100644 index 000000000..4dde3fdbf --- /dev/null +++ b/packages/app/e2e/browser/chat-file-attachment-types.spec.ts @@ -0,0 +1,164 @@ +import type { DaemonClient } from "@clisbot/client/internal/daemon-client"; +import { randomBytes } from "node:crypto"; +import type { Page } from "@playwright/test"; +import { expect, test } from "../support/fixtures"; +import { attachFileFromMenu } from "../support/helpers/composer"; +import { connectDaemonClient } from "../support/helpers/daemon-client-loader"; +import { getServerId } from "../support/helpers/server-id"; +import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent"; + +test.use({ e2eDaemonEnvironment: { CLISBOT_TAILSCALE_BIN: process.execPath } }); +test.describe.configure({ timeout: 180_000 }); + +type BotClient = Pick< + DaemonClient, + | "createBot" + | "archiveBot" + | "createChat" + | "archiveChat" + | "removeProject" + | "fetchChatTranscript" + | "readFile" + | "connect" + | "close" +>; + +const MB = 1024 * 1024; +// Synthetic bytes only; names and types match the reported cases. +const FILES = [ + { name: "clip.mp4", mimeType: "video/mp4", buffer: randomBytes(6 * MB) }, + { name: "report.pdf", mimeType: "application/pdf", buffer: randomBytes(300 * 1024) }, + { name: "bundle.zip", mimeType: "application/zip", buffer: randomBytes(2 * MB) }, + { name: "data.json", mimeType: "application/json", buffer: Buffer.from('{"ok":true}') }, +]; + +async function withDirectBotChat(run: (client: BotClient, chatId: string) => Promise) { + const client = await connectDaemonClient({ clientIdPrefix: "chat-file-types" }); + let botId: string | undefined; + let projectId: string | undefined; + let chatId: string | undefined; + try { + const created = await client.createBot({ + name: "File types fixture", + launch: { + provider: "mock", + model: "e2e-fast-stream", + modeId: "load-test", + featureValues: { mockAssistantResponses: ["Got it"] }, + }, + }); + if (!created.bot) throw new Error(created.error ?? "Bot creation failed"); + botId = created.bot.id; + projectId = created.bot.projectId; + const chat = await client.createChat({ botIds: [botId], kind: "direct" }); + if (!chat.chat) throw new Error(chat.error ?? "Chat creation failed"); + chatId = chat.chat.id; + await run(client, chatId); + } finally { + if (chatId) await client.archiveChat({ chatId }).catch(() => undefined); + if (botId) await client.archiveBot({ botId }); + if (projectId) await client.removeProject(projectId); + await client.close(); + } +} + +// One drop carrying every file, as a user drags a Finder selection; the composer refuses a +// second drop while the first one is still uploading. +async function dropFilesOnComposer(page: Page, files: typeof FILES): Promise { + const dataTransfer = await page.evaluateHandle( + (entries) => { + const transfer = new DataTransfer(); + for (const { name, mimeType, base64 } of entries) { + const bytes = Uint8Array.from(atob(base64), (char) => char.charCodeAt(0)); + transfer.items.add(new File([bytes], name, { type: mimeType })); + } + return transfer; + }, + files.map((file) => ({ + name: file.name, + mimeType: file.mimeType, + base64: file.buffer.toString("base64"), + })), + ); + const composerRoot = page.getByTestId("message-input-root").filter({ visible: true }).first(); + await expect(composerRoot).toBeVisible({ timeout: 10_000 }); + for (const type of ["dragenter", "dragover", "drop"]) { + await composerRoot.dispatchEvent(type, { dataTransfer }); + } + await dataTransfer.dispose(); +} + +async function sentLine(client: BotClient, chatId: string) { + const { lines } = await client.fetchChatTranscript({ chatId }); + return lines.find((line) => line.text === "Files attached"); +} + +function collectErrors(page: Page): string[] { + const errors: string[] = []; + page.on("console", (message) => { + if (message.type() === "error") errors.push(message.text()); + }); + return errors; +} + +for (const via of ["menu", "drop"] as const) { + test(`Bot chat sends MP4, PDF, zip and JSON chosen by ${via}`, async ({ page }) => { + const errors = collectErrors(page); + await withDirectBotChat(async (client, chatId) => { + await page.goto(`/h/${getServerId()}/chat/${chatId}`); + const scope = page.getByTestId("chat-composer"); + const input = scope.getByRole("textbox").last(); + await expect(input).toBeEditable({ timeout: 30_000 }); + if (via === "menu") for (const file of FILES) await attachFileFromMenu(page, file); + else await dropFilesOnComposer(page, FILES); + await expect(scope.getByTestId("composer-file-attachment-pill")).toHaveCount(FILES.length, { + timeout: 30_000, + }); + await input.fill("Files attached"); + await scope.getByTestId("composer-send-button").click(); + await expect + .poll(async () => (await sentLine(client, chatId))?.attachments?.length ?? 0, { + timeout: 60_000, + }) + .toBe(FILES.length); + const line = (await sentLine(client, chatId))!; + for (const file of FILES) { + const stored = line.attachments!.find( + (attachment) => attachment.type === "uploaded_file" && attachment.fileName === file.name, + ); + if (stored?.type !== "uploaded_file") throw new Error(`Missing ${file.name}`); + expect(stored.mimeType).toBe(file.mimeType); + expect(stored.size).toBe(file.buffer.byteLength); + const read = await client.readFile( + stored.path.slice(0, stored.path.lastIndexOf("/")), + stored.path, + ); + expect(Buffer.from(read.bytes).equals(file.buffer)).toBe(true); + } + }); + expect(errors.filter((text) => /upload|attachment/i.test(text))).toEqual([]); + }); +} + +test("agent composer sends MP4, PDF, zip and JSON chosen from the menu", async ({ page }) => { + const errors = collectErrors(page); + const agent = await seedMockAgentWorkspace({ repoPrefix: "file-types-", title: "File types" }); + try { + await openAgentRoute(page, agent); + const input = page.getByTestId("message-input-root").filter({ visible: true }).first(); + await expect(input).toBeVisible({ timeout: 30_000 }); + for (const file of FILES) await attachFileFromMenu(page, file); + await expect( + page.getByTestId("composer-file-attachment-pill").filter({ visible: true }), + ).toHaveCount(FILES.length, { timeout: 30_000 }); + await input.getByRole("textbox").fill("Agent files attached"); + await input.getByRole("textbox").press("Enter"); + const sent = page.getByTestId("user-message").filter({ hasText: "Agent files attached" }); + await expect(sent).toBeVisible({ timeout: 30_000 }); + for (const file of FILES) await expect(sent).toContainText(file.name); + await expect(page.getByText(/upload failed|failed to upload/i)).toHaveCount(0); + } finally { + await agent.cleanup(); + } + expect(errors.filter((text) => /upload|attachment/i.test(text))).toEqual([]); +}); diff --git a/packages/app/src/clisbot/bots/chat/chat-screen.test.tsx b/packages/app/src/clisbot/bots/chat/chat-screen.test.tsx index 7ad445aae..029cd1254 100644 --- a/packages/app/src/clisbot/bots/chat/chat-screen.test.tsx +++ b/packages/app/src/clisbot/bots/chat/chat-screen.test.tsx @@ -82,6 +82,9 @@ vi.mock("react-native", () => ({ ), })); +vi.mock("@/components/file-drop/file-drop-zone", () => ({ + FileDropZone: ({ children }: { children: ReactNode }) =>
{children}
, +})); vi.mock("@/components/headers/menu-header", () => ({ MenuHeader: ({ title }: { title: string }) =>

{title}

, })); diff --git a/packages/app/src/clisbot/bots/chat/chat-screen.tsx b/packages/app/src/clisbot/bots/chat/chat-screen.tsx index 2ce884a3e..e15d69151 100644 --- a/packages/app/src/clisbot/bots/chat/chat-screen.tsx +++ b/packages/app/src/clisbot/bots/chat/chat-screen.tsx @@ -2,6 +2,7 @@ import type { MessagePayload } from "@/composer/types"; import { profileNativeRegion } from "@/utils/render-profiler"; import { ToolCallSheetProvider } from "@/components/tool-call-sheet"; import { ComposerDock } from "@/composer/dock"; +import { FileDropZone } from "@/components/file-drop/file-drop-zone"; import { useMemo, type ReactNode } from "react"; import { useTranslation } from "react-i18next"; import { View } from "react-native"; @@ -73,36 +74,40 @@ export function ChatScreen({ return ( {!hideHeader ? : null} - - - {profileNativeRegion( - "BotChatHistory", - , - )} - - head.turnActive)} - hasPermission={[...liveHeads.values()].some((head) => head.permissions.length > 0)} - onSubmitMessage={onSubmitMessage} - mentionMembers={group ? members : undefined} - /> - + {/* The composer's file drop sink needs this zone; without it a dropped file was ignored. */} + + + + {profileNativeRegion( + "BotChatHistory", + , + )} + + head.turnActive)} + hasPermission={[...liveHeads.values()].some((head) => head.permissions.length > 0)} + onSubmitMessage={onSubmitMessage} + mentionMembers={group ? members : undefined} + /> + + ); } const styles = StyleSheet.create((theme) => ({ container: { flex: 1, backgroundColor: theme.colors.surface0 }, + dropZone: { flex: 1 }, })); diff --git a/packages/server/src/server/file-upload/index.test.ts b/packages/server/src/server/file-upload/index.test.ts index d1d5002f9..9eb9bf9ca 100644 --- a/packages/server/src/server/file-upload/index.test.ts +++ b/packages/server/src/server/file-upload/index.test.ts @@ -281,6 +281,26 @@ describe("file uploads", () => { }); expect(readFileSync(path, "utf8")).toBe("hello world"); }); + + it("keeps an unsent draft linkable while other uploads fill the store", async () => { + const clisbotHome = makeClisbotHome(); + const uploads = new FileUploadStore({ clisbotHome }); + // New workspace drafts send the upload made when the file was chosen. + const draft = await uploadNamed(uploads, "draft.pdf"); + for (let index = 1; index < 256; index++) await uploadNamed(uploads, `other-${index}.pdf`); + + expect(() => + uploads.beginUpload({ + type: "file.upload.request", + fileName: "one-more.pdf", + mimeType: "text/plain", + size: 11, + modifiedAt: "2026-05-02T00:00:00.000Z", + requestId: "req-over-cap", + }), + ).toThrow("Upload storage overloaded: too many unfinished drafts"); + expect(draft && uploads.ownsUploadedFile(draft)).toBe(true); + }); }); let uploadCount = 0; diff --git a/packages/server/src/server/file-upload/index.ts b/packages/server/src/server/file-upload/index.ts index 2d9a73fbf..fb643116d 100644 --- a/packages/server/src/server/file-upload/index.ts +++ b/packages/server/src/server/file-upload/index.ts @@ -152,6 +152,26 @@ export class FileUploadStore { return operation; } + hasPendingUpload(requestId: string, source: object = this.defaultSource): boolean { + return this.pending.get(source)?.has(requestId) === true; + } + + /** Ends an unfinished upload and answers its request with `error`. */ + async failUpload( + requestId: string, + source: object = this.defaultSource, + error: string, + ): Promise { + const upload = this.pending.get(source)?.get(requestId); + if (upload) await this.cancel(upload, buildUploadResponse(upload, error)); + } + + /** Ends an unfinished upload without answering; its request was already answered. */ + async discardUpload(requestId: string, source: object = this.defaultSource): Promise { + const upload = this.pending.get(source)?.get(requestId); + if (upload) await this.cancel(upload); + } + ownsUploadedFile(attachment: UploadedFileAttachment): boolean { const completed = this.completed.get(attachment.id); return ( diff --git a/packages/server/src/server/managed-access/file-upload-order.e2e.test.ts b/packages/server/src/server/managed-access/file-upload-order.e2e.test.ts new file mode 100644 index 000000000..d1058f392 --- /dev/null +++ b/packages/server/src/server/managed-access/file-upload-order.e2e.test.ts @@ -0,0 +1,342 @@ +import { mkdir, mkdtemp, readFile, readdir, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { randomBytes } from "node:crypto"; +import { afterEach, expect, test } from "vitest"; +import { + decodeFileTransferFrame, + encodeFileTransferFrame, + FileTransferOpcode, + type FileTransferFrame, +} from "@clisbot/protocol/binary-frames/index"; +import { CreationService } from "../creation/index.js"; +import { MessageReceipts } from "../message-receipts/index.js"; +import { Session, type SessionOptions } from "../session.js"; +import type { SessionOutboundMessage } from "../messages.js"; +import { FileBackedProjectRegistry, FileBackedWorkspaceRegistry } from "../workspace-registry.js"; +import { WorkspaceGitServiceImpl } from "../workspace-git-service.js"; +import { createTestClisbotDaemon } from "../test-utils/clisbot-daemon.js"; +import { createStub } from "../test-utils/class-mocks.js"; +import { createProviderSnapshotManagerStub } from "../test-utils/session-stubs.js"; +import { heldUploadFrameBytes } from "../session/files/workspace-files-session.js"; + +function frame(input: Parameters[0]): FileTransferFrame { + const decoded = decodeFileTransferFrame(encodeFileTransferFrame(input)); + if (!decoded) throw new Error("Invalid file transfer frame"); + return decoded; +} + +const MiB = 1024 * 1024; +const CHUNK = 128 * 1024; +const PERMISSION_DENIED = + "You do not have permission to perform this action. Ask your administrator for access."; +const cleanups: Array<() => Promise> = []; + +afterEach(async () => { + for (const cleanup of cleanups.splice(0)) await cleanup(); +}); + +// A Project-scoped (Hub Member) session, the way the Hub admits a Member's app. +async function scopedSession( + options: { granted?: boolean; authorization?: Promise; heldUploadTimeoutMs?: number } = {}, +) { + const root = await mkdtemp(path.join(tmpdir(), "managed-upload-order-")); + const repo = path.join(root, "repo"); + await mkdir(repo); + const daemon = await createTestClisbotDaemon(); + cleanups.push(async () => { + await daemon.close(); + await rm(root, { recursive: true, force: true }); + }); + const logger = pino({ level: "silent" }); + const projects = new FileBackedProjectRegistry(path.join(root, "projects.json"), logger); + const workspaces = new FileBackedWorkspaceRegistry(path.join(root, "workspaces.json"), logger); + await Promise.all([projects.initialize(), workspaces.initialize()]); + const project = await projects.getOrCreateActiveByRoot({ + rootPath: repo, + kind: "non_git", + displayName: "Repo", + timestamp: new Date().toISOString(), + }); + // Resource authorization first loads the Project registry; this is where it can stall. + if (options.authorization) { + const list = projects.list.bind(projects); + const gate = options.authorization; + projects.list = async () => { + await gate; + return list(); + }; + } + const grant = { + privileges: new Set(["project.use", "agent.interact"] as const), + agentConfigurations: [], + }; + const messages: SessionOutboundMessage[] = []; + const session = new Session({ + clientId: "managed-upload-client", + permissions: ["daemon.read", "workspace.read", "workspace.write"], + resourceAuthorization: { + resourceMode: "projects", + leaseId: "managed-lease", + leaseExpiresAt: Date.now() + 60_000, + projects: new Map(options.granted === false ? [] : [[project.projectId, grant]]), + }, + ...(options.heldUploadTimeoutMs ? { heldUploadTimeoutMs: options.heldUploadTimeoutMs } : {}), + logger, + clisbotHome: daemon.clisbotHome, + creationService: new CreationService(path.join(root, "creation-receipts"), logger), + messageReceipts: new MessageReceipts(path.join(root, "message-receipts")), + onMessage: (message) => messages.push(message), + downloadTokenStore: createStub({}), + pushNotifications: createStub({}), + agentManager: daemon.daemon.agentManager, + agentStorage: daemon.daemon.agentStorage, + projectRegistry: projects, + workspaceRegistry: workspaces, + workspaceGitService: new WorkspaceGitServiceImpl({ logger, clisbotHome: daemon.clisbotHome }), + terminalManager: null, + scheduleService: createStub({}), + checkoutDiffManager: createStub({ + scheduleRefreshForCwd: () => {}, + onWorkspaceStateMayHaveChanged: () => {}, + }), + workspaceAutoName: createStub({ + scheduleForWorktree: () => {}, + scheduleForDirectory: () => {}, + }), + daemonConfigStore: createStub({ + get: () => ({ mcp: { injectIntoAgents: false }, providers: {} }), + onChange: () => () => {}, + }), + providerSnapshotManager: createProviderSnapshotManagerStub().manager, + stt: null, + tts: null, + }); + const connect = () => { + const source = {}; + session.updateClientCapabilities({}, source); + return source; + }; + return { session, messages, connect, clisbotHome: daemon.clisbotHome }; +} + +type Scoped = Awaited>; + +function request(scoped: Scoped, source: object, requestId: string, size: number) { + return scoped.session.handleMessage( + { + type: "file.upload.request", + requestId, + fileName: "clip.mp4", + mimeType: "video/mp4", + size, + modifiedAt: "2026-11-04T00:00:00.000Z", + }, + source, + ); +} + +/** The client's frames for `bytes`: begin, 128 KiB chunks, end. */ +function uploadFrames(requestId: string, bytes: Uint8Array): FileTransferFrame[] { + const frames = [ + frame({ + opcode: FileTransferOpcode.FileBegin, + requestId, + metadata: { + mime: "video/mp4", + size: bytes.byteLength, + encoding: "binary", + modifiedAt: "2026-11-04T00:00:00.000Z", + fileName: "clip.mp4", + }, + }), + ]; + for (let offset = 0; offset < bytes.byteLength; offset += CHUNK) + frames.push( + frame({ + opcode: FileTransferOpcode.FileChunk, + requestId, + payload: bytes.subarray(offset, offset + CHUNK), + }), + ); + frames.push(frame({ opcode: FileTransferOpcode.FileEnd, requestId })); + return frames; +} + +/** Delivered the way the websocket server does: neither awaited nor ordered against JSON. */ +function deliver(scoped: Scoped, source: object, frames: FileTransferFrame[]) { + return frames.map((next) => + scoped.session.handleBinaryFrame({ kind: "file_transfer", frame: next }, source), + ); +} + +async function outcome(scoped: Scoped, requestId: string) { + await expect.poll(() => findOutcome(scoped, requestId), { timeout: 20_000 }).toBeTruthy(); + return findOutcome(scoped, requestId)!; +} + +function answers(scoped: Scoped, requestId: string) { + return scoped.messages.filter( + (message) => + (message.type === "file.upload.response" || message.type === "rpc_error") && + message.payload.requestId === requestId, + ); +} + +function findOutcome(scoped: Scoped, requestId: string) { + for (const message of scoped.messages) { + if (message.type === "file.upload.response" && message.payload.requestId === requestId) + return { error: message.payload.error, path: message.payload.file?.path ?? null }; + if (message.type === "rpc_error" && message.payload.requestId === requestId) + return { error: message.payload.error, path: null }; + } + return undefined; +} + +async function uploadedFiles(clisbotHome: string): Promise { + return await readdir(path.join(clisbotHome, "uploads")).catch(() => []); +} + +for (const order of ["after the request settles", "before the request is authorized"] as const) + test(`a Project-scoped session stores a small file whose frames arrive ${order}`, async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const bytes = new TextEncoder().encode('{"tokens":true}'); + const pending = request(scoped, source, "small", bytes.byteLength); + if (order === "after the request settles") await pending; + await Promise.all([pending, ...deliver(scoped, source, uploadFrames("small", bytes))]); + const result = await outcome(scoped, "small"); + expect(result.error).toBeNull(); + expect(await readFile(result.path!, "utf8")).toBe('{"tokens":true}'); + }); + +test("a 49 MiB file whose every frame arrives before authorization is stored whole", async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const bytes = randomBytes(49 * MiB); + const pending = request(scoped, source, "whole", bytes.byteLength); + const frames = uploadFrames("whole", bytes); + expect(frames.length).toBeGreaterThan(64); + await Promise.all([pending, ...deliver(scoped, source, frames)]); + const result = await outcome(scoped, "whole"); + expect(result.error).toBeNull(); + expect((await readFile(result.path!)).equals(bytes)).toBe(true); +}, 60_000); + +test("a 12 MiB file split across authorization keeps its frame order", async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const bytes = randomBytes(12 * MiB); + const frames = uploadFrames("split", bytes); + const pending = request(scoped, source, "split", bytes.byteLength); + const early = deliver(scoped, source, frames.slice(0, 80)); + await pending; + // The rest lands while the held frames are still being written. + await Promise.all([...early, ...deliver(scoped, source, frames.slice(80))]); + const result = await outcome(scoped, "split"); + expect(result.error).toBeNull(); + expect((await readFile(result.path!)).equals(bytes)).toBe(true); +}, 60_000); + +test("held bytes beyond the declared size end the upload with an error, not a hang", async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const pending = request(scoped, source, "oversize", 10); + await Promise.all([ + pending, + ...deliver(scoped, source, uploadFrames("oversize", randomBytes(MiB))), + ]); + expect((await outcome(scoped, "oversize")).error).toMatch(/overloaded/); + expect(await uploadedFiles(scoped.clisbotHome)).toEqual([]); +}); + +test("a refused upload request discards its held frames and writes nothing", async () => { + const scoped = await scopedSession({ granted: false }); + const source = scoped.connect(); + const bytes = randomBytes(MiB); + const pending = request(scoped, source, "refused", bytes.byteLength); + await Promise.all([pending, ...deliver(scoped, source, uploadFrames("refused", bytes))]); + // The request is answered by the ordinary refusal; its frames reach nothing. + expect((await outcome(scoped, "refused")).error).toBeTruthy(); + expect(await uploadedFiles(scoped.clisbotHome)).toEqual([]); +}); + +test("losing workspace.write while frames are held ends the upload and stores nothing", async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const bytes = randomBytes(MiB); + const pending = request(scoped, source, "revoked", bytes.byteLength); + const frames = deliver(scoped, source, uploadFrames("revoked", bytes)); + scoped.session.setPermissions(["daemon.read", "workspace.read"]); + await Promise.all([pending, ...frames]); + expect((await outcome(scoped, "revoked")).error).toBe(PERMISSION_DENIED); + expect(await uploadedFiles(scoped.clisbotHome)).toEqual([]); +}); + +test("frames from another connection with the same request id do not join a held upload", async () => { + const scoped = await scopedSession(); + const source = scoped.connect(); + const other = scoped.connect(); + const bytes = randomBytes(2 * MiB); + const pending = request(scoped, source, "shared-id", bytes.byteLength); + const intruder = deliver(scoped, other, uploadFrames("shared-id", randomBytes(2 * MiB))); + await Promise.all([ + pending, + ...intruder, + ...deliver(scoped, source, uploadFrames("shared-id", bytes)), + ]); + const result = await outcome(scoped, "shared-id"); + expect(result.error).toBeNull(); + expect((await readFile(result.path!)).equals(bytes)).toBe(true); +}); + +test("an upload whose authorization never finishes is answered once within the timeout", async () => { + const scoped = await scopedSession({ + authorization: new Promise(() => {}), + heldUploadTimeoutMs: 300, + }); + const source = scoped.connect(); + const bytes = randomBytes(3 * MiB); + void request(scoped, source, "stalled", bytes.byteLength); + void Promise.all(deliver(scoped, source, uploadFrames("stalled", bytes))); + expect((await outcome(scoped, "stalled")).error).toMatch(/timed out waiting for authorization/); + await new Promise((resolve) => setTimeout(resolve, 600)); + expect(answers(scoped, "stalled")).toHaveLength(1); + expect(heldUploadFrameBytes()).toBe(0); + expect(await uploadedFiles(scoped.clisbotHome)).toEqual([]); +}); + +for (const late of ["admitted", "refused"] as const) + test(`authorization that arrives after the timeout answer (${late}) adds no answer and no upload`, async () => { + let authorize!: () => void; + const scoped = await scopedSession({ + granted: late === "admitted", + authorization: new Promise((resolve) => (authorize = resolve)), + heldUploadTimeoutMs: 300, + }); + const source = scoped.connect(); + const bytes = randomBytes(2 * MiB); + const pending = request(scoped, source, "late", bytes.byteLength); + const frames = uploadFrames("late", bytes); + void Promise.all(deliver(scoped, source, frames.slice(0, 4))); + expect((await outcome(scoped, "late")).error).toMatch(/timed out/); + // The client stopped at the answer; frames already on the wire still land. + void Promise.all(deliver(scoped, source, frames.slice(4))); + authorize(); + await pending; + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(answers(scoped, "late")).toHaveLength(1); + expect(heldUploadFrameBytes()).toBe(0); + expect(await uploadedFiles(scoped.clisbotHome)).toEqual([]); + if (late === "refused") return; + // The cancelled late upload holds no slot: the next file goes through. + const next = randomBytes(MiB); + await Promise.all([ + request(scoped, source, "after", next.byteLength), + ...deliver(scoped, source, uploadFrames("after", next)), + ]); + const result = await outcome(scoped, "after"); + expect(result.error).toBeNull(); + expect((await readFile(result.path!)).equals(next)).toBe(true); + }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 95f0bdbe2..8685cd4c0 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -497,6 +497,8 @@ export interface SessionOptions { clientId: string; permissions: readonly DaemonPermission[]; resourceAuthorization?: SessionResourceAuthorization; + /** Tests shorten how long an upload request may wait for authorization. */ + heldUploadTimeoutMs?: number; appVersion?: string | null; clientCapabilities?: Record | null; onMessage: (msg: SessionOutboundMessage) => void; @@ -1037,6 +1039,7 @@ export class Session { clisbotHome, logger: this.sessionLogger, sessionStorageEnabled: () => this.agentManager.sessionStorageEnabled, + heldUploadTimeoutMs: options.heldUploadTimeoutMs, resolveForkSource: async (source) => { if (!(await this.resourceAuthorizer.allowsAgent(source.agentId, "project.use"))) throw new Error("Fork source not found"); @@ -1531,6 +1534,7 @@ export class Session { } clearAgentTimelineSubscription(source: object): void { + this.workspaceFilesSession.releaseSource(source); this.clientCapabilitiesBySource.delete(source); void this.delivery .detach(source) @@ -2490,8 +2494,25 @@ export class Session { ); } + /** A timed-out upload request already has its answer; a late outcome must not add another. */ + private uploadRequestAnswered(msg: SessionInboundMessage, source?: object): boolean { + return ( + msg.type === "file.upload.request" && + this.workspaceFilesSession.uploadRequestAnswered(msg.requestId, source) + ); + } + public async handleMessage(msg: SessionInboundMessage, source?: object): Promise { - return this.delivery.request(source, msg, () => this.handleRequest(msg, source)); + // Registered before the first await: the upload's binary frames may already be next. + const releaseUploadFrames = + msg.type === "file.upload.request" + ? this.workspaceFilesSession.holdUploadFrames(msg, source) + : undefined; + try { + return await this.delivery.request(source, msg, () => this.handleRequest(msg, source)); + } finally { + releaseUploadFrames?.(); + } } private async handleRequest(msg: SessionInboundMessage, source?: object): Promise { @@ -2516,7 +2537,7 @@ export class Session { : true; if (!operationAllowed || !resourceAllowed) { const requestId = sessionRequestId(msg); - if (requestId) { + if (requestId && !this.uploadRequestAnswered(msg, source)) { const code = resourceAllowed ? "access_denied" : await this.resourceAuthorizer.denialCode(msg); @@ -2546,7 +2567,7 @@ export class Session { const requestId = "requestId" in msg && typeof msg.requestId === "string" ? msg.requestId : undefined; - if (typeof requestId === "string") { + if (typeof requestId === "string" && !this.uploadRequestAnswered(msg, source)) { try { this.emit({ type: "rpc_error", @@ -3823,16 +3844,22 @@ export class Session { this.connectedHostSession?.binary(binaryFrame.frame, source); return; } + if (binaryFrame.kind === "file_transfer") { + await this.workspaceFilesSession.handleFileTransferFrame( + binaryFrame.frame, + source, + () => + this.authorization.isLeaseActive() && + this.authorization.allowsPermission("workspace.write") && + this.resourceAuthorizer.allowsAnyProject(), + ); + return; + } if ( !this.authorization.isLeaseActive() || !this.authorization.allowsPermission("workspace.write") ) return; - if (binaryFrame.kind === "file_transfer") { - if (!this.resourceAuthorizer.allowsAnyProject()) return; - await this.workspaceFilesSession.handleFileTransferFrame(binaryFrame.frame, source); - return; - } this.terminalController.handleBinaryFrame(binaryFrame.frame, source); } diff --git a/packages/server/src/server/session/files/held-upload-frames.test.ts b/packages/server/src/server/session/files/held-upload-frames.test.ts new file mode 100644 index 000000000..a59ab4fb0 --- /dev/null +++ b/packages/server/src/server/session/files/held-upload-frames.test.ts @@ -0,0 +1,272 @@ +import { mkdtempSync, realpathSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import pino from "pino"; +import { afterEach, expect, test, vi } from "vitest"; +import { FileTransferOpcode, type FileTransferFrame } from "@clisbot/protocol/binary-frames/index"; +import { DownloadTokenStore } from "../../file-download/token-store.js"; +import type { SessionOutboundMessage } from "../../messages.js"; +import { SessionDelivery } from "../owned-subscriptions/index.js"; +import { heldUploadFrameBytes, WorkspaceFilesSession } from "./workspace-files-session.js"; + +const MiB = 1024 * 1024; +const CHUNK = 128 * 1024; +// One shared payload keeps large holds cheap: the hold counts bytes, it never copies them. +const PAYLOAD = new Uint8Array(CHUNK); +const homes: string[] = []; +const subsystems: WorkspaceFilesSession[] = []; + +afterEach(() => { + vi.useRealTimers(); + for (const subsystem of subsystems.splice(0)) subsystem.dispose(); + for (const home of homes.splice(0)) rmSync(home, { recursive: true, force: true }); + // Every test must leave the process-wide budget as it found it. + expect(heldUploadFrameBytes()).toBe(0); +}); + +function setup(options: { heldUploadTimeoutMs?: number } = {}) { + const home = realpathSync(mkdtempSync(join(tmpdir(), "held-upload-"))); + homes.push(home); + const emitted: SessionOutboundMessage[] = []; + const subsystem = new WorkspaceFilesSession({ + host: { + emit: (message) => emitted.push(message), + emitBinary: async () => {}, + hasBinaryChannel: () => false, + }, + downloadTokenStore: new DownloadTokenStore({ ttlMs: 60_000 }), + clisbotHome: home, + logger: pino({ level: "silent" }), + ...options, + }); + subsystems.push(subsystem); + const ownership = new SessionDelivery((_source, message) => emitted.push(message)); + const connect = () => { + const source = {}; + ownership.attach(source, true); + return source; + }; + const admit = (source: object, requestId: string, size: number) => { + const request = { + type: "file.upload.request", + fileName: "clip.mp4", + mimeType: "video/mp4", + size, + modifiedAt: "2026-11-04T00:00:00.000Z", + requestId, + } as const; + return ownership.request(source, request, async () => + subsystem.handleFileUploadRequest(request, ownership), + ); + }; + const send = (source: object, batch: FileTransferFrame[]) => + batch.map((frame) => subsystem.handleFileTransferFrame(frame, source, () => true)); + const answer = (requestId: string) => + emitted.find( + (message) => + (message.type === "file.upload.response" || message.type === "rpc_error") && + message.payload.requestId === requestId, + ); + const answers = (requestId: string) => + emitted.filter( + (message) => + (message.type === "file.upload.response" || message.type === "rpc_error") && + message.payload.requestId === requestId, + ); + return { subsystem, ownership, connect, admit, send, answer, answers }; +} + +function frames(requestId: string, size: number): FileTransferFrame[] { + const result: FileTransferFrame[] = [ + { + opcode: FileTransferOpcode.FileBegin, + requestId, + metadata: { + mime: "video/mp4", + size, + encoding: "binary", + modifiedAt: "2026-11-04T00:00:00.000Z", + fileName: "clip.mp4", + }, + }, + ]; + for (let offset = 0; offset < size; offset += CHUNK) + result.push({ + opcode: FileTransferOpcode.FileChunk, + requestId, + payload: PAYLOAD.subarray(0, Math.min(CHUNK, size - offset)), + }); + result.push({ opcode: FileTransferOpcode.FileEnd, requestId }); + return result; +} + +test("a drained hold returns every byte it counted", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "drained", size: 3 * MiB }, source); + const delivered = t.send(source, frames("drained", 3 * MiB)); + expect(heldUploadFrameBytes()).toBe(3 * MiB); + await t.admit(source, "drained", 3 * MiB); + release(); + await Promise.all(delivered); + await expect.poll(() => t.answer("drained")?.type).toBe("file.upload.response"); + expect(heldUploadFrameBytes()).toBe(0); +}); + +test("a refused request returns its held bytes and writes nothing", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "refused", size: MiB }, source); + const delivered = t.send(source, frames("refused", MiB)); + release(); // The request was refused: no upload was begun. + await Promise.all(delivered); + expect(heldUploadFrameBytes()).toBe(0); + expect(t.answer("refused")).toBeUndefined(); +}); + +test("a connection that closes while its request waits releases the hold", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "closed", size: 2 * MiB }, source); + void Promise.all(t.send(source, frames("closed", 2 * MiB))); + expect(heldUploadFrameBytes()).toBe(2 * MiB); + t.subsystem.releaseSource(source); + expect(heldUploadFrameBytes()).toBe(0); + // A late frame finds no hold, and the late settle has nothing left to drain. + await Promise.all(t.send(source, frames("closed", CHUNK).slice(1, 2))); + release(); + expect(heldUploadFrameBytes()).toBe(0); +}); + +test("a request that never settles gives its hold up", async () => { + vi.useFakeTimers(); + const t = setup(); + const source = t.connect(); + t.subsystem.holdUploadFrames({ requestId: "stuck", size: MiB }, source); + void Promise.all(t.send(source, frames("stuck", MiB))); + expect(heldUploadFrameBytes()).toBe(MiB); + await vi.advanceTimersByTimeAsync(60_000); + expect(heldUploadFrameBytes()).toBe(0); + expect(t.answers("stuck")).toEqual([ + expect.objectContaining({ + type: "rpc_error", + payload: expect.objectContaining({ code: "upload_authorization_timeout" }), + }), + ]); + expect(t.subsystem.uploadRequestAnswered("stuck", source)).toBe(true); + // Frames still on the wire are dropped by the marker, not held again. + await Promise.all(t.send(source, frames("stuck", CHUNK).slice(1, 2))); + expect(heldUploadFrameBytes()).toBe(0); +}); + +test("a connection that closes while held frames drain stops the drain and returns the bytes", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "midway", size: 16 * MiB }, source); + const delivered = t.send(source, frames("midway", 16 * MiB)); + await t.admit(source, "midway", 16 * MiB); + release(); + // Socket close, as the websocket server does it: release the hold, then the delivery source. + t.subsystem.releaseSource(source); + await t.ownership.detach(source); + await Promise.all(delivered); + expect(heldUploadFrameBytes()).toBe(0); + expect( + t + .answers("midway") + .some((message) => message.type === "file.upload.response" && message.payload.error === null), + ).toBe(false); +}); + +test("the authorization timeout cannot fire once held frames are draining", async () => { + const t = setup({ heldUploadTimeoutMs: 50 }); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "slow", size: 32 * MiB }, source); + const delivered = t.send(source, frames("slow", 32 * MiB)); + await t.admit(source, "slow", 32 * MiB); + release(); + await new Promise((resolve) => setTimeout(resolve, 150)); + await Promise.all(delivered); + await expect.poll(() => t.answer("slow")?.type, { timeout: 30_000 }).toBe("file.upload.response"); + expect(t.answers("slow")).toHaveLength(1); + const answer = t.answer("slow"); + expect(answer?.type === "file.upload.response" && answer.payload.error).toBeNull(); + await expect.poll(() => heldUploadFrameBytes()).toBe(0); +}, 60_000); + +test("disposing the session releases every connection's holds", () => { + const t = setup(); + const [first, second] = [t.connect(), t.connect()]; + t.subsystem.holdUploadFrames({ requestId: "a", size: MiB }, first); + t.subsystem.holdUploadFrames({ requestId: "b", size: MiB }, second); + void Promise.all([...t.send(first, frames("a", MiB)), ...t.send(second, frames("b", MiB))]); + expect(heldUploadFrameBytes()).toBe(2 * MiB); + t.subsystem.dispose(); + expect(heldUploadFrameBytes()).toBe(0); +}); + +test("the shared budget refuses a third full hold with an answer, then recovers", async () => { + const sessions = [setup(), setup(), setup()]; + const sources = sessions.map((t) => t.connect()); + const releases = sessions.map((t, index) => + t.subsystem.holdUploadFrames({ requestId: `full-${index}`, size: 64 * MiB }, sources[index]!), + ); + const delivered = sessions.flatMap((t, index) => + t.send(sources[index]!, frames(`full-${index}`, 64 * MiB)), + ); + expect(heldUploadFrameBytes()).toBe(128 * MiB); + const third = sessions[2]!; + await third.admit(sources[2]!, "full-2", 64 * MiB); + releases[2]!(); + await expect.poll(() => third.answer("full-2")?.type).toBe("file.upload.response"); + const answer = third.answer("full-2"); + expect(answer?.type === "file.upload.response" && answer.payload.error).toMatch(/overloaded/); + for (const [index, t] of sessions.slice(0, 2).entries()) { + await t.admit(sources[index]!, `full-${index}`, 64 * MiB); + releases[index]!(); + } + await Promise.all(delivered); + // The two admitted holds drain to disk in the background. + await expect.poll(() => heldUploadFrameBytes(), { timeout: 30_000 }).toBe(0); + for (const [index, t] of sessions.slice(0, 2).entries()) { + await expect + .poll(() => t.answer(`full-${index}`)?.type, { timeout: 30_000 }) + .toBe("file.upload.response"); + const admitted = t.answer(`full-${index}`); + expect(admitted?.type === "file.upload.response" && admitted.payload.error).toBeNull(); + } +}, 60_000); + +test("a second request with the same id on the same connection joins the first hold", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "dup", size: MiB }, source); + const releaseDuplicate = t.subsystem.holdUploadFrames({ requestId: "dup", size: MiB }, source); + const delivered = t.send(source, frames("dup", MiB)); + expect(heldUploadFrameBytes()).toBe(MiB); + releaseDuplicate(); + expect(heldUploadFrameBytes()).toBe(MiB); + await t.admit(source, "dup", MiB); + release(); + await Promise.all(delivered); + await expect.poll(() => t.answer("dup")?.type).toBe("file.upload.response"); + expect(heldUploadFrameBytes()).toBe(0); +}); + +test("overflow while held frames drain ends the upload and returns the bytes", async () => { + const t = setup(); + const source = t.connect(); + const release = t.subsystem.holdUploadFrames({ requestId: "spill", size: MiB }, source); + const early = t.send(source, frames("spill", MiB).slice(0, 5)); + await t.admit(source, "spill", MiB); + release(); + // Past the declared megabyte while the first frames are still being written. + const late = t.send(source, frames("spill", 2 * MiB).slice(1)); + await Promise.all([...early, ...late]); + await expect.poll(() => t.answer("spill")?.type).toBe("file.upload.response"); + const answer = t.answer("spill"); + expect(answer?.type === "file.upload.response" && answer.payload.error).toMatch( + /overloaded|exceeded declared size/, + ); + expect(heldUploadFrameBytes()).toBe(0); +}); diff --git a/packages/server/src/server/session/files/workspace-files-session.test.ts b/packages/server/src/server/session/files/workspace-files-session.test.ts index df80dfdf2..0c2a0e698 100644 --- a/packages/server/src/server/session/files/workspace-files-session.test.ts +++ b/packages/server/src/server/session/files/workspace-files-session.test.ts @@ -596,6 +596,7 @@ describe("WorkspaceFilesSession", () => { }, }), source, + () => true, ); await subsystem.handleFileTransferFrame( uploadFrame({ @@ -604,6 +605,7 @@ describe("WorkspaceFilesSession", () => { payload: new TextEncoder().encode("hello world"), }), source, + () => true, ); await subsystem.handleFileTransferFrame( uploadFrame({ @@ -611,6 +613,7 @@ describe("WorkspaceFilesSession", () => { requestId: "req-upload", }), source, + () => true, ); const message = emitted.find((entry) => entry.type === "file.upload.response"); diff --git a/packages/server/src/server/session/files/workspace-files-session.ts b/packages/server/src/server/session/files/workspace-files-session.ts index 62fa63645..75a14ba41 100644 --- a/packages/server/src/server/session/files/workspace-files-session.ts +++ b/packages/server/src/server/session/files/workspace-files-session.ts @@ -56,6 +56,38 @@ export interface WorkspaceFilesSessionHost { hasBinaryChannel(): boolean; } +// A held upload may be the whole file: the app allows 50 MiB, sent in 128 KiB chunks. +const MAX_HELD_UPLOAD_BYTES = 64 * 1024 * 1024; +const MAX_HELD_UPLOAD_FRAMES = 1024; +// Bounds memory across every session while their upload requests are authorized. +const MAX_HELD_UPLOAD_BYTES_TOTAL = 128 * 1024 * 1024; +const UPLOAD_OVERLOADED = "Upload storage overloaded: pending frame/byte limit exceeded"; +const UPLOAD_NOT_PERMITTED = + "You do not have permission to perform this action. Ask your administrator for access."; +// A request that never settles must not pin its frames, or the shared budget, for good. +const HELD_UPLOAD_TIMEOUT_MS = 60_000; +const UPLOAD_CONNECTION_CLOSED = "Upload connection closed"; +const UPLOAD_AUTHORIZATION_TIMED_OUT = "Upload request timed out waiting for authorization"; +let heldUploadBytesTotal = 0; + +/** Bytes currently held for upload requests still being authorized, across all sessions. */ +export function heldUploadFrameBytes(): number { + return heldUploadBytesTotal; +} + +interface HeldUploadFrames { + queue: Array<{ frame: FileTransferFrame; bytes: number; allowed: () => boolean }>; + /** Declared size, capped: more than this cannot be a valid upload. */ + limit: number; + bytes: number; + released: boolean; + draining: boolean; + failure: string | null; + /** The client already has its one answer; a late admission or refusal must stay silent. */ + answered: boolean; + timeout: ReturnType; +} + export interface WorkspaceFilesSessionOptions { host: WorkspaceFilesSessionHost; downloadTokenStore: DownloadTokenStore; @@ -66,6 +98,8 @@ export interface WorkspaceFilesSessionOptions { resolveAgentDirectory?: (agentId: string) => Promise; resolveAgentReadDirectory?: (agentId: string) => Promise; resolveForkSource?: ResolveForkSource; + /** How long an upload request may wait for authorization before it is answered with an error. */ + heldUploadTimeoutMs?: number; } /** @@ -81,6 +115,7 @@ export class WorkspaceFilesSession { private readonly logger: pino.Logger; private readonly fileUploads: FileUploadStore; private readonly fileObserver: FileObserver; + private readonly heldUploadFrames = new Map>(); constructor(private readonly options: WorkspaceFilesSessionOptions) { this.host = options.host; @@ -263,6 +298,7 @@ export class WorkspaceFilesSession { } dispose(): void { + for (const source of Array.from(this.heldUploadFrames.keys())) this.releaseSource(source); void this.fileUploads .dispose() .catch((error) => @@ -495,8 +531,174 @@ export class WorkspaceFilesSession { ); } - async handleFileTransferFrame(frame: FileTransferFrame, source: object): Promise { - await this.fileUploads.receiveFrame(frame, source); + /** + * A client sends the begin frame right behind its upload request, and the socket hands binary + * frames over at once while a Project-scoped request is still being authorized. Without this + * hold the begin frame found no upload and was dropped, so every chunk then failed with + * "Upload chunks arrived before file begin". Frames wait, in arrival order, until the request + * has been admitted or refused, then reach the store one at a time. A hold that outgrows its + * budget, or whose sender loses authority, ends the upload with an error instead of a hang. + */ + holdUploadFrames( + request: { requestId: string; size: number }, + source: object | undefined, + ): () => void { + if (!source) return () => {}; + const holds = this.heldUploadFrames.get(source) ?? new Map(); + if (holds.has(request.requestId)) return () => {}; + const hold: HeldUploadFrames = { + queue: [], + limit: Math.min(Math.max(request.size, 0), MAX_HELD_UPLOAD_BYTES), + bytes: 0, + released: false, + draining: false, + failure: null, + answered: false, + // The hold stays as a marker until the request settles, so late frames are dropped and + // a late admission is cancelled instead of left behind. + timeout: setTimeout(() => { + if (hold.released) return; + hold.failure = UPLOAD_AUTHORIZATION_TIMED_OUT; + hold.answered = true; + this.discardHeldFrames(hold); + this.host.emit( + { + type: "rpc_error", + payload: { + requestId: request.requestId, + requestType: "file.upload.request", + error: UPLOAD_AUTHORIZATION_TIMED_OUT, + code: "upload_authorization_timeout", + }, + }, + source, + ); + }, this.options.heldUploadTimeoutMs ?? HELD_UPLOAD_TIMEOUT_MS), + }; + hold.timeout.unref?.(); + holds.set(request.requestId, hold); + this.heldUploadFrames.set(source, holds); + return () => { + hold.released = true; + clearTimeout(hold.timeout); + void this.drainHeldUploadFrames(request.requestId, source, hold); + }; + } + + /** True once a timed-out hold answered this request; Session then emits no refusal for it. */ + uploadRequestAnswered(requestId: string, source: object | undefined): boolean { + return source ? this.heldUploadFrames.get(source)?.get(requestId)?.answered === true : false; + } + + /** A closed connection's held frames go; the store cancels its uploads with the operation. */ + releaseSource(source: object): void { + for (const [requestId, hold] of this.heldUploadFrames.get(source) ?? []) { + hold.failure = UPLOAD_CONNECTION_CLOSED; + clearTimeout(hold.timeout); + this.discardHeldFrames(hold); + this.forgetHold(requestId, source, hold); + } + } + + /** `allowed` is read when the frame is applied, after any hold: authority can change meanwhile. */ + async handleFileTransferFrame( + frame: FileTransferFrame, + source: object, + allowed: () => boolean, + ): Promise { + const hold = this.heldUploadFrames.get(source)?.get(frame.requestId); + if (!hold) { + if (allowed()) await this.fileUploads.receiveFrame(frame, source); + return; + } + if (hold.failure) return; + const bytes = frame.opcode === FileTransferOpcode.FileChunk ? frame.payload.byteLength : 0; + if ( + hold.bytes + bytes > hold.limit || + hold.queue.length >= MAX_HELD_UPLOAD_FRAMES || + heldUploadBytesTotal + bytes > MAX_HELD_UPLOAD_BYTES_TOTAL + ) { + await this.failHeldUpload(frame.requestId, source, hold, UPLOAD_OVERLOADED); + return; + } + hold.bytes += bytes; + heldUploadBytesTotal += bytes; + hold.queue.push({ frame, bytes, allowed }); + if (hold.released) await this.drainHeldUploadFrames(frame.requestId, source, hold); + } + + private async drainHeldUploadFrames( + requestId: string, + source: object, + hold: HeldUploadFrames, + ): Promise { + if (hold.draining) return; + hold.draining = true; + try { + if (hold.failure) return await this.failHeldUpload(requestId, source, hold, hold.failure); + // A refused request registered nothing; its frames are discarded unread. + if (!this.fileUploads.hasPendingUpload(requestId, source)) { + this.discardHeldFrames(hold); + return; + } + for (let next = hold.queue.shift(); next; next = hold.queue.shift()) { + hold.bytes -= next.bytes; + heldUploadBytesTotal -= next.bytes; + if (!next.allowed()) + return await this.failHeldUpload(requestId, source, hold, UPLOAD_NOT_PERMITTED); + await this.fileUploads.receiveFrame(next.frame, source).catch(() => null); + } + } finally { + hold.draining = false; + // No await separates the empty check from the removal, so a later frame goes straight + // to the store behind every frame drained here. + if (hold.queue.length === 0 || hold.failure) this.forgetHold(requestId, source, hold); + } + } + + private async failHeldUpload( + requestId: string, + source: object, + hold: HeldUploadFrames, + failure: string, + ): Promise { + hold.failure = failure; + this.discardHeldFrames(hold); + // Before admission the request may still be refused; the drain at release reports then. + if (!hold.released) return; + this.forgetHold(requestId, source, hold); + if (hold.answered) { + await this.fileUploads.discardUpload(requestId, source); + return; + } + await this.fileUploads.failUpload(requestId, source, failure); + // Without workspace.write the upload response is no longer delivered; rpc_error always is. + if (failure === UPLOAD_NOT_PERMITTED) + this.host.emit( + { + type: "rpc_error", + payload: { + requestId, + requestType: "file.upload.request", + error: failure, + code: "access_denied", + }, + }, + source, + ); + } + + private discardHeldFrames(hold: HeldUploadFrames): void { + heldUploadBytesTotal -= hold.bytes; + hold.bytes = 0; + hold.queue = []; + } + + private forgetHold(requestId: string, source: object, hold: HeldUploadFrames): void { + const holds = this.heldUploadFrames.get(source); + if (holds?.get(requestId) !== hold) return; + holds.delete(requestId); + if (holds.size === 0) this.heldUploadFrames.delete(source); } async handleProjectIconRequest(