diff --git a/.oxlintrc.json b/.oxlintrc.json index 5fe29499a5..0ce51bfa06 100644 --- a/.oxlintrc.json +++ b/.oxlintrc.json @@ -84,6 +84,31 @@ "max-nested-callbacks": ["error", { "max": 3 }] }, "overrides": [ + { + "files": ["packages/server/src/**/*.{ts,tsx}"], + "rules": { + "no-restricted-imports": [ + "error", + { + "patterns": [ + { + "group": ["**/workspace-helper/integration/**", "**/workspace-helper/internal/**"], + "message": "Normal callers consume WorkspaceFiles from workspace-helper/index. Only the workspace-runtime parent integration and workspace-helper tests may assemble helper clients." + } + ] + } + ] + } + }, + { + "files": [ + "packages/server/src/server/workspace-helper/**/*.{ts,tsx}", + "packages/server/src/server/workspace-runtime/internal/service.ts" + ], + "rules": { + "no-restricted-imports": "off" + } + }, { "files": ["packages/app/src/**/*.{ts,tsx}"], "rules": { diff --git a/packages/app/src/file-pane/live-file/hook.ts b/packages/app/src/file-pane/live-file/hook.ts index e318f7c0c6..f97275a9bc 100644 --- a/packages/app/src/file-pane/live-file/hook.ts +++ b/packages/app/src/file-pane/live-file/hook.ts @@ -4,6 +4,7 @@ import { LiveFileModel, type LiveFileSession } from "./model"; export function useLiveFile(input: { client: DaemonClient | null; + workspaceId: string; cwd: string | null; path: string | null; enabled: boolean; @@ -15,13 +16,13 @@ export function useLiveFile(input: { const client = input.client; return { subscribe(target, onVersion) { - return client.subscribeFile(target, onVersion); + return client.subscribeFile({ ...target, workspaceId: input.workspaceId }, onVersion); }, read(target) { - return client.readFile(target.cwd, target.path); + return client.readFile(target.cwd, target.path, undefined, input.workspaceId); }, }; - }, [input.client]); + }, [input.client, input.workspaceId]); useEffect(() => { if (!input.enabled || !session || !input.cwd || !input.path) { diff --git a/packages/app/src/file-pane/pane.tsx b/packages/app/src/file-pane/pane.tsx index 07b0ee2671..a04c969b63 100644 --- a/packages/app/src/file-pane/pane.tsx +++ b/packages/app/src/file-pane/pane.tsx @@ -400,11 +400,13 @@ function FilePreviewBody({ export function FilePane({ serverId, + workspaceId, workspaceRoot, location, navigationRevision, }: { serverId: string; + workspaceId: string; workspaceRoot: string; location: WorkspaceFileLocation; navigationRevision: number; @@ -448,6 +450,7 @@ export function FilePane({ }); const liveFile = useLiveFile({ client, + workspaceId, cwd: readTarget?.cwd ?? null, path: readTarget?.path ?? null, enabled, @@ -488,6 +491,7 @@ export function FilePane({ ({ write(input: { content: string; expectedModifiedAt: string; expectedRevision?: string }) { - return client.writeFile({ cwd, path, ...input }); + return client.writeFile({ cwd, path, workspaceId, ...input }); }, }), - [client, cwd, path], + [client, cwd, path, workspaceId], ); const [model] = useState(() => { return new FileEditorModel({ diff --git a/packages/app/src/hooks/use-file-explorer-actions.ts b/packages/app/src/hooks/use-file-explorer-actions.ts index d1bb4c239a..b908996728 100644 --- a/packages/app/src/hooks/use-file-explorer-actions.ts +++ b/packages/app/src/hooks/use-file-explorer-actions.ts @@ -137,7 +137,12 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor } try { - const directory = await client.listDirectory(normalizedWorkspaceRoot, normalizedPath); + const directory = await client.listDirectory( + normalizedWorkspaceRoot, + normalizedPath, + undefined, + workspaceId ?? undefined, + ); updateExplorerState((state) => { const nextState: AgentFileExplorerState = { ...state, @@ -168,7 +173,7 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor return null; } }, - [client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceStateKey], + [client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceId, workspaceStateKey], ); const requestFilePreview = useCallback( @@ -205,7 +210,12 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor } try { - const file = await client.readFile(normalizedWorkspaceRoot, normalizedPath); + const file = await client.readFile( + normalizedWorkspaceRoot, + normalizedPath, + undefined, + workspaceId ?? undefined, + ); updateExplorerState((state) => { const nextState: AgentFileExplorerState = { ...state, @@ -232,7 +242,7 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor })); } }, - [client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceStateKey], + [client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceId, workspaceStateKey], ); const requestFileDownloadToken = useCallback( @@ -243,13 +253,18 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor if (!client) { throw new Error(t("workspace.terminal.hostDisconnected")); } - const payload = await client.requestDownloadToken(normalizedWorkspaceRoot, path); + const payload = await client.requestDownloadToken( + normalizedWorkspaceRoot, + path, + undefined, + workspaceId ?? undefined, + ); if (payload.error) { throw new Error(payload.error); } return payload; }, - [client, normalizedWorkspaceRoot, t], + [client, normalizedWorkspaceRoot, t, workspaceId], ); const selectExplorerEntry = useCallback( diff --git a/packages/app/src/panels/file-panel.tsx b/packages/app/src/panels/file-panel.tsx index b4b6975e58..18b2fc6796 100644 --- a/packages/app/src/panels/file-panel.tsx +++ b/packages/app/src/panels/file-panel.tsx @@ -43,6 +43,7 @@ function FilePanel() { return ( (); private fileSubscriptions = new Map< string, - { cwd: string; path: string; onUpdate: (version: FileVersion) => void } + { cwd: string; path: string; workspaceId?: string; onUpdate: (version: FileVersion) => void } >(); private readonly terminalStreams = new TerminalStreamRouter(); private pendingBinaryFileReads = new Map(); @@ -2429,6 +2429,7 @@ export class DaemonClient { cwd: subscription.cwd, path: subscription.path, subscriptionId, + ...(subscription.workspaceId ? { workspaceId: subscription.workspaceId } : {}), }, responseType: "fs.file.subscribe.response", }) @@ -4231,6 +4232,7 @@ export class DaemonClient { mode: "list" | "file", requestId?: string, acceptBinary = false, + workspaceId?: string, ): Promise { return this.sendCorrelatedSessionRequest({ requestId, @@ -4240,6 +4242,7 @@ export class DaemonClient { path, mode, ...(acceptBinary ? { acceptBinary: true } : {}), + ...(workspaceId ? { workspaceId } : {}), }, responseType: "file_explorer_response", }); @@ -4249,8 +4252,16 @@ export class DaemonClient { cwd: string, path: string, requestId?: string, + workspaceId?: string, ): Promise { - const payload = await this.requestFileExplorer(cwd, path, "list", requestId); + const payload = await this.requestFileExplorer( + cwd, + path, + "list", + requestId, + false, + workspaceId, + ); if (payload.error) { throw new Error(payload.error); } @@ -4260,11 +4271,23 @@ export class DaemonClient { return payload.directory; } - async readFile(cwd: string, path: string, requestId?: string): Promise { + async readFile( + cwd: string, + path: string, + requestId?: string, + workspaceId?: string, + ): Promise { const resolvedRequestId = this.createRequestId(requestId); this.pendingBinaryFileReads.set(resolvedRequestId, { cwd, path }); try { - const payload = await this.requestFileExplorer(cwd, path, "file", resolvedRequestId, true); + const payload = await this.requestFileExplorer( + cwd, + path, + "file", + resolvedRequestId, + true, + workspaceId, + ); if (payload.error) { throw new Error(payload.error); } @@ -4284,7 +4307,7 @@ export class DaemonClient { } async subscribeFile( - input: { cwd: string; path: string }, + input: { cwd: string; path: string; workspaceId?: string }, onUpdate: (version: FileVersion) => void, ): Promise<{ initial: FileVersion; unsubscribe: () => void }> { const subscriptionId = this.createRequestId(); @@ -4296,6 +4319,7 @@ export class DaemonClient { cwd: input.cwd, path: input.path, subscriptionId, + ...(input.workspaceId ? { workspaceId: input.workspaceId } : {}), }, responseType: "fs.file.subscribe.response", }); @@ -4321,6 +4345,7 @@ export class DaemonClient { content: string; expectedModifiedAt: string; expectedRevision?: string; + workspaceId?: string; }): Promise { const payload = await this.sendCorrelatedSessionRequest({ message: { type: "fs.file.write.request", ...input }, @@ -4389,6 +4414,7 @@ export class DaemonClient { cwd: string, path: string, requestId?: string, + workspaceId?: string, ): Promise { return this.sendCorrelatedSessionRequest({ requestId, @@ -4396,6 +4422,7 @@ export class DaemonClient { type: "file_download_token_request", cwd, path, + ...(workspaceId ? { workspaceId } : {}), }, responseType: "file_download_token_response", }); @@ -4404,12 +4431,14 @@ export class DaemonClient { async requestProjectIcon( cwd: string, requestId?: string, + workspaceId?: string, ): Promise { return this.sendCorrelatedSessionRequest({ requestId, message: { type: "project_icon_request", cwd, + ...(workspaceId ? { workspaceId } : {}), }, responseType: "project_icon_response", }); diff --git a/packages/docker-workspace-runtime/Dockerfile b/packages/docker-workspace-runtime/Dockerfile index a20c804114..02896deb7e 100644 --- a/packages/docker-workspace-runtime/Dockerfile +++ b/packages/docker-workspace-runtime/Dockerfile @@ -4,6 +4,9 @@ RUN apt-get update \ && apt-get install -y --no-install-recommends git procps \ && rm -rf /var/lib/apt/lists/* -COPY src/workload.mjs /opt/paseo-workspace-runtime/workload.mjs +COPY packages/docker-workspace-runtime/src/workload.mjs /opt/paseo-workspace-runtime/workload.mjs +COPY packages/server/src/server/workspace-helper/executable.mjs /opt/paseo-workspace-helper/executable.mjs +COPY packages/server/src/server/workspace-helper/executable.mjs /usr/bin/paseo-workspace-helper +RUN chmod 755 /usr/bin/paseo-workspace-helper WORKDIR /workspace diff --git a/packages/protocol/src/messages.file-editing.test.ts b/packages/protocol/src/messages.file-editing.test.ts index 9e2c16d99b..46642516e5 100644 --- a/packages/protocol/src/messages.file-editing.test.ts +++ b/packages/protocol/src/messages.file-editing.test.ts @@ -28,8 +28,9 @@ describe("workspace file editing messages", () => { path: "file.ts", subscriptionId: "subscription-1", requestId: "request-1", - }).subscriptionId, - ).toBe("subscription-1"); + workspaceId: "workspace-1", + }), + ).toMatchObject({ subscriptionId: "subscription-1", workspaceId: "workspace-1" }); expect( FileSubscribeResponseSchema.parse({ type: "fs.file.subscribe.response", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 4421080c96..3a8385d36a 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2243,6 +2243,7 @@ export const FileExplorerRequestSchema = z.object({ mode: z.enum(["list", "file"]), requestId: z.string(), acceptBinary: z.boolean().optional(), + workspaceId: z.string().optional(), }); export const FileVersionSchema = z.discriminatedUnion("status", [ @@ -2273,6 +2274,7 @@ export const FileSubscribeRequestSchema = z.object({ path: z.string(), subscriptionId: z.string(), requestId: z.string(), + workspaceId: z.string().optional(), }); export const FileUnsubscribeRequestSchema = z.object({ @@ -2289,12 +2291,14 @@ export const FileWriteRequestSchema = z.object({ expectedModifiedAt: z.string(), expectedRevision: z.string().optional(), requestId: z.string(), + workspaceId: z.string().optional(), }); export const ProjectIconRequestSchema = z.object({ type: z.literal("project_icon_request"), cwd: z.string(), requestId: z.string(), + workspaceId: z.string().optional(), }); export const ProjectIconGetRequestSchema = z.object({ @@ -2308,6 +2312,7 @@ export const FileDownloadTokenRequestSchema = z.object({ cwd: z.string(), path: z.string(), requestId: z.string(), + workspaceId: z.string().optional(), }); export const FileUploadRequestSchema = z.object({ diff --git a/packages/server/package.json b/packages/server/package.json index f487a83eb6..cd3f7d681c 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -38,7 +38,7 @@ "clean": "node ../../scripts/clean-package-dist.mjs", "build": "npm run build:lib && npm run build:scripts", "build:clean": "npm run clean && npm run build", - "build:lib": "tsc -p tsconfig.server.json --incremental false && node -e \"const fs=require('node:fs'); fs.mkdirSync('dist/server/server/speech/providers/local/sherpa/assets',{recursive:true}); fs.copyFileSync('src/server/speech/providers/local/sherpa/assets/silero_vad.onnx','dist/server/server/speech/providers/local/sherpa/assets/silero_vad.onnx'); fs.cpSync('src/terminal/shell-integration','dist/server/terminal/shell-integration',{recursive:true}); fs.cpSync('src/terminal/shell-integration','dist/src/terminal/shell-integration',{recursive:true}); fs.copyFileSync('src/terminal/terminal-ts-loader.mjs','dist/server/terminal/terminal-ts-loader.mjs');\"", + "build:lib": "tsc -p tsconfig.server.json --incremental false && node -e \"const fs=require('node:fs'); fs.mkdirSync('dist/server/server/speech/providers/local/sherpa/assets',{recursive:true}); fs.copyFileSync('src/server/speech/providers/local/sherpa/assets/silero_vad.onnx','dist/server/server/speech/providers/local/sherpa/assets/silero_vad.onnx'); fs.cpSync('src/terminal/shell-integration','dist/server/terminal/shell-integration',{recursive:true}); fs.cpSync('src/terminal/shell-integration','dist/src/terminal/shell-integration',{recursive:true}); fs.copyFileSync('src/terminal/terminal-ts-loader.mjs','dist/server/terminal/terminal-ts-loader.mjs'); fs.copyFileSync('src/server/workspace-helper/executable.mjs','dist/server/server/workspace-helper/executable.mjs');\"", "build:scripts": "tsc -p tsconfig.scripts.json --incremental false && node -e \"const fs=require('node:fs'); fs.mkdirSync('dist/scripts',{recursive:true}); fs.copyFileSync('scripts/mcp-stdio-socket-bridge-cli.mjs','dist/scripts/mcp-stdio-socket-bridge-cli.mjs');\"", "prepack": "npm run build:clean && npm --prefix ../.. run build:daemon-web-ui", "start": "node dist/scripts/supervisor-entrypoint.js", diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 38cce0e84f..63d70f61c7 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1,5 +1,7 @@ import express from "express"; import { createServer as createHTTPServer, type IncomingMessage, type ServerResponse } from "http"; +import { Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; import { constants, existsSync, unlinkSync } from "fs"; import { open } from "fs/promises"; import { randomUUID } from "node:crypto"; @@ -396,6 +398,7 @@ export interface PaseoDaemonConfig { type: "command"; command: readonly [string, ...string[]]; options?: Readonly>; + helperCommand?: readonly [string, ...string[]]; } > >; @@ -776,6 +779,27 @@ export async function createPaseoDaemon( let fileHandle: Awaited> | null = null; try { + if (entry.open) { + const opened = await entry.open(); + res.setHeader("Content-Type", entry.mimeType); + res.setHeader( + "Content-Disposition", + `attachment; filename="${entry.fileName.replace(/["\r\n]/g, "_")}"`, + ); + res.setHeader("Content-Length", opened.size.toString()); + const stream = Readable.from(opened.chunks); + try { + await pipeline(stream, res); + } catch (err) { + logger.error({ err }, "Failed to stream runtime download"); + if (!res.headersSent) res.status(500).json({ error: "Failed to read file" }); + else res.destroy(err instanceof Error ? err : new Error(String(err))); + } finally { + await opened.chunks.cancel(); + } + return; + } + if (!entry.absolutePath) throw new Error("Download source is unavailable"); fileHandle = await open(entry.absolutePath, DOWNLOAD_OPEN_FLAGS); const fileStats = await fileHandle.stat(); if (!fileStats.isFile()) { diff --git a/packages/server/src/server/config.test.ts b/packages/server/src/server/config.test.ts index e92a60347d..c8a12c0569 100644 --- a/packages/server/src/server/config.test.ts +++ b/packages/server/src/server/config.test.ts @@ -40,6 +40,7 @@ describe("server config", () => { type: "command", command: ["/trusted/runtime", "--mode", "fixture"], options: { image: "fixture:test" }, + helperCommand: ["/runtime/bin/workspace-helper"], }, }, }), @@ -51,6 +52,7 @@ describe("server config", () => { type: "command", command: ["/trusted/runtime", "--mode", "fixture"], options: { image: "fixture:test" }, + helperCommand: ["/runtime/bin/workspace-helper"], }, }); }); @@ -74,6 +76,10 @@ describe("server config", () => { type: "command", command: [process.execPath, fixtureExecutable], options: { stateDirectory }, + helperCommand: [ + process.execPath, + fileURLToPath(new URL("./workspace-helper/executable.mjs", import.meta.url)), + ], }, }, }), diff --git a/packages/server/src/server/daemon-e2e/workspace-runtime-characterization.e2e.test.ts b/packages/server/src/server/daemon-e2e/workspace-runtime-characterization.e2e.test.ts index e172f71ed3..1d589428dd 100644 --- a/packages/server/src/server/daemon-e2e/workspace-runtime-characterization.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/workspace-runtime-characterization.e2e.test.ts @@ -67,6 +67,7 @@ function createRepository(): string { }); execFileSync("git", ["config", "user.name", "Paseo Test"], { cwd: repo, stdio: "pipe" }); writeFileSync(path.join(repo, "characterized.txt"), "before\n"); + writeFileSync(path.join(repo, "binary.bin"), Buffer.alloc(700_000, 0xa5)); writeFileSync( path.join(repo, "paseo.json"), JSON.stringify({ @@ -131,10 +132,15 @@ async function waitForTerminalOutput(terminalId: string, marker: string): Promis } async function expectFileEditAndWatch(workspace: CharacterizedWorkspace): Promise { - const listing = await client.listDirectory(workspace.cwd, "."); + const listing = await client.listDirectory(workspace.cwd, ".", undefined, workspace.id); expect(listing.entries.map((entry) => entry.name)).toContain("characterized.txt"); - const initialRead = await client.readFile(workspace.cwd, "characterized.txt"); + const initialRead = await client.readFile( + workspace.cwd, + "characterized.txt", + undefined, + workspace.id, + ); expect(new TextDecoder().decode(initialRead.bytes)).toBe("before\n"); let resolveUpdate!: (version: FileVersion) => void; @@ -142,7 +148,7 @@ async function expectFileEditAndWatch(workspace: CharacterizedWorkspace): Promis resolveUpdate = resolve; }); const fileSubscription = await client.subscribeFile( - { cwd: workspace.cwd, path: "characterized.txt" }, + { cwd: workspace.cwd, path: "characterized.txt", workspaceId: workspace.id }, resolveUpdate, ); expect(fileSubscription.initial).toMatchObject({ status: "ready", size: 7 }); @@ -156,13 +162,34 @@ async function expectFileEditAndWatch(workspace: CharacterizedWorkspace): Promis content: "after\n", expectedModifiedAt: fileSubscription.initial.modifiedAt, expectedRevision: fileSubscription.initial.revision, + workspaceId: workspace.id, }); expect(write).toMatchObject({ status: "written", size: 6 }); await expect(updated).resolves.toMatchObject({ status: "ready", size: 6 }); fileSubscription.unsubscribe(); - const writtenRead = await client.readFile(workspace.cwd, "characterized.txt"); + const writtenRead = await client.readFile( + workspace.cwd, + "characterized.txt", + undefined, + workspace.id, + ); expect(new TextDecoder().decode(writtenRead.bytes)).toBe("after\n"); + const binaryRead = await client.readFile(workspace.cwd, "binary.bin", undefined, workspace.id); + expect(binaryRead.bytes).toEqual(new Uint8Array(Buffer.alloc(700_000, 0xa5))); + + const download = await client.requestDownloadToken( + workspace.cwd, + "binary.bin", + undefined, + workspace.id, + ); + expect(download.error).toBeNull(); + const response = await fetch( + `http://127.0.0.1:${daemon.port}/api/files/download?token=${download.token}`, + ); + expect(response.status).toBe(200); + expect(new Uint8Array(await response.arrayBuffer())).toEqual(binaryRead.bytes); } async function expectGitObservation(workspace: CharacterizedWorkspace): Promise { @@ -180,6 +207,15 @@ async function expectGitObservation(workspace: CharacterizedWorkspace): Promise< } async function expectTerminalCommand(workspace: CharacterizedWorkspace): Promise { + let resolveFileUpdate!: (version: FileVersion) => void; + const fileUpdated = new Promise((resolve) => { + resolveFileUpdate = resolve; + }); + const fileSubscription = await client.subscribeFile( + { cwd: workspace.cwd, path: "terminal-edit.txt", workspaceId: workspace.id }, + resolveFileUpdate, + ); + expect(fileSubscription.initial).toMatchObject({ status: "missing" }); const terminal = await client.createTerminal( workspace.cwd, `${workspace.kind} terminal`, @@ -191,14 +227,18 @@ async function expectTerminalCommand(workspace: CharacterizedWorkspace): Promise const terminalId = terminal.terminal?.id; if (!terminalId) throw new Error(terminal.error ?? "Failed to create terminal"); const marker = "runtime-terminal-ok"; - const command = "printf '%s%s\\n' runtime-terminal- ok"; + const command = "printf terminal-edit > terminal-edit.txt; printf '%s%s\\n' runtime-terminal- ok"; expect(command).not.toContain(marker); try { const terminalOutput = waitForTerminalOutput(terminalId, marker); await client.subscribeTerminal(terminalId); client.sendTerminalInput(terminalId, { type: "input", data: `${command}\r` }); await expect(terminalOutput).resolves.toContain(marker); + await expect(fileUpdated).resolves.toMatchObject({ status: "ready", size: 13 }); + const edit = await client.readFile(workspace.cwd, "terminal-edit.txt", undefined, workspace.id); + expect(new TextDecoder().decode(edit.bytes)).toBe("terminal-edit"); } finally { + fileSubscription.unsubscribe(); await client.killTerminal(terminalId); } } diff --git a/packages/server/src/server/file-download/token-store.ts b/packages/server/src/server/file-download/token-store.ts index 7b82f3fc66..672158072f 100644 --- a/packages/server/src/server/file-download/token-store.ts +++ b/packages/server/src/server/file-download/token-store.ts @@ -3,13 +3,18 @@ import { randomUUID } from "node:crypto"; export interface DownloadTokenEntry { token: string; path: string; - absolutePath: string; + absolutePath?: string; + open?: () => Promise<{ chunks: DownloadContent; size: number }>; fileName: string; mimeType: string; size: number; expiresAt: number; } +export interface DownloadContent extends AsyncIterable { + cancel(): Promise; +} + interface DownloadTokenStoreOptions { ttlMs: number; now?: () => number; diff --git a/packages/server/src/server/persisted-config.ts b/packages/server/src/server/persisted-config.ts index 1dac79bf86..79c04f5a27 100644 --- a/packages/server/src/server/persisted-config.ts +++ b/packages/server/src/server/persisted-config.ts @@ -92,6 +92,11 @@ const WorkspaceRuntimesConfigSchema = z.record( .min(1) .transform((command) => command as [string, ...string[]]), options: z.record(z.string(), z.unknown()).optional(), + helperCommand: z + .array(z.string().min(1)) + .min(1) + .transform((command) => command as [string, ...string[]]) + .optional(), }) .strict(), ); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 24afed06de..d89b38cef7 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -769,6 +769,8 @@ export class Session { downloadTokenStore, paseoHome, logger: this.sessionLogger, + workspaceRuntime, + workspaceRegistry, }); this.agentManager = agentManager; this.agentStorage = agentStorage; @@ -990,6 +992,7 @@ export class Session { serviceProxy: this.serviceProxy, scriptRuntimeStore: this.scriptRuntimeStore, terminalManager: this.terminalManager, + workspaceRuntime: this.workspaceRuntime, workspaceRegistry: this.workspaceRegistry, projectRegistry: this.projectRegistry, workspaceGitService: this.workspaceGitService, @@ -2197,8 +2200,7 @@ export class Session { case "fs.file.subscribe.request": return this.workspaceFilesSession.handleFileSubscribeRequest(msg); case "fs.file.unsubscribe.request": - this.workspaceFilesSession.handleFileUnsubscribeRequest(msg); - return undefined; + return this.workspaceFilesSession.handleFileUnsubscribeRequest(msg); case "fs.file.write.request": return this.workspaceFilesSession.handleFileWriteRequest(msg); case "project_icon_request": @@ -4476,7 +4478,7 @@ export class Session { statusEnteredAt: null, activityAt: null, diffStat, - scripts: this.buildWorkspaceScriptPayloadSnapshot(workspace, resolvedProjectRecord), + scripts: await this.buildWorkspaceScriptPayloadSnapshot(workspace, resolvedProjectRecord), ...(resolvedProjectRecord ? { project: await this.buildProjectPlacementForWorkspace(workspace, resolvedProjectRecord), @@ -5854,11 +5856,11 @@ export class Session { // Named accessor: the workspace descriptor builder and the git-watch test both read a workspace's // scripts snapshot through here; the workspace-scripts module owns the payload assembly. - private buildWorkspaceScriptPayloadSnapshot( + private async buildWorkspaceScriptPayloadSnapshot( workspace: PersistedWorkspaceRecord, project: PersistedProjectRecord | null, - ): WorkspaceDescriptorPayload["scripts"] { - return this.workspaceScripts.buildSnapshot(workspace, project); + ): Promise { + return await this.workspaceScripts.buildSnapshot(workspace, project); } private handleStartWorkspaceScriptRequest(request: StartWorkspaceScriptRequest): Promise { @@ -7002,7 +7004,7 @@ export class Session { this.checkoutSession.cleanup(); this.workspaceGitObserver.dispose(); - this.workspaceFilesSession.dispose(); + await this.workspaceFilesSession.dispose(); } } 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 0b09c87743..c9f14e9cb5 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 @@ -15,6 +15,8 @@ import { } from "./workspace-files-session.js"; import { DownloadTokenStore } from "../../file-download/token-store.js"; import type { SessionOutboundMessage } from "../../messages.js"; +import { createWorkspaceRuntimeService } from "../../workspace-runtime/index.js"; +import type { PersistedWorkspaceRecord } from "../../workspace-registry.js"; const tempDirs: string[] = []; @@ -34,6 +36,10 @@ function makeSubsystem( options: { hasBinaryChannel?: boolean; emitBinary?: (frame: Uint8Array) => Promise | void; + selectedWorkspace?: { + record: PersistedWorkspaceRecord; + runtime: ReturnType; + }; } = {}, ) { const emitted: SessionOutboundMessage[] = []; @@ -48,17 +54,31 @@ function makeSubsystem( hasBinaryChannel: () => hasBinary, }; const paseoHome = makeDir("workspace-files-home-"); + const downloadTokenStore = new DownloadTokenStore({ ttlMs: 60_000 }); const subsystem = new WorkspaceFilesSession({ host, - downloadTokenStore: new DownloadTokenStore({ ttlMs: 60_000 }), + downloadTokenStore, paseoHome, logger: pino({ level: "silent" }), + ...(options.selectedWorkspace + ? { + workspaceRuntime: options.selectedWorkspace.runtime, + workspaceRegistry: { + get: async (workspaceId: string) => + workspaceId === options.selectedWorkspace?.record.workspaceId + ? options.selectedWorkspace.record + : null, + list: async () => [options.selectedWorkspace!.record], + }, + } + : {}), }); return { subsystem, emitted, binary, paseoHome, + downloadTokenStore, setHasBinary: (value: boolean) => { hasBinary = value; }, @@ -73,7 +93,135 @@ function uploadFrame(args: Parameters[0]): FileT return frame; } +async function collectUpload(chunks: AsyncIterable): Promise { + const buffers: Buffer[] = []; + for await (const chunk of chunks) buffers.push(Buffer.from(chunk)); + return Buffer.concat(buffers); +} + describe("WorkspaceFilesSession", () => { + test("routes the complete selected-workspace file surface through workspace runtime", async () => { + const cwd = makeDir("workspace-files-selected-"); + writeFileSync(join(cwd, "runtime.txt"), "before\n"); + writeFileSync(join(cwd, "favicon.svg"), ''); + const runtimeIds = new Map(); + const runtime = createWorkspaceRuntimeService({ + paseoHome: makeDir("workspace-files-runtime-home-"), + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await runtime.create({ + workspaceId: "selected-workspace", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: cwd } }, + placement: { kind: "existing" }, + }); + const now = new Date().toISOString(); + const record: PersistedWorkspaceRecord = { + workspaceId: "selected-workspace", + projectId: "project", + cwd, + kind: "directory", + displayName: "selected", + title: null, + branch: null, + worktreeRoot: null, + baseBranch: null, + isPaseoOwnedWorktree: false, + mainRepoRoot: null, + runtime: { runtimeId: "local" }, + createdAt: now, + updatedAt: now, + archivedAt: null, + autoArchivedChangeRequestUrl: null, + pinnedAt: null, + }; + const { subsystem, emitted, binary, downloadTokenStore } = makeSubsystem({ + hasBinaryChannel: true, + selectedWorkspace: { record, runtime }, + }); + + await subsystem.handleFileExplorerRequest({ + type: "file_explorer_request", + workspaceId: record.workspaceId, + cwd, + path: ".", + mode: "list", + requestId: "selected-list", + }); + const listingMessage = emitted.at(-1); + if (listingMessage?.type !== "file_explorer_response") { + throw new Error("Expected selected file listing"); + } + expect(listingMessage.payload.directory?.entries.map((entry) => entry.name)).toContain( + "runtime.txt", + ); + + await subsystem.handleFileExplorerRequest({ + type: "file_explorer_request", + workspaceId: record.workspaceId, + cwd, + path: "runtime.txt", + mode: "file", + acceptBinary: true, + requestId: "selected-read", + }); + expect(binary.map((frame) => decodeFileTransferFrame(frame)?.opcode)).toEqual([ + FileTransferOpcode.FileBegin, + FileTransferOpcode.FileChunk, + FileTransferOpcode.FileEnd, + ]); + + const version = await runtime.files(record.workspaceId).stat("runtime.txt"); + if (version.status !== "ready") throw new Error("Expected runtime.txt to exist"); + await subsystem.handleFileWriteRequest({ + type: "fs.file.write.request", + workspaceId: record.workspaceId, + cwd, + path: "runtime.txt", + content: "after\n", + expectedModifiedAt: version.modifiedAt, + expectedRevision: version.revision, + requestId: "selected-write", + }); + expect(emitted.at(-1)).toMatchObject({ + type: "fs.file.write.response", + payload: { result: { status: "written", size: 6 } }, + }); + + await subsystem.handleFileDownloadTokenRequest({ + type: "file_download_token_request", + workspaceId: record.workspaceId, + cwd, + path: "runtime.txt", + requestId: "selected-download", + }); + const tokenMessage = emitted.at(-1); + if (tokenMessage?.type !== "file_download_token_response" || !tokenMessage.payload.token) { + throw new Error("Expected selected download token"); + } + const download = downloadTokenStore.consumeToken(tokenMessage.payload.token); + expect(download?.absolutePath).toBeUndefined(); + const opened = await download?.open?.(); + expect(opened && Buffer.from(await collectUpload(opened.chunks)).toString("utf8")).toBe( + "after\n", + ); + await subsystem.handleProjectIconRequest({ + type: "project_icon_request", + workspaceId: record.workspaceId, + cwd, + requestId: "selected-icon", + }); + expect(emitted.at(-1)).toMatchObject({ + type: "project_icon_response", + payload: { icon: { mimeType: "image/svg+xml" }, error: null }, + }); + await subsystem.dispose(); + await runtime.destroy(record.workspaceId); + }); + test("lists directory entries", async () => { const cwd = makeDir("workspace-files-list-"); writeFileSync(join(cwd, "a.txt"), "alpha"); 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 dd7d7d04d4..387a4ec025 100644 --- a/packages/server/src/server/session/files/workspace-files-session.ts +++ b/packages/server/src/server/session/files/workspace-files-session.ts @@ -15,6 +15,10 @@ import type { SessionInboundMessage, SessionOutboundMessage, } from "../../messages.js"; +import type { WorkspaceFileKind, WorkspaceFiles } from "../../workspace-helper/index.js"; +import type { WorkspaceRuntimeService } from "../../workspace-runtime/index.js"; +import type { WorkspaceRegistry } from "../../workspace-registry.js"; +import { areEquivalentPaths } from "../../../utils/path.js"; import { FileUploadStore } from "../../file-upload/index.js"; import type { DownloadTokenStore } from "../../file-download/token-store.js"; import { @@ -25,7 +29,14 @@ import { writeExplorerFile, } from "../../file-explorer/service.js"; import { workspaceFileObserver, type FileObserver } from "../../file-explorer/observer.js"; -import { getProjectIcon } from "../../../utils/project-icon.js"; +import { + getProjectIcon, + ICON_PATTERNS, + IGNORED_DIRS, + MONOREPO_PACKAGE_DIRS, + PRIORITY_DIRS, + projectIconFromBytes, +} from "../../../utils/project-icon.js"; /** * What a workspace file-access request reaches outside its own domain: the @@ -45,6 +56,8 @@ export interface WorkspaceFilesSessionOptions { paseoHome: string; logger: pino.Logger; fileObserver?: FileObserver; + workspaceRuntime?: WorkspaceRuntimeService; + workspaceRegistry?: Pick; } /** @@ -60,7 +73,9 @@ export class WorkspaceFilesSession { private readonly logger: pino.Logger; private readonly fileUploads: FileUploadStore; private readonly fileObserver: FileObserver; - private readonly fileSubscriptions = new Map void>(); + private readonly workspaceRuntime: WorkspaceRuntimeService | null; + private readonly workspaceRegistry: Pick | null; + private readonly fileSubscriptions = new Map Promise>(); constructor(options: WorkspaceFilesSessionOptions) { this.host = options.host; @@ -68,11 +83,68 @@ export class WorkspaceFilesSession { this.logger = options.logger; this.fileUploads = new FileUploadStore({ paseoHome: options.paseoHome }); this.fileObserver = options.fileObserver ?? workspaceFileObserver; + this.workspaceRuntime = options.workspaceRuntime ?? null; + this.workspaceRegistry = options.workspaceRegistry ?? null; } async handleFileSubscribeRequest(request: FileSubscribeRequest): Promise { - this.fileSubscriptions.get(request.subscriptionId)?.(); + await this.fileSubscriptions.get(request.subscriptionId)?.(); try { + const selectedFiles = await this.resolveSelectedFiles(request.cwd, request.workspaceId); + if (selectedFiles) { + const subscription = await selectedFiles.subscribe({ paths: [request.path] }, (event) => { + if (event.type === "overflow") return; + if (event.type === "error") { + this.host.emit({ + type: "fs.file.update", + payload: { + subscriptionId: request.subscriptionId, + version: { + status: "error", + cwd: request.cwd, + path: request.path, + error: event.error, + }, + }, + }); + return; + } + void selectedFiles.stat(request.path).then( + (version) => + this.host.emit({ + type: "fs.file.update", + payload: { + subscriptionId: request.subscriptionId, + version: toProtocolVersion(request.cwd, version), + }, + }), + (error) => + this.host.emit({ + type: "fs.file.update", + payload: { + subscriptionId: request.subscriptionId, + version: { + status: "error", + cwd: request.cwd, + path: request.path, + error: getErrorMessage(error), + }, + }, + }), + ); + }); + const initial = toProtocolVersion(request.cwd, await selectedFiles.stat(request.path)); + this.fileSubscriptions.set(request.subscriptionId, () => subscription.unsubscribe()); + this.host.emit({ + type: "fs.file.subscribe.response", + payload: { + subscriptionId: request.subscriptionId, + initial, + requestId: request.requestId, + }, + }); + return; + } const subscription = await this.fileObserver.subscribe( { cwd: request.cwd, path: request.path }, (version) => { @@ -82,7 +154,7 @@ export class WorkspaceFilesSession { }); }, ); - this.fileSubscriptions.set(request.subscriptionId, subscription.unsubscribe); + this.fileSubscriptions.set(request.subscriptionId, async () => subscription.unsubscribe()); this.host.emit({ type: "fs.file.subscribe.response", payload: { @@ -108,8 +180,8 @@ export class WorkspaceFilesSession { } } - handleFileUnsubscribeRequest(request: FileUnsubscribeRequest): void { - this.fileSubscriptions.get(request.subscriptionId)?.(); + async handleFileUnsubscribeRequest(request: FileUnsubscribeRequest): Promise { + await this.fileSubscriptions.get(request.subscriptionId)?.(); this.fileSubscriptions.delete(request.subscriptionId); this.host.emit({ type: "fs.file.unsubscribe.response", @@ -118,21 +190,32 @@ export class WorkspaceFilesSession { } async handleFileWriteRequest(request: FileWriteRequest): Promise { - const result = await writeExplorerFile({ - root: request.cwd, - relativePath: request.path, - content: request.content, - expectedModifiedAt: request.expectedModifiedAt, - expectedRevision: request.expectedRevision, - }); + const selectedFiles = await this.resolveSelectedFiles(request.cwd, request.workspaceId); + const result = selectedFiles + ? toProtocolWriteResult( + request.cwd, + await selectedFiles.write({ + path: request.path, + contents: Buffer.from(request.content, "utf8"), + expectedModifiedAt: request.expectedModifiedAt, + expectedRevision: request.expectedRevision, + }), + ) + : await writeExplorerFile({ + root: request.cwd, + relativePath: request.path, + content: request.content, + expectedModifiedAt: request.expectedModifiedAt, + expectedRevision: request.expectedRevision, + }); this.host.emit({ type: "fs.file.write.response", payload: { result, requestId: request.requestId }, }); } - dispose(): void { - for (const unsubscribe of this.fileSubscriptions.values()) unsubscribe(); + async dispose(): Promise { + await Promise.all([...this.fileSubscriptions.values()].map((unsubscribe) => unsubscribe())); this.fileSubscriptions.clear(); } @@ -159,11 +242,11 @@ export class WorkspaceFilesSession { } try { + const selectedFiles = await this.resolveSelectedFiles(cwd, request.workspaceId); if (mode === "list") { - const directory = await listDirectoryEntries({ - root: cwd, - relativePath: requestedPath, - }); + const directory = selectedFiles + ? await selectedFiles.list(requestedPath) + : await listDirectoryEntries({ root: cwd, relativePath: requestedPath }); this.host.emit( { @@ -181,6 +264,39 @@ export class WorkspaceFilesSession { source, ); } else { + if (selectedFiles) { + const file = await selectedFiles.read(requestedPath); + if (request.acceptBinary && this.host.hasBinaryChannel()) { + await this.emitStream(requestId, file, source); + } else { + const bytes = await collectBytes(file.chunks); + this.host.emit( + { + type: "file_explorer_response", + payload: { + cwd, + path: file.path, + mode, + directory: null, + file: { + path: file.path, + kind: file.kind, + encoding: explorerEncoding(file.kind), + ...explorerContent(file.kind, bytes), + mimeType: file.mimeType, + size: file.size, + modifiedAt: file.modifiedAt, + revision: file.revision, + }, + error: null, + requestId, + }, + }, + source, + ); + } + return; + } if (request.acceptBinary && this.host.hasBinaryChannel()) { await streamExplorerFile({ root: cwd, relativePath: requestedPath }, async (file) => { await this.host.emitBinary( @@ -278,7 +394,10 @@ export class WorkspaceFilesSession { const { cwd, requestId } = request; try { - const icon = await getProjectIcon(cwd); + const selectedFiles = await this.resolveSelectedFiles(cwd, request.workspaceId); + const icon = selectedFiles + ? await getWorkspaceProjectIcon(selectedFiles) + : await getProjectIcon(cwd); this.host.emit({ type: "project_icon_response", payload: { @@ -327,6 +446,37 @@ export class WorkspaceFilesSession { ); try { + const selectedFiles = await this.resolveSelectedFiles(cwd, request.workspaceId); + if (selectedFiles) { + const info = await selectedFiles.stat(requestedPath); + if (info.status !== "ready") { + throw new Error(info.status === "error" ? info.error : "File not found"); + } + const entry = this.downloadTokenStore.issueToken({ + path: info.path, + fileName: info.path.split("/").at(-1) ?? "download", + mimeType: info.mimeType, + size: info.size, + open: async () => { + const file = await selectedFiles.read(requestedPath); + return { chunks: file.chunks, size: file.size }; + }, + }); + this.host.emit({ + type: "file_download_token_response", + payload: { + cwd, + path: info.path, + token: entry.token, + fileName: entry.fileName, + mimeType: entry.mimeType, + size: entry.size, + error: null, + requestId, + }, + }); + return; + } const info = await getDownloadableFileInfo({ root: cwd, relativePath: requestedPath, @@ -373,4 +523,130 @@ export class WorkspaceFilesSession { }); } } + + private async resolveSelectedFiles( + cwd: string, + workspaceId: string | undefined, + ): Promise { + if (!this.workspaceRuntime || !this.workspaceRegistry) return null; + if (workspaceId) { + const workspace = await this.workspaceRegistry.get(workspaceId); + if (!workspace) throw new Error(`Workspace not found: ${workspaceId}`); + return workspace.runtime ? this.workspaceRuntime.files(workspaceId) : null; + } + const matches = (await this.workspaceRegistry.list()).filter( + (workspace) => + workspace.archivedAt === null && + workspace.runtime && + areEquivalentPaths(workspace.cwd, cwd), + ); + if (matches.length > 1) { + throw new Error("workspaceId is required because the workspace path is not unique"); + } + return matches[0] ? this.workspaceRuntime.files(matches[0].workspaceId) : null; + } + + private async emitStream( + requestId: string, + file: Awaited>, + source?: object, + ): Promise { + await this.host.emitBinary( + encodeFileTransferFrame({ + opcode: FileTransferOpcode.FileBegin, + requestId, + metadata: { + mime: file.mimeType, + size: file.size, + encoding: file.encoding, + modifiedAt: file.modifiedAt, + revision: file.revision, + }, + }), + source, + ); + for await (const chunk of file.chunks) { + await this.host.emitBinary( + encodeFileTransferFrame({ + opcode: FileTransferOpcode.FileChunk, + requestId, + payload: chunk, + }), + source, + ); + } + await this.host.emitBinary( + encodeFileTransferFrame({ opcode: FileTransferOpcode.FileEnd, requestId }), + source, + ); + } +} + +function explorerEncoding(kind: WorkspaceFileKind): "base64" | "none" | "utf-8" { + if (kind === "image") return "base64"; + if (kind === "binary") return "none"; + return "utf-8"; +} + +function explorerContent(kind: WorkspaceFileKind, bytes: Uint8Array): { content?: string } { + if (kind === "image") return { content: Buffer.from(bytes).toString("base64") }; + if (kind === "text") return { content: Buffer.from(bytes).toString("utf8") }; + return {}; +} + +function toProtocolVersion(cwd: string, version: Awaited>) { + return { ...version, cwd }; +} + +function toProtocolWriteResult(cwd: string, result: Awaited>) { + return result.status === "conflict" + ? { ...result, version: toProtocolVersion(cwd, result.version) } + : result; +} + +async function collectBytes(chunks: AsyncIterable): Promise { + const buffers: Buffer[] = []; + for await (const chunk of chunks) buffers.push(Buffer.from(chunk)); + return Buffer.concat(buffers); +} + +async function getWorkspaceProjectIcon(files: WorkspaceFiles) { + const candidates = [...PRIORITY_DIRS, ...MONOREPO_PACKAGE_DIRS, "."]; + const ignored = new Set(IGNORED_DIRS); + for (const root of candidates) { + const icon = await findWorkspaceIcon(files, root, ignored, root === "." ? 0 : 2); + if (!icon) continue; + const read = await files.read(icon); + if (read.size > 32 * 1024) continue; + return projectIconFromBytes(icon, await collectBytes(read.chunks)); + } + return null; +} + +async function findWorkspaceIcon( + files: WorkspaceFiles, + directory: string, + ignored: ReadonlySet, + depth: number, +): Promise { + let listing: Awaited>; + try { + listing = await files.list(directory); + } catch { + return null; + } + for (const pattern of ICON_PATTERNS) { + const expression = new RegExp(`^${pattern.replace(/\./g, "\\.").replace(/\*/g, ".*")}$`); + const match = listing.entries.find( + (entry) => entry.kind === "file" && expression.test(entry.name), + ); + if (match) return match.path; + } + if (depth === 0) return null; + for (const entry of listing.entries) { + if (entry.kind !== "directory" || ignored.has(entry.name)) continue; + const nested = await findWorkspaceIcon(files, entry.path, ignored, depth - 1); + if (nested) return nested; + } + return null; } diff --git a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts index c377639747..c35022e413 100644 --- a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts +++ b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts @@ -21,6 +21,10 @@ import type { import type { WorkspaceGitService } from "../../workspace-git-service.js"; import { createWorkspaceScriptsService } from "./workspace-scripts-service.js"; import { deriveProjectServiceSlug } from "../../workspace-git-metadata.js"; +import { + createWorkspaceRuntimeService, + type WorkspaceRuntimeService, +} from "../../workspace-runtime/index.js"; // The production module reads only WorkspaceGitService.{peekSnapshot,getProjectSlug}, // WorkspaceRegistry.get, and forwards the launcher + opaque managers to the injected @@ -77,6 +81,7 @@ interface BuildOptions { project?: PersistedProjectRecord | null; spawnThrows?: string; gitService?: Pick; + workspaceRuntime?: WorkspaceRuntimeService; } function buildService(options: BuildOptions = {}) { @@ -101,6 +106,7 @@ function buildService(options: BuildOptions = {}) { workspaceRegistry: fakeWorkspaceRegistry(workspace), projectRegistry: fakeProjectRegistry(options.project ?? null), workspaceGitService: options.gitService ?? fakeGitService(), + workspaceRuntime: options.workspaceRuntime, getDaemonTcpPort: () => 6767, getDaemonTcpHost: () => "127.0.0.1", serviceProxyPublicBaseUrl: null, @@ -146,14 +152,20 @@ describe("buildSnapshot", () => { test("returns no scripts when the service proxy is unavailable", async () => { const { service } = buildService({ serviceProxy: null }); expect( - service.buildSnapshot({ workspaceId: "ws-1", cwd: "/tmp/repo" } as PersistedWorkspaceRecord), + await service.buildSnapshot({ + workspaceId: "ws-1", + cwd: "/tmp/repo", + } as PersistedWorkspaceRecord), ).toEqual([]); }); test("returns no scripts when the runtime store is unavailable", async () => { const { service } = buildService({ scriptRuntimeStore: null }); expect( - service.buildSnapshot({ workspaceId: "ws-1", cwd: "/tmp/repo" } as PersistedWorkspaceRecord), + await service.buildSnapshot({ + workspaceId: "ws-1", + cwd: "/tmp/repo", + } as PersistedWorkspaceRecord), ).toEqual([]); }); @@ -162,10 +174,46 @@ describe("buildSnapshot", () => { tempDirs.push(dir); const { service } = buildService(); expect( - service.buildSnapshot({ workspaceId: "ws-1", cwd: dir } as PersistedWorkspaceRecord), + await service.buildSnapshot({ workspaceId: "ws-1", cwd: dir } as PersistedWorkspaceRecord), ).toEqual([]); }); + test("discovers paseo.json inside a selected workspace runtime", async () => { + const dir = mkdtempSync(join(tmpdir(), "workspace-scripts-runtime-")); + const paseoHome = mkdtempSync(join(tmpdir(), "workspace-scripts-runtime-home-")); + tempDirs.push(dir, paseoHome); + writeFileSync( + join(dir, "paseo.json"), + JSON.stringify({ scripts: { app: { command: "npm run app" } } }), + ); + const runtimeIds = new Map(); + const workspaceRuntime = createWorkspaceRuntimeService({ + paseoHome, + resolveRuntimeId: async (workspaceId) => runtimeIds.get(workspaceId) ?? null, + persistRuntimeId: async (workspaceId, runtimeId) => { + runtimeIds.set(workspaceId, runtimeId); + }, + }); + await workspaceRuntime.create({ + workspaceId: "ws-runtime-config", + runtimeId: "local", + project: { id: "project", source: { kind: "host-directory", path: dir } }, + placement: { kind: "existing" }, + }); + const workspace = { + workspaceId: "ws-runtime-config", + projectId: "project", + cwd: dir, + runtime: { runtimeId: "local" }, + } as PersistedWorkspaceRecord; + const { service } = buildService({ workspace, workspaceRuntime }); + + await expect(service.buildSnapshot(workspace)).resolves.toEqual([ + expect.objectContaining({ scriptName: "app", type: "script" }), + ]); + await workspaceRuntime.destroy(workspace.workspaceId); + }); + test("projects service hostnames without a Git snapshot", async () => { const directory = mkdtempSync(join(tmpdir(), "workspace-scripts-")); tempDirs.push(directory); @@ -197,7 +245,7 @@ describe("buildSnapshot", () => { gitService: { peekSnapshot: () => undefined }, }); - expect(service.buildSnapshot(workspace, project)[0]?.hostname).toBe( + expect((await service.buildSnapshot(workspace, project))[0]?.hostname).toBe( serviceProxy.projectWorkspaceService({ projectSlug: deriveProjectServiceSlug(project), branchName: workspace.branch, @@ -416,7 +464,7 @@ describe("start", () => { const serviceProxy = createServiceProxySubsystem({ logger }); const { service, spawnCalls } = buildService({ workspace, project, serviceProxy }); - const snapshot = service.buildSnapshot(workspace, project); + const snapshot = await service.buildSnapshot(workspace, project); await service.start({ ...request, workspaceId: workspace.workspaceId }); const started = spawnCalls[0]!; diff --git a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts index b43e13548b..03a760c1bb 100644 --- a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts +++ b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts @@ -26,6 +26,8 @@ import { } from "../../script-status-projection.js"; import { deriveProjectServiceSlug, deriveProjectSlug } from "../../workspace-git-metadata.js"; import type { PaseoServicePortAllocation } from "@getpaseo/protocol/paseo-config-schema"; +import { PaseoConfigSchema, type PaseoConfig } from "@getpaseo/protocol/paseo-config-schema"; +import type { WorkspaceRuntimeService } from "../../workspace-runtime/index.js"; type WorkspaceScriptsPayload = WorkspaceDescriptorPayload["scripts"]; @@ -42,7 +44,7 @@ export interface WorkspaceScriptsService { buildSnapshot( workspace: PersistedWorkspaceRecord, project?: PersistedProjectRecord | null, - ): WorkspaceScriptsPayload; + ): Promise; emitStatusUpdate(workspaceId: string, workspaceDirectory: string): Promise; list(workspaceId: string): Promise; launch(input: { workspaceId: string; scriptName: string }): Promise; @@ -59,6 +61,7 @@ export function createWorkspaceScriptsService(deps: { workspaceRegistry: Pick; projectRegistry: Pick; workspaceGitService: WorkspaceScriptsGitSource; + workspaceRuntime?: WorkspaceRuntimeService; getDaemonTcpPort: (() => number | null) | null; getDaemonTcpHost: (() => string | null) | null; serviceProxyPublicBaseUrl: string | null; @@ -75,6 +78,7 @@ export function createWorkspaceScriptsService(deps: { workspaceRegistry, projectRegistry, workspaceGitService, + workspaceRuntime, getDaemonTcpPort, getDaemonTcpHost, serviceProxyPublicBaseUrl, @@ -106,17 +110,17 @@ export function createWorkspaceScriptsService(deps: { }; } - function buildSnapshot( + async function buildSnapshot( workspace: PersistedWorkspaceRecord, project: PersistedProjectRecord | null = null, - ): WorkspaceScriptsPayload { + ): Promise { if (!serviceProxy || !scriptRuntimeStore) { return []; } return buildWorkspaceScriptPayloads({ workspaceId: workspace.workspaceId, workspaceDirectory: workspace.cwd, - paseoConfig: readPaseoConfigForProjection(workspace.cwd, logger), + paseoConfig: await readWorkspaceConfig(workspace), serviceProxy, runtimeStore: scriptRuntimeStore, daemonPort: getDaemonTcpPort?.() ?? null, @@ -126,6 +130,37 @@ export function createWorkspaceScriptsService(deps: { }); } + async function readWorkspaceConfig( + workspace: PersistedWorkspaceRecord, + ): Promise { + if (!workspace.runtime) return readPaseoConfigForProjection(workspace.cwd, logger); + if (!workspaceRuntime) { + throw new Error(`Workspace runtime is not available: ${workspace.workspaceId}`); + } + const files = workspaceRuntime.files(workspace.workspaceId); + const version = await files.stat("paseo.json"); + if (version.status === "missing") return null; + if (version.status === "error") { + logger.warn( + { workspaceId: workspace.workspaceId, error: version.error }, + "Failed to read runtime paseo.json; treating workspace as having no scripts", + ); + return null; + } + try { + const file = await files.read("paseo.json"); + const chunks: Buffer[] = []; + for await (const chunk of file.chunks) chunks.push(Buffer.from(chunk)); + return PaseoConfigSchema.parse(JSON.parse(Buffer.concat(chunks).toString("utf8"))); + } catch (error) { + logger.warn( + { workspaceId: workspace.workspaceId, err: error }, + "Failed to parse runtime paseo.json; treating workspace as having no scripts", + ); + return null; + } + } + async function emitStatusUpdate(workspaceId: string, _workspaceDirectory: string): Promise { try { const workspace = await workspaceRegistry.get(workspaceId); @@ -133,7 +168,7 @@ export function createWorkspaceScriptsService(deps: { const project = await projectRegistry.get(workspace.projectId); emit({ type: "script_status_update", - payload: { workspaceId, scripts: buildSnapshot(workspace, project) }, + payload: { workspaceId, scripts: await buildSnapshot(workspace, project) }, }); } catch (error) { logger.warn({ err: error, workspaceId }, "Failed to project workspace script status"); @@ -163,7 +198,7 @@ export function createWorkspaceScriptsService(deps: { requireAvailable(); const workspace = await getWorkspace(workspaceId); const project = await projectRegistry.get(workspace.projectId); - return buildSnapshot(workspace, project); + return await buildSnapshot(workspace, project); } async function launchProcess(input: { workspaceId: string; scriptName: string }) { @@ -197,13 +232,13 @@ export function createWorkspaceScriptsService(deps: { scriptName: string; }): Promise { const { workspace, project } = await launchProcess(input); - const script = buildSnapshot(workspace, project).find( + const script = (await buildSnapshot(workspace, project)).find( (entry) => entry.scriptName === input.scriptName, ); if (!script) { throw new Error(`Script '${input.scriptName}' did not produce a status record`); } - void emitStatusUpdate(workspace.workspaceId, workspace.cwd); + await emitStatusUpdate(workspace.workspaceId, workspace.cwd); return script; } @@ -225,20 +260,20 @@ export function createWorkspaceScriptsService(deps: { // The launcher's terminal exit listener owns route removal and runtime state updates. await available.terminalManager.killTerminalAndWait(runtime.terminalId); - const script = buildSnapshot(workspace, project).find( + const script = (await buildSnapshot(workspace, project)).find( (entry) => entry.scriptName === input.scriptName, ); if (!script) { throw new Error(`Script '${input.scriptName}' did not produce a status record`); } - void emitStatusUpdate(workspace.workspaceId, workspace.cwd); + await emitStatusUpdate(workspace.workspaceId, workspace.cwd); return script; } async function start(request: StartWorkspaceScriptRequest): Promise { try { const { workspace, terminalId } = await launchProcess(request); - void emitStatusUpdate(workspace.workspaceId, workspace.cwd); + await emitStatusUpdate(workspace.workspaceId, workspace.cwd); emit({ type: "start_workspace_script_response", payload: { diff --git a/packages/server/src/server/workspace-helper/executable.mjs b/packages/server/src/server/workspace-helper/executable.mjs new file mode 100644 index 0000000000..18d4cc263d --- /dev/null +++ b/packages/server/src/server/workspace-helper/executable.mjs @@ -0,0 +1,281 @@ +#!/usr/local/bin/node +import { randomUUID } from "node:crypto"; +import { constants, watch } from "node:fs"; +import { open, realpath, readdir, rename, stat, unlink } from "node:fs/promises"; +import path from "node:path"; +import readline from "node:readline"; + +const outsideMessage = "Access outside of workspace is not allowed"; +const imageTypes = new Map([ + [".png", "image/png"], + [".jpg", "image/jpeg"], + [".jpeg", "image/jpeg"], + [".gif", "image/gif"], + [".webp", "image/webp"], + [".svg", "image/svg+xml"], +]); +const argv = process.argv.slice(2); +const operation = argv.shift(); +const argument = (name, fallback) => { + const index = argv.indexOf(name); + return index < 0 ? fallback : argv[index + 1]; +}; +const root = argument("--root"); + +try { + if (!operation || !root) throw new Error("Usage: workspace-helper --root "); + if (operation === "fs-stat") print(await fileStat(root, argument("--path", "."))); + else if (operation === "fs-list") print(await list(root, argument("--path", "."))); + else if (operation === "fs-read") await read(root, argument("--path")); + else if (operation === "fs-write") await write(root, argument("--path")); + else if (operation === "watch") await runWatcher(root); + else if (operation === "describe") print({ version: 1, capabilities: ["files", "watch"] }); + else throw new Error(`Unknown workspace-helper command: ${operation}`); +} catch (error) { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exitCode = 1; +} + +async function fileStat(workspaceRoot, relativePath) { + try { + const scoped = await resolveScoped(workspaceRoot, relativePath); + const info = await stat(scoped.absolute, { bigint: true }); + if (!info.isFile()) + return { status: "error", path: scoped.relative, error: "Requested path is not a file" }; + const type = await classify(scoped.absolute, Number(info.size)); + return { + status: "ready", + path: scoped.relative, + ...type, + size: Number(info.size), + modifiedAt: info.mtime.toISOString(), + revision: revision(info), + }; + } catch (error) { + if (isMissing(error)) return { status: "missing", path: relativePath }; + return { + status: "error", + path: relativePath, + error: error instanceof Error ? error.message : String(error), + }; + } +} + +async function list(workspaceRoot, relativePath) { + const directory = await resolveScoped(workspaceRoot, relativePath); + const info = await stat(directory.absolute); + if (!info.isDirectory()) throw new Error("Requested path is not a directory"); + const entries = []; + for (const dirent of await readdir(directory.absolute, { withFileTypes: true })) { + try { + const child = await resolveScoped( + workspaceRoot, + path.posix.join(directory.relative, dirent.name), + ); + const childInfo = await stat(child.absolute); + entries.push({ + name: dirent.name, + path: child.relative, + kind: childInfo.isDirectory() ? "directory" : "file", + size: childInfo.size, + modifiedAt: childInfo.mtime.toISOString(), + }); + } catch (error) { + if (!isMissing(error) && error?.message !== outsideMessage) throw error; + } + } + entries.sort( + (left, right) => + Date.parse(right.modifiedAt) - Date.parse(left.modifiedAt) || + left.name.localeCompare(right.name), + ); + return { path: directory.relative, entries }; +} + +async function read(workspaceRoot, relativePath) { + const scoped = await resolveScoped(workspaceRoot, relativePath); + const handle = await open( + scoped.absolute, + process.platform === "win32" ? constants.O_RDONLY : constants.O_RDONLY | constants.O_NOFOLLOW, + ); + try { + const info = await handle.stat(); + if (!info.isFile()) throw new Error("Requested path is not a file"); + for await (const chunk of handle.createReadStream({ + autoClose: false, + highWaterMark: 256 * 1024, + })) { + if (!process.stdout.write(chunk)) + await new Promise((resolve) => process.stdout.once("drain", resolve)); + } + } finally { + await handle.close(); + } +} + +async function write(workspaceRoot, relativePath) { + const scoped = await resolveScoped(workspaceRoot, relativePath); + const before = await stat(scoped.absolute, { bigint: true }).catch((error) => + isMissing(error) ? null : Promise.reject(error), + ); + if (!before) + return print({ status: "conflict", version: { status: "missing", path: relativePath } }); + if (!before.isFile()) return print({ status: "error", error: "Requested path is not a file" }); + const expectedRevision = argument("--expected-revision"); + const expectedModifiedAt = argument("--expected-modified-at"); + if ( + (expectedRevision && revision(before) !== expectedRevision) || + (!expectedRevision && expectedModifiedAt && before.mtime.toISOString() !== expectedModifiedAt) + ) { + return print({ status: "conflict", version: await fileStat(workspaceRoot, relativePath) }); + } + const temporary = path.join( + path.dirname(scoped.absolute), + `.${path.basename(scoped.absolute)}.paseo-${randomUUID()}.tmp`, + ); + let handle; + try { + handle = await open(temporary, "wx", Number(before.mode)); + for await (const chunk of process.stdin) await handle.write(chunk); + await handle.sync(); + await handle.close(); + handle = undefined; + const latest = await stat(scoped.absolute, { bigint: true }); + if (revision(latest) !== revision(before)) + return print({ status: "conflict", version: await fileStat(workspaceRoot, relativePath) }); + await rename(temporary, scoped.absolute); + const result = await stat(scoped.absolute, { bigint: true }); + print({ + status: "written", + modifiedAt: result.mtime.toISOString(), + size: Number(result.size), + revision: revision(result), + }); + } finally { + await handle?.close().catch(() => undefined); + await unlink(temporary).catch(() => undefined); + } +} + +async function runWatcher(workspaceRoot) { + const subscriptions = new Map(); + print({ type: "ready" }); + const lines = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); + try { + for await (const line of lines) { + const command = JSON.parse(line); + if (command.protocolVersion !== 1) throw new Error("Unsupported workspace-helper protocol"); + if ("root" in command) throw new Error("Watch subscriptions cannot supply a root"); + if (command.type === "close") break; + if (command.type === "unsubscribe") { + closeSubscription(subscriptions.get(command.id)); + subscriptions.delete(command.id); + print({ type: "unsubscribed", subscriptionId: command.id }); + continue; + } + if (command.type !== "subscribe") continue; + closeSubscription(subscriptions.get(command.id)); + const observation = { watchers: [], timers: new Map() }; + subscriptions.set(command.id, observation); + for (const requested of command.paths) { + const scoped = await resolveScoped(workspaceRoot, requested); + const expectedName = path.basename(scoped.absolute); + const watcher = watch(path.dirname(scoped.absolute), (_event, filename) => { + if (filename != null && filename.toString() !== expectedName) return; + clearTimeout(observation.timers.get(scoped.relative)); + observation.timers.set( + scoped.relative, + setTimeout(() => { + observation.timers.delete(scoped.relative); + print({ type: "changed", subscriptionId: command.id, paths: [scoped.relative] }); + }, 50), + ); + }); + watcher.on("error", (error) => + print({ type: "error", subscriptionId: command.id, message: error.message }), + ); + observation.watchers.push(watcher); + } + print({ type: "subscribed", subscriptionId: command.id }); + } + } finally { + for (const observation of subscriptions.values()) closeSubscription(observation); + } +} + +function closeSubscription(observation) { + if (!observation) return; + for (const watcher of observation.watchers) watcher.close(); + for (const timer of observation.timers.values()) clearTimeout(timer); +} + +async function resolveScoped(workspaceRoot, relativePath = ".") { + if (path.isAbsolute(relativePath)) throw new Error(outsideMessage); + const normalizedRoot = path.resolve(workspaceRoot); + const requested = path.resolve(normalizedRoot, relativePath); + if (!contains(normalizedRoot, requested)) throw new Error(outsideMessage); + const canonicalRoot = await realpath(normalizedRoot); + try { + const canonical = await realpath(requested); + if (!contains(canonicalRoot, canonical)) throw new Error(outsideMessage); + return { absolute: canonical, relative: normalize(path.relative(normalizedRoot, requested)) }; + } catch (error) { + if (!isMissing(error)) throw error; + const canonicalParent = await realpath(path.dirname(requested)); + if (!contains(canonicalRoot, canonicalParent)) + throw new Error(outsideMessage, { cause: error }); + return { absolute: requested, relative: normalize(path.relative(normalizedRoot, requested)) }; + } +} + +function contains(rootPath, candidate) { + const relative = path.relative(rootPath, candidate); + return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative)); +} +function normalize(relative) { + return relative === "" ? "." : relative.split(path.sep).join("/"); +} +function revision(info) { + return `${info.dev}:${info.ino}:${info.size}:${info.mtimeNs}`; +} +function isMissing(error) { + return ["ENOENT", "ENOTDIR", "ELOOP"].includes(error?.code); +} +function print(value) { + process.stdout.write(`${JSON.stringify({ protocolVersion: 1, ...value })}\n`); +} + +async function classify(file, size) { + const extension = path.extname(file).toLowerCase(); + if (imageTypes.has(extension)) + return { kind: "image", encoding: "binary", mimeType: imageTypes.get(extension) }; + const handle = await open(file, "r"); + try { + const sample = Buffer.alloc(Math.min(8192, size)); + const { bytesRead } = await handle.read(sample, 0, sample.length, 0); + const bytes = sample.subarray(0, bytesRead); + if (binary(bytes)) + return { kind: "binary", encoding: "binary", mimeType: "application/octet-stream" }; + return { + kind: "text", + encoding: "utf-8", + mimeType: extension === ".json" ? "application/json" : "text/plain", + }; + } finally { + await handle.close(); + } +} +function binary(bytes) { + if (bytes.length === 0) return false; + let suspicious = 0; + for (const byte of bytes) { + if (byte === 0) return true; + if ((byte < 32 && byte !== 9 && byte !== 10 && byte !== 13) || byte === 127) suspicious++; + } + try { + new TextDecoder("utf-8", { fatal: true }).decode(bytes); + } catch { + return true; + } + return suspicious / bytes.length > 0.3; +} diff --git a/packages/server/src/server/workspace-helper/fixtures/adversarial-helper.mjs b/packages/server/src/server/workspace-helper/fixtures/adversarial-helper.mjs new file mode 100644 index 0000000000..7ff6898642 --- /dev/null +++ b/packages/server/src/server/workspace-helper/fixtures/adversarial-helper.mjs @@ -0,0 +1,66 @@ +#!/usr/bin/env node +import readline from "node:readline"; + +const mode = process.argv[2]; +const operation = process.argv[3]; + +if (operation === "fs-stat") { + process.stdout.write( + `${JSON.stringify({ + protocolVersion: 1, + status: "ready", + path: "large.bin", + kind: "binary", + encoding: "binary", + mimeType: "application/octet-stream", + size: Number.MAX_SAFE_INTEGER, + modifiedAt: "2026-01-01T00:00:00.000Z", + revision: "fixture-revision", + })}\n`, + ); +} else if (operation === "fs-read" && mode === "endless-read") { + const chunk = Buffer.alloc(256 * 1024, 0xa5); + const write = () => { + while (process.stdout.write(chunk)) continue; + process.stdout.once("drain", write); + }; + write(); +} else if (operation === "watch") { + const safetyExit = setTimeout(() => process.exit(86), 4_000); + const keepAlive = setInterval(() => undefined, 1_000); + const finish = () => { + clearTimeout(safetyExit); + clearInterval(keepAlive); + }; + process.once("exit", finish); + if (mode === "malformed-watch") { + process.stdout.write("not-json\n"); + } else if (mode !== "no-ready") { + print({ type: "ready" }); + const lines = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); + for await (const line of lines) { + handleWatchCommand(JSON.parse(line), finish); + } + } +} else { + process.stderr.write(`Unsupported fixture operation: ${mode} ${operation}\n`); + process.exitCode = 1; +} + +function print(value) { + process.stdout.write(`${JSON.stringify({ protocolVersion: 1, ...value })}\n`); +} + +function handleWatchCommand(command, finish) { + if (command.type === "subscribe" && mode !== "no-subscribed") { + print({ type: "subscribed", subscriptionId: command.id }); + return; + } + if (command.type === "unsubscribe" && mode !== "no-unsubscribed") { + print({ type: "unsubscribed", subscriptionId: command.id }); + return; + } + if (command.type !== "close") return; + if (mode === "resists-close") process.on("SIGTERM", () => undefined); + else finish(); +} diff --git a/packages/server/src/server/workspace-helper/index.ts b/packages/server/src/server/workspace-helper/index.ts new file mode 100644 index 0000000000..150c4b07ff --- /dev/null +++ b/packages/server/src/server/workspace-helper/index.ts @@ -0,0 +1,74 @@ +export type WorkspaceFileKind = "text" | "image" | "binary"; + +export type WorkspaceFileStat = + | { + status: "ready"; + path: string; + kind: WorkspaceFileKind; + encoding: "utf-8" | "binary"; + mimeType: string; + size: number; + modifiedAt: string; + revision: string; + } + | { status: "missing"; path: string } + | { status: "error"; path: string; error: string }; + +export interface WorkspaceDirectoryEntry { + name: string; + path: string; + kind: "file" | "directory"; + size: number; + modifiedAt: string; +} + +export interface WorkspaceDirectory { + path: string; + entries: WorkspaceDirectoryEntry[]; +} + +export interface WorkspaceFileRead { + path: string; + kind: WorkspaceFileKind; + encoding: "utf-8" | "binary"; + mimeType: string; + size: number; + modifiedAt: string; + revision: string; + chunks: WorkspaceFileContent; +} + +export interface WorkspaceFileContent extends AsyncIterable { + /** Stops an incomplete transfer and resolves only after its helper process exits. */ + cancel(): Promise; +} + +export type WorkspaceFileWriteResult = + | { status: "written"; modifiedAt: string; size: number; revision: string } + | { status: "conflict"; version: WorkspaceFileStat } + | { status: "error"; error: string }; + +export type WorkspaceWatchEvent = + | { type: "changed"; paths: string[] } + | { type: "overflow" } + | { type: "error"; error: string }; + +export interface WorkspaceFilesSubscription { + unsubscribe(): Promise; +} + +export interface WorkspaceFiles { + stat(path: string): Promise; + list(path: string): Promise; + read(path: string): Promise; + write(input: { + path: string; + contents: Uint8Array | AsyncIterable; + expectedModifiedAt?: string; + expectedRevision?: string; + }): Promise; + subscribe( + input: { paths: readonly string[] }, + listener: (event: WorkspaceWatchEvent) => void, + ): Promise; +} diff --git a/packages/server/src/server/workspace-helper/integration/index.ts b/packages/server/src/server/workspace-helper/integration/index.ts new file mode 100644 index 0000000000..bb2775e0e6 --- /dev/null +++ b/packages/server/src/server/workspace-helper/integration/index.ts @@ -0,0 +1,27 @@ +import type { Readable, Writable } from "node:stream"; + +import type { WorkspaceFiles } from "../index.js"; +import { createClient } from "../internal/client.js"; + +export interface WorkspaceHelperProcess { + stdin: Writable; + stdout: Readable; + stderr: Readable; + exited: Promise<{ code: number | null; signal: NodeJS.Signals | null }>; + kill(signal?: NodeJS.Signals): void; +} + +export interface WorkspaceFilesOwner { + files: WorkspaceFiles; + verify(): Promise; + close(): Promise; +} + +/** @package Parent integration SPI. Normal callers consume WorkspaceFiles from the public module. */ +export function bindWorkspaceHelper(options: { + root: string; + command: readonly [string, ...string[]]; + launch(argv: readonly [string, ...string[]]): Promise; +}): WorkspaceFilesOwner { + return createClient(options); +} diff --git a/packages/server/src/server/workspace-helper/internal/client.ts b/packages/server/src/server/workspace-helper/internal/client.ts new file mode 100644 index 0000000000..337674eec0 --- /dev/null +++ b/packages/server/src/server/workspace-helper/internal/client.ts @@ -0,0 +1,443 @@ +import type { WorkspaceFileContent, WorkspaceFileRead, WorkspaceWatchEvent } from "../index.js"; +import type { WorkspaceFilesOwner, WorkspaceHelperProcess } from "../integration/index.js"; +import { + describeSchema, + directorySchema, + fileStatSchema, + watchEventSchema, + writeResultSchema, +} from "./protocol.js"; +import type { z } from "zod"; + +interface WatchState { + process: WorkspaceHelperProcess; + subscriptions: Map void>; + acknowledgements: Map; + closed: boolean; + cleanup: Promise | null; + diagnostics: Promise; +} + +const WATCH_ACKNOWLEDGEMENT_TIMEOUT_MS = 750; +const WATCH_CLOSE_TIMEOUT_MS = 500; +const PROCESS_STOP_TIMEOUT_MS = 1_000; + +export function createClient(options: { + root: string; + launch(argv: readonly [string, ...string[]]): Promise; + command: readonly [string, ...string[]]; +}): WorkspaceFilesOwner { + let watcher: WatchState | null = null; + let nextSubscriptionId = 0; + let closing = false; + let closePromise: Promise | null = null; + const processes = new Set(); + + const command = async (name: string, args: readonly string[] = []) => { + if (closing) throw new Error("Workspace files client is closed"); + const child = await options.launch([...options.command, name, "--root", options.root, ...args]); + processes.add(child); + void child.exited.then( + () => processes.delete(child), + () => processes.delete(child), + ); + if (closing) { + await terminate(child); + throw new Error("Workspace files client is closed"); + } + return child; + }; + + async function json( + name: string, + args: readonly string[], + schema: T, + ): Promise> { + const child = await command(name, args); + child.stdin.end(); + const [stdout, stderr, exit] = await Promise.all([ + collect(child.stdout), + collect(child.stderr), + child.exited, + ]); + if (exit.code !== 0 || exit.signal !== null) throw helperError(stderr, exit); + return schema.parse(JSON.parse(stdout)); + } + + return { + files: { + stat(path) { + return json("fs-stat", ["--path", path], fileStatSchema); + }, + list(path) { + return json("fs-list", ["--path", path], directorySchema); + }, + async read(path) { + const metadata = await json("fs-stat", ["--path", path], fileStatSchema); + if (metadata.status !== "ready") { + throw new Error(metadata.status === "error" ? metadata.error : `File not found: ${path}`); + } + const child = await command("fs-read", ["--path", path]); + child.stdin.end(); + const stderr = collect(child.stderr); + const chunks = ownedRead(child, stderr, async () => { + const final = await json("fs-stat", ["--path", path], fileStatSchema); + if (final.status !== "ready" || final.revision !== metadata.revision) { + throw new Error("File changed during transfer"); + } + }); + return { ...metadata, chunks } satisfies WorkspaceFileRead; + }, + async write(input) { + const args = ["--path", input.path]; + if (input.expectedModifiedAt) args.push("--expected-modified-at", input.expectedModifiedAt); + if (input.expectedRevision) args.push("--expected-revision", input.expectedRevision); + const child = await command("fs-write", args); + const stdout = collect(child.stdout); + const stderr = collect(child.stderr); + try { + if (input.contents instanceof Uint8Array) { + child.stdin.end(input.contents); + } else { + for await (const chunk of input.contents) { + if (!child.stdin.write(chunk)) await onceDrain(child.stdin); + } + child.stdin.end(); + } + } catch (error) { + await terminate(child); + throw error; + } + const [body, diagnostics, exit] = await Promise.all([stdout, stderr, child.exited]); + if (exit.code !== 0 || exit.signal !== null) throw helperError(diagnostics, exit); + return writeResultSchema.parse(JSON.parse(body)); + }, + async subscribe(input, listener) { + if (closing) throw new Error("Workspace files client is closed"); + const state = watcher ?? (await startWatcher()); + if (state.closed) throw new Error("Workspace helper watcher is not running"); + const id = `subscription-${++nextSubscriptionId}`; + state.subscriptions.set(id, listener); + const acknowledged = acknowledgement(state, `subscribed:${id}`); + state.process.stdin.write( + `${JSON.stringify({ protocolVersion: 1, type: "subscribe", id, paths: input.paths })}\n`, + ); + await waitForAcknowledgement( + state, + `subscribed:${id}`, + "Workspace helper subscribe acknowledgement timed out", + acknowledged, + ); + let active = true; + return { + async unsubscribe() { + if (!active) return; + active = false; + state.subscriptions.delete(id); + if (!state.closed) { + const removed = acknowledgement(state, `unsubscribed:${id}`); + state.process.stdin.write( + `${JSON.stringify({ protocolVersion: 1, type: "unsubscribe", id })}\n`, + ); + await waitForAcknowledgement( + state, + `unsubscribed:${id}`, + "Workspace helper unsubscribe acknowledgement timed out", + removed, + ); + } + if (state.subscriptions.size === 0) await stopWatcher(state); + }, + }; + }, + }, + async verify() { + const description = await json("describe", [], describeSchema); + if ( + !description.capabilities.includes("files") || + !description.capabilities.includes("watch") + ) { + throw new Error("Workspace helper does not support files and watching"); + } + }, + async close() { + closePromise ??= (async () => { + closing = true; + if (watcher) { + await stopWatcher(watcher, new Error("Workspace files client is closed")); + } + await Promise.all([...processes].map((child) => terminate(child))); + })(); + await closePromise; + }, + }; + + async function startWatcher(): Promise { + const process = await command("watch"); + const state: WatchState = { + process, + subscriptions: new Map(), + acknowledgements: new Map(), + closed: false, + cleanup: null, + diagnostics: collect(process.stderr), + }; + watcher = state; + process.stdin.on("error", () => undefined); + const ready = acknowledgement(state, "ready"); + void consumeLines(process.stdout, (line) => handleWatchEvent(state, line)) + .then( + () => failWatcher(state, new Error("Workspace helper watcher output ended")), + (error) => failWatcher(state, toError(error)), + ) + .catch(() => undefined); + void process.exited + .then( + (exit) => { + return failWatcher( + state, + new Error(`Workspace helper watcher exited: ${exit.code ?? exit.signal}`), + ); + }, + (error) => { + return failWatcher(state, toError(error)); + }, + ) + .catch(() => undefined); + await waitForAcknowledgement( + state, + "ready", + "Workspace helper ready acknowledgement timed out", + ready, + ); + return state; + } + + async function stopWatcher(state: WatchState, reason?: Error): Promise { + if (state.cleanup) return state.cleanup; + if (state.closed) return; + state.closed = true; + if (watcher === state) watcher = null; + if (reason) { + for (const listener of state.subscriptions.values()) { + listener({ type: "error", error: reason.message }); + } + } + state.subscriptions.clear(); + const closedError = reason ?? new Error("Workspace helper watcher is closed"); + for (const pending of state.acknowledgements.values()) pending.reject(closedError); + state.acknowledgements.clear(); + state.process.stdin.end(`${JSON.stringify({ protocolVersion: 1, type: "close" })}\n`); + state.cleanup = (async () => { + const gracefulExit = await exitWithin(state.process.exited, WATCH_CLOSE_TIMEOUT_MS); + if (!gracefulExit) await terminate(state.process); + state.process.stdin.destroy(); + state.process.stdout.destroy(); + await state.diagnostics.catch(() => undefined); + return undefined; + })(); + await state.cleanup; + } + + async function waitForAcknowledgement( + state: WatchState, + key: string, + timeoutMessage: string, + pending: Promise, + ): Promise { + try { + await withTimeout(pending, WATCH_ACKNOWLEDGEMENT_TIMEOUT_MS, timeoutMessage); + } catch (error) { + state.acknowledgements.delete(key); + const failure = toError(error); + await failWatcher(state, failure); + throw failure; + } + } + + async function failWatcher(state: WatchState, error: Error): Promise { + if (state.cleanup) return state.cleanup; + if (state.closed) return; + state.closed = true; + if (watcher === state) watcher = null; + for (const listener of state.subscriptions.values()) { + listener({ type: "error", error: error.message }); + } + state.subscriptions.clear(); + for (const pending of state.acknowledgements.values()) pending.reject(error); + state.acknowledgements.clear(); + state.cleanup = terminate(state.process).then(async () => { + state.process.stdin.destroy(); + state.process.stdout.destroy(); + await state.diagnostics.catch(() => undefined); + return undefined; + }); + return state.cleanup; + } +} + +function handleWatchEvent(state: WatchState, line: string): void { + const event = watchEventSchema.parse(JSON.parse(line)); + const subscriptionId = "subscriptionId" in event ? event.subscriptionId : undefined; + let acknowledgementKey: string | null = null; + if (event.type === "ready") acknowledgementKey = "ready"; + else if (subscriptionId) acknowledgementKey = `${event.type}:${subscriptionId}`; + if (acknowledgementKey) { + state.acknowledgements.get(acknowledgementKey)?.resolve(); + state.acknowledgements.delete(acknowledgementKey); + } + if (!subscriptionId) return; + const listener = state.subscriptions.get(subscriptionId); + if (!listener) return; + if (event.type === "changed") listener({ type: "changed", paths: event.paths ?? [] }); + if (event.type === "overflow") listener({ type: "overflow" }); + if (event.type === "error") listener({ type: "error", error: event.message ?? "Watcher failed" }); +} + +function acknowledgement(state: WatchState, key: string): Promise { + return new Promise((resolve, reject) => state.acknowledgements.set(key, { resolve, reject })); +} + +async function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function toError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +async function collect(stream: NodeJS.ReadableStream): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + return Buffer.concat(chunks).toString("utf8"); +} + +function ownedRead( + child: WorkspaceHelperProcess, + stderr: Promise, + verify: () => Promise, +): WorkspaceFileContent { + const source = child.stdout[Symbol.asyncIterator](); + let iteratorCreated = false; + let finished = false; + let cancellation: Promise | null = null; + + async function cancel(): Promise { + if (finished) return; + cancellation ??= terminate(child).then(async () => { + finished = true; + child.stdout.destroy(); + child.stdin.destroy(); + await stderr.catch(() => undefined); + return undefined; + }); + await cancellation; + } + + return { + cancel, + [Symbol.asyncIterator]() { + if (iteratorCreated) throw new Error("Workspace file content can only be consumed once"); + iteratorCreated = true; + return { + async next() { + if (finished) return { done: true, value: undefined }; + try { + const result = await source.next(); + if (!result.done) { + return { + done: false, + value: Buffer.isBuffer(result.value) ? result.value : Buffer.from(result.value), + }; + } + const [diagnostics, exit] = await Promise.all([stderr, child.exited]); + finished = true; + if (exit.code !== 0 || exit.signal !== null) throw helperError(diagnostics, exit); + await verify(); + return { done: true, value: undefined }; + } catch (error) { + await cancel(); + throw error; + } + }, + async return() { + await cancel(); + return { done: true, value: undefined }; + }, + async throw(error) { + await cancel(); + throw error; + }, + }; + }, + }; +} + +async function terminate(child: WorkspaceHelperProcess): Promise { + child.kill("SIGTERM"); + let exit = await exitWithin(child.exited, PROCESS_STOP_TIMEOUT_MS); + if (!exit) { + child.kill("SIGKILL"); + exit = await exitWithin(child.exited, PROCESS_STOP_TIMEOUT_MS); + } + if (!exit) throw new Error("Workspace helper process did not exit after SIGKILL"); +} + +async function exitWithin( + exited: WorkspaceHelperProcess["exited"], + timeoutMs: number, +): Promise | null> { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + exited, + new Promise((resolve) => { + timer = setTimeout(() => resolve(null), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function helperError( + stderr: string, + exit: { code: number | null; signal: NodeJS.Signals | null }, +): Error { + return new Error(stderr.trim() || `Workspace helper failed: ${exit.code ?? exit.signal}`); +} + +function onceDrain(stream: NodeJS.WritableStream): Promise { + return new Promise((resolve, reject) => { + stream.once("drain", resolve); + stream.once("error", reject); + }); +} + +async function consumeLines( + stream: NodeJS.ReadableStream, + consume: (line: string) => void, +): Promise { + let buffered = ""; + for await (const chunk of stream) { + buffered += chunk.toString(); + for (;;) { + const newline = buffered.indexOf("\n"); + if (newline < 0) break; + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + if (line) consume(line); + } + } +} diff --git a/packages/server/src/server/workspace-helper/internal/protocol.ts b/packages/server/src/server/workspace-helper/internal/protocol.ts new file mode 100644 index 0000000000..63dadf49d4 --- /dev/null +++ b/packages/server/src/server/workspace-helper/internal/protocol.ts @@ -0,0 +1,91 @@ +import { z } from "zod"; + +const version = { protocolVersion: z.literal(1) }; +const fileKind = z.enum(["text", "image", "binary"]); + +export const describeSchema = z.object({ + ...version, + version: z.literal(1), + capabilities: z.array(z.enum(["files", "watch"])), +}); + +const readyFile = z.object({ + status: z.literal("ready"), + path: z.string(), + kind: fileKind, + encoding: z.enum(["utf-8", "binary"]), + mimeType: z.string(), + size: z.number().int().nonnegative(), + modifiedAt: z.string(), + revision: z.string(), +}); + +const fileStatValueSchema = z.discriminatedUnion("status", [ + readyFile, + z.object({ status: z.literal("missing"), path: z.string() }), + z.object({ + status: z.literal("error"), + path: z.string(), + error: z.string(), + }), +]); + +export const fileStatSchema = fileStatValueSchema.and(z.object(version)).transform((result) => { + const { protocolVersion: _protocolVersion, ...value } = result; + return value; +}); + +export const directorySchema = z + .object({ + ...version, + path: z.string(), + entries: z.array( + z.object({ + name: z.string(), + path: z.string(), + kind: z.enum(["file", "directory"]), + size: z.number().int().nonnegative(), + modifiedAt: z.string(), + }), + ), + }) + .transform((result) => { + const { protocolVersion: _protocolVersion, ...value } = result; + return value; + }); + +export const writeResultSchema = z + .discriminatedUnion("status", [ + z.object({ + ...version, + status: z.literal("written"), + modifiedAt: z.string(), + size: z.number().int().nonnegative(), + revision: z.string(), + }), + z.object({ ...version, status: z.literal("conflict"), version: fileStatValueSchema }), + z.object({ ...version, status: z.literal("error"), error: z.string() }), + ]) + .transform((result) => { + const { protocolVersion: _protocolVersion, ...value } = result; + return value; + }); + +export const watchEventSchema = z.discriminatedUnion("type", [ + z.object({ ...version, type: z.literal("ready") }), + z.object({ ...version, type: z.literal("subscribed"), subscriptionId: z.string() }), + z.object({ ...version, type: z.literal("unsubscribed"), subscriptionId: z.string() }), + z.object({ + ...version, + type: z.literal("changed"), + subscriptionId: z.string(), + paths: z.array(z.string()), + }), + z.object({ ...version, type: z.literal("overflow"), subscriptionId: z.string() }), + z.object({ + ...version, + type: z.literal("error"), + subscriptionId: z.string().optional(), + message: z.string(), + }), +]); diff --git a/packages/server/src/server/workspace-helper/workspace-helper.posix.test.ts b/packages/server/src/server/workspace-helper/workspace-helper.posix.test.ts new file mode 100644 index 0000000000..32dbb1b1f7 --- /dev/null +++ b/packages/server/src/server/workspace-helper/workspace-helper.posix.test.ts @@ -0,0 +1,319 @@ +import { spawn } from "node:child_process"; +import { mkdir, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +import { afterEach, describe, expect, test } from "vitest"; + +import { bindWorkspaceHelper, type WorkspaceHelperProcess } from "./integration/index.js"; + +const posixDescribe = describe.runIf(process.platform !== "win32"); +const cleanupRoots: string[] = []; +const adversarialHelper = fileURLToPath( + new URL("./fixtures/adversarial-helper.mjs", import.meta.url), +); +const workspaceHelper = fileURLToPath(new URL("./executable.mjs", import.meta.url)); + +afterEach(async () => { + await Promise.all( + cleanupRoots.splice(0).map((root) => rm(root, { recursive: true, force: true })), + ); +}); + +async function fixture() { + const parent = await mkdtemp(path.join(tmpdir(), "paseo-workspace-helper-")); + cleanupRoots.push(parent); + const root = path.join(parent, "workspace"); + await mkdir(root); + const client = bindWorkspaceHelper({ + root, + command: [process.execPath, workspaceHelper], + launch: launchTestProcess, + }); + return { client, parent, root }; +} + +async function collect(chunks: AsyncIterable): Promise { + const collected: Buffer[] = []; + for await (const chunk of chunks) collected.push(Buffer.from(chunk)); + return Buffer.concat(collected); +} + +posixDescribe("workspace-helper public capability", () => { + test("watch rejects a second root authority supplied by a subscriber", async () => { + const parent = await mkdtemp(path.join(tmpdir(), "paseo-workspace-helper-root-")); + cleanupRoots.push(parent); + const trustedRoot = path.join(parent, "trusted"); + const untrustedRoot = path.join(parent, "untrusted"); + await mkdir(trustedRoot); + await mkdir(untrustedRoot); + await writeFile(path.join(untrustedRoot, "outside.txt"), "outside\n"); + const child = spawn(process.execPath, [workspaceHelper, "watch", "--root", trustedRoot], { + stdio: ["pipe", "pipe", "pipe"], + }); + const stdout = collect(child.stdout); + const stderr = collect(child.stderr); + child.stdin.end( + `${JSON.stringify({ + protocolVersion: 1, + type: "subscribe", + id: "malicious", + root: untrustedRoot, + paths: ["outside.txt"], + })}\n${JSON.stringify({ protocolVersion: 1, type: "close" })}\n`, + ); + + await expect( + new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => resolve({ code, signal })); + }), + ).resolves.toEqual({ code: 1, signal: null }); + expect((await stderr).toString()).toContain("Watch subscriptions cannot supply a root"); + expect((await stdout).toString()).not.toContain("subscribed"); + }); + + test("multi-path watches preserve exact paths for matching basenames", async () => { + const { client, root } = await fixture(); + await mkdir(path.join(root, "a")); + await mkdir(path.join(root, "b")); + await writeFile(path.join(root, "a", "file.txt"), "a\n"); + await writeFile(path.join(root, "b", "file.txt"), "b\n"); + let resolveChanged!: (paths: string[]) => void; + const changed = new Promise((resolve) => { + resolveChanged = resolve; + }); + const subscription = await client.files.subscribe( + { paths: ["a/file.txt", "b/file.txt"] }, + (event) => { + if (event.type === "changed") resolveChanged(event.paths); + }, + ); + + await writeFile(path.join(root, "a", "file.txt"), "changed\n"); + await expect(changed).resolves.toEqual(["a/file.txt"]); + await subscription.unsubscribe(); + await client.close(); + }); + + test("cancelling a partially-consumed read waits for the helper to exit", async () => { + const parent = await mkdtemp(path.join(tmpdir(), "paseo-workspace-helper-cancel-")); + cleanupRoots.push(parent); + const children: ReturnType[] = []; + const exits: WorkspaceHelperProcess["exited"][] = []; + const client = bindWorkspaceHelper({ + root: parent, + command: [process.execPath, adversarialHelper, "endless-read"], + launch: async (argv): Promise => { + const child = spawn(argv[0], argv.slice(1), { stdio: ["pipe", "pipe", "pipe"] }); + children.push(child); + const exited = new Promise>((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => resolve({ code, signal })); + }); + exits.push(exited); + return { + stdin: child.stdin, + stdout: child.stdout, + stderr: child.stderr, + exited, + kill: (signal) => child.kill(signal), + }; + }, + }); + + try { + const file = await client.files.read("large.bin"); + const iterator = file.chunks[Symbol.asyncIterator](); + await expect(iterator.next()).resolves.toMatchObject({ done: false }); + await file.chunks.cancel(); + await expect(exits[1]).resolves.toMatchObject({ signal: "SIGTERM" }); + expect(() => process.kill(children[1]?.pid ?? 0, 0)).toThrow(); + + const returned = await client.files.read("large.bin"); + for await (const _chunk of returned.chunks) break; + await expect(exits[3]).resolves.toMatchObject({ signal: "SIGTERM" }); + expect(() => process.kill(children[3]?.pid ?? 0, 0)).toThrow(); + } finally { + for (const child of children) child.kill("SIGKILL"); + await client.close(); + } + }); + + test("malformed watcher output rejects only after terminating the helper", async () => { + const recorded = await recordedClient("malformed-watch"); + try { + await expect( + recorded.client.files.subscribe({ paths: ["notes.txt"] }, () => undefined), + ).rejects.toThrow(/JSON|Unexpected token/); + await expect(recorded.exits[0]).resolves.toMatchObject({ signal: "SIGTERM" }); + expect(() => process.kill(recorded.pids[0] ?? 0, 0)).toThrow(); + } finally { + await recorded.cleanup(); + } + }); + + test.each([ + ["no-ready", "ready acknowledgement timed out"], + ["no-subscribed", "subscribe acknowledgement timed out"], + ["no-unsubscribed", "unsubscribe acknowledgement timed out"], + ])("%s watcher acknowledgements fail after bounded cleanup", async (mode, expectedError) => { + const recorded = await recordedClient(mode); + try { + if (mode === "no-unsubscribed") { + const subscription = await recorded.client.files.subscribe( + { paths: ["notes.txt"] }, + () => undefined, + ); + await expect(subscription.unsubscribe()).rejects.toThrow(expectedError); + } else { + await expect( + recorded.client.files.subscribe({ paths: ["notes.txt"] }, () => undefined), + ).rejects.toThrow(expectedError); + } + await expect(recorded.exits[0]).resolves.toMatchObject({ signal: "SIGTERM" }); + expect(() => process.kill(recorded.pids[0] ?? 0, 0)).toThrow(); + } finally { + await recorded.cleanup(); + } + }); + + test("watcher close escalates through TERM to KILL before unsubscribe settles", async () => { + const recorded = await recordedClient("resists-close"); + try { + const subscription = await recorded.client.files.subscribe( + { paths: ["notes.txt"] }, + () => undefined, + ); + await subscription.unsubscribe(); + await expect(recorded.exits[0]).resolves.toMatchObject({ signal: "SIGKILL" }); + expect(() => process.kill(recorded.pids[0] ?? 0, 0)).toThrow(); + } finally { + await recorded.cleanup(); + } + }); + + test("lists, streams, writes optimistically, confines symlinks, and observes changes", async () => { + const { client, parent, root } = await fixture(); + const binary = Buffer.alloc(700_000, 0xa5); + await writeFile(path.join(root, "large.bin"), binary); + await writeFile(path.join(root, "notes.txt"), "before\n"); + await writeFile(path.join(root, "upload.bin"), new Uint8Array()); + + const listing = await client.files.list("."); + expect(listing.entries.map((entry) => entry.name).sort()).toEqual([ + "large.bin", + "notes.txt", + "upload.bin", + ]); + + const streamed = await client.files.read("large.bin"); + expect(streamed).toMatchObject({ path: "large.bin", kind: "binary", size: binary.byteLength }); + expect(await collect(streamed.chunks)).toEqual(binary); + + const upload = await client.files.write({ + path: "upload.bin", + contents: (async function* () { + for (let offset = 0; offset < binary.byteLength; offset += 64 * 1024) { + yield binary.subarray(offset, Math.min(offset + 64 * 1024, binary.byteLength)); + } + })(), + }); + expect(upload).toMatchObject({ status: "written", size: binary.byteLength }); + expect(await readFile(path.join(root, "upload.bin"))).toEqual(binary); + + const initial = await client.files.stat("notes.txt"); + expect(initial).toMatchObject({ status: "ready", path: "notes.txt", size: 7 }); + if (initial.status !== "ready") throw new Error("Expected notes.txt to exist"); + + let resolveChanged!: () => void; + const changed = new Promise((resolve) => { + resolveChanged = resolve; + }); + const changeSubscription = await client.files.subscribe({ paths: ["notes.txt"] }, (event) => { + if (event.type === "changed" && event.paths.includes("notes.txt")) resolveChanged(); + }); + const subscription = await client.files.subscribe({ paths: ["notes.txt"] }, () => undefined); + const written = await client.files.write({ + path: "notes.txt", + contents: Buffer.from("after\n"), + expectedRevision: initial.revision, + }); + expect(written).toMatchObject({ status: "written", size: 6 }); + await expect(changed).resolves.toBeUndefined(); + expect(await readFile(path.join(root, "notes.txt"), "utf8")).toBe("after\n"); + + const conflict = await client.files.write({ + path: "notes.txt", + contents: Buffer.from("stale\n"), + expectedRevision: initial.revision, + }); + expect(conflict).toMatchObject({ status: "conflict" }); + + await writeFile(path.join(parent, "outside.txt"), "secret\n"); + await symlink(path.join(parent, "outside.txt"), path.join(root, "escape.txt")); + await expect(client.files.read("escape.txt")).rejects.toThrow( + "Access outside of workspace is not allowed", + ); + + await subscription.unsubscribe(); + await changeSubscription.unsubscribe(); + await client.close(); + }); +}); + +async function recordedClient(mode: string) { + const root = await mkdtemp(path.join(tmpdir(), `paseo-workspace-helper-${mode}-`)); + cleanupRoots.push(root); + const children: ReturnType[] = []; + const exits: WorkspaceHelperProcess["exited"][] = []; + const pids: number[] = []; + const client = bindWorkspaceHelper({ + root, + command: [process.execPath, adversarialHelper, mode], + launch: async (argv): Promise => { + const child = spawn(argv[0], argv.slice(1), { stdio: ["pipe", "pipe", "pipe"] }); + const exited = new Promise>((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => resolve({ code, signal })); + }); + children.push(child); + exits.push(exited); + if (child.pid) pids.push(child.pid); + return { + stdin: child.stdin, + stdout: child.stdout, + stderr: child.stderr, + exited, + kill: (signal) => child.kill(signal), + }; + }, + }); + return { + client, + exits, + pids, + async cleanup() { + for (const child of children) child.kill("SIGKILL"); + await Promise.allSettled(exits); + await client.close().catch(() => undefined); + }, + }; +} + +async function launchTestProcess( + argv: readonly [string, ...string[]], +): Promise { + const child = spawn(argv[0], argv.slice(1), { stdio: ["pipe", "pipe", "pipe"] }); + return { + stdin: child.stdin, + stdout: child.stdout, + stderr: child.stderr, + exited: new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => resolve({ code, signal })); + }), + kill: (signal) => child.kill(signal), + }; +} diff --git a/packages/server/src/server/workspace-runtime/command/index.ts b/packages/server/src/server/workspace-runtime/command/index.ts index 3ed44d4ed5..b952f51ab6 100644 --- a/packages/server/src/server/workspace-runtime/command/index.ts +++ b/packages/server/src/server/workspace-runtime/command/index.ts @@ -4,6 +4,7 @@ import { createCommandRuntime } from "./internal/command-runtime.js"; export interface CommandRuntimeAdapterConfig { command: readonly [string, ...string[]]; options?: Readonly>; + helperCommand?: readonly [string, ...string[]]; } export function createCommandRuntimeAdapter( diff --git a/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts b/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts index d7e740bb2f..3a3c228d96 100644 --- a/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts +++ b/packages/server/src/server/workspace-runtime/command/internal/command-runtime.ts @@ -23,6 +23,7 @@ const COMMAND_RUNTIME_CLEANUP_TIMEOUT_MS = 1_000; export interface CommandRuntimeConfig { command: readonly [string, ...string[]]; options?: Readonly>; + helperCommand?: readonly [string, ...string[]]; } export function createCommandRuntime( @@ -57,6 +58,7 @@ export function createCommandRuntime( return { id: runtimeId, + workspaceHelperCommand: config.helperCommand ?? ["paseo-workspace-helper"], async create(input) { const response = await lifecycle("create", input.workspaceId, input); if (response.type !== "state") throw new Error(`Invalid create response from ${runtimeId}`); diff --git a/packages/server/src/server/workspace-runtime/drivers/index.ts b/packages/server/src/server/workspace-runtime/drivers/index.ts index fe9f658bc6..ccd093b0cc 100644 --- a/packages/server/src/server/workspace-runtime/drivers/index.ts +++ b/packages/server/src/server/workspace-runtime/drivers/index.ts @@ -82,6 +82,8 @@ export type WorkspaceDriverProcess = WorkspacePipeProcess | WorkspacePtyProcess; export interface WorkspaceRuntimeDriver { readonly id: WorkspaceRuntimeId; + /** Runtime-local command for Paseo's compatible workspace helper. */ + readonly workspaceHelperCommand: readonly [string, ...string[]]; create(input: WorkspaceDriverCreateInput): Promise; inspect(workspaceId: WorkspaceId): Promise; spawn(input: WorkspaceDriverSpawnInput): Promise; diff --git a/packages/server/src/server/workspace-runtime/index.ts b/packages/server/src/server/workspace-runtime/index.ts index 3882f5e4c7..1eb8b51da4 100644 --- a/packages/server/src/server/workspace-runtime/index.ts +++ b/packages/server/src/server/workspace-runtime/index.ts @@ -1,5 +1,7 @@ import type { Readable, Writable } from "node:stream"; +import type { WorkspaceFiles } from "../workspace-helper/index.js"; + import { createCommandRuntimeAdapter } from "./command/index.js"; import { createLocalRuntime } from "./internal/local-runtime.js"; import { createService } from "./internal/service.js"; @@ -80,6 +82,7 @@ export interface WorkspaceRuntimeService { create(input: CreateWorkspaceInput): Promise<{ workspaceId: string; runtimeId: string }>; run(input: WorkspaceProcessInput): Promise; openTerminal(input: WorkspaceTerminalInput): Promise; + files(workspaceId: string): WorkspaceFiles; pause(workspaceId: string): Promise; resume(workspaceId: string): Promise; destroy(workspaceId: string): Promise; @@ -94,6 +97,7 @@ export interface ExternalWorkspaceRuntime { type: "command"; command: readonly [string, ...string[]]; options?: Readonly>; + helperCommand?: readonly [string, ...string[]]; } export interface WorkspaceRuntimeOptions extends WorkspaceRuntimeRecordStore { diff --git a/packages/server/src/server/workspace-runtime/internal/host-helper.ts b/packages/server/src/server/workspace-runtime/internal/host-helper.ts new file mode 100644 index 0000000000..e8d8bb9568 --- /dev/null +++ b/packages/server/src/server/workspace-runtime/internal/host-helper.ts @@ -0,0 +1,6 @@ +import { fileURLToPath } from "node:url"; + +export const hostWorkspaceHelperCommand = [ + process.execPath, + fileURLToPath(new URL("../../workspace-helper/executable.mjs", import.meta.url)), +] as const; diff --git a/packages/server/src/server/workspace-runtime/internal/local-runtime.ts b/packages/server/src/server/workspace-runtime/internal/local-runtime.ts index 5d7b8806d3..d0cf918648 100644 --- a/packages/server/src/server/workspace-runtime/internal/local-runtime.ts +++ b/packages/server/src/server/workspace-runtime/internal/local-runtime.ts @@ -9,6 +9,7 @@ import type { WorkspaceRuntimeDriver, } from "../drivers/index.js"; import { resolveRuntimeCwd, spawnHostProcess, spawnHostPty } from "./host-process.js"; +import { hostWorkspaceHelperCommand } from "./host-helper.js"; import { createRuntimeStateStore } from "./runtime-state.js"; export function createLocalRuntime(paseoHome: string): WorkspaceRuntimeDriver { @@ -36,6 +37,7 @@ export function createLocalRuntime(paseoHome: string): WorkspaceRuntimeDriver { return { id: "local", + workspaceHelperCommand: hostWorkspaceHelperCommand, async create(input: WorkspaceDriverCreateInput) { const existing = await inspect(input.workspaceId); if (existing.status === "ready" || existing.status === "paused") return existing.state; diff --git a/packages/server/src/server/workspace-runtime/internal/service.ts b/packages/server/src/server/workspace-runtime/internal/service.ts index a1163713bd..71bdc37491 100644 --- a/packages/server/src/server/workspace-runtime/internal/service.ts +++ b/packages/server/src/server/workspace-runtime/internal/service.ts @@ -12,6 +12,11 @@ import type { WorkspaceDriverCreateInput, WorkspaceRuntimeDriver, } from "../drivers/index.js"; +import type { WorkspaceFiles } from "../../workspace-helper/index.js"; +import { + bindWorkspaceHelper, + type WorkspaceFilesOwner, +} from "../../workspace-helper/integration/index.js"; const gracefulStopMilliseconds = 1_000; const forcedStopMilliseconds = 1_000; @@ -23,6 +28,12 @@ export function createService( const driversById = new Map(drivers.map((driver) => [driver.id, driver])); const processesByWorkspaceId = new Map>(); const workspaceTails = new Map>(); + const fileClients = new Map< + string, + { runtimeId: string; revision: string; client: WorkspaceFilesOwner } + >(); + const boundFiles = new Map(); + const unavailableFiles = new Map(); function requireRegistered(runtimeId: string): WorkspaceRuntimeDriver { const driver = driversById.get(runtimeId); @@ -86,8 +97,20 @@ export function createService( const before = await driver.inspect(input.workspaceId); let ownsNewResource = false; try { - await driver.create(toDriverCreateInput(input)); + const state = await driver.create(toDriverCreateInput(input)); ownsNewResource = before.status === "missing"; + if (state.lifecycle === "ready") { + const helper = bindWorkspaceHelper({ + root: state.root, + command: driver.workspaceHelperCommand, + launch: (argv) => launchHelper(driver, input.workspaceId, argv), + }); + try { + await helper.verify(); + } finally { + await helper.close(); + } + } if (ownsNewResource) { for (const command of input.setup ?? []) { const process = await runWithDriver(driver, { @@ -107,6 +130,7 @@ export function createService( } } await records.persistRuntimeId(input.workspaceId, input.runtimeId); + unavailableFiles.delete(input.workspaceId); return { workspaceId: input.workspaceId, runtimeId: input.runtimeId }; } catch (error) { if (ownsNewResource) { @@ -133,9 +157,19 @@ export function createService( openTerminalWithDriver(await resolve(input.workspaceId), input), ); }, + files(workspaceId) { + let files = boundFiles.get(workspaceId); + if (!files) { + files = bindFiles(workspaceId); + boundFiles.set(workspaceId, files); + } + return files; + }, async pause(workspaceId) { await sequence(workspaceId, async () => { const driver = await resolve(workspaceId); + unavailableFiles.set(workspaceId, "paused"); + await closeFiles(workspaceId); await stopProcesses(workspaceId); await driver.pause(workspaceId); }); @@ -143,17 +177,93 @@ export function createService( async resume(workspaceId) { await sequence(workspaceId, async () => { await (await resolve(workspaceId)).resume(workspaceId); + unavailableFiles.delete(workspaceId); }); }, async destroy(workspaceId) { await sequence(workspaceId, async () => { const driver = await resolve(workspaceId); + unavailableFiles.set(workspaceId, "destroyed"); + await closeFiles(workspaceId); await stopProcesses(workspaceId); await driver.destroy(workspaceId); }); }, }; + function bindFiles(workspaceId: string): WorkspaceFiles { + return { + async stat(path) { + return (await requireFiles(workspaceId)).stat(path); + }, + async list(path) { + return (await requireFiles(workspaceId)).list(path); + }, + async read(path) { + return (await requireFiles(workspaceId)).read(path); + }, + async write(input) { + return (await requireFiles(workspaceId)).write(input); + }, + async subscribe(input, listener) { + return (await requireFiles(workspaceId)).subscribe(input, listener); + }, + }; + } + + async function requireFiles(workspaceId: string): Promise { + const unavailable = unavailableFiles.get(workspaceId); + if (unavailable) throw new Error(`Workspace runtime is ${unavailable}: ${workspaceId}`); + const driver = await resolve(workspaceId); + const inspection = await driver.inspect(workspaceId); + if (inspection.status !== "ready") { + throw new Error(`Workspace runtime is ${inspection.status}: ${workspaceId}`); + } + const cached = fileClients.get(workspaceId); + if (cached && cached.runtimeId === driver.id && cached.revision === inspection.state.revision) { + return cached.client.files; + } + if (cached) await cached.client.close(); + const client = bindWorkspaceHelper({ + root: inspection.state.root, + command: driver.workspaceHelperCommand, + launch: async (argv) => { + return launchHelper(driver, workspaceId, argv); + }, + }); + fileClients.set(workspaceId, { + runtimeId: driver.id, + revision: inspection.state.revision, + client, + }); + return client.files; + } + + async function closeFiles(workspaceId: string): Promise { + const cached = fileClients.get(workspaceId); + fileClients.delete(workspaceId); + if (cached) await cached.client.close(); + } + + async function launchHelper( + driver: WorkspaceRuntimeDriver, + workspaceId: string, + argv: readonly [string, ...string[]], + ) { + const runtimeProcess = await driver.spawn({ + workspaceId, + argv, + env: {}, + purpose: { kind: "workspace-helper" }, + stdio: { kind: "pipes" }, + }); + if (runtimeProcess.kind !== "pipes") { + throw new Error(`Workspace runtime returned PTY mode for its helper: ${driver.id}`); + } + trackProcess(workspaceId, runtimeProcess); + return runtimeProcess; + } + async function sequence(workspaceId: string, operation: () => Promise): Promise { const previous = workspaceTails.get(workspaceId) ?? Promise.resolve(); let release!: () => void; diff --git a/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts b/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts index 240ef1cda1..52775f89f9 100644 --- a/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts +++ b/packages/server/src/server/workspace-runtime/internal/worktree-runtime.ts @@ -15,6 +15,7 @@ import type { WorkspaceRuntimeDriver, } from "../drivers/index.js"; import { resolveRuntimeCwd, spawnHostProcess, spawnHostPty } from "./host-process.js"; +import { hostWorkspaceHelperCommand } from "./host-helper.js"; import { createRuntimeStateStore } from "./runtime-state.js"; interface WorktreeRuntimeState extends WorkspaceDriverState { @@ -50,6 +51,7 @@ export function createWorktreeRuntime(options: { return { id: "worktree", + workspaceHelperCommand: hostWorkspaceHelperCommand, async create(input: WorkspaceDriverCreateInput) { const existing = await inspect(input.workspaceId); if (existing.status === "ready" || existing.status === "paused") return existing.state; diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts index 0f55a75d67..361a079f82 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.command.test.ts @@ -1,4 +1,4 @@ -import { existsSync } from "node:fs"; +import { existsSync, watch } from "node:fs"; import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -11,6 +11,9 @@ import { createWorkspaceRuntimeService } from "./index.js"; const fixtureExecutable = fileURLToPath( new URL("../test-utils/fixtures/workspace-runtime-command-fixture.mjs", import.meta.url), ); +const helperExecutable = fileURLToPath( + new URL("../workspace-helper/executable.mjs", import.meta.url), +); const cleanupRoots: string[] = []; afterEach(async () => { @@ -62,6 +65,56 @@ test("a trusted registered command is selected and receives secret launch data o await fixture.service.destroy(fixture.workspaceId); }); +test("a command runtime tears down and reconstructs its bound files capability", async () => { + const fixture = await createFixture("files-reconstruction"); + await writeFile(path.join(fixture.source, "watched.txt"), "before\n"); + await fixture.service.create(fixture.createInput); + const files = fixture.service.files(fixture.workspaceId); + const firstEvents: Array<{ type: string; error?: string }> = []; + const first = await files.subscribe({ paths: ["watched.txt"] }, (event) => { + firstEvents.push(event); + }); + + await fixture.service.pause(fixture.workspaceId); + expect(firstEvents).toContainEqual({ + type: "error", + error: "Workspace files client is closed", + }); + await first.unsubscribe(); + await fixture.service.resume(fixture.workspaceId); + await expect(files.list(".")).resolves.toMatchObject({ path: "." }); + + const reconstructedEvents: Array<{ type: string; error?: string }> = []; + await files.subscribe({ paths: ["watched.txt"] }, (event) => { + reconstructedEvents.push(event); + }); + await fixture.service.destroy(fixture.workspaceId); + expect(reconstructedEvents).toContainEqual({ + type: "error", + error: "Workspace files client is closed", + }); +}); + +test("a file operation racing pause cannot reconstruct the closing helper client", async () => { + const fixture = await createFixture("files-pause-race", true); + await writeFile(path.join(fixture.source, "watched.txt"), "before\n"); + await fixture.service.create(fixture.createInput); + const files = fixture.service.files(fixture.workspaceId); + await files.list("."); + await writeFile(path.join(fixture.barrierDirectory, "block-next-inspect"), "block"); + const inspectEntered = nextFile(fixture.barrierDirectory, "inspect-entered"); + + const racingList = files.list("."); + await inspectEntered; + await fixture.service.pause(fixture.workspaceId); + await writeFile(path.join(fixture.barrierDirectory, "release-inspect"), "release"); + + await expect(racingList).rejects.toThrow(`Workspace runtime is paused: ${fixture.workspaceId}`); + await fixture.service.resume(fixture.workspaceId); + await expect(files.list(".")).resolves.toMatchObject({ path: "." }); + await fixture.service.destroy(fixture.workspaceId); +}); + test("the command runtime transports PTY input, Unicode output, resize, and signals", async () => { const fixture = await createFixture("pty"); await fixture.service.create(fixture.createInput); @@ -329,6 +382,7 @@ async function createFixture( fixtureExecutable, ...(modes === "pipes" ? ["--modes", "pipes"] : []), ], + helperCommand: [processExecPath(), helperExecutable], options: { stateDirectory, ...runtimeOptions, @@ -364,3 +418,14 @@ function processExists(pid: number): boolean { return (error as NodeJS.ErrnoException).code !== "ESRCH"; } } + +function nextFile(directory: string, expectedName: string): Promise { + return new Promise((resolve, reject) => { + const watcher = watch(directory, (_event, filename) => { + if (filename?.toString() !== expectedName) return; + watcher.close(); + resolve(); + }); + watcher.once("error", reject); + }); +} diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts index 108a0862cb..a1f580ff8a 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts @@ -16,7 +16,11 @@ const image = "paseo-workspace-runtime-slice1:test"; const cleanupRoots: string[] = []; beforeAll(() => { - execFileSync("docker", ["build", "-t", image, dockerRuntimeRoot], { stdio: "pipe" }); + execFileSync( + "docker", + ["build", "-t", image, "-f", path.join(dockerRuntimeRoot, "Dockerfile"), repositoryRoot], + { stdio: "pipe" }, + ); }, 120_000); afterAll(async () => { @@ -67,6 +71,42 @@ test("the command runtime owns Docker workspace materialization, pipes, and life expect(workspace).toEqual({ workspaceId, runtimeId: "docker" }); expect("cwd" in workspace).toBe(false); + const files = service.files(workspaceId); + const listing = await files.list("."); + expect(listing.entries.map((entry) => entry.name)).toContain("committed.txt"); + const initialFile = await files.stat("committed.txt"); + expect(initialFile).toMatchObject({ status: "ready", size: 10 }); + let resolveFileChange!: () => void; + const fileChanged = new Promise((resolve) => { + resolveFileChange = resolve; + }); + const fileSubscription = await files.subscribe({ paths: ["committed.txt"] }, (event) => { + if (event.type === "changed" && event.paths.includes("committed.txt")) resolveFileChange(); + }); + const terminalFileEdit = await service.run({ + workspaceId, + argv: ["/bin/sh", "-c", "printf container-change > committed.txt"], + env: {}, + purpose: { kind: "terminal", terminalId: "docker-file-edit" }, + }); + terminalFileEdit.stdin.end(); + await expect(terminalFileEdit.exited).resolves.toEqual({ code: 0, signal: null }); + await expect(fileChanged).resolves.toBeUndefined(); + const containerFile = await files.read("committed.txt"); + await expect(collect(containerFile.chunks)).resolves.toBe("container-change"); + const conflict = await files.write({ + path: "committed.txt", + contents: Buffer.from("stale"), + ...(initialFile.status === "ready" ? { expectedRevision: initialFile.revision } : {}), + }); + expect(conflict).toMatchObject({ status: "conflict" }); + const large = await files.read("large.bin"); + expect(await collectBytes(large.chunks)).toEqual(Buffer.alloc(700_000, 0xa5)); + await expect(files.read("outside-link")).rejects.toThrow( + "Access outside of workspace is not allowed", + ); + await fileSubscription.unsubscribe(); + const write = await service.run({ workspaceId, argv: ["/bin/sh", "-c", "cat > runtime-owned.txt; printf runtime-stderr >&2; exit 19"], @@ -410,6 +450,8 @@ async function createRepository(root: string): Promise { execFileSync("git", ["config", "user.email", "test@getpaseo.local"], { cwd: repo }); execFileSync("git", ["config", "user.name", "Paseo Test"], { cwd: repo }); await writeFile(path.join(repo, "committed.txt"), "committed\n"); + await writeFile(path.join(repo, "large.bin"), Buffer.alloc(700_000, 0xa5)); + await symlink("/etc/passwd", path.join(repo, "outside-link")); execFileSync("git", ["add", "."], { cwd: repo }); execFileSync("git", ["-c", "commit.gpgsign=false", "commit", "-m", "fixture"], { cwd: repo, @@ -417,6 +459,12 @@ async function createRepository(root: string): Promise { return repo; } +async function collectBytes(chunks: AsyncIterable): Promise { + const buffers: Buffer[] = []; + for await (const chunk of chunks) buffers.push(Buffer.from(chunk)); + return Buffer.concat(buffers); +} + async function waitForOutput(stream: NodeJS.ReadableStream, marker: string): Promise { return new Promise((resolve, reject) => { let output = ""; diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts index 651cf62318..03e43ddd34 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts @@ -23,6 +23,58 @@ afterEach(async () => { }); posixDescribe.each(["local", "worktree"] as const)("%s runtime public contract", (runtimeId) => { + test("binds streaming files and live observation to the selected runtime", async () => { + const fixture = await createFixture(runtimeId); + await fixture.service.create(fixture.createInput); + const files = fixture.service.files(fixture.workspaceId); + + const listing = await files.list("."); + expect(listing.entries.map((entry) => entry.name)).toContain("committed.txt"); + const initial = await files.stat("committed.txt"); + expect(initial).toMatchObject({ status: "ready", size: 10 }); + if (initial.status !== "ready") throw new Error("Expected committed.txt to exist"); + + let resolveChanged!: () => void; + const watcherEvents: Array<{ type: string; error?: string }> = []; + const changed = new Promise((resolve) => { + resolveChanged = resolve; + }); + const subscription = await files.subscribe({ paths: ["committed.txt"] }, (event) => { + watcherEvents.push(event); + if (event.type === "changed" && event.paths.includes("committed.txt")) resolveChanged(); + }); + const terminalEdit = await fixture.service.run({ + workspaceId: fixture.workspaceId, + argv: ["/bin/sh", "-c", "printf changed > committed.txt"], + env: {}, + purpose: { kind: "terminal", terminalId: `${runtimeId}-file-edit` }, + }); + terminalEdit.stdin.end(); + await expect(terminalEdit.exited).resolves.toEqual({ code: 0, signal: null }); + await expect(changed).resolves.toBeUndefined(); + + const streamed = await files.read("committed.txt"); + await expect(collect(streamed.chunks)).resolves.toBe("changed"); + await fixture.service.pause(fixture.workspaceId); + expect(watcherEvents).toContainEqual({ + type: "error", + error: "Workspace files client is closed", + }); + await expect(files.list(".")).rejects.toThrow(`Workspace runtime is paused`); + await subscription.unsubscribe(); + await fixture.service.resume(fixture.workspaceId); + await expect(files.list(".")).resolves.toMatchObject({ path: "." }); + const reconstructedEvents: Array<{ type: string; error?: string }> = []; + await files.subscribe({ paths: ["committed.txt"] }, (event) => { + reconstructedEvents.push(event); + }); + await fixture.service.destroy(fixture.workspaceId); + expect(reconstructedEvents).toContainEqual({ + type: "error", + error: "Workspace files client is closed", + }); + }); + test("opens an interactive PTY with input, Unicode output, resize, EOF, and signals", async () => { const fixture = await createFixture(runtimeId); await fixture.service.create(fixture.createInput); diff --git a/packages/server/src/utils/project-icon.ts b/packages/server/src/utils/project-icon.ts index 259e293773..db7091a875 100644 --- a/packages/server/src/utils/project-icon.ts +++ b/packages/server/src/utils/project-icon.ts @@ -496,13 +496,18 @@ export async function getProjectIcon(projectDir: string): Promise MAX_ICON_SIZE) { - return null; - } - const fileBuffer = await readFile(iconPath); - let mimeType = sniffMimeType(fileBuffer) ?? getMimeType(iconPath); + return projectIconFromBytes(iconPath, fileBuffer); + } catch { + return null; + } +} + +export function projectIconFromBytes(fileName: string, bytes: Uint8Array): ProjectIcon | null { + try { + if (bytes.byteLength > MAX_ICON_SIZE) return null; + const fileBuffer = Buffer.from(bytes); + let mimeType = sniffMimeType(fileBuffer) ?? getMimeType(fileName); let buffer: Buffer = fileBuffer; if (mimeType === "image/x-icon") { const pngFrame = extractIcoPngFrame(fileBuffer); @@ -517,8 +522,7 @@ export async function getProjectIcon(projectDir: string): Promise