fix(opencode-local): preserve deadlines in Daytona process sessions (#15905)

## Thinking Path > - Paperclip manages work done by AI agents. > - The OpenCode adapter can run its CLI in a provider sandbox. > - A long CLI run must remain observable until it exits or the host stops it. > - The legacy main command does not request a provider process session. A launch without an active run step uses one synchronous command request. > - That request can lose its response before the resolved execution budget ends. > - This PR requests the existing asynchronous session path for the main Daytona OpenCode command. It also carries the resolved command deadline through the host and provider contracts. The host still applies the effective capability gate. > - The change removes that synchronous request from eligible sandbox runs. ## Linked Issues or Issue Description Related reporting fix: #15789. A search of open and closed GitHub issues and PRs found no duplicate transport fix. **What happened?** A hosted Linux legacy OpenCode run retained a gateway-timeout response about one hour after CLI startup. It had a four-hour resolved budget and recent agent output. The earlier reporting fix correctly kept the configured policy separate from the timeout message. The original response owner and specific timer were not retained, so this PR does not claim to prove that provider cause. **Expected behavior** A Daytona target that supports persistent process sessions should dispatch the long agent command asynchronously. Paperclip should observe its output and exit through the existing session path. A single command HTTP request should not need to stay open for the full agent run. **Steps to reproduce** 1. Use legacy OpenCode with a Daytona sandbox target and a four-hour execution budget. 2. Launch outside an active measured step. The current main CLI call does not request a persistent process session. 3. In the SDK fixture, make the synchronous command request return a provider timeout before the budget ends. Keep the asynchronous session healthy for more than one hour. 4. The regression tests require session dispatch, incremental output, one command dispatch, a real exit result, and session cleanup. 5. Set a shorter command deadline and stall dispatch, a status read or a log snapshot. The command must return a timeout at its own deadline. Healthy output must not extend that deadline. **Paperclip version or commit** Observed source: `e733bddc45e00efb1819fffa8677f7b0d07802c5`. PR base: `4b59e2a2ecc67e98b336b8eb4c09959681672d54`. **Deployment mode** Hosted Linux server, legacy OpenCode adapter, Daytona sandbox. The host gate retains the old path when a provider cannot keep process sessions. ## What Changed - Add an optional sandbox session choice to the shared process helper. Leave the choice absent for existing callers. - Request that session only at the main Daytona OpenCode CLI invocation. Keep probes, preparation and other providers on their existing paths. - Carry a separate optional command deadline through the shared runner, host service and plugin SDK. Existing session callers can omit it. Zero leaves the command deadline disabled. - Bound dispatch, live output and recovery observations with one command timer. Clear it on settlement. Fence late stream callbacks. Remove deadline listeners after each read and clear interrupted polling waits. Preserve a known exit code if only the final logs time out. A timeout result does not claim that the process stopped. - Add adapter and process-helper regression coverage. Test the existing Daytona session for a virtual run beyond one hour, without command replay, with release cleanup. - Document transport selection and the limits of the timeout evidence. ## Verification - `pnpm -r typecheck` passed on Node 25.6.1 and pnpm 9.15.4 with the command deadline changes. - `pnpm build` passed with the updated contracts. The separately installed Daytona package also passes its own typecheck and build against the pinned SDK 0.203.0 and TypeScript 5.9.3. - Five focused process, adapter, host and provider suites passed all 349 tests. They cover a run beyond one hour, five stalled observation phases, healthy-stream deadlines, zero/absent budgets, late callbacks, session release, SSH and capability fallback. - The real environment-runtime forwarding tests passed: 19 selected tests, including the separate command budget and forced session flag in the worker RPC. - An earlier full local `pnpm test:run` began before a clean build and overlapped the correction. That mixed-version run was stopped and is not passing evidence. After the build, the two previously uncollected suites pass (1,659 tests). Two unchanged workspace fixtures fail with this machine's Git default branch `main`; both pass with a test-process-only `master` override. One unchanged native recovery fixture fails on both this branch and an independently installed and built checkout of base `4b59e2a2ecc67e98b336b8eb4c09959681672d54`, with the same terminal-result mismatch. It is not introduced by this change. The full local suite is not claimed green. These limitations do not replace fresh hosted CI. - Final head `8e96638cdef8ca6e91f7a42a52c83a075180c137` passes all hosted checks: 54 passed and 2 expected Storybook skips. Greptile reviewed this exact head at 18:12 UTC and scored it 5/5. Both actionable findings are fixed and resolved. GitHub reports no conflicts and a clean merge state. Earlier-head results are not used as final-head proof. - The standalone provider test command passes 373 tests, with 14 expected live-probe skips. - SDK tests use a controlled virtual clock. No live provider experiment, task replay, manual deployment, or public PR merge was performed. ## Risks The main legacy OpenCode command uses Daytona's existing session route when the effective capability gate permits it. A quiet log stream can use the existing status and snapshot fallback. Those observations add provider reads. The new optional command budget is separate from each read's timeout and returns a timeout without the host RPC's extra 30-second allowance. Existing session callers that omit it keep their previous lifetime. Callback bridge control commands retain their explicit session bypass. The capability gate retains the previous path for unsupported targets. Other providers, local and SSH execution keep their existing path. Retry policy, Stop acknowledgement and lease ownership rules stay the same. The SDK exposes no cancellation handle for an abandoned stream; existing Stop and lease teardown still own command termination. This fix does not establish the original gateway owner or prove recovery in a live environment. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, repository tools and code execution. The exact backend identifier and context window size are not exposed by this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (changed paths; the independently reproduced local baseline failure is documented above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-21 08:22:20 +02:00 · 2026-10-11 12:08:45 -07:00
1 parent 1ac48aff8d
commit 0ba88d2195
@@ -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`,
@@ -81,6 +81,9 @@ export interface CommandManagedRuntimeRunner {
    env?: Record<string, string>;
    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<void>;
    onSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
    /**
@@ -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<void> }) => {
+      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");
  });
 });

@@ -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<void>;
  onRuntimeProgress?: RuntimeStatusSink;
  onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise<void>;
@@ -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,
@@ -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<typeof import("@paperclipai/adapter-utils/execution-target")>("@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<void> }) => {
+        // 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<typeof import("@paperclipai/adapter-utils/execution-target")>("@paperclipai/adapter-utils/execution-target");
    mocks.process.mockImplementation(actual.runAdapterExecutionTargetProcess);
@@ -666,6 +666,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
      });
      const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
        onProcessStopped: providerStop.beginInvocation(),
+        // Keep a long sandbox run off the synchronous command HTTP request.
+        // Probes and preparation keep their existing execution paths.
+        ...(runtimeExecutionTarget?.kind === "remote" && runtimeExecutionTarget.transport === "sandbox"
+          && runtimeExecutionTarget.providerKey === "daytona" ? { useSession: true } : {}),
        cwd,
        env: preparedRuntimeConfig.env,
        stdin: prompt,
@@ -2512,6 +2512,143 @@ describe("Daytona sandbox provider plugin", () => {
        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<typeof setInterval> | 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();
@@ -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 <T>(action: () => Promise<T>): Promise<T> => {
+      if (expired) throw new SessionCommandDeadlineError();
+      let rejectDeadline!: (error: SessionCommandDeadlineError) => void;
+      const deadline = new Promise<never>((_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<typeof createCommandDeadline>;
+function withinCommandDeadline<T>(deadline: CommandDeadline, action: () => Promise<T>): Promise<T> {
+  return deadline ? deadline.race(action) : action();
+}
+
+async function sleepWithinCommandDeadline(deadline: CommandDeadline, delayMs: number): Promise<void> {
+  if (!deadline) return await sleep(delayMs);
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    await deadline.race(() => new Promise<void>((resolve) => { timer = setTimeout(resolve, delayMs); }));
+  } finally {
+    if (timer !== undefined) clearTimeout(timer);
+  }
+}
+
 async function beforeSessionDeadline<T>(deadlineMs: number, action: () => Promise<T>): Promise<T> {
  const remainingMs = deadlineMs - Date.now();
  if (remainingMs <= 0) throw new SessionCommandDeadlineError();
@@ -1943,12 +1990,18 @@ async function beforeSessionDeadline<T>(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<T>(timeoutMs: number, action: () => Promise<T>): Promise<T> {
+// 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<T>(timeoutMs: number, action: () => Promise<T>, commandDeadline?: CommandDeadline): Promise<T> {
  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<typeof createSessionStreamBuffer>,
+  commandDeadline?: CommandDeadline,
 ): Promise<SessionLogStreamResult> {
  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<number | null> {
-  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 <T>(action: () => Promise<T>): Promise<T> => {
      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);
    }
@@ -713,6 +713,10 @@ export interface PluginEnvironmentExecuteParams extends PluginEnvironmentDriverB
  env?: Record<string, string>;
  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.
   *
@@ -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 () => {
@@ -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);
  });

@@ -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
@@ -539,6 +539,9 @@ export interface EnvironmentDriverExecuteInput extends EnvironmentDriverLeaseInp
  env?: Record<string, string>;
  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