From 8bc9f8160115648bb3fe1dbf72b5e579d00427fe Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 21:34:05 +1000 Subject: [PATCH 1/7] fix(server): a Project-scoped session no longer loses the frames of a file it uploads A client sends the begin frame right behind file.upload.request. The socket hands binary frames to the session at once, while a Project-scoped (Hub Member) request awaits resource authorization before beginUpload registers the upload. Frames that arrived in that window found no upload and were dropped: the file then failed with "Upload chunks arrived before file begin." or never finished uploading. Images travel inside the message and were unaffected. The session now holds that request's frames until the request is admitted or refused and applies them in arrival order; file-transfer authority is checked when a frame is applied. --- .../file-upload-order.e2e.test.ts | 163 ++++++++++++++++++ packages/server/src/server/session.ts | 27 ++- .../files/workspace-files-session.test.ts | 3 + .../session/files/workspace-files-session.ts | 64 ++++++- 4 files changed, 249 insertions(+), 8 deletions(-) create mode 100644 packages/server/src/server/managed-access/file-upload-order.e2e.test.ts 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 0000000000..4bb027cbe2 --- /dev/null +++ b/packages/server/src/server/managed-access/file-upload-order.e2e.test.ts @@ -0,0 +1,163 @@ +import { mkdir, mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { 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"; + +function frame(input: Parameters[0]): FileTransferFrame { + const decoded = decodeFileTransferFrame(encodeFileTransferFrame(input)); + if (!decoded) throw new Error("Invalid file transfer frame"); + return decoded; +} + +// A Project-scoped (Hub Member) session authorizes the JSON request asynchronously, while +// the socket hands the binary frames that follow it to the session at once. +for (const order of ["after the request settles", "before the request is authorized"] as const) + test(`a Project-scoped session stores file frames sent ${order}`, async () => { + const root = await mkdtemp(path.join(tmpdir(), "managed-upload-order-")); + const repo = path.join(root, "repo"); + await mkdir(repo); + const daemon = await createTestClisbotDaemon(); + 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(), + }); + const messages: SessionOutboundMessage[] = []; + const snapshot = createProviderSnapshotManagerStub(); + 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([ + [ + project.projectId, + { + privileges: new Set(["project.use", "agent.interact"] as const), + agentConfigurations: [], + }, + ], + ]), + }, + 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: snapshot.manager, + stt: null, + tts: null, + }); + const source = {}; + session.updateClientCapabilities({}, source); + const bytes = new TextEncoder().encode('{"tokens":true}'); + try { + // The websocket server neither awaits nor orders these against each other. + const request = session.handleMessage( + { + type: "file.upload.request", + requestId: "upload-1", + fileName: "Light.tokens.json", + mimeType: "application/json", + size: bytes.byteLength, + modifiedAt: "2026-11-04T00:00:00.000Z", + }, + source, + ); + if (order === "after the request settles") await request; + const frames = [ + session.handleBinaryFrame( + { + kind: "file_transfer", + frame: frame({ + opcode: FileTransferOpcode.FileBegin, + requestId: "upload-1", + metadata: { + mime: "application/json", + size: bytes.byteLength, + encoding: "binary", + modifiedAt: "2026-11-04T00:00:00.000Z", + fileName: "Light.tokens.json", + }, + }), + }, + source, + ), + session.handleBinaryFrame( + { + kind: "file_transfer", + frame: frame({ + opcode: FileTransferOpcode.FileChunk, + requestId: "upload-1", + payload: bytes, + }), + }, + source, + ), + session.handleBinaryFrame( + { + kind: "file_transfer", + frame: frame({ opcode: FileTransferOpcode.FileEnd, requestId: "upload-1" }), + }, + source, + ), + ]; + await Promise.all([request, ...frames]); + await expect + .poll(() => messages.find((message) => message.type === "file.upload.response"), { + timeout: 5_000, + }) + .toBeTruthy(); + const response = messages.find((message) => message.type === "file.upload.response"); + if (response?.type !== "file.upload.response") throw new Error(JSON.stringify(messages)); + expect(response.payload.error).toBeNull(); + expect(await readFile(response.payload.file!.path, "utf8")).toBe('{"tokens":true}'); + } finally { + await daemon.close(); + await rm(root, { recursive: true, force: true }); + } + }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 95f0bdbe2a..738d3f6684 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -2491,7 +2491,16 @@ export class Session { } 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.requestId, source) + : undefined; + try { + return await this.delivery.request(source, msg, () => this.handleRequest(msg, source)); + } finally { + releaseUploadFrames?.(); + } } private async handleRequest(msg: SessionInboundMessage, source?: object): Promise { @@ -3823,16 +3832,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/workspace-files-session.test.ts b/packages/server/src/server/session/files/workspace-files-session.test.ts index df80dfdf2f..0c2a0e698e 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 62fa636450..d83b0cf1b4 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,14 @@ export interface WorkspaceFilesSessionHost { hasBinaryChannel(): boolean; } +const MAX_HELD_UPLOAD_FRAMES = 64; + +interface HeldUploadFrames { + source: object; + tail: Promise; + frames: number; +} + export interface WorkspaceFilesSessionOptions { host: WorkspaceFilesSessionHost; downloadTokenStore: DownloadTokenStore; @@ -81,6 +89,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; @@ -495,8 +504,59 @@ 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. + */ + holdUploadFrames(requestId: string, source: object | undefined): () => void { + if (!source || this.heldUploadFrames.has(requestId)) return () => {}; + let release!: () => void; + const hold: HeldUploadFrames = { + source, + tail: new Promise((resolve) => (release = resolve)), + frames: 0, + }; + this.heldUploadFrames.set(requestId, hold); + const forget = (): void => { + const tail = hold.tail; + void tail.then(() => { + if (hold.tail !== tail) return forget(); + if (this.heldUploadFrames.get(requestId) === hold) this.heldUploadFrames.delete(requestId); + return undefined; + }); + }; + return () => { + release(); + forget(); + }; + } + + /** `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 apply = async () => { + if (allowed()) await this.fileUploads.receiveFrame(frame, source); + }; + const hold = this.heldUploadFrames.get(frame.requestId); + if (!hold || hold.source !== source) { + await apply(); + return; + } + // Past the store's own per-upload frame budget the upload fails with a size mismatch. + if (hold.frames >= MAX_HELD_UPLOAD_FRAMES) return; + hold.frames += 1; + const received = hold.tail.then(apply); + hold.tail = received.then( + () => undefined, + () => undefined, + ); + await received; } async handleProjectIconRequest( From 1bbd2a1e88bfa59bf2e51f6182ecd4de70b2fd88 Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 21:35:21 +1000 Subject: [PATCH 2/7] fix(server): uploads that were never sent no longer block every later file The app uploads a chosen file when it is attached and again from its local copy at send; only the second copy is linked to the message. Each attached file therefore left one completed upload behind, and after 256 in one connection session every further file failed with "Upload storage overloaded: too many unfinished drafts" while images still sent. At the cap the store now forgets its oldest unlinked upload instead of refusing. The bytes stay on disk because a terminal may still hold the pasted path. --- .../src/server/file-upload/index.test.ts | 18 ++++++++++++++++++ .../server/src/server/file-upload/index.ts | 9 ++++++++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/packages/server/src/server/file-upload/index.test.ts b/packages/server/src/server/file-upload/index.test.ts index d1d5002f9d..ab0fa7a2dd 100644 --- a/packages/server/src/server/file-upload/index.test.ts +++ b/packages/server/src/server/file-upload/index.test.ts @@ -281,6 +281,24 @@ describe("file uploads", () => { }); expect(readFileSync(path, "utf8")).toBe("hello world"); }); + + it("keeps accepting files after many uploads that were never sent", async () => { + const clisbotHome = makeClisbotHome(); + const uploads = new FileUploadStore({ clisbotHome }); + // The app uploads a chosen file once at pick time and again at send; only the second is linked. + const unsent = []; + for (let index = 0; index < 256; index++) { + unsent.push(await uploadNamed(uploads, `draft-${index}.pdf`)); + } + + const next = await uploadNamed(uploads, "clip.mp4"); + + expect(next && uploads.ownsUploadedFile(next)).toBe(true); + expect(unsent[0] && uploads.ownsUploadedFile(unsent[0])).toBe(false); + expect(unsent[1] && uploads.ownsUploadedFile(unsent[1])).toBe(true); + // Untracking does not delete bytes: a terminal may still hold the pasted path. + expect(existsSync(unsent[0]!.path)).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 2d9a73fbfc..bba1a635b8 100644 --- a/packages/server/src/server/file-upload/index.ts +++ b/packages/server/src/server/file-upload/index.ts @@ -66,8 +66,15 @@ export class FileUploadStore { finished: (response: FileUploadResponse | null) => void = () => {}, ): () => Promise { const count = [...this.pending.values()].reduce((total, uploads) => total + uploads.size, 0); - if (count >= MAX_PENDING_UPLOADS || this.completed.size >= MAX_COMPLETED_UPLOADS) + if (count >= MAX_PENDING_UPLOADS) throw new Error("Upload storage overloaded: too many unfinished drafts"); + // Apps upload a chosen file again at send, so most completed uploads are never linked. + // Refusing at the cap blocked every later file for the rest of the session; forget the + // oldest instead. Its bytes stay: a terminal may hold its pasted path. + if (this.completed.size >= MAX_COMPLETED_UPLOADS) { + const oldest = this.completed.keys().next(); + if (!oldest.done) this.completed.delete(oldest.value); + } const existingUpload = this.pending.get(source)?.get(request.requestId); if (existingUpload) void this.cancel(existingUpload).catch(() => {}); const fileName = sanitizeFileName(request.fileName); From ea9978e69ff67912eeced953039fd7a4d9e80049 Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 21:36:32 +1000 Subject: [PATCH 3/7] fix(app): files dropped onto a Bot chat are attached The Bot chat screen had no FileDropZone, so its composer's drop sink never registered and a dropped file, image or not, was ignored on web and desktop. The screen now provides the zone the agent panel already has. A browser journey sends MP4, PDF, zip and JSON through the Upload file menu and through one drop in a Bot chat, and through the menu in an agent composer, and checks the bytes the Host stored. --- .../chat-file-attachment-types.spec.ts | 164 ++++++++++++++++++ .../clisbot/bots/chat/chat-screen.test.tsx | 3 + .../app/src/clisbot/bots/chat/chat-screen.tsx | 57 +++--- 3 files changed, 198 insertions(+), 26 deletions(-) create mode 100644 packages/app/e2e/browser/chat-file-attachment-types.spec.ts 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 0000000000..4dde3fdbfc --- /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 7ad445aaeb..029cd12543 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 2ce884a3ec..e15d69151f 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 }, })); From 000a85adaad9743275785bb31d89246a2b57e350 Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 22:11:12 +1000 Subject: [PATCH 4/7] fix(server): a held upload stays bounded and always gets an answer Review of the frame hold found that it counted every frame for as long as the hold existed and dropped frames past 64 without telling anyone. A file over 8 MiB whose frames arrived during a slow authorization, or while held frames were still being written, lost its end frame and the composer spun forever. Held frames now drain to the store one at a time, and the hold counts only what it still holds: up to the declared size and at most 64 MiB, so a whole 50 MiB file can wait for authorization, with 128 MiB across all sessions. If the hold overflows, the upload fails with the store's overload error. A refused request's frames are discarded unread. If the sender loses authority, the upload ends with an access_denied rpc_error, because a file.upload.response no longer reaches a session without workspace.write. Holds are keyed per connection. --- .../server/src/server/file-upload/index.ts | 14 + .../file-upload-order.e2e.test.ts | 376 +++++++++++------- packages/server/src/server/session.ts | 2 +- .../session/files/workspace-files-session.ts | 157 ++++++-- 4 files changed, 377 insertions(+), 172 deletions(-) diff --git a/packages/server/src/server/file-upload/index.ts b/packages/server/src/server/file-upload/index.ts index bba1a635b8..589da21489 100644 --- a/packages/server/src/server/file-upload/index.ts +++ b/packages/server/src/server/file-upload/index.ts @@ -159,6 +159,20 @@ 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)); + } + 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 index 4bb027cbe2..1cbcfe130f 100644 --- 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 @@ -1,8 +1,9 @@ -import { mkdir, mkdtemp, readFile, rm } from "node:fs/promises"; +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 { expect, test } from "vitest"; +import { randomBytes } from "node:crypto"; +import { afterEach, expect, test } from "vitest"; import { decodeFileTransferFrame, encodeFileTransferFrame, @@ -25,139 +26,246 @@ function frame(input: Parameters[0]): FileTransf return decoded; } -// A Project-scoped (Hub Member) session authorizes the JSON request asynchronously, while -// the socket hands the binary frames that follow it to the session at once. -for (const order of ["after the request settles", "before the request is authorized"] as const) - test(`a Project-scoped session stores file frames sent ${order}`, async () => { - const root = await mkdtemp(path.join(tmpdir(), "managed-upload-order-")); - const repo = path.join(root, "repo"); - await mkdir(repo); - const daemon = await createTestClisbotDaemon(); - 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(), - }); - const messages: SessionOutboundMessage[] = []; - const snapshot = createProviderSnapshotManagerStub(); - 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([ - [ - project.projectId, - { - privileges: new Set(["project.use", "agent.interact"] as const), - agentConfigurations: [], - }, - ], - ]), - }, - 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: snapshot.manager, - stt: null, - tts: null, - }); +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 } = {}) { + 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(), + }); + 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]]), + }, + 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 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}'); - try { - // The websocket server neither awaits nor orders these against each other. - const request = session.handleMessage( - { - type: "file.upload.request", - requestId: "upload-1", - fileName: "Light.tokens.json", - mimeType: "application/json", - size: bytes.byteLength, - modifiedAt: "2026-11-04T00:00:00.000Z", - }, - source, - ); - if (order === "after the request settles") await request; - const frames = [ - session.handleBinaryFrame( - { - kind: "file_transfer", - frame: frame({ - opcode: FileTransferOpcode.FileBegin, - requestId: "upload-1", - metadata: { - mime: "application/json", - size: bytes.byteLength, - encoding: "binary", - modifiedAt: "2026-11-04T00:00:00.000Z", - fileName: "Light.tokens.json", - }, - }), - }, - source, - ), - session.handleBinaryFrame( - { - kind: "file_transfer", - frame: frame({ - opcode: FileTransferOpcode.FileChunk, - requestId: "upload-1", - payload: bytes, - }), - }, - source, - ), - session.handleBinaryFrame( - { - kind: "file_transfer", - frame: frame({ opcode: FileTransferOpcode.FileEnd, requestId: "upload-1" }), - }, - source, - ), - ]; - await Promise.all([request, ...frames]); - await expect - .poll(() => messages.find((message) => message.type === "file.upload.response"), { - timeout: 5_000, - }) - .toBeTruthy(); - const response = messages.find((message) => message.type === "file.upload.response"); - if (response?.type !== "file.upload.response") throw new Error(JSON.stringify(messages)); - expect(response.payload.error).toBeNull(); - expect(await readFile(response.payload.file!.path, "utf8")).toBe('{"tokens":true}'); - } finally { - await daemon.close(); - await rm(root, { recursive: true, force: 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); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 738d3f6684..3f6709c845 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -2494,7 +2494,7 @@ export class Session { // 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.requestId, source) + ? this.workspaceFilesSession.holdUploadFrames(msg, source) : undefined; try { return await this.delivery.request(source, msg, () => this.handleRequest(msg, source)); 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 d83b0cf1b4..949934bb7d 100644 --- a/packages/server/src/server/session/files/workspace-files-session.ts +++ b/packages/server/src/server/session/files/workspace-files-session.ts @@ -56,12 +56,24 @@ export interface WorkspaceFilesSessionHost { hasBinaryChannel(): boolean; } -const MAX_HELD_UPLOAD_FRAMES = 64; +// 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."; +let heldUploadBytesTotal = 0; interface HeldUploadFrames { - source: object; - tail: Promise; - frames: number; + 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; } export interface WorkspaceFilesSessionOptions { @@ -89,7 +101,7 @@ export class WorkspaceFilesSession { private readonly logger: pino.Logger; private readonly fileUploads: FileUploadStore; private readonly fileObserver: FileObserver; - private readonly heldUploadFrames = new Map(); + private readonly heldUploadFrames = new WeakMap>(); constructor(private readonly options: WorkspaceFilesSessionOptions) { this.host = options.host; @@ -509,28 +521,29 @@ export class WorkspaceFilesSession { * 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. + * 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(requestId: string, source: object | undefined): () => void { - if (!source || this.heldUploadFrames.has(requestId)) return () => {}; - let release!: () => void; + 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 = { - source, - tail: new Promise((resolve) => (release = resolve)), - frames: 0, - }; - this.heldUploadFrames.set(requestId, hold); - const forget = (): void => { - const tail = hold.tail; - void tail.then(() => { - if (hold.tail !== tail) return forget(); - if (this.heldUploadFrames.get(requestId) === hold) this.heldUploadFrames.delete(requestId); - return undefined; - }); + queue: [], + limit: Math.min(Math.max(request.size, 0), MAX_HELD_UPLOAD_BYTES), + bytes: 0, + released: false, + draining: false, + failure: null, }; + holds.set(request.requestId, hold); + this.heldUploadFrames.set(source, holds); return () => { - release(); - forget(); + hold.released = true; + void this.drainHeldUploadFrames(request.requestId, source, hold); }; } @@ -540,23 +553,93 @@ export class WorkspaceFilesSession { source: object, allowed: () => boolean, ): Promise { - const apply = async () => { + const hold = this.heldUploadFrames.get(source)?.get(frame.requestId); + if (!hold) { if (allowed()) await this.fileUploads.receiveFrame(frame, source); - }; - const hold = this.heldUploadFrames.get(frame.requestId); - if (!hold || hold.source !== source) { - await apply(); return; } - // Past the store's own per-upload frame budget the upload fails with a size mismatch. - if (hold.frames >= MAX_HELD_UPLOAD_FRAMES) return; - hold.frames += 1; - const received = hold.tail.then(apply); - hold.tail = received.then( - () => undefined, - () => undefined, - ); - await received; + 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); + 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) holds.delete(requestId); } async handleProjectIconRequest( From 216e67e87337cbbca31641cf00044bf694825f85 Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 22:12:18 +1000 Subject: [PATCH 5/7] Revert "fix(server): uploads that were never sent no longer block every later file" Review found the eviction could invalidate a live attachment: New workspace, the workspace setup dialog and the draft create flow send the upload made when the file was chosen, so forgetting the oldest unlinked upload could fail that draft's send with "Uploaded file does not belong to this connection". The store refuses at its cap again; a test keeps an unsent draft linkable while other uploads fill the store. The cap itself remains: each file a composer sends is uploaded twice, so one connection session can still reach 256 unlinked uploads. Ending that needs the client to reuse or release the first upload. --- .../src/server/file-upload/index.test.ts | 30 ++++++++++--------- .../server/src/server/file-upload/index.ts | 9 +----- 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/packages/server/src/server/file-upload/index.test.ts b/packages/server/src/server/file-upload/index.test.ts index ab0fa7a2dd..9eb9bf9ca5 100644 --- a/packages/server/src/server/file-upload/index.test.ts +++ b/packages/server/src/server/file-upload/index.test.ts @@ -282,22 +282,24 @@ describe("file uploads", () => { expect(readFileSync(path, "utf8")).toBe("hello world"); }); - it("keeps accepting files after many uploads that were never sent", async () => { + it("keeps an unsent draft linkable while other uploads fill the store", async () => { const clisbotHome = makeClisbotHome(); const uploads = new FileUploadStore({ clisbotHome }); - // The app uploads a chosen file once at pick time and again at send; only the second is linked. - const unsent = []; - for (let index = 0; index < 256; index++) { - unsent.push(await uploadNamed(uploads, `draft-${index}.pdf`)); - } - - const next = await uploadNamed(uploads, "clip.mp4"); - - expect(next && uploads.ownsUploadedFile(next)).toBe(true); - expect(unsent[0] && uploads.ownsUploadedFile(unsent[0])).toBe(false); - expect(unsent[1] && uploads.ownsUploadedFile(unsent[1])).toBe(true); - // Untracking does not delete bytes: a terminal may still hold the pasted path. - expect(existsSync(unsent[0]!.path)).toBe(true); + // 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); }); }); diff --git a/packages/server/src/server/file-upload/index.ts b/packages/server/src/server/file-upload/index.ts index 589da21489..c17ba3645c 100644 --- a/packages/server/src/server/file-upload/index.ts +++ b/packages/server/src/server/file-upload/index.ts @@ -66,15 +66,8 @@ export class FileUploadStore { finished: (response: FileUploadResponse | null) => void = () => {}, ): () => Promise { const count = [...this.pending.values()].reduce((total, uploads) => total + uploads.size, 0); - if (count >= MAX_PENDING_UPLOADS) + if (count >= MAX_PENDING_UPLOADS || this.completed.size >= MAX_COMPLETED_UPLOADS) throw new Error("Upload storage overloaded: too many unfinished drafts"); - // Apps upload a chosen file again at send, so most completed uploads are never linked. - // Refusing at the cap blocked every later file for the rest of the session; forget the - // oldest instead. Its bytes stay: a terminal may hold its pasted path. - if (this.completed.size >= MAX_COMPLETED_UPLOADS) { - const oldest = this.completed.keys().next(); - if (!oldest.done) this.completed.delete(oldest.value); - } const existingUpload = this.pending.get(source)?.get(request.requestId); if (existingUpload) void this.cancel(existingUpload).catch(() => {}); const fileName = sanitizeFileName(request.fileName); From f2103e95be582d24460243ebbfb6c54ba786ebb3 Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 22:21:11 +1000 Subject: [PATCH 6/7] fix(server): held upload frames are released when their request cannot finish Review asked what happens to a hold whose request never settles or whose connection closes first. The hold stayed in place, and so did its bytes in the process-wide budget, so later uploads in every session could be refused as overloaded. A hold now gives up after 60 s without an answer. It is released when its connection closes and when the session is disposed. Tests cover byte accounting after a drain, a refused request, a closed connection, a request that never settles, dispose, the shared 128 MiB budget across sessions, a duplicate request id, and an overflow during the drain. --- packages/server/src/server/session.ts | 1 + .../session/files/held-upload-frames.test.ts | 222 ++++++++++++++++++ .../session/files/workspace-files-session.ts | 34 ++- 3 files changed, 255 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/server/session/files/held-upload-frames.test.ts diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3f6709c845..2786d8308c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1531,6 +1531,7 @@ export class Session { } clearAgentTimelineSubscription(source: object): void { + this.workspaceFilesSession.releaseSource(source); this.clientCapabilitiesBySource.delete(source); void this.delivery .detach(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 0000000000..af839eef83 --- /dev/null +++ b/packages/server/src/server/session/files/held-upload-frames.test.ts @@ -0,0 +1,222 @@ +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() { + 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" }), + }); + 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, + ); + return { subsystem, connect, admit, send, answer }; +} + +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); + await Promise.all(t.send(source, frames("stuck", CHUNK).slice(1, 2))); + expect(heldUploadFrameBytes()).toBe(0); +}); + +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.ts b/packages/server/src/server/session/files/workspace-files-session.ts index 949934bb7d..be45685fc3 100644 --- a/packages/server/src/server/session/files/workspace-files-session.ts +++ b/packages/server/src/server/session/files/workspace-files-session.ts @@ -64,8 +64,16 @@ 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"; 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. */ @@ -74,6 +82,7 @@ interface HeldUploadFrames { released: boolean; draining: boolean; failure: string | null; + timeout: ReturnType; } export interface WorkspaceFilesSessionOptions { @@ -101,7 +110,7 @@ export class WorkspaceFilesSession { private readonly logger: pino.Logger; private readonly fileUploads: FileUploadStore; private readonly fileObserver: FileObserver; - private readonly heldUploadFrames = new WeakMap>(); + private readonly heldUploadFrames = new Map>(); constructor(private readonly options: WorkspaceFilesSessionOptions) { this.host = options.host; @@ -284,6 +293,7 @@ export class WorkspaceFilesSession { } dispose(): void { + for (const source of Array.from(this.heldUploadFrames.keys())) this.releaseSource(source); void this.fileUploads .dispose() .catch((error) => @@ -538,15 +548,33 @@ export class WorkspaceFilesSession { released: false, draining: false, failure: null, + timeout: setTimeout(() => { + if (hold.released) return; + hold.failure = UPLOAD_OVERLOADED; + this.discardHeldFrames(hold); + this.forgetHold(request.requestId, source, hold); + }, 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); }; } + /** 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, @@ -639,7 +667,9 @@ export class WorkspaceFilesSession { private forgetHold(requestId: string, source: object, hold: HeldUploadFrames): void { const holds = this.heldUploadFrames.get(source); - if (holds?.get(requestId) === hold) holds.delete(requestId); + if (holds?.get(requestId) !== hold) return; + holds.delete(requestId); + if (holds.size === 0) this.heldUploadFrames.delete(source); } async handleProjectIconRequest( From 82306d543b0dbd6339b6c5c56b9d4e1b6c6809cd Mon Sep 17 00:00:00 2001 From: Long Luong Date: Wed, 4 Nov 2026 22:36:10 +1000 Subject: [PATCH 7/7] fix(server): an upload whose authorization stalls is answered once Review found that the 60 s hold timeout only discarded frames. A client whose request never settled still waited forever. The timeout now sends one rpc_error (upload_authorization_timeout) to that connection and keeps a marker until the request settles. Frames still on the wire are dropped. A late admission is cancelled without a second answer and leaves no upload behind. A late refusal emits no second rpc_error. The timer is cleared at release, so it cannot fire while held frames drain. Session tests cover a permanently stalled authorization and late admission or refusal after the timeout answer, checking a single answer, no upload left on disk, and that the next file still goes through. Unit tests cover a connection closing mid-drain and a timeout during a long drain. --- .../server/src/server/file-upload/index.ts | 6 ++ .../file-upload-order.e2e.test.ts | 73 ++++++++++++++++++- packages/server/src/server/session.ts | 15 +++- .../session/files/held-upload-frames.test.ts | 54 +++++++++++++- .../session/files/workspace-files-session.ts | 35 ++++++++- 5 files changed, 175 insertions(+), 8 deletions(-) diff --git a/packages/server/src/server/file-upload/index.ts b/packages/server/src/server/file-upload/index.ts index c17ba3645c..fb643116db 100644 --- a/packages/server/src/server/file-upload/index.ts +++ b/packages/server/src/server/file-upload/index.ts @@ -166,6 +166,12 @@ export class FileUploadStore { 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 index 1cbcfe130f..d1058f392b 100644 --- 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 @@ -19,6 +19,7 @@ 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)); @@ -37,7 +38,9 @@ afterEach(async () => { }); // A Project-scoped (Hub Member) session, the way the Hub admits a Member's app. -async function scopedSession(options: { granted?: boolean } = {}) { +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); @@ -56,6 +59,15 @@ async function scopedSession(options: { granted?: boolean } = {}) { 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: [], @@ -70,6 +82,7 @@ async function scopedSession(options: { granted?: boolean } = {}) { 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), @@ -163,6 +176,14 @@ async function outcome(scoped: Scoped, requestId: string) { 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) @@ -269,3 +290,53 @@ test("frames from another connection with the same request id do not join a held 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 2786d8308c..8685cd4c06 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"); @@ -2491,6 +2494,14 @@ 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 { // Registered before the first await: the upload's binary frames may already be next. const releaseUploadFrames = @@ -2526,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); @@ -2556,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", 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 index af839eef83..a59ab4fb0b 100644 --- a/packages/server/src/server/session/files/held-upload-frames.test.ts +++ b/packages/server/src/server/session/files/held-upload-frames.test.ts @@ -24,7 +24,7 @@ afterEach(() => { expect(heldUploadFrameBytes()).toBe(0); }); -function setup() { +function setup(options: { heldUploadTimeoutMs?: number } = {}) { const home = realpathSync(mkdtempSync(join(tmpdir(), "held-upload-"))); homes.push(home); const emitted: SessionOutboundMessage[] = []; @@ -37,6 +37,7 @@ function setup() { 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)); @@ -66,7 +67,13 @@ function setup() { (message.type === "file.upload.response" || message.type === "rpc_error") && message.payload.requestId === requestId, ); - return { subsystem, connect, admit, send, answer }; + 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[] { @@ -140,10 +147,53 @@ test("a request that never settles gives its hold up", async () => { 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()]; 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 be45685fc3..75a14ba410 100644 --- a/packages/server/src/server/session/files/workspace-files-session.ts +++ b/packages/server/src/server/session/files/workspace-files-session.ts @@ -67,6 +67,7 @@ const UPLOAD_NOT_PERMITTED = // 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. */ @@ -82,6 +83,8 @@ interface HeldUploadFrames { 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; } @@ -95,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; } /** @@ -548,12 +553,27 @@ export class WorkspaceFilesSession { 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_OVERLOADED; + hold.failure = UPLOAD_AUTHORIZATION_TIMED_OUT; + hold.answered = true; this.discardHeldFrames(hold); - this.forgetHold(request.requestId, source, hold); - }, HELD_UPLOAD_TIMEOUT_MS), + 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); @@ -565,6 +585,11 @@ export class WorkspaceFilesSession { }; } + /** 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) ?? []) { @@ -642,6 +667,10 @@ export class WorkspaceFilesSession { // 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)