diff --git a/doc/observability.md b/doc/observability.md index 7d17a9f564c..327f588c517 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -643,6 +643,19 @@ resolved policy describes the configured limit, not proof that its timer fired; stderr remains in the instance run log and is not used as the timeout exception message. An earlier provider timeout does not establish its upstream cause. +Legacy OpenCode requests Daytona's persistent process session for +its main CLI command. Daytona targets with the effective persistent-session capability +use asynchronous dispatch and log observation instead of holding one synchronous +command HTTP request open for the whole run. The command's resolved timeout is +forwarded separately from individual status/log observation timeouts. One timer +bounds dispatch and observation, fences late log callbacks, and clears at settlement. +A timeout does not prove command termination; the existing stop/lease cleanup +path still owns that evidence. Existing session callers without this deadline +keep their previous lifetime. Probes, preparation and other providers +keep their existing paths. Local and SSH execution do not use this option. +The resolved execution budget and lease cleanup remain unchanged; the session +request does not prove that a previously observed gateway timeout had this cause. + When settlement records a workspace restore failure, `run_execution` also includes `workspaceRestoreFailure` with one of the shared, path-free codes: `restore_permission_denied`, `restore_lock_timeout`, `restore_unsafe_archive`, diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index b46b2476f1c..30d09ab72f7 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -81,6 +81,9 @@ export interface CommandManagedRuntimeRunner { env?: Record; stdin?: string; timeoutMs?: number; + /** Optional execution deadline for a session command, separate from an + * individual provider observation timeout. Zero means no command deadline. */ + commandTimeoutMs?: number; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; /** diff --git a/packages/adapter-utils/src/execution-target.test.ts b/packages/adapter-utils/src/execution-target.test.ts index 46a4df7d6d5..54fc73c639d 100644 --- a/packages/adapter-utils/src/execution-target.test.ts +++ b/packages/adapter-utils/src/execution-target.test.ts @@ -227,6 +227,35 @@ describe("runAdapterExecutionTargetProcess", () => { vi.unstubAllEnvs(); }); + it.each([true, false, undefined])("preserves the sandbox session choice %s and command outcome", async (useSession) => { + const stopped = vi.fn(); + const onLog = vi.fn(async (_stream: "stdout" | "stderr", _chunk: string) => {}); + const completed = { + exitCode: 7, signal: null, timedOut: false, stdout: "partial output", + stderr: "fixture failure", pid: null, startedAt: "2020-01-01T00:00:00.000Z", + }; + const execute = vi.fn(async (input: { onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise }) => { + await input.onLog?.("stdout", completed.stdout); + return completed; + }); + const result = await runAdapterExecutionTargetProcess("run-fixture", { + kind: "remote", transport: "sandbox", remoteCwd: "/workspace", runner: { execute }, + }, "opencode", ["run"], { + cwd: "/local", env: {}, timeoutSec: 14400, graceSec: 20, + onLog, onProcessStopped: stopped, ...(useSession === undefined ? {} : { useSession }), + }); + const call = execute.mock.calls[0]![0]; + expect(call).toMatchObject({ command: "opencode", args: ["run"], cwd: "/workspace", timeoutMs: 14400000 }); + if (useSession === undefined) expect(call).not.toHaveProperty("useSession"); + else expect(call).toHaveProperty("useSession", useSession); + if (useSession === true) expect(call).toHaveProperty("commandTimeoutMs", 14400000); + else expect(call).not.toHaveProperty("commandTimeoutMs"); + expect(result).toEqual(completed); + expect(execute).toHaveBeenCalledTimes(1); + expect(onLog.mock.calls).toEqual([["stdout", "partial output"]]); + expect(stopped).toHaveBeenCalledTimes(1); + }); + it("sanitizes inherited host env before SSH process execution", async () => { vi.stubEnv("PATH", "/host/bin:/usr/bin"); vi.stubEnv("HOME", "/Users/local"); @@ -269,6 +298,7 @@ describe("runAdapterExecutionTargetProcess", () => { }, timeoutSec: 5, graceSec: 1, + useSession: true, onLog: async () => {}, }, ); @@ -283,6 +313,7 @@ describe("runAdapterExecutionTargetProcess", () => { }, }), ); + expect(runChildProcessSpy.mock.calls[0]![3]).not.toHaveProperty("useSession"); }); }); diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index c70c02d7b91..df93ac848f5 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -278,6 +278,9 @@ export interface AdapterExecutionTargetProcessOptions { stdin?: string; timeoutSec: number; graceSec: number; + /** Sandbox-only: request the provider's persistent process session for a + * long-lived command. The host still gates this on provider capabilities. */ + useSession?: boolean; onLog: (stream: "stdout" | "stderr", chunk: string) => Promise; onRuntimeProgress?: RuntimeStatusSink; onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise; @@ -881,6 +884,8 @@ export async function runAdapterExecutionTargetProcess( env, stdin: options.stdin, timeoutMs: options.timeoutSec > 0 ? options.timeoutSec * 1000 : target.timeoutMs ?? undefined, + ...(options.useSession !== undefined ? { useSession: options.useSession } : {}), + ...(options.useSession === true ? { commandTimeoutMs: Math.max(0, options.timeoutSec) * 1000 } : {}), // The tail loop already streams incremental chunks; suppress the // runner's end-of-run batched onLog to avoid duplicate log bytes. onLog: runLogTail ? undefined : options.onLog, diff --git a/packages/adapters/opencode-local/src/server/execute.timeout.test.ts b/packages/adapters/opencode-local/src/server/execute.timeout.test.ts index 748e88d5c11..eac949db088 100644 --- a/packages/adapters/opencode-local/src/server/execute.timeout.test.ts +++ b/packages/adapters/opencode-local/src/server/execute.timeout.test.ts @@ -49,7 +49,7 @@ describe("OpenCode timeout reporting", () => { config: { command: "opencode", model: "openai/fixture", cwd: dir, timeoutSec, env: { OPENCODE_ALLOW_ALL_MODELS: "1" } }, context: {}, onLog: async () => {}, - ...(remote ? { executionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace" } } : {}), + ...(remote ? { executionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", remoteCwd: "/workspace" } } : {}), }); it.each([ @@ -80,6 +80,40 @@ describe("OpenCode timeout reporting", () => { expect(mocks.stop).toHaveBeenCalledTimes(1); }); + it("runs the sandbox CLI through the actual process seam with session dispatch and incremental output", async () => { + const actual = await vi.importActual("@paperclipai/adapter-utils/execution-target"); + mocks.process.mockImplementation(actual.runAdapterExecutionTargetProcess); + const logs = vi.fn(async () => {}); + const runner = { + execute: vi.fn(async (input: { useSession?: boolean; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise }) => { + // A synchronous command can lose its HTTP response before the run's + // configured deadline. Session dispatch must keep that request off the + // main CLI path; it must not replay the agent to recover the response. + if (!input.useSession) return failed; + await input.onLog?.("stdout", "streamed fixture output"); + return { ...failed, timedOut: false, exitCode: 0, stdout: "streamed fixture output", stderr: "" }; + }), + }; + const ctx = context(0); + ctx.onLog = logs; + ctx.executionTarget = { kind: "remote", transport: "sandbox", providerKey: "daytona", remoteCwd: "/workspace", runner }; + const result = await execute(ctx); + expect(result).toMatchObject({ timedOut: false, exitCode: 0, errorMessage: null, + resultJson: { stdout: "streamed fixture output", adapterExecutionTimeout: { timeoutSec: 14400, source: "sandbox_default" } } }); + expect(runner.execute).toHaveBeenCalledTimes(1); + expect(runner.execute.mock.calls[0]![0]).toMatchObject({ command: "opencode", useSession: true, timeoutMs: 14400000, commandTimeoutMs: 14400000 }); + expect(logs).toHaveBeenCalledWith("stdout", "streamed fixture output"); + expect(mocks.restore).toHaveBeenCalledTimes(1); + expect(mocks.stop).toHaveBeenCalledTimes(1); + }); + + it("keeps other sandbox providers on their existing command path", async () => { + const ctx = context(12); + ctx.executionTarget = { kind: "remote", transport: "sandbox", providerKey: "fixture-provider", remoteCwd: "/workspace" }; + await execute(ctx); + expect(mocks.process.mock.calls[0]![4]).not.toHaveProperty("useSession"); + }); + it("retains real local deadline and termination behavior without claiming provider origin", async () => { const actual = await vi.importActual("@paperclipai/adapter-utils/execution-target"); mocks.process.mockImplementation(actual.runAdapterExecutionTargetProcess); diff --git a/packages/adapters/opencode-local/src/server/execute.ts b/packages/adapters/opencode-local/src/server/execute.ts index 7171067d005..bc643aef8ee 100644 --- a/packages/adapters/opencode-local/src/server/execute.ts +++ b/packages/adapters/opencode-local/src/server/execute.ts @@ -666,6 +666,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise { finally { clearInterval(timer); } } + it("completes a session command beyond a synchronous gateway deadline without redispatch", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + const sandbox = createMockSandbox(); + sandbox.process.executeCommand.mockRejectedValue(new MockDaytonaTimeoutError("opaque gateway timeout")); + sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: 0 }); + sandbox.process.getSessionCommandLogs.mockImplementation( + async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => { + if (!onStdout) return { stdout: "first;last;", stderr: "" }; + onStdout("first;"); + await streamFor(3_600_001, onStderr); + onStdout("last;"); + }, + ); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const params = streamExecParams({ timeoutMs: 14_400_000, commandTimeoutMs: 14_400_000 }); + // Characterize the one-shot transport failure with the same command + // and budget, then exercise the existing asynchronous session path. + expect(await plugin.definition.onEnvironmentExecute?.({ ...params, bypassSession: true })) + .toMatchObject({ timedOut: true, exitCode: null }); + sandbox.process.executeCommand.mockClear(); + const pending = plugin.definition.onEnvironmentExecute?.(params); + await vi.advanceTimersByTimeAsync(3_600_002); + expect(await pending).toMatchObject({ timedOut: false, exitCode: 0, stdout: "first;last;" }); + expect(sandbox.process.executeCommand).not.toHaveBeenCalled(); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(sandbox.process.executeSessionCommand.mock.calls[0]![1]).toMatchObject({ runAsync: true }); + expect(sandbox.process.executeSessionCommand.mock.calls[0]![2]).toBe(14400); + await plugin.definition.onEnvironmentReleaseLease?.({ + driverKey: "daytona", companyId: "company-1", environmentId: "env-1", + providerLeaseId: "sandbox-123", config: params.config, + }); + expect(sandbox.process.deleteSession).toHaveBeenCalledTimes(1); + } finally { vi.useRealTimers(); } + }); + + 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(); + 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: "" }; + await streamFor(60_000, stderr); + stdout("done;"); + }); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const pending = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 1000, commandTimeoutMs })); + await vi.advanceTimersByTimeAsync(60_000); + expect(await pending).toMatchObject({ timedOut: false, exitCode: 0, stdout: "done;" }); + expect(vi.getTimerCount()).toBe(0); + } finally { vi.useRealTimers(); } + }); + + it("removes deadline listeners from completed reads during a quiet polling run", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + const sandbox = createMockSandbox(); + sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: undefined }); + sandbox.process.getSessionCommandLogs.mockImplementation(async (_s: string, _c: string, stdout?: (chunk: string) => void) => { + if (stdout) { stdout("partial;"); return; } + return { stdout: "partial;", stderr: "" }; + }); + mockGet.mockResolvedValue(sandbox); + const added = vi.spyOn(AbortSignal.prototype, "addEventListener"); + const removed = vi.spyOn(AbortSignal.prototype, "removeEventListener"); + vi.useFakeTimers(); + try { + const pending = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 60_000, commandTimeoutMs: 60_000 })); + await vi.advanceTimersByTimeAsync(30_000); + const signal = added.mock.contexts.find((value, index) => added.mock.calls[index]?.[0] === "abort"); + expect(signal).toBeDefined(); + const count = (spy: typeof added | typeof removed) => spy.mock.calls.filter((call, index) => call[0] === "abort" && spy.mock.contexts[index] === signal).length; + expect(count(added)).toBeGreaterThan(50); + expect(count(added) - count(removed)).toBeLessThanOrEqual(1); + await vi.advanceTimersByTimeAsync(30_000); + expect(await pending).toMatchObject({ timedOut: true, exitCode: null, stdout: "partial;" }); + expect(count(removed)).toBe(count(added)); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { added.mockRestore(); removed.mockRestore(); vi.useRealTimers(); } + }); + + it.each(["stream", "status", "snapshot", "dispatch", "final snapshot"])("preserves a command deadline during a healthy or stalled %s observation", async (phase) => { + process.env.DAYTONA_API_KEY = "host-key"; + const executionLog = vi.fn(); + const restore = __setDaytonaPluginContextForTest({ execution: { log: executionLog } } as unknown as PluginContext); + const sandbox = createMockSandbox(); + let lateOutput: ((chunk: string) => void) | undefined; + let rejectStream: ((error: Error) => void) | undefined; + let keepAlive: ReturnType | undefined; + sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: phase === "final snapshot" ? 0 : undefined }); + sandbox.process.getSessionCommandLogs.mockImplementation(async (_s: string, _c: string, stdout?: (chunk: string) => void) => { + if (!stdout) { + if (phase === "snapshot" || phase === "final snapshot") return new Promise(() => {}); + return { stdout: "first;", stderr: "" }; + } + stdout("first;"); + if (phase === "stream") { + lateOutput = stdout; + keepAlive = setInterval(() => stdout("."), 10000); + return new Promise((_resolve, reject) => { rejectStream = reject; }); + } + }); + if (phase === "status") sandbox.process.getSessionCommand.mockImplementation(async () => new Promise(() => {})); + if (phase === "dispatch") sandbox.process.executeSessionCommand.mockImplementation(async () => new Promise(() => {})); + mockGet.mockResolvedValue(sandbox); + vi.useFakeTimers(); + try { + const pending = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 14_400_000, commandTimeoutMs: 60000 })); + let settled = false; + void pending?.then(() => { settled = true; }); + await vi.advanceTimersByTimeAsync(59999); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + const result = await pending; + expect(result).toMatchObject({ timedOut: true, exitCode: null }); + if (phase === "final snapshot") { + expect(result?.metadata).toMatchObject({ timeoutScope: "session_log_observation", commandExitConfirmed: true, observedExitCode: 0 }); + expect(result?.stderr).toContain("Session log observation timed out after 60 s; final output may be incomplete."); + } else expect(result?.stderr).toContain("Command timed out after 60 s."); + expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1); + expect(sandbox.process.executeCommand).not.toHaveBeenCalled(); + const logged = executionLog.mock.calls.length; + lateOutput?.("late private output"); + rejectStream?.(new Error("late transport rejection")); + await vi.advanceTimersByTimeAsync(1); + expect(executionLog.mock.calls.length).toBe(logged); + expect(JSON.stringify(result)).not.toContain("late private"); + clearInterval(keepAlive); + await plugin.definition.onEnvironmentReleaseLease?.({ driverKey: "daytona", companyId: "company-1", environmentId: "env-1", providerLeaseId: "sandbox-123", config: streamExecParams().config }); + expect(sandbox.process.deleteSession).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { clearInterval(keepAlive); restore(); vi.useRealTimers(); } + }); + it.each(["", "prefix;🙂;"])("recovers unseen output and cancellation from a stuck socket after prefix %j", async (prefix) => { process.env.DAYTONA_API_KEY = "host-key"; const executionLog = vi.fn(); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 1f9ddd10189..6813867075a 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -1927,6 +1927,53 @@ type SessionLogStreamResult = { class SessionCommandDeadlineError extends Error {} class SessionObservationTimeoutError extends Error {} +function createCommandDeadline(timeoutMs: number | undefined) { + if (timeoutMs === undefined || timeoutMs === 0) return undefined; + if (!Number.isFinite(timeoutMs) || timeoutMs < 0 || timeoutMs > 2_147_483_647) { + throw new Error("Invalid session command timeout"); + } + let expired = false; + const dueAt = Date.now() + timeoutMs; + const controller = new AbortController(); + const timer = setTimeout(() => { + expired = true; + controller.abort(); + }, timeoutMs); + return { + expired: () => expired, + remainingMs: () => Math.max(0, dueAt - Date.now()), + race: async (action: () => Promise): Promise => { + if (expired) throw new SessionCommandDeadlineError(); + let rejectDeadline!: (error: SessionCommandDeadlineError) => void; + const deadline = new Promise((_resolve, reject) => { rejectDeadline = reject; }); + const onAbort = () => rejectDeadline(new SessionCommandDeadlineError()); + controller.signal.addEventListener("abort", onAbort, { once: true }); + try { + return await Promise.race([action(), deadline]); + } finally { + // Retain only active observations. A shared pending promise would keep + // every finished poll and its result until a long command ends. + controller.signal.removeEventListener("abort", onAbort); + } + }, + clear: () => clearTimeout(timer), + }; +} +type CommandDeadline = ReturnType; +function withinCommandDeadline(deadline: CommandDeadline, action: () => Promise): Promise { + return deadline ? deadline.race(action) : action(); +} + +async function sleepWithinCommandDeadline(deadline: CommandDeadline, delayMs: number): Promise { + if (!deadline) return await sleep(delayMs); + let timer: ReturnType | undefined; + try { + await deadline.race(() => new Promise((resolve) => { timer = setTimeout(resolve, delayMs); })); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + async function beforeSessionDeadline(deadlineMs: number, action: () => Promise): Promise { const remainingMs = deadlineMs - Date.now(); if (remainingMs <= 0) throw new SessionCommandDeadlineError(); @@ -1943,12 +1990,18 @@ async function beforeSessionDeadline(deadlineMs: number, action: () => Promis } } -// 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 { +// A provider operation timeout bounds one observation. An explicit command +// deadline also bounds the stream; existing callers retain their run/Stop guard. +async function observeSessionCommand(timeoutMs: number, action: () => Promise, commandDeadline?: CommandDeadline): Promise { try { + // Use the earlier shared command timer instead of leaving a longer + // observation timer alive after the command has already settled. + if (commandDeadline && commandDeadline.remainingMs() <= timeoutMs) { + return await commandDeadline.race(action); + } return await beforeSessionDeadline(Date.now() + timeoutMs, action); } catch (error) { + if (commandDeadline?.expired()) throw new SessionCommandDeadlineError(); if (error instanceof SessionCommandDeadlineError || error instanceof DaytonaTimeoutError) { throw new SessionObservationTimeoutError(); } @@ -1966,6 +2019,7 @@ async function runSessionLogStream( commandId: string, timeoutMs: number, buffer: ReturnType, + commandDeadline?: CommandDeadline, ): Promise { let reconnects = 0; let pollUntilExit = false; @@ -1980,14 +2034,14 @@ async function runSessionLogStream( idleTimer = setTimeout(signalIdle, SESSION_LOG_STREAM_IDLE_MS); }; const receive = (stream: "stdout" | "stderr", chunk: string) => { - if (!acceptingOutput || !chunk) return; + if (!acceptingOutput || commandDeadline?.expired() || !chunk) return; armIdleTimer(); if (stream === "stdout") buffer.onStdout(chunk); else buffer.onStderr(chunk); }; try { armIdleTimer(); - const outcome = await Promise.race([ + const outcome = await withinCommandDeadline(commandDeadline, () => Promise.race([ sandbox.process.getSessionCommandLogs( sessionId, commandId, @@ -1995,7 +2049,7 @@ async function runSessionLogStream( (chunk) => receive("stderr", chunk), ).then(() => "closed" as const), idle, - ]); + ])); if (outcome === "idle") { // Do not wait for the abandoned socket or open another one. The SDK // exposes no cancellation handle. Fence its callbacks before the @@ -2004,7 +2058,8 @@ async function runSessionLogStream( } streamClosed = true; pollUntilExit = true; - } catch { + } catch (error) { + if (error instanceof SessionCommandDeadlineError) throw error; // The command can still be running after a log transport failure. } finally { // The SDK exposes no cancellation handle for this socket. Session @@ -2013,7 +2068,7 @@ async function runSessionLogStream( clearTimeout(idleTimer); } if (streamClosed) { - const exitCode = await readSessionExitCode(sandbox, sessionId, commandId, timeoutMs); + const exitCode = await readSessionExitCode(sandbox, sessionId, commandId, timeoutMs, commandDeadline); if (exitCode !== null) return { exitCode, pollUntilExit, buffer }; } if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) { @@ -2033,14 +2088,15 @@ async function readSessionExitCode( sessionId: string, commandId: string, timeoutMs: number, + commandDeadline?: CommandDeadline, ): Promise { - const first = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId)); + const first = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId), commandDeadline); if (typeof first.exitCode === "number") { return first.exitCode; } for (const delayMs of SESSION_EXIT_CODE_RETRY_DELAYS_MS) { - await sleep(delayMs); - const status = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId)); + await sleepWithinCommandDeadline(commandDeadline, delayMs); + const status = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId), commandDeadline); if (typeof status.exitCode === "number") { return status.exitCode; } @@ -2083,6 +2139,7 @@ async function executeInSession( // attributed to the provider boundary. let execStart: number | null = null; let confirmedExitCode: number | null = null; + let commandDeadline: CommandDeadline; const streamBuffer = createSessionStreamBuffer((stream, text) => pluginContext?.execution.log(stream, text)); try { @@ -2102,11 +2159,12 @@ async function executeInSession( const command = `( ${loginScript} )`; execStart = timingNow(); - const dispatched = await sandbox.process.executeSessionCommand( + commandDeadline = createCommandDeadline(params.commandTimeoutMs); + const dispatched = await withinCommandDeadline(commandDeadline, () => sandbox.process.executeSessionCommand( sessionId, { command, runAsync: true }, timeoutSeconds, - ); + )); const commandId = dispatched.cmdId; // Log-stream path. A session command always tries the stream first: it @@ -2124,12 +2182,13 @@ async function executeInSession( commandId, effectiveTimeoutMs, streamBuffer, + commandDeadline, ); 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)); + const logs = await observeSessionCommand(effectiveTimeoutMs, () => sandbox.process.getSessionCommandLogs(sessionId, commandId), commandDeadline); streamBuffer.resetConnectionCursors(); streamBuffer.onStdout(logs?.stdout ?? ""); streamBuffer.onStderr(logs?.stderr ?? ""); @@ -2155,8 +2214,8 @@ async function executeInSession( const observe = async (action: () => Promise): Promise => { try { return await (streamResult.pollUntilExit - ? observeSessionCommand(effectiveTimeoutMs, action) - : beforeSessionDeadline(deadlineMs, action)); + ? observeSessionCommand(effectiveTimeoutMs, action, commandDeadline) + : commandDeadline ? commandDeadline.race(action) : beforeSessionDeadline(deadlineMs, action)); } catch (error) { if (error instanceof DaytonaTimeoutError) throw new SessionObservationTimeoutError(); throw error; @@ -2183,11 +2242,12 @@ async function executeInSession( // 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.pollUntilExit ? SESSION_LOG_POLL_INTERVAL_MS + await sleepWithinCommandDeadline(commandDeadline, streamResult.pollUntilExit ? SESSION_LOG_POLL_INTERVAL_MS : Math.max(0, Math.min(SESSION_LOG_POLL_INTERVAL_MS, deadlineMs - Date.now()))); } } catch (error) { - if (error instanceof SessionObservationTimeoutError) { + if (error instanceof SessionObservationTimeoutError || + (error instanceof SessionCommandDeadlineError && confirmedExitCode !== null)) { // Preserve undecided decoder suffixes in the partial result, without // injecting a possibly incomplete character into the live protocol. streamBuffer.finish(false); @@ -2195,7 +2255,7 @@ async function executeInSession( 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`, + stderr: `${streamBuffer.stderr}Session log observation timed out after ${Math.round((commandDeadline?.expired() ? params.commandTimeoutMs! : 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", @@ -2209,7 +2269,7 @@ async function executeInSession( 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 instanceof SessionCommandDeadlineError - ? `Command timed out after ${Math.round(effectiveTimeoutMs / 1000)} s.` + ? `Command timed out after ${Math.round((commandDeadline?.expired() ? params.commandTimeoutMs! : effectiveTimeoutMs) / 1000)} s.` : error.message.trim(); const durationMs = execStart != null ? timingNow() - execStart : undefined; return { @@ -2222,6 +2282,7 @@ async function executeInSession( } throw error; } finally { + commandDeadline?.clear(); if (stdinPath) { await sandbox.fs.deleteFile(stdinPath).catch(() => undefined); } diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 8f2ad9ce351..a851294d7ad 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -713,6 +713,10 @@ export interface PluginEnvironmentExecuteParams extends PluginEnvironmentDriverB env?: Record; stdin?: string; timeoutMs?: number; + /** Optional total command execution budget, in milliseconds. A session + * provider must keep this distinct from individual observation timeouts. + * Zero disables this deadline. Absence keeps the existing session lifetime. */ + commandTimeoutMs?: number; /** * Run this command outside the lease's persistent session. * diff --git a/server/src/__tests__/environment-execution-target-capabilities.test.ts b/server/src/__tests__/environment-execution-target-capabilities.test.ts index 5a35dec8467..22992d53253 100644 --- a/server/src/__tests__/environment-execution-target-capabilities.test.ts +++ b/server/src/__tests__/environment-execution-target-capabilities.test.ts @@ -374,9 +374,10 @@ describe("effective snapshot gates the persistent-session execution decision", ( it("forces the persistent session when the snapshot grants persistent process sessions", async () => { const { target, execute } = await buildSandboxTarget({ snapshot: FULL_GRANT, supportsSync: false }); - await target.runner?.execute({ command: "node", args: ["agent.js"], useSession: true }); + await target.runner?.execute({ command: "node", args: ["agent.js"], useSession: true, commandTimeoutMs: 12000 }); const call = execute.mock.calls[0]![0] as { forceSession?: boolean }; expect(call.forceSession).toBe(true); + expect(call).toHaveProperty("commandTimeoutMs", 12000); }); it("never forces the persistent session when the snapshot removes persistent process sessions", async () => { diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 9c456d8e83a..18d7279abdd 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -3822,6 +3822,8 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { cwd: "/workspace", env: {}, timeoutMs: 1000, + commandTimeoutMs: 900, + forceSession: true, }); await environmentService(db).update(environment.id, { @@ -3835,7 +3837,11 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(released[0]?.lease.status).toBe("released"); // The execute call carries the optional log sink as the fifth argument; it // is undefined when the caller passes no sink. - expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentExecute", expect.anything(), 31000, undefined); + expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentExecute", expect.objectContaining({ + timeoutMs: 1000, + commandTimeoutMs: 900, + bypassSession: false, + }), 31000, undefined); expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentReleaseLease", expect.anything(), 31234); }); diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index d719535f032..7303e4ad0eb 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -484,6 +484,7 @@ export async function resolveEnvironmentExecutionTarget(input: { env: commandInput.env, stdin: commandInput.stdin, timeoutMs: commandInput.timeoutMs, + commandTimeoutMs: commandInput.commandTimeoutMs, onLog: commandInput.onLog ? onIncrementalLog : undefined, // The ACP process session bridge sets `useSession` so its // long-lived agent command opens the persistent session and diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 1d63a948523..07bad8dea9c 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -539,6 +539,9 @@ export interface EnvironmentDriverExecuteInput extends EnvironmentDriverLeaseInp env?: Record; stdin?: string; timeoutMs?: number; + /** Total command execution budget for an explicitly requested session. + * Separate from the timeout of an individual status/log observation. */ + commandTimeoutMs?: number; /** * Run this command outside the lease's persistent session. The run * orchestrator sets this on the workspace provision command, which runs before @@ -2962,6 +2965,7 @@ function createSandboxEnvironmentDriver( env: input.env, stdin: input.stdin, timeoutMs: input.timeoutMs, + commandTimeoutMs: input.commandTimeoutMs, // Forward the effective session-bypass flag so a provider that opens // a persistent session skips it for a pre-run or context-less command // (the workspace provision command, the CLI install command, the