diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 1e276816779..6c5feb138d3 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3038,6 +3038,18 @@ Older packs without the declaration retain content verification until the image is updated. Selected ACPX agent profile checks remain in force. Daytona detects gzip/zstd headers and skips redundant transport compression. +### Daytona duplex close receipts + +Lease teardown removes duplex routes from command admission before it closes +their transports. The provider retains up to 4096 route/session bindings and +their close promises, so a later host close joins the same cleanup and returns +the original binding. These records grant no write, stop, or open authority. +Unknown or evicted routes cannot supply a session acknowledgement. A pending +close still waits; an unconfirmed close still retires the worker through the +existing host guard. The acknowledgement identifies the channel. Provider stop +receipts and unresolved external-work checks still determine process termination. +No command is replayed or worker restarted by this mechanism. + ### Remote GitHub launcher staging Managed GitHub launchers stage their nine text files in one bounded control diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 7a3180bb1bb..2eb882a76c3 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -40,6 +40,7 @@ import type { PluginContext } from "@paperclipai/plugin-sdk"; import { environmentSyncErrorData, readEnvironmentSyncErrorDiagnostic, withEnvironmentSyncErrorCapture, environmentCreationCleanupErrorData, readEnvironmentCreationCleanupError, readEnvironmentAcquisitionDiagnostic } from "@paperclipai/plugin-sdk"; import manifest from "./manifest.js"; import { parseTarVerboseListingLine, splitLinkEntryOnce } from "./file-sync.js"; +import * as duplexCommands from "./duplex-command-stream.js"; describe("Daytona read-only resource inspection", () => { const params = { driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment", @@ -729,6 +730,8 @@ describe("Daytona sandbox provider plugin", () => { }); expect(killed).toBeGreaterThanOrEqual(1); expect(disconnected).toBe(1); + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: "route-1" })) + .resolves.toEqual({ hostRouteId: "route-1", workerSessionId }); // The channel entry is gone, so a later write is a no-op and reaches no // process. @@ -746,6 +749,94 @@ describe("Daytona sandbox provider plugin", () => { } }); + describe("retired duplex ownership", () => { + async function fixture() { + const params = { driverKey: "daytona", companyId: "company-1", environmentId: "env-1", + config: { apiKey: "fixture-key", image: "fixture-image", timeoutMs: 3_000, livenessTimeoutMs: 1_000, reuseLease: true } }; + const pty = { waitForConnection: vi.fn(async () => {}), sendInput: vi.fn(async () => {}), + wait: vi.fn(() => new Promise<{ exitCode: number }>(() => {})), + kill: vi.fn(async () => {}), disconnect: vi.fn(async () => {}) }; + const sandbox = createMockSandbox(); + const createPty = vi.fn(async () => pty); + Object.assign(sandbox.process, { createPty }); + mockCreate.mockResolvedValue(sandbox); + await plugin.definition.onEnvironmentAcquireLease!({ ...params, runId: "run-1" }); + const opening = { driverKey: params.driverKey, companyId: params.companyId, environmentId: params.environmentId, + providerLeaseId: sandbox.id, command: ["fixture-bridge"] }; + return { params: { ...params, providerLeaseId: sandbox.id }, opening, pty, createPty }; + } + + it("joins pending teardown for concurrent closes while all channel authority stays revoked", async () => { + const f = await fixture(); + const opened = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "pending-route" }); + let finishKill!: () => void, enteredKill!: () => void; + const entered = new Promise(resolve => { enteredKill = resolve; }); + const killing = new Promise(resolve => { finishKill = resolve; }); + f.pty.kill.mockImplementation(async () => { enteredKill(); await killing; }); + const requests: Promise[] = []; + try { + requests.push(plugin.definition.onEnvironmentReleaseLease!(f.params)); + await entered; + let replies = 0; + for (let index = 0; index < 2; index++) requests.push(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId }) + .then(value => { replies++; return value; })); + await new Promise(resolve => setImmediate(resolve)); + expect(replies).toBe(0); + expect(f.pty.kill).toHaveBeenCalledTimes(1); + expect(f.pty.disconnect).not.toHaveBeenCalled(); + await plugin.definition.onDuplexChannelWrite!({ ...opened, data: Buffer.from("closed").toString("base64") }); + await plugin.definition.onDuplexChannelStop!(opened); + expect(f.pty.sendInput).toHaveBeenCalledTimes(1); + expect(f.pty.kill).toHaveBeenCalledTimes(1); + await expect(plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: opened.hostRouteId })) + .rejects.toThrow("already been used"); + expect(f.createPty).toHaveBeenCalledTimes(1); + finishKill(); + const result = await Promise.all(requests); + expect(result.slice(1)).toEqual([opened, opened]); + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId })).resolves.toEqual(opened); + expect(f.pty.kill).toHaveBeenCalledTimes(1); + expect(f.pty.disconnect).toHaveBeenCalledTimes(1); + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: "unknown-route", workerSessionId: opened.workerSessionId } as never)) + .resolves.toEqual({ hostRouteId: "unknown-route" }); + } finally { finishKill(); await Promise.allSettled(requests); } + }); + + it("does not turn a rejected retired close into an acknowledgement", async () => { + const f = await fixture(); + const close = vi.fn().mockRejectedValue(new Error("fixture closure remains unknown")); + const restore = vi.spyOn(duplexCommands, "openDaytonaDuplexChannelSession").mockResolvedValueOnce({ + onData() {}, write: vi.fn(), kill: vi.fn(), wait: () => new Promise(() => {}), close, + }); + try { + const opened = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "failed-route" }); + await plugin.definition.onEnvironmentReleaseLease!(f.params); + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: opened.hostRouteId })) + .rejects.toThrow("fixture closure remains unknown"); + expect(close).toHaveBeenCalledTimes(1); + await expect(plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: opened.hostRouteId })) + .rejects.toThrow("already been used"); + } finally { restore.mockRestore(); } + }); + + it("bounds retained acknowledgements and never echoes a supplied session id for evicted routes", async () => { + const f = await fixture(); + const first = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: "oldest-route" }); + await plugin.definition.onDuplexChannelClose!({ hostRouteId: first.hostRouteId }); + let latest = first; + for (let index = 0; index < 4_096; index++) { + latest = await plugin.definition.onDuplexChannelOpen!({ ...f.opening, hostRouteId: `bounded-route-${index}` }); + await plugin.definition.onDuplexChannelClose!({ hostRouteId: latest.hostRouteId }); + } + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: first.hostRouteId, workerSessionId: first.workerSessionId } as never)) + .resolves.toEqual({ hostRouteId: first.hostRouteId }); + await expect(plugin.definition.onDuplexChannelClose!({ hostRouteId: latest.hostRouteId, workerSessionId: "forged-session" } as never)) + .resolves.toEqual(latest); + expect(f.pty.kill).toHaveBeenCalledTimes(4_097); + expect(f.pty.disconnect).toHaveBeenCalledTimes(4_097); + }); + }); + it("normalizes config and validates the API key fallback", async () => { process.env.DAYTONA_API_KEY = "host-key"; diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 4300b21f938..85674cf0bcb 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -1728,6 +1728,7 @@ export function __resetDaytonaSandboxHandleCacheForTest(): void { sandboxHandleWritableDirs.reset(); sandboxHandleSessionStore.reset(); runnerIngressGenerationStore.reset(); + retiredDaytonaDuplexChannelByRoute.clear(); } /** @@ -2489,12 +2490,24 @@ interface DaytonaDuplexChannelEntry { } const daytonaDuplexChannelByRoute = new Map(); const daytonaDuplexChannelBySession = new Map(); +const retiredDaytonaDuplexChannelByRoute = new Map }>(); +const MAX_RETIRED_DAYTONA_DUPLEX_CHANNELS = 4096; function forgetDaytonaDuplexChannel(entry: DaytonaDuplexChannelEntry): void { daytonaDuplexChannelByRoute.delete(entry.hostRouteId); daytonaDuplexChannelBySession.delete(entry.workerSessionId); } +function closeDaytonaDuplexChannel(entry: DaytonaDuplexChannelEntry): Promise { + forgetDaytonaDuplexChannel(entry); + const closed = Promise.resolve().then(() => entry.session.close()); + if (retiredDaytonaDuplexChannelByRoute.size >= MAX_RETIRED_DAYTONA_DUPLEX_CHANNELS) { + retiredDaytonaDuplexChannelByRoute.delete(retiredDaytonaDuplexChannelByRoute.keys().next().value!); + } + retiredDaytonaDuplexChannelByRoute.set(entry.hostRouteId, { workerSessionId: entry.workerSessionId, closed }); + return closed; +} + // Close every open duplex channel that belongs to one provider lease and drop its // entry. The lease teardown hooks (release, destroy, resume) call this, so a // channel never outlives the sandbox that carries it. The close kills the child @@ -2504,8 +2517,7 @@ async function closeDaytonaDuplexChannelsForLease(providerLeaseId: string): Prom const matches = [...daytonaDuplexChannelByRoute.values()].filter( (entry) => entry.providerLeaseId === providerLeaseId, ); - for (const entry of matches) forgetDaytonaDuplexChannel(entry); - await Promise.allSettled(matches.map(entry => entry.session.close())); + await Promise.allSettled(matches.map(closeDaytonaDuplexChannel)); } async function stopLease(params: PluginEnvironmentReleaseLeaseParams): Promise { @@ -3716,6 +3728,9 @@ const plugin = definePlugin(trackDaytonaHooks({ // `ctx.duplexChannel`, bound to the returned worker session id. Fail closed when // no cached sandbox matches the lease. async onDuplexChannelOpen(params) { + if (daytonaDuplexChannelByRoute.has(params.hostRouteId) || retiredDaytonaDuplexChannelByRoute.has(params.hostRouteId)) { + throw new Error("Daytona duplex channel: host route has already been used."); + } const sandbox = await sandboxHandleCache.findByProviderLeaseId(params.providerLeaseId); if (!sandbox) { throw new Error( @@ -3794,10 +3809,14 @@ const plugin = definePlugin(trackDaytonaHooks({ const entry = daytonaDuplexChannelByRoute.get(params.hostRouteId); if (entry) { const boundWorkerSessionId = entry.workerSessionId; - forgetDaytonaDuplexChannel(entry); - await entry.session.close().catch(() => undefined); + await closeDaytonaDuplexChannel(entry).catch(() => undefined); return { hostRouteId: params.hostRouteId, workerSessionId: boundWorkerSessionId }; } + const retired = retiredDaytonaDuplexChannelByRoute.get(params.hostRouteId); + if (retired) { + await retired.closed; + return { hostRouteId: params.hostRouteId, workerSessionId: retired.workerSessionId }; + } return { hostRouteId: params.hostRouteId }; }, @@ -3814,6 +3833,7 @@ const plugin = definePlugin(trackDaytonaHooks({ const openChannels = [...daytonaDuplexChannelByRoute.values()]; daytonaDuplexChannelByRoute.clear(); daytonaDuplexChannelBySession.clear(); + retiredDaytonaDuplexChannelByRoute.clear(); for (const entry of openChannels) { await entry.session.close().catch(() => undefined); } diff --git a/server/src/__tests__/fixtures/daytona-worker-lifecycle-sdk.mjs b/server/src/__tests__/fixtures/daytona-worker-lifecycle-sdk.mjs new file mode 100644 index 00000000000..83e4ae3f0a7 --- /dev/null +++ b/server/src/__tests__/fixtures/daytona-worker-lifecycle-sdk.mjs @@ -0,0 +1,77 @@ +import { appendFileSync, readFileSync } from "node:fs"; + +const sandboxes = new Map(); +let nextId = 0; + +function record(event, fields = {}) { + appendFileSync(process.env.PAPERCLIP_DAYTONA_FIXTURE_JOURNAL, `${JSON.stringify({ event, ...fields })}\n`); +} + +function sandbox(params) { + const id = `fixture-sandbox-${++nextId}`; + const files = new Map(); + const resource = { + id, + name: params.name, + labels: params.labels, + state: "started", + target: "us", + autoDeleteInterval: -1, + async getWorkDir() { return "/home/daytona"; }, + async getUserHomeDir() { return "/home/daytona"; }, + async refreshData() {}, + async setTtl() {}, + async setAutoDeleteInterval(value) { resource.autoDeleteInterval = value; }, + async start() { resource.state = "started"; record("start", { id }); }, + async stop() { resource.state = "stopped"; record("stop", { id }); }, + async delete() { sandboxes.delete(id); record("delete", { id }); }, + fs: { + async createFolder() {}, + async uploadFile(content, destination) { files.set(destination, content.toString("utf8")); }, + async deleteFile() {}, + }, + process: { + async executeCommand(command) { + const afterRelease = command.includes("monitor-after-release"); + record("execute", { id, afterRelease }); + if (command.includes("fixture-provider-stop")) resource.state = "stopped"; + const sentinel = command.match(/^cat '([^']+)'/); + const result = afterRelease ? readFileSync(process.env.PAPERCLIP_DAYTONA_FIXTURE_SAVED_FILE, "utf8") + : sentinel ? files.get(sentinel[1]) ?? "" : "bash"; + return { exitCode: 0, result, artifacts: { stdout: result } }; + }, + async createPty() { + let finish; + const exited = new Promise(resolve => { finish = resolve; }); + record("pty.open", { id }); + return { + async waitForConnection() {}, + async sendInput() { record("pty.write", { id }); }, + wait() { return exited; }, + async kill() { record("pty.kill", { id }); finish({ exitCode: 0 }); }, + async disconnect() { + record("pty.disconnect", { id }); + if (process.env.PAPERCLIP_DAYTONA_FIXTURE_HANG_CLOSE === "1" && id === "fixture-sandbox-1") { + await new Promise(() => {}); + } + }, + }; + }, + }, + }; + sandboxes.set(id, resource); + record("create", { id }); + return resource; +} + +export class Daytona { + async create(params) { return sandbox(params); } + async get(id) { + const result = sandboxes.get(id); + if (!result) throw new DaytonaNotFoundError("Fixture sandbox is absent"); + return result; + } +} + +export class DaytonaNotFoundError extends Error {} +export class DaytonaTimeoutError extends Error {} diff --git a/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts b/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts new file mode 100644 index 00000000000..c9e5c649bad --- /dev/null +++ b/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts @@ -0,0 +1,97 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { build } from "esbuild"; +import { expect, it } from "vitest"; +import { createPluginWorkerManager } from "../services/plugin-worker-manager.js"; + +const repoRoot = fileURLToPath(new URL("../../..", import.meta.url)); +const providerRoot = path.join(repoRoot, "packages/plugins/sandbox-providers/daytona"); +const pluginId = "fixture-daytona-provider"; + +it.each(["release", "cancel", "stop", "destroy", "resume", "stalled-close"] as const)("preserves worker ownership after %s of a bound duplex channel", async mode => { + const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-daytona-worker-close-")); + const journal = path.join(root, "provider.jsonl"), savedFile = path.join(root, "saved.txt"); + const workerFile = path.join(root, "worker.mjs"); + const manager = createPluginWorkerManager(); + try { + await writeFile(savedFile, "saved work remains intact", { mode: 0o600 }); + await build({ + entryPoints: { worker: path.join(providerRoot, "src/worker.ts"), manifest: path.join(providerRoot, "src/manifest.ts") }, + outdir: root, + outExtension: { ".js": ".mjs" }, + platform: "node", + format: "esm", + target: "node24", + bundle: true, + logLevel: "silent", + nodePaths: [path.join(repoRoot, "server/node_modules")], + alias: { + "@daytonaio/sdk": path.join(repoRoot, "server/src/__tests__/fixtures/daytona-worker-lifecycle-sdk.mjs"), + "@paperclipai/plugin-sdk": path.join(repoRoot, "packages/plugins/sdk/dist/index.js"), + }, + banner: { js: 'import { createRequire } from "node:module"; const require = createRequire(import.meta.url);' }, + }); + const { default: manifest } = await import(pathToFileURL(path.join(root, "manifest.mjs")).href); + const worker = await manager.startWorker(pluginId, { + entrypointPath: workerFile, + manifest, + config: {}, + instanceInfo: { instanceId: "fixture-instance", hostVersion: "0.0.0" }, + apiVersion: 1, + hostHandlers: {}, + env: { PAPERCLIP_DAYTONA_FIXTURE_JOURNAL: journal, PAPERCLIP_DAYTONA_FIXTURE_SAVED_FILE: savedFile, + PAPERCLIP_DAYTONA_FIXTURE_HANG_CLOSE: mode === "stalled-close" ? "1" : "0" }, + duplexChannelLimits: { closeTimeoutMs: 500 }, + }); + const pid = worker.diagnostics().pid; + const exited = new Promise(resolve => worker.on("exit", () => resolve())); + const base = { driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment", + config: { apiKey: "fixture-not-a-credential", image: "fixture-image", timeoutMs: 2_000, livenessTimeoutMs: 250, reuseLease: true } }; + const acquisition = { ...base, agentId: "fixture-agent", executionWorkspaceId: "fixture-workspace" }; + const first = await manager.call(pluginId, "environmentAcquireLease", { ...acquisition, runId: "fixture-first-run" }); + const next = await manager.call(pluginId, "environmentAcquireLease", { ...acquisition, runId: "fixture-next-run" }); + const channel = await worker.openDuplexChannel({ + driverKey: base.driverKey, companyId: base.companyId, environmentId: base.environmentId, + providerLeaseId: first.providerLeaseId!, command: ["fixture-bridge"], + }); + const teardown = { ...base, providerLeaseId: first.providerLeaseId! }; + if (mode === "resume") { + await manager.call(pluginId, "environmentExecute", { ...base, lease: first, + command: "fixture-provider-stop", bypassSession: true }); + await expect(manager.call(pluginId, "environmentResumeLease", { ...teardown, leaseMetadata: first.metadata })) + .resolves.toMatchObject({ providerLeaseId: first.providerLeaseId, metadata: { resumedFromState: "stopped" } }); + } else if (mode === "stop") await manager.call(pluginId, "environmentStopLease", teardown); + else if (mode === "destroy") await manager.call(pluginId, "environmentDestroyLease", teardown); + else await manager.call(pluginId, "environmentReleaseLease", { ...teardown, cancelActiveWork: mode === "cancel" }); + await channel.close(); + if (mode === "stalled-close") await exited; + else await Promise.race([exited, new Promise(resolve => setTimeout(resolve, 100))]); + const execution = manager.call(pluginId, "environmentExecute", { + ...base, lease: next, command: "printf", args: ["monitor-after-release"], bypassSession: true, + }); + if (mode === "stalled-close") await expect(execution).rejects.toThrow("is stopped"); + else await expect(execution).resolves.toMatchObject({ exitCode: 0, stdout: "saved work remains intact" }); + expect(worker.status).toBe(mode === "stalled-close" ? "stopped" : "running"); + expect(manager.getWorker(pluginId)).toBe(worker); + expect(worker.diagnostics().pid).toBe(mode === "stalled-close" ? null : pid); + channel.write(new TextEncoder().encode("closed channel must remain closed")); + channel.kill(); + if (mode !== "resume") await expect(manager.call(pluginId, "environmentExecute", { + ...base, lease: first, command: "printf", args: ["closed-lease-after-release"], bypassSession: true, + })).rejects.toThrow(); + const events = (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + expect(events.filter(event => event.event === "create")).toHaveLength(2); + expect(events.filter(event => event.event === "pty.open")).toHaveLength(1); + expect(events.filter(event => event.event === "pty.write")).toHaveLength(1); + expect(events.filter(event => event.event === "pty.kill")).toHaveLength(1); + expect(events.filter(event => event.event === "pty.disconnect")).toHaveLength(1); + expect(events.filter(event => event.event === "execute" && event.afterRelease)) + .toEqual(mode === "stalled-close" ? [] : [{ event: "execute", id: next.providerLeaseId, afterRelease: true }]); + expect(await readFile(savedFile, "utf8")).toBe("saved work remains intact"); + } finally { + await manager.stopAll(); + await rm(root, { recursive: true, force: true }); + } +});