diff --git a/packages/plugins/sandbox-providers/daytona/README.md b/packages/plugins/sandbox-providers/daytona/README.md index 76edf25e51d..967c5452075 100644 --- a/packages/plugins/sandbox-providers/daytona/README.md +++ b/packages/plugins/sandbox-providers/daytona/README.md @@ -30,6 +30,8 @@ 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 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. Healthy initial and reconnected streams retain their existing lifetime under the callerโ€™s RPC/run guard. After a clean close, 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. ## Local development diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 198ff6ffb30..63369685bed 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -2460,8 +2460,8 @@ describe("Daytona sandbox provider plugin", () => { const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams()); - // The callback stream form ran (four args), not the 50-ms snapshot poll. - expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(1); + // One callback stream plus a final snapshot to reconcile unseen bytes. + expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(2); const streamCall = sandbox.process.getSessionCommandLogs.mock.calls[0]!; expect(typeof streamCall[2]).toBe("function"); expect(typeof streamCall[3]).toBe("function"); @@ -2492,6 +2492,391 @@ describe("Daytona sandbox provider plugin", () => { expect(result).toMatchObject({ exitCode: 5, timedOut: false, stdout: "done" }); }); + it("keeps a running command alive after a clean log-stream close", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest( + { execution: { log: executionLog } } as unknown as PluginContext, + ); + try { + const sandbox = createMockSandbox(); + let streams = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (!onStdout) return { stdout: "first;last;", stderr: "" }; + streams += 1; + onStdout?.("first;"); + if (streams === 2) onStdout?.("last;"); + // The SDK resolves on WebSocket close, including an idle close. + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => ({ + id: "cmd-1", command: "", exitCode: streams === 1 ? undefined : 7, + })); + mockGet.mockResolvedValue(sandbox); + + const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 })); + + expect(result).toMatchObject({ exitCode: 7, timedOut: false, stdout: "first;last;" }); + expect(executionLog.mock.calls).toEqual([["stdout", "first;"], ["stdout", "last;"]]); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(streams).toBe(2); + } finally { + restore(); + } + }); + + it.each(["first;live;done;", "first;"])("polls live output after repeated clean stream closes with terminal snapshot %j", async (finalSnapshot) => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest( + { execution: { log: executionLog } } as unknown as PluginContext, + ); + try { + const sandbox = createMockSandbox(); + let snapshots = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (onStdout) { + onStdout("first;"); + return; + } + snapshots += 1; + return { stdout: snapshots === 1 ? "first;live;" : finalSnapshot, stderr: "" }; + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => { + if (snapshots > 0) { + // New output must reach the bridge while the command still runs. + expect(executionLog).toHaveBeenCalledWith("stdout", "live;"); + } + return { id: "cmd-1", command: "", exitCode: snapshots > 0 ? 9 : undefined }; + }); + mockGet.mockResolvedValue(sandbox); + + const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 })); + + expect(result).toMatchObject({ + exitCode: 9, timedOut: false, stdout: finalSnapshot.length > 11 ? finalSnapshot : "first;live;", + }); + expect(executionLog.mock.calls).toEqual([ + ["stdout", "first;"], ["stdout", "live;"], + ...(finalSnapshot.endsWith("done;") ? [["stdout", "done;"]] : []), + ]); + expect(sandbox.process.getSessionCommandLogs.mock.calls.filter((call) => call[2]).length).toBe(2); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + } finally { + restore(); + } + }); + + it.each(["close", "reject"])("preserves healthy stream lifetimes after an aged %s", async (ending) => { + process.env.DAYTONA_API_KEY = "host-key"; + const sandbox = createMockSandbox(); + let streams = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (!onStdout) return { stdout: "first;last;", stderr: "" }; + streams += 1; + onStdout("first;"); + await new Promise((resolve) => setTimeout(resolve, 3_600_000)); + if (streams === 1 && ending === "reject") throw new Error("socket error"); + if (streams === 2) onStdout("last;"); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: streams === 2 ? 7 : undefined })); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 300_000 })); + let settled = false; + void promise?.then(() => { settled = true; }); + await vi.advanceTimersByTimeAsync(3_600_000); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(350); + expect(streams).toBe(2); + await vi.advanceTimersByTimeAsync(300_001); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(3_300_000); + expect(await promise).toMatchObject({ exitCode: 7, timedOut: false, stdout: "first;last;" }); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + 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(); + let streams = 0; + let snapshots = 0; + let exited = false; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (!onStdout) { snapshots += 1; return { stdout: "partial;", stderr: "" }; } + onStdout("partial;"); + if (++streams === 1) await new Promise((resolve) => setTimeout(resolve, 3_600_000)); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: exited ? 0 : undefined })); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 })); + let settled = false; + void promise?.then(() => { settled = true; }); + await vi.advanceTimersByTimeAsync(3_602_700); + expect(settled).toBe(false); + expect(snapshots).toBe(3); + exited = true; + await vi.advanceTimersByTimeAsync(1000); + expect(await promise).toMatchObject({ exitCode: 0, timedOut: false, stdout: "partial;" }); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + 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(); + let streams = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (onStdout) { + onStdout("partial;"); + if (++streams === 1 && streamLifetime > 0) await new Promise((resolve) => setTimeout(resolve, streamLifetime)); + throw new Error("socket error"); + } + return { stdout: "partial;", stderr: "" }; + }, + ); + sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: undefined }); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 })); + let settled = false; + void promise?.then(() => { settled = true; }); + await vi.advanceTimersByTimeAsync(streamLifetime + 399); + expect(settled).toBe(false); + expect(streams).toBe(2); + await vi.advanceTimersByTimeAsync(1); + expect(await promise).toMatchObject({ exitCode: null, timedOut: true, stdout: "partial;" }); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("preserves provider status errors after a clean stream close", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + const sandbox = createMockSandbox(); + sandbox.process.getSessionCommandLogs.mockResolvedValue(undefined); + sandbox.process.getSessionCommand.mockRejectedValue(new Error("session no longer exists")); + mockGet.mockResolvedValue(sandbox); + + await expect(plugin.definition.onEnvironmentExecute?.(streamExecParams())) + .rejects.toThrow("session no longer exists"); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + }); + + it("reads bytes written after socket close before the exit code becomes visible", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + const sandbox = createMockSandbox(); + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => { + if (onStdout) onStdout("prefix;"); + else return { stdout: "prefix;tail;", stderr: "final error output" }; + }, + ); + sandbox.process.getSessionCommand + .mockResolvedValueOnce({ exitCode: undefined }) + .mockResolvedValue({ exitCode: 3 }); + mockGet.mockResolvedValue(sandbox); + + const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams()); + + expect(result).toMatchObject({ exitCode: 3, stdout: "prefix;tail;", stderr: "final error output" }); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + }); + + it.each(["status", "snapshot", "final snapshot"])("bounds a stalled %s observation after clean close and ignores late callbacks", async (stall) => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest( + { execution: { log: executionLog } } as unknown as PluginContext, + ); + vi.useFakeTimers(); + try { + const sandbox = createMockSandbox(); + let streams = 0; + let lateOutput: ((chunk: string) => void) | undefined; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { + if (onStdout) { + streams += 1; + lateOutput = onStdout; + onStdout("partial;"); + onStderr?.("partial error;"); + return; + } + return new Promise(() => {}); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => { + if (stall === "status") return new Promise(() => {}); + return { exitCode: stall === "final snapshot" ? 0 : undefined }; + }); + mockGet.mockResolvedValue(sandbox); + const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 800 })); + let settled = false; + void promise?.then(() => { settled = true; }); + await vi.advanceTimersByTimeAsync(799); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(701); + const result = await promise; + + expect(result).toMatchObject({ + exitCode: null, + timedOut: true, + stdout: "partial;", + metadata: { + timeoutScope: "session_log_observation", + commandExitConfirmed: stall === "final snapshot", + ...(stall === "final snapshot" ? { observedExitCode: 0 } : {}), + }, + }); + expect(result!.stderr).toMatch(/^partial error;Session log observation timed out/); + const received = [...executionLog.mock.calls]; + lateOutput?.("late output must be ignored"); + expect(executionLog.mock.calls).toEqual(received); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + restore(); + } + }); + + 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(); + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { + if (!onStdout) throw new MockDaytonaTimeoutError("provider timeout"); + onStdout("partial;"); + onStderr?.("partial error;"); + if (stage === "fallback snapshot") throw new Error("socket error"); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => { + if (stage === "status") throw new MockDaytonaTimeoutError("provider timeout"); + return { exitCode: stage === "final snapshot" ? 0 : undefined }; + }); + mockGet.mockResolvedValue(sandbox); + + const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams()); + + expect(result).toMatchObject({ + exitCode: null, timedOut: true, stdout: "partial;", + metadata: { + timeoutScope: "session_log_observation", + commandExitConfirmed: stage === "final snapshot", + ...(stage === "final snapshot" ? { observedExitCode: 0 } : {}), + }, + }); + expect(result!.stderr).toMatch(/^partial error;Session log observation timed out/); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + }); + + it.each(["observation", "legacy fallback", "SDK"])("retains pending replacement characters in a %s timeout result only", async (stage) => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest( + { execution: { log: executionLog } } as unknown as PluginContext, + ); + vi.useFakeTimers(); + try { + const sandbox = createMockSandbox(); + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { + onStdout?.("A\uFFFD"); + onStderr?.("E\uFFFD\uFFFD"); + if (stage === "legacy fallback") throw new Error("socket error"); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => { + if (stage === "SDK") throw new MockDaytonaTimeoutError("provider timeout"); + return new Promise(() => {}); + }); + mockGet.mockResolvedValue(sandbox); + const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 })); + await vi.advanceTimersByTimeAsync(400); + const result = await promise; + expect(result).toMatchObject({ exitCode: null, timedOut: true, stdout: "A\uFFFD" }); + expect(result!.stderr).toMatch(/^E\uFFFD\uFFFD/); + expect(executionLog.mock.calls).toEqual([["stdout", "A"], ["stderr", "E"]]); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + restore(); + } + }); + + it.each([ + { text: "A๐Ÿ™‚Z", snapshot: "A๐Ÿ™‚Z" }, + { text: "A\uFFFD", snapshot: "" }, + { text: "A\uFFFD\uFFFD", snapshot: "A\uFFFD" }, + { text: "A\uFFFD\uFFFD", snapshot: "" }, + ])("preserves UTF-8 when the SDK decoder flushes on clean close: $text", async ({ text, snapshot }) => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest( + { execution: { log: executionLog } } as unknown as PluginContext, + ); + try { + const sandbox = createMockSandbox(); + let streams = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { + if (!onStdout) return { stdout: snapshot, stderr: "E๐Ÿ™‚Z" }; + streams += 1; + // The pinned SDK streams bytes through TextDecoder and flushes + // it on clean socket close. Model that native decoder boundary + // without importing the optional provider package in root CI. + const emitClosedStream = (bytes: Buffer, emit: (chunk: string) => void) => { + const decoder = new TextDecoder(); + const text = decoder.decode(bytes, { stream: true }); + if (text) emit(text); + const tail = decoder.decode(); + if (tail) emit(tail); + }; + const stdout = Buffer.from(text); + const stderr = Buffer.from("E๐Ÿ™‚Z"); + emitClosedStream(streams === 1 && text.includes("๐Ÿ™‚") ? stdout.subarray(0, 2) : stdout, onStdout); + emitClosedStream(streams === 1 ? stderr.subarray(0, 2) : stderr, onStderr!); + }, + ); + sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: streams === 1 ? undefined : 0 })); + mockGet.mockResolvedValue(sandbox); + + const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 })); + + expect(result).toMatchObject({ exitCode: 0, stdout: text, stderr: "E๐Ÿ™‚Z" }); + for (const [stream, expected] of [["stdout", text], ["stderr", "E๐Ÿ™‚Z"]]) { + expect(executionLog.mock.calls.filter(([name]) => name === stream).map(([, chunk]) => chunk).join("")).toBe(expected); + } + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + } finally { + restore(); + } + }); + 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(); @@ -2507,7 +2892,7 @@ describe("Daytona sandbox provider plugin", () => { onStdout("partial"); throw new Error("socket error"); } - return { stdout: "poll-out", stderr: "poll-err" }; + return { stdout: "partialpoll-out", stderr: "poll-err" }; }, ); // The command still runs to its exit on the server, so the poll reads it. @@ -2518,7 +2903,7 @@ describe("Daytona sandbox provider plugin", () => { // The poll fallback served the final result, and the command still // yielded its exit code. - expect(result).toMatchObject({ exitCode: 9, timedOut: false, stdout: "poll-out", stderr: "poll-err" }); + expect(result).toMatchObject({ exitCode: 9, timedOut: false, stdout: "partialpoll-out", stderr: "poll-err" }); // The snapshot form (two args) ran for the fallback read. const snapshotCalls = sandbox.process.getSessionCommandLogs.mock.calls.filter( (call) => call[2] === undefined, @@ -2559,7 +2944,7 @@ describe("Daytona sandbox provider plugin", () => { // The pre-disconnect bytes appear one time, not two. expect(result).toMatchObject({ exitCode: 0, timedOut: false, stdout: "AAABBB", stderr: "EEEFFF" }); - expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(2); + expect(sandbox.process.getSessionCommandLogs.mock.calls.filter((call) => call[2])).toHaveLength(2); }); it("emits each new chunk to the host and drops a replayed prefix (test_log_stream_emits_execute_log_per_chunk)", async () => { diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 84babf49860..347564f2f0b 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -1772,10 +1772,9 @@ async function executeOneShot( } } -// Poll interval for a session command's exit code. The live spike measured a -// session command resolving in about 260-300 ms, so a short interval keeps the -// poll responsive without a busy loop. -const SESSION_POLL_INTERVAL_MS = 50; +// 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; function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -1787,10 +1786,10 @@ function sleep(ms: number): Promise { // where the first read has no code yet. const SESSION_EXIT_CODE_RETRY_DELAYS_MS = [50, 100, 200]; -// A bounded reconnect for the log stream. A disconnect settles the stream -// promise as a rejection while the command still runs on the server. One -// reconnect replays the log from byte 0; the stream buffer drops the replayed -// prefix by byte offset. After this many reconnects the dispatch falls back to +// A bounded reconnect for the log stream. The SDK resolves on WebSocket close, +// which does not prove that the command exited. A reconnect replays the log +// from byte 0; the stream buffer drops the replayed prefix by byte offset. +// After this many reconnects the dispatch falls back to // the poll path. const MAX_SESSION_STREAM_RECONNECTS = 1; @@ -1812,19 +1811,39 @@ const MAX_SESSION_STREAM_RECONNECTS = 1; function createSessionStreamBuffer( onNewTail?: (stream: "stdout" | "stderr", text: string) => void, ) { + type Stream = { + chunks: Buffer[]; + length: number; + connectionBytes: number; + pendingReplacement: string; + previousReplacement: { offset: number; text: string } | null; + }; + const createStream = (): Stream => ({ + chunks: [], length: 0, connectionBytes: 0, pendingReplacement: "", previousReplacement: null, + }); const streams = { - stdout: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 }, - stderr: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 }, + stdout: createStream(), + stderr: createStream(), }; function append( streamName: "stdout" | "stderr", - stream: { chunks: Buffer[]; length: number; connectionBytes: number }, + stream: Stream, chunk: string, + final = false, + publish = true, ): void { - const buf = Buffer.from(chunk, "utf8"); + // The SDK flushes an incomplete UTF-8 character as U+FFFD on socket close. + // Do not publish that suffix or count its replacement bytes until a replay + // confirms it. A real final U+FFFD is flushed at confirmed command exit. + const text = stream.pendingReplacement + chunk; + stream.pendingReplacement = final ? "" : (text.match(/\uFFFD+$/u)?.[0] ?? ""); + const buf = Buffer.from(text.slice(0, text.length - stream.pendingReplacement.length), "utf8"); const start = stream.connectionBytes; stream.connectionBytes = start + buf.length; + if (stream.previousReplacement && stream.connectionBytes > stream.previousReplacement.offset) { + stream.previousReplacement = null; + } // The whole chunk falls before the delivered byte count, so it is a replay. if (start + buf.length <= stream.length) { return; @@ -1837,7 +1856,7 @@ function createSessionStreamBuffer( stream.length += tail.length; // Deliver only the genuinely new tail to the live sink, so a replayed // prefix on a reconnect never reaches the host twice. - if (onNewTail && tail.length > 0) { + if (publish && onNewTail && tail.length > 0) { onNewTail(streamName, tail.toString("utf8")); } } @@ -1848,8 +1867,34 @@ function createSessionStreamBuffer( // Reset the per-connection read cursors after a reconnect, so the replayed // prefix drops against the already-delivered byte count. resetConnectionCursors(): void { - streams.stdout.connectionBytes = 0; - streams.stderr.connectionBytes = 0; + for (const stream of Object.values(streams)) { + if (stream.pendingReplacement && + (!stream.previousReplacement || stream.connectionBytes > stream.previousReplacement.offset || + (stream.connectionBytes === stream.previousReplacement.offset && + stream.pendingReplacement.length > stream.previousReplacement.text.length))) { + stream.previousReplacement = { offset: stream.connectionBytes, text: stream.pendingReplacement }; + } + stream.pendingReplacement = ""; + stream.connectionBytes = 0; + } + }, + finish(publish = true): void { + for (const name of ["stdout", "stderr"] as const) { + const stream = streams[name]; + if (stream.previousReplacement?.offset === stream.connectionBytes && + stream.previousReplacement.text.length > stream.pendingReplacement.length) { + stream.pendingReplacement = stream.previousReplacement.text; + } + append(name, stream, "", true, publish); + // A short terminal snapshot must not discard a genuine trailing + // replacement character already received from a longer stream. + const previous = stream.previousReplacement; + if (previous && previous.offset >= stream.length) { + stream.connectionBytes = previous.offset; + append(name, stream, previous.text, true, publish); + } + stream.previousReplacement = null; + } }, get stdout(): string { return Buffer.concat(streams.stdout.chunks).toString("utf8"); @@ -1860,34 +1905,85 @@ function createSessionStreamBuffer( }; } -type SessionLogStreamResult = - | { ok: true; stdout: string; stderr: string } - | { ok: false }; +type SessionLogStreamResult = { + exitCode: number | null; + closedStream: boolean; + buffer: ReturnType; +}; + +class SessionCommandDeadlineError extends Error {} +class SessionObservationTimeoutError extends Error {} + +async function beforeSessionDeadline(deadlineMs: number, action: () => Promise): Promise { + const remainingMs = deadlineMs - Date.now(); + if (remainingMs <= 0) throw new SessionCommandDeadlineError(); + let timer: ReturnType | undefined; + try { + return await Promise.race([ + action(), + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new SessionCommandDeadlineError()), remainingMs); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + +// A provider operation timeout bounds one observation, not a healthy stream's +// lifetime. The caller's RPC/run guard and session teardown own that lifetime. +async function observeSessionCommand(timeoutMs: number, action: () => Promise): Promise { + try { + return await beforeSessionDeadline(Date.now() + timeoutMs, action); + } catch (error) { + if (error instanceof SessionCommandDeadlineError || error instanceof DaytonaTimeoutError) { + throw new SessionObservationTimeoutError(); + } + throw error; + } +} // Stream stdout and stderr of one session command from the callback log form. -// The stream buffer drops a replayed prefix by byte offset on a reconnect. A -// disconnect rejects the stream promise; the dispatch reconnects a bounded -// number of times, then reports failure so the caller falls back to the poll -// path. +// The stream buffer drops a replayed prefix by byte offset on a reconnect. +// Both a rejected stream and a clean close without a confirmed process exit +// use bounded reconnects, then polling. async function runSessionLogStream( sandbox: Sandbox, sessionId: string, commandId: string, - onNewTail?: (stream: "stdout" | "stderr", text: string) => void, + timeoutMs: number, + buffer: ReturnType, ): Promise { - const buffer = createSessionStreamBuffer(onNewTail); let reconnects = 0; + let closedStream = false; while (true) { + let streamClosed = false; + let acceptingOutput = true; try { - await sandbox.process.getSessionCommandLogs(sessionId, commandId, buffer.onStdout, buffer.onStderr); - return { ok: true, stdout: buffer.stdout, stderr: buffer.stderr }; + await sandbox.process.getSessionCommandLogs( + sessionId, + commandId, + (chunk) => { if (acceptingOutput) buffer.onStdout(chunk); }, + (chunk) => { if (acceptingOutput) buffer.onStderr(chunk); }, + ); + streamClosed = true; + closedStream = true; } catch { - if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) { - return { ok: false }; - } - reconnects += 1; - buffer.resetConnectionCursors(); + // The command can still be running after a log transport failure. + } finally { + // The SDK exposes no cancellation handle for this socket. Session + // teardown owns closing it; late data cannot change a settled execution. + acceptingOutput = false; } + if (streamClosed) { + const exitCode = await readSessionExitCode(sandbox, sessionId, commandId, timeoutMs); + if (exitCode !== null) return { exitCode, closedStream, buffer }; + } + if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) { + return { exitCode: null, closedStream, buffer }; + } + reconnects += 1; + buffer.resetConnectionCursors(); } } @@ -1899,14 +1995,15 @@ async function readSessionExitCode( sandbox: Sandbox, sessionId: string, commandId: string, + timeoutMs: number, ): Promise { - const first = await sandbox.process.getSessionCommand(sessionId, commandId); + const first = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId)); if (typeof first.exitCode === "number") { return first.exitCode; } for (const delayMs of SESSION_EXIT_CODE_RETRY_DELAYS_MS) { await sleep(delayMs); - const status = await sandbox.process.getSessionCommand(sessionId, commandId); + const status = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId)); if (typeof status.exitCode === "number") { return status.exitCode; } @@ -1948,6 +2045,8 @@ async function executeInSession( // the exec wall-time spent before the abort, so a slow command is still // attributed to the provider boundary. let execStart: number | null = null; + let confirmedExitCode: number | null = null; + const streamBuffer = createSessionStreamBuffer((stream, text) => pluginContext?.execution.log(stream, text)); try { if (stdinPath) { @@ -1975,8 +2074,8 @@ async function executeInSession( // 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 - // code one time. On a stream failure, fall through to the poll path below, - // because the command still runs to its exit on the server. + // code. A log socket closing without a recorded exit leaves the command + // running, so reconnect and then poll the same command without replaying it. // // Emit each genuinely new output chunk to the host during the active execute // call. The host routes it to the runner log sink by the host-issued @@ -1986,71 +2085,101 @@ async function executeInSession( sandbox, sessionId, commandId, - (stream, text) => pluginContext?.execution.log(stream, text), + effectiveTimeoutMs, + streamBuffer, ); - if (streamResult.ok) { - const exitCode = await readSessionExitCode(sandbox, sessionId, commandId); + if (streamResult.exitCode !== null) { + confirmedExitCode = streamResult.exitCode; + // A log socket can close before the process emits its final bytes. Even + // a now-visible exit code does not prove the streamed prefix is complete. + const logs = await observeSessionCommand(effectiveTimeoutMs, () => sandbox.process.getSessionCommandLogs(sessionId, commandId)); + streamBuffer.resetConnectionCursors(); + streamBuffer.onStdout(logs?.stdout ?? ""); + streamBuffer.onStderr(logs?.stderr ?? ""); + streamBuffer.finish(); const durationMs = timingNow() - execStart; return { - exitCode, + exitCode: streamResult.exitCode, timedOut: false, - stdout: streamResult.stdout, - stderr: streamResult.stderr, + stdout: streamResult.buffer.stdout, + stderr: streamResult.buffer.stderr, metadata: { durationMs }, }; } - // Poll for the exit code; the SDK has no wait method. The poll deadline uses - // the wall clock, separate from the injected timing clock that measures the - // reported `durationMs`. The poll path is the fallback when the log stream - // fails. + // Keep forwarding output while waiting for the actual command exit. ACP + // peers may need a tool response before they can finish the command. Waiting + // for exit before reading logs would strand those peers after a stream loss. + // An EOF after a live stream does not end the command's lifetime. Bound + // each recovery read independently; the caller still owns stop/teardown. + // Preserve the legacy budget starting at fallback entry when both stream + // attempts fail, including a stream that fails after running for hours. const deadlineMs = Date.now() + effectiveTimeoutMs; - let exitCode: number | null = null; - while (true) { - const status = await sandbox.process.getSessionCommand(sessionId, commandId); - if (typeof status.exitCode === "number") { - exitCode = status.exitCode; - break; + const observe = async (action: () => Promise): Promise => { + try { + return await (streamResult.closedStream + ? observeSessionCommand(effectiveTimeoutMs, action) + : beforeSessionDeadline(deadlineMs, action)); + } catch (error) { + if (error instanceof DaytonaTimeoutError) throw new SessionObservationTimeoutError(); + throw error; } - if (Date.now() >= deadlineMs) { + }; + while (true) { + const status = await observe(() => sandbox.process.getSessionCommand(sessionId, commandId)); + if (typeof status.exitCode === "number") confirmedExitCode = status.exitCode; + const logs = await observe(() => sandbox.process.getSessionCommandLogs(sessionId, commandId)); + streamResult.buffer.resetConnectionCursors(); + streamResult.buffer.onStdout(logs.stdout ?? ""); + streamResult.buffer.onStderr(logs.stderr ?? ""); + if (confirmedExitCode !== null) { + streamResult.buffer.finish(); const durationMs = timingNow() - execStart; - const timeoutMessage = gitNet - ? `Git network operation timed out after ${Math.round(effectiveTimeoutMs / 1000)} s โ€” the remote may be unreachable or noninteractive credentials are not configured.` - : `Command timed out after ${Math.round(effectiveTimeoutMs / 1000)} s.`; return { - exitCode: null, - timedOut: true, - stdout: "", - stderr: `${timeoutMessage}\n`, + exitCode: confirmedExitCode, + timedOut: false, + stdout: streamResult.buffer.stdout, + stderr: streamResult.buffer.stderr, metadata: { durationMs }, }; } - await sleep(SESSION_POLL_INTERVAL_MS); + // Full log snapshots are heavier than a status read. Limit fallback + // traffic to one snapshot per second. Failed-stream fallback retains + // its original deadline, including the time spent waiting between reads. + await sleep(streamResult.closedStream ? SESSION_LOG_POLL_INTERVAL_MS + : Math.max(0, Math.min(SESSION_LOG_POLL_INTERVAL_MS, deadlineMs - Date.now()))); } - - // Read true, separated stdout and stderr from the logs endpoint. The - // synchronous dispatch response fields are optional, so the logs endpoint is - // the source of truth. - const logs = await sandbox.process.getSessionCommandLogs(sessionId, commandId); - const durationMs = timingNow() - execStart; - return { - exitCode, - timedOut: false, - stdout: logs.stdout ?? "", - stderr: logs.stderr ?? "", - metadata: { durationMs }, - }; } catch (error) { - if (error instanceof DaytonaTimeoutError) { + if (error instanceof SessionObservationTimeoutError) { + // Preserve undecided decoder suffixes in the partial result, without + // injecting a possibly incomplete character into the live protocol. + streamBuffer.finish(false); + return { + exitCode: null, + timedOut: true, + stdout: streamBuffer.stdout, + stderr: `${streamBuffer.stderr}Session log observation timed out after ${Math.round(effectiveTimeoutMs / 1000)} s; ${confirmedExitCode === null ? "command exit remains unconfirmed" : "final output may be incomplete"}.\n`, + metadata: { + ...(execStart != null ? { durationMs: timingNow() - execStart } : {}), + timeoutScope: "session_log_observation", + commandExitConfirmed: confirmedExitCode !== null, + ...(confirmedExitCode !== null ? { observedExitCode: confirmedExitCode } : {}), + }, + }; + } + if (error instanceof DaytonaTimeoutError || error instanceof SessionCommandDeadlineError) { + streamBuffer.finish(false); const timeoutMessage = gitNet ? `Git network operation timed out after ${Math.round(effectiveTimeoutMs / 1000)} s โ€” the remote may be unreachable or noninteractive credentials are not configured.` - : error.message.trim(); + : error instanceof SessionCommandDeadlineError + ? `Command timed out after ${Math.round(effectiveTimeoutMs / 1000)} s.` + : error.message.trim(); const durationMs = execStart != null ? timingNow() - execStart : undefined; return { exitCode: null, timedOut: true, - stdout: "", - stderr: `${timeoutMessage}\n`, + stdout: streamBuffer.stdout, + stderr: `${streamBuffer.stderr}${timeoutMessage}\n`, ...(durationMs != null ? { metadata: { durationMs } } : {}), }; }