diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 8ad2fee0f96..0f1ea78a082 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -44,6 +44,60 @@ const json = (value: unknown, status = 200, headers?: Record) => headers: { "content-type": "application/json", ...headers }, }); describe("Boat transport validation", () => { + it("uses only the file budget remaining after transport preparation", async () => { + let clock = 1_000_000; + const time = vi.spyOn(Date, "now").mockImplementation(() => clock); + const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` }; + const execute = vi.fn(async () => ({ exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" })); + sshFactory.mockReturnValue({ execute }); + const backend = boatBackend(async () => "fixture", vi.fn(async () => { + clock += 90_000; + return json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" }); + })); + try { + await backend.remote(scoped, { action: "read", path: "AGENTS.md" }, { deadlineMs: 1_120_000 }); + expect(execute).toHaveBeenCalledWith(expect.objectContaining({ timeoutMs: 30_000 })); + clock = 1_120_000; + await expect(backend.remote(scoped, { action: "write" }, { deadlineMs: clock })).rejects.toThrow("timed out"); + expect(execute).toHaveBeenCalledOnce(); + } finally { time.mockRestore(); sshFactory.mockReset(); } + }); + it("expires only the file waiter while shared SSH initialization remains usable", async () => { + vi.useFakeTimers(); + const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` }; + let initialized!: () => void; + const execute = vi.fn(async () => ({ exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" })); + sshFactory.mockReturnValue({ execute }); + const fetcher = vi.fn(() => new Promise(resolve => { + initialized = () => resolve(json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" })); + })); + const backend = boatBackend(async () => "fixture", fetcher); + try { + const short = backend.remote(scoped, { action: "read" }, { deadlineMs: Date.now() + 20 }); + const failure = expect(short).rejects.toThrow("timed out"); + const other = backend.remote(scoped, { action: "owned-port" }); + await vi.advanceTimersByTimeAsync(21); + await failure; + expect(execute).not.toHaveBeenCalled(); + initialized(); + await other; + await backend.remote(scoped, { action: "read" }, { deadlineMs: Date.now() + 1000 }); + expect(fetcher).toHaveBeenCalledOnce(); + expect(execute).toHaveBeenCalledTimes(2); + } finally { vi.useRealTimers(); sshFactory.mockReset(); } + }); + it("does not resume after the file readiness budget has expired", async () => { + let clock = 1_000_000; + const time = vi.spyOn(Date, "now").mockImplementation(() => clock); + const fetcher = vi.fn(async () => { + clock += 120_000; + return json({ sandbox: { id: "bx_test", state: "archived", snapshots: true, stop: null } }); + }); + try { + await expect(boatBackend(async () => "fixture", fetcher).ready(record, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out"); + expect(fetcher).toHaveBeenCalledOnce(); + } finally { time.mockRestore(); } + }); it("reads outgoing and listening TCP ports across IPv4 and IPv6", () => { const root = mkdtempSync(join(tmpdir(), "boat-port-inventory-")); try { diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index c578dc17b81..0cf1eeddb2c 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -4,7 +4,7 @@ import { shellQuote, } from "@paperclipai/adapter-utils/ssh"; import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; -import type { ComputerBackend } from "../application/ports.js"; +import type { ComputerBackend, ComputerOperationOptions } from "../application/ports.js"; import { ComputerError, segment, @@ -195,11 +195,17 @@ export function boatBackend( resolveKey: (record: ComputerRecord) => Promise, fetcher: typeof fetch = fetch, ): ComputerBackend { + function remaining(options: ComputerOperationOptions | undefined, limit: number) { + const ms = options?.deadlineMs === undefined ? limit : Math.min(limit, options.deadlineMs - Date.now()); + if (ms <= 0) throw new ComputerError("provider_error", "Computer file operation timed out"); + return ms; + } async function api( record: ComputerRecord, method: string, suffix = "", body?: unknown, + options?: ComputerOperationOptions, ): Promise { const response = await fetcher( `${apiOrigin}/sandboxes/${segment(record.providerId)}${suffix}`, @@ -210,7 +216,7 @@ export function boatBackend( "content-type": "application/json", }, ...(body === undefined ? {} : { body: JSON.stringify(body) }), - signal: AbortSignal.timeout(90_000), + signal: AbortSignal.timeout(remaining(options, 90_000)), }, ); if (!response.ok) @@ -223,8 +229,8 @@ export function boatBackend( throw new ComputerError("provider_error", "Invalid Boat response"); return data; } - async function inspect(record: ComputerRecord) { - const data = await api(record, "GET"); + async function inspect(record: ComputerRecord, options?: ComputerOperationOptions) { + const data = await api(record, "GET", "", undefined, options); const sandbox = data.sandbox ?? data; if (sandbox.id !== record.providerId || typeof sandbox.state !== "string") throw new ComputerError( @@ -344,21 +350,39 @@ export function boatBackend( const execute = admission; return { ...raw, execute: (input) => execute(raw, input, options?.control === true) }; } + async function fileTransport( + record: ComputerRecord, + control: boolean, + options?: ComputerOperationOptions, + ) { + remaining(options, 120_000); + const pending = runner(record, { control }); + if (options?.deadlineMs === undefined) return pending; + // Initialization is shared with unrelated owners. Expire this waiter only; + // the cached promise retains its normal success/eviction behavior. + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new ComputerError( + "provider_error", "Computer file operation timed out", + )), remaining(options, 120_000)); + pending.then(resolve, reject).finally(() => clearTimeout(timer)); + }); + } async function execute( record: ComputerRecord, code: string, payload: unknown, control = true, + options?: ComputerOperationOptions, ) { const result = await ( - await runner(record, { control }) + await fileTransport(record, control, options) ).execute({ command: "python3", args: ["-c", code], stdin: JSON.stringify(payload), - timeoutMs: 120_000, + timeoutMs: remaining(options, 120_000), }); - if (result.exitCode !== 0) + if (result.exitCode !== 0 || result.timedOut) throw new ComputerError("provider_error", "Computer operation failed"); let value: any; try { @@ -422,8 +446,8 @@ export function boatBackend( return { inspect, runner, - async ready(record) { - const state = await inspect(record); + async ready(record, options) { + const state = await inspect(record, options); if (!state.snapshots) throw new ComputerError("invalid", "Boat snapshots must be enabled"); if (state.stop && ["pending", "failing"].includes(state.stop.status)) @@ -432,19 +456,19 @@ export function boatBackend( "Boat is still saving a previous stop", ); if (!["ready", "idle", "running"].includes(state.state)) { - await api(record, "POST", "/resume", { ttlSeconds: 300 }); + await api(record, "POST", "/resume", { ttlSeconds: 300 }, options); transportCache.delete( `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`, ); for (let i = 0; i < 90; i++) { - const status = await inspect(record); + const status = await inspect(record, options); if (["ready", "idle", "running"].includes(status.state)) return; - await new Promise((r) => setTimeout(r, 1000)); + await new Promise((r) => setTimeout(r, remaining(options, 1000))); } throw new ComputerError("provider_error", "Boat did not become ready"); } }, - async claim(record) { + async claim(record, options) { await execute( record, String.raw` @@ -462,6 +486,8 @@ print('{}') companyId: record.companyId, computerId: record.id, }, + true, + options, ); const retiredOwners = record.ledger.owners .filter((owner) => owner.kind === "runner" && owner.phase === "retired") @@ -472,7 +498,7 @@ print('{}') await execute(record, processProgram, { action: "cleanup-retired", owners: retiredOwners, - }); + }, true, options); } }, async runnerPorts(record) { @@ -566,10 +592,10 @@ print('{}') url.pathname = path; return { url: url.toString(), secretHeaders: { Cookie: cookie } }; }, - async remote(record, input) { + async remote(record, input, options) { // Git setup and file I/O may wait on locks or transfer substantial data. // Keep lifecycle capacity reserved; only the short owned-port probe uses it. - return execute(record, remoteProgram, input, input.action === "owned-port"); + return execute(record, remoteProgram, input, input.action === "owned-port", options); }, }; } diff --git a/server/src/modules/computers/application/ports.ts b/server/src/modules/computers/application/ports.ts index b04c386eada..609fc2e73ca 100644 --- a/server/src/modules/computers/application/ports.ts +++ b/server/src/modules/computers/application/ports.ts @@ -23,14 +23,16 @@ export interface ComputerRepository { agentId: string, ): Promise<"active" | "terminal" | "missing">; } +export type ComputerOperationOptions = { deadlineMs?: number }; + export interface ComputerBackend { inspect(record: ComputerRecord): Promise<{ state: string; snapshots: boolean; stop: null | { id: string; status: string }; }>; - ready(record: ComputerRecord): Promise; - claim(record: ComputerRecord): Promise; + ready(record: ComputerRecord, options?: ComputerOperationOptions): Promise; + claim(record: ComputerRecord, options?: ComputerOperationOptions): Promise; advance(record: ComputerRecord, owner: Owner): Promise; runnerPorts(record: ComputerRecord): Promise; runner(record: ComputerRecord, options?: { control?: boolean }): Promise; @@ -66,6 +68,6 @@ export interface ComputerBackend { path: string, ): Promise<{ url: string; secretHeaders: Record }>; preview(record: ComputerRecord, port: number): Promise<{ url: string }>; - remote(record: ComputerRecord, input: Record): Promise; + remote(record: ComputerRecord, input: Record, options?: ComputerOperationOptions): Promise; } export type { Ledger }; diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index 69106e41edd..d7211d9a7ba 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -512,6 +512,66 @@ describe("computer ownership", () => { ).resolves.toHaveProperty("owner"); expect(f.backend.stop).toHaveBeenCalledOnce(); }); + it("shares one file budget across readiness, claim, and command without early retirement", async () => { + const f = fixture(); + await f.attach(); + const files = await f.service.files({ ...f.scope, agentId: "agent" }); + const expectedDeadline = Date.parse("2026-10-10T12:02:00Z"); + vi.mocked(f.backend.ready).mockImplementation(async (_record, options) => { + expect(options?.deadlineMs).toBe(expectedDeadline); + f.advance(90_000); + }); + vi.mocked(f.backend.claim).mockImplementation(async (_record, options) => { + expect(options?.deadlineMs).toBe(expectedDeadline); + f.advance(20_000); + }); + vi.mocked(f.backend.remote).mockImplementation(async (_record, _input, options) => { + expect(options?.deadlineMs).toBe(expectedDeadline); + f.advance(9999); + await f.service.reconcile(); + expect(f.backend.stop).not.toHaveBeenCalled(); + f.advance(1); + return {}; + }); + await files.stat("AGENTS.md"); + f.advance(5001); + await f.service.reconcile(); + expect(f.backend.stop).toHaveBeenCalledOnce(); + }); + it("does not launch file work after slow readiness consumes its budget", async () => { + const f = fixture(); + await f.attach(); + const files = await f.service.files({ ...f.scope, agentId: "agent" }); + vi.mocked(f.backend.claim).mockClear(); + vi.mocked(f.backend.ready).mockImplementation(async () => { f.advance(120_000); }); + await expect(files.stat("AGENTS.md")).rejects.toThrow("timed out"); + expect(f.backend.claim).not.toHaveBeenCalled(); + expect(f.backend.remote).not.toHaveBeenCalled(); + f.advance(5001); + await f.service.reconcile(); + expect(f.backend.stop).toHaveBeenCalledOnce(); + }); + it("bounds an abandoned file hold and fences a late readiness continuation", async () => { + const f = fixture(); + await f.attach(); + const files = await f.service.files({ ...f.scope, agentId: "agent" }); + let finish!: () => void; + let entered!: () => void; + const readyEntered = new Promise(resolve => { entered = resolve; }); + vi.mocked(f.backend.claim).mockClear(); + vi.mocked(f.backend.ready).mockImplementation(() => new Promise(resolve => { finish = resolve; entered(); })); + const operation = files.stat("AGENTS.md"); + const rejected = expect(operation).rejects.toThrow("timed out"); + await readyEntered; + f.advance(125_001); + await f.service.reconcile(); + expect(f.backend.stop).toHaveBeenCalledOnce(); + finish(); + await rejected; + expect(f.backend.claim).not.toHaveBeenCalled(); + expect(f.backend.remote).not.toHaveBeenCalled(); + expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true); + }); it("keeps a short bounded file batch hold between editor operations", async () => { const f = fixture(); await f.attach(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 54450301cc7..094abab7394 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -26,6 +26,8 @@ class ComputerSessionBusyError extends ComputerError { // This is shutdown time only; admission and ordinary work stop at the warm deadline. const RETIREMENT_GRACE_MS = 30_000; +const FILE_OPERATION_BUDGET_MS = 120_000; +const FILE_OPERATION_SETTLEMENT_MS = 5000; type Scope = { companyId: string; environmentId: string }; export function createComputerService( repository: ComputerRepository, @@ -697,26 +699,39 @@ finally: async function withFiles( input: Scope, root: string, - fn: (record: ComputerRecord) => Promise, + fn: (record: ComputerRecord, deadlineMs: number) => Promise, ): Promise { const data = await admitRecord(input, (record) => { + const deadlineMs = now().getTime() + FILE_OPERATION_BUDGET_MS; const owner: Owner = { id: randomUUID(), generation: 1, kind: "file-operation", phase: "active", port: 0, - deadline: new Date(now().getTime() + 120_000).toISOString(), + // Transport cancellation has a bounded settlement window; it is not + // additional time in which commands may start or continue. + deadline: new Date(deadlineMs + FILE_OPERATION_SETTLEMENT_MS).toISOString(), absoluteDeadline: null, process: null, }; record.ledger.owners.push(owner); - return { record: structuredClone(record), owner }; + return { record: structuredClone(record), owner, deadlineMs }; }); + const assertLive = async () => { + if (now().getTime() >= data.deadlineMs) + throw new ComputerError("provider_error", "Computer file operation timed out"); + const current = await repository.get(input); + if (exactOwner(current, ref(data.record, data.owner)).phase !== "active") + throw new ComputerError("conflict", "Computer file operation is no longer active"); + }; try { - await backend.ready(data.record); - await backend.claim(data.record); - return await fn(data.record); + await assertLive(); + await backend.ready(data.record, { deadlineMs: data.deadlineMs }); + await assertLive(); + await backend.claim(data.record, { deadlineMs: data.deadlineMs }); + await assertLive(); + return await fn(data.record, data.deadlineMs); } finally { await repository.update(input, (record) => { const owner = record.ledger.owners.find( @@ -724,15 +739,18 @@ finally: ); if (owner?.phase === "active") { owner.phase = "warm"; - owner.deadline = new Date(now().getTime() + 5000).toISOString(); + owner.deadline = new Date(Math.min( + now().getTime() + FILE_OPERATION_SETTLEMENT_MS, + data.deadlineMs + FILE_OPERATION_SETTLEMENT_MS, + )).toISOString(); } }); } } function fileAccess(input: Scope, root: string) { const remote = (payload: Record) => - withFiles(input, root, (record) => - backend.remote(record, { ...payload, root }), + withFiles(input, root, (record, deadlineMs) => + backend.remote(record, { ...payload, root }, { deadlineMs }), ); const readBytes = async ( path: string,