diff --git a/packages/plugins/sandbox-providers/daytona/README.md b/packages/plugins/sandbox-providers/daytona/README.md index b384c57549d..76dde8c5897 100644 --- a/packages/plugins/sandbox-providers/daytona/README.md +++ b/packages/plugins/sandbox-providers/daytona/README.md @@ -30,6 +30,7 @@ Notes: - Each cold create uses a unique provider name and ownership labels. If creation fails after Daytona has allocated a sandbox, the driver looks up that exact name, checks every ownership label, and waits for deletion. Failed, missing, or timed-out lookups and failed deletions report unconfirmed cleanup with the provider name; they never return a usable lease. A missing name lookup after an uncertain create is not proof that a delayed provider request cannot create a resource. - For plugin-backed sandbox lease acquisition, unresolved creation cleanup crosses the worker RPC as a validated ownership envelope. The host records `pending_cleanup` before retrying deletion. If only a creation name is known, the provider first returns an ownership-verified sandbox ID without deleting it. The host saves that observation before authorizing deletion. Its existing cleanup sweep and durable spool preserve retries across controller restarts and environment deletion. A missing observed ID confirms cleanup after a lost deletion reply or database update; a name that has never been observed remains unresolved. Provider exceptions and resolved credentials are excluded from that envelope. This requires the matching host and plugin SDK update; probe and custom-image interactive-setup calls still report unconfirmed immediate cleanup without this lease-recovery path. - Reusable leases map to Daytona stop/start semantics. Non-reusable leases are deleted on release. A provider-resolved `target` does not change the identity of an existing sandbox. Release closes the same scoped lease that a later sentinel-verified resume reopens. +- A command that has already exited returns its final separated log snapshot without opening a follow socket. The initial status check is bounded to 250ms and the caller's command deadline. An unfinished command, failed status read, or slow status read keeps the existing streaming and recovery path; the command is dispatched only once. - A session log socket can close while its command still runs. The driver requires a recorded command exit before returning a completed execution. It reconnects the log stream once, then polls status and log snapshots at most once per second. Replayed output is removed by byte offset, new output continues to reach the host, and the command is never dispatched again. A socket that delivers no output for 15 seconds switches directly to polling, even if the socket stays open or its initial connection never completes. Late callbacks from that socket are ignored. This recovers pending permission requests and cancellation results from saved output without replaying the command. Healthy initial and reconnected streams retain their existing lifetime under the caller’s RPC/run guard. After a clean close or an idle socket, each status or snapshot read gets a fresh operation timeout; successful recovery polling does not impose a new command lifetime. When both stream attempts fail without a clean close, fallback retains its original timeout budget starting after the stream attempts. - The SDK returns full log snapshots, so fallback bandwidth grows with retained output; it has no paged snapshot or stream cancellation API. A recovery-read timeout keeps partial output and explicitly reports whether command exit remains unconfirmed; it does not prove the remote command stopped. Provider/session teardown remains responsible for closing an outstanding socket. A final snapshot reconciles bytes missed by the stream before returning a recorded exit. - A sandbox record can survive the loss of its underlying container. Resume treats it as expired only when a fresh provider read confirms the exact missing-container error for that sandbox and marks it unrecoverable. Unknown errors and failed confirmation reads preserve the lease. The host still requires a verified native-runner backup before replacement. diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 362021051dc..894a6237c93 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -2461,6 +2461,7 @@ describe("Daytona sandbox provider plugin", () => { process.env.DAYTONA_API_KEY = "host-key"; const sandbox = createMockSandbox(); sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 7 }); + sandbox.process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined }); // A session command tries the log stream first: it delivers stdout and // stderr from the callback log form. sandbox.process.getSessionCommandLogs.mockImplementation( @@ -2561,6 +2562,12 @@ describe("Daytona sandbox provider plugin", () => { }); describe("log stream (default)", () => { + const createStreamingSandbox = () => { + const sandbox = createMockSandbox(); + sandbox.process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined }); + return sandbox; + }; + // A session command tries the log stream first. It streams stdout and // stderr from the callback log form instead of the 50-ms poll, and falls // back to the poll only when the stream fails. @@ -2577,7 +2584,7 @@ describe("Daytona sandbox provider plugin", () => { it("completes a session command beyond a synchronous gateway deadline without redispatch", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.executeCommand.mockRejectedValue(new MockDaytonaTimeoutError("opaque gateway timeout")); sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: 0 }); sandbox.process.getSessionCommandLogs.mockImplementation( @@ -2614,7 +2621,7 @@ describe("Daytona sandbox provider plugin", () => { it.each([undefined, 0])("preserves the existing healthy session lifetime with command timeout %s", async (commandTimeoutMs) => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: 0 }); sandbox.process.getSessionCommandLogs.mockImplementation(async (_s: string, _c: string, stdout?: (chunk: string) => void, stderr?: (chunk: string) => void) => { if (!stdout) return { stdout: "done;", stderr: "" }; @@ -2633,7 +2640,7 @@ describe("Daytona sandbox provider plugin", () => { it("removes deadline listeners from completed reads during a quiet polling run", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: undefined }); sandbox.process.getSessionCommandLogs.mockImplementation(async (_s: string, _c: string, stdout?: (chunk: string) => void) => { if (stdout) { stdout("partial;"); return; } @@ -2663,7 +2670,7 @@ describe("Daytona sandbox provider plugin", () => { process.env.DAYTONA_API_KEY = "host-key"; const executionLog = vi.fn(); const restore = __setDaytonaPluginContextForTest({ execution: { log: executionLog } } as unknown as PluginContext); - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let lateOutput: ((chunk: string) => void) | undefined; let rejectStream: ((error: Error) => void) | undefined; let keepAlive: ReturnType | undefined; @@ -2718,7 +2725,7 @@ describe("Daytona sandbox provider plugin", () => { const restore = __setDaytonaPluginContextForTest( { execution: { log: executionLog } } as unknown as PluginContext, ); - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let lateOutput!: (chunk: string) => void; let rejectStream!: (error: Error) => void; let snapshots = 0; @@ -2772,7 +2779,7 @@ describe("Daytona sandbox provider plugin", () => { it.each(["status", "snapshot"])("bounds a stuck %s read after an idle socket without claiming exit", async (stalledRead) => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let releaseRead!: () => void; const stalled = new Promise((resolve) => { releaseRead = resolve; }); const executionLog = vi.fn(); @@ -2816,7 +2823,7 @@ describe("Daytona sandbox provider plugin", () => { it("streams ordered stdout and stderr from the callback log form (test_log_stream_delivers_ordered_chunks)", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 0 }); // The callback form emits stdout and stderr chunks in order. The plugin // keeps each stream in its own arrival order. @@ -2853,7 +2860,7 @@ describe("Daytona sandbox provider plugin", () => { it("reads the exit code once after the stream ends (test_log_stream_reads_exit_code_once_after_stream_end)", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 5 }); sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { @@ -2865,7 +2872,7 @@ describe("Daytona sandbox provider plugin", () => { const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams()); // The exit code read runs one time after the stream promise resolves. - expect(sandbox.process.getSessionCommand).toHaveBeenCalledTimes(1); + expect(sandbox.process.getSessionCommand).toHaveBeenCalledTimes(2); expect(result).toMatchObject({ exitCode: 5, timedOut: false, stdout: "done" }); }); @@ -2876,7 +2883,7 @@ describe("Daytona sandbox provider plugin", () => { { execution: { log: executionLog } } as unknown as PluginContext, ); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { @@ -2910,7 +2917,7 @@ describe("Daytona sandbox provider plugin", () => { { execution: { log: executionLog } } as unknown as PluginContext, ); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let snapshots = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { @@ -2949,7 +2956,7 @@ describe("Daytona sandbox provider plugin", () => { it.each(["close", "reject"])("preserves healthy stream lifetimes after an aged %s", async (ending) => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { @@ -2985,7 +2992,7 @@ describe("Daytona sandbox provider plugin", () => { it("polls an aged running command after repeated EOF without inventing an exit or lifetime cap", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; let snapshots = 0; let exited = false; @@ -3018,7 +3025,7 @@ describe("Daytona sandbox provider plugin", () => { it.each([0, 3_600_000])("retains the fallback budget and partial output after a failed stream lasting %i ms", async (streamLifetime) => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { @@ -3051,7 +3058,7 @@ describe("Daytona sandbox provider plugin", () => { it("preserves provider status errors after a clean stream close", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommandLogs.mockResolvedValue(undefined); sandbox.process.getSessionCommand.mockRejectedValue(new Error("session no longer exists")); mockGet.mockResolvedValue(sandbox); @@ -3063,7 +3070,7 @@ describe("Daytona sandbox provider plugin", () => { it("reads bytes written after socket close before the exit code becomes visible", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { if (onStdout) onStdout("prefix;"); @@ -3089,7 +3096,7 @@ describe("Daytona sandbox provider plugin", () => { ); vi.useFakeTimers(); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; let lateOutput: ((chunk: string) => void) | undefined; sandbox.process.getSessionCommandLogs.mockImplementation( @@ -3141,7 +3148,7 @@ describe("Daytona sandbox provider plugin", () => { it.each(["status", "final snapshot", "fallback snapshot"])("preserves observation state when the SDK %s times out", async (stage) => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { if (!onStdout) throw new MockDaytonaTimeoutError("provider timeout"); @@ -3178,7 +3185,7 @@ describe("Daytona sandbox provider plugin", () => { ); vi.useFakeTimers(); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { onStdout?.("A\uFFFD"); @@ -3217,7 +3224,7 @@ describe("Daytona sandbox provider plugin", () => { { execution: { log: executionLog } } as unknown as PluginContext, ); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let streams = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { @@ -3256,7 +3263,7 @@ describe("Daytona sandbox provider plugin", () => { it("falls back to the poll path when the stream promise rejects (test_log_stream_disconnect_rejects_and_falls_back_to_poll)", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); // The callback form disconnects and rejects. The snapshot form (no // callbacks) serves the poll fallback read. sandbox.process.getSessionCommandLogs.mockImplementation( @@ -3290,7 +3297,7 @@ describe("Daytona sandbox provider plugin", () => { it("drops the replayed prefix by byte offset on a reconnect (test_log_stream_reconnect_drops_replayed_prefix_by_byte_offset)", async () => { process.env.DAYTONA_API_KEY = "host-key"; - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let attempt = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async ( @@ -3331,7 +3338,7 @@ describe("Daytona sandbox provider plugin", () => { { execution: { log: executionLog } } as unknown as PluginContext, ); try { - const sandbox = createMockSandbox(); + const sandbox = createStreamingSandbox(); let attempt = 0; sandbox.process.getSessionCommandLogs.mockImplementation( async ( diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index b8e71c2ab8f..c9276cf7942 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -1808,6 +1808,7 @@ async function executeOneShot( // Fallback reads full log snapshots as well as status. Limit the heavier // polling path while keeping output available to interactive commands. const SESSION_LOG_POLL_INTERVAL_MS = 1000; +const SESSION_INITIAL_STATUS_TIMEOUT_MS = 250; // A socket can stay open after it stops delivering output. Switch to saved // logs after this much silence; silence never proves that the command exited. const SESSION_LOG_STREAM_IDLE_MS = 15_000; @@ -2189,6 +2190,17 @@ async function executeInSession( timeoutSeconds, )); const commandId = dispatched.cmdId; + let initialExitCode: number | null = null; + try { + const status = await observeSessionCommand( + Math.min(effectiveTimeoutMs, SESSION_INITIAL_STATUS_TIMEOUT_MS), + () => sandbox.process.getSessionCommand(sessionId, commandId), + commandDeadline, + ); + if (typeof status.exitCode === "number") initialExitCode = status.exitCode; + } catch (error) { + if (error instanceof SessionCommandDeadlineError) throw error; + } // Log-stream path. A session command always tries the stream first: it // streams stdout and stderr from the callback log form, then reads the exit @@ -2199,7 +2211,9 @@ async function executeInSession( // call. The host routes it to the runner log sink by the host-issued // invocation id. This is a no-op when no plugin context is set (a direct // test call) or when the host has no active execute route. - const streamResult = await runSessionLogStream( + const streamResult = initialExitCode !== null + ? { exitCode: initialExitCode, pollUntilExit: false, buffer: streamBuffer } + : await runSessionLogStream( sandbox, sessionId, commandId, diff --git a/packages/plugins/sandbox-providers/daytona/src/session-command-completion.test.ts b/packages/plugins/sandbox-providers/daytona/src/session-command-completion.test.ts new file mode 100644 index 00000000000..47ad191601f --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/session-command-completion.test.ts @@ -0,0 +1,107 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const getSandbox = vi.hoisted(() => vi.fn()); +vi.mock("@daytonaio/sdk", () => ({ + Daytona: class { get = getSandbox; }, + DaytonaNotFoundError: class extends Error {}, + DaytonaTimeoutError: class extends Error {}, +})); + +import plugin, { __resetDaytonaSandboxHandleCacheForTest } from "./plugin.js"; + +function fixture() { + const process = { + createSession: vi.fn().mockResolvedValue(undefined), + executeSessionCommand: vi.fn().mockResolvedValue({ cmdId: "fixture-command" }), + getSessionCommand: vi.fn<(...args: string[]) => Promise<{ exitCode: number | undefined }>>().mockResolvedValue({ exitCode: 0 }), + getSessionCommandLogs: vi.fn<(session: string, command: string, stdout?: (chunk: string) => void, stderr?: (chunk: string) => void) => Promise<{ stdout: string; stderr: string } | void>>().mockImplementation(async (_session, _command, stdout) => { + if (stdout) return new Promise(() => {}); + return { stdout: "complete stdout\n", stderr: "complete stderr\n" }; + }), + }; + getSandbox.mockResolvedValue({ id: "fixture-sandbox", state: "started", process }); + const execute = (commandTimeoutMs = 2_000) => plugin.definition.onEnvironmentExecute!({ + driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment", + config: { apiKey: "fixture-not-a-key", timeoutMs: 30_000 }, + lease: { providerLeaseId: "fixture-sandbox" }, command: "printf", args: ["fixture"], + commandTimeoutMs, + }); + return { process, execute }; +} + +describe("Daytona completed session commands", () => { + beforeEach(() => { + __resetDaytonaSandboxHandleCacheForTest(); + getSandbox.mockReset(); + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + __resetDaytonaSandboxHandleCacheForTest(); + }); + + it("returns a finished command's final separated logs without opening a follow socket", async () => { + const { process, execute } = fixture(); + let settled = false; + const pending = execute().then(result => { settled = true; return result; }); + await vi.advanceTimersByTimeAsync(500); + expect(settled).toBe(true); + expect(await pending).toMatchObject({ exitCode: 0, timedOut: false, stdout: "complete stdout\n", stderr: "complete stderr\n" }); + expect(process.getSessionCommandLogs).toHaveBeenCalledExactlyOnceWith(expect.any(String), "fixture-command"); + expect(process.executeSessionCommand).toHaveBeenCalledTimes(1); + }); + + it("streams a still-running command and reconciles its final snapshot", async () => { + const { process, execute } = fixture(); + process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined }); + process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => { + if (stdout) { stdout("prefix "); return; } + return { stdout: "prefix complete\n", stderr: "error\n" }; + }); + expect(await execute()).toMatchObject({ exitCode: 0, timedOut: false, stdout: "prefix complete\n", stderr: "error\n" }); + expect(process.getSessionCommandLogs.mock.calls.filter(call => call[2])).toHaveLength(1); + }); + + it("bounds an unavailable initial status read and retains the working stream", async () => { + const { process, execute } = fixture(); + process.getSessionCommand.mockImplementationOnce(() => new Promise(() => {})); + process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => { + if (stdout) { stdout("complete stdout\n"); return; } + return { stdout: "complete stdout\n", stderr: "complete stderr\n" }; + }); + const pending = execute(); + await vi.advanceTimersByTimeAsync(300); + expect(await pending).toMatchObject({ exitCode: 0, timedOut: false, stdout: "complete stdout\n", stderr: "complete stderr\n" }); + expect(process.getSessionCommandLogs.mock.calls.filter(call => call[2])).toHaveLength(1); + }); + + it("retains stream recovery after the initial status request rejects", async () => { + const { process, execute } = fixture(); + process.getSessionCommand.mockRejectedValueOnce(new Error("status transport unavailable")); + process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => { + if (stdout) { stdout("complete stdout\n"); return; } + return { stdout: "complete stdout\n", stderr: "complete stderr\n" }; + }); + expect(await execute()).toMatchObject({ exitCode: 0, timedOut: false, stdout: "complete stdout\n", stderr: "complete stderr\n" }); + expect(process.executeSessionCommand).toHaveBeenCalledTimes(1); + }); + + it("does not treat a visible exit as proof that final logs were read", async () => { + const { process, execute } = fixture(); + process.getSessionCommandLogs.mockImplementation(() => new Promise(() => {})); + const pending = execute(100); + await vi.advanceTimersByTimeAsync(101); + expect(await pending).toMatchObject({ exitCode: null, timedOut: true, + metadata: { timeoutScope: "session_log_observation", commandExitConfirmed: true, observedExitCode: 0 } }); + }); + + it("keeps the total command deadline while the initial status is unavailable", async () => { + const { process, execute } = fixture(); + process.getSessionCommand.mockImplementation(() => new Promise(() => {})); + const pending = execute(100); + await vi.advanceTimersByTimeAsync(101); + expect(await pending).toMatchObject({ exitCode: null, timedOut: true }); + expect(process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(process.getSessionCommandLogs).not.toHaveBeenCalled(); + }); +});