perf(daytona): observe command exit before log-follow closure (#16561)

Return completed Daytona session commands after recorded exit and a final log snapshot. Keep bounded early status reads, command deadlines, late callback fences and stream recovery.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/plugins/sandbox-providers/daytona/README.md

@@ -30,7 +30,7 @@ Notes:
 - Each cold create uses a unique provider name and ownership labels. If creation fails after Daytona has allocated a sandbox, the driver looks up that exact name, checks every ownership label, and waits for deletion. Failed, missing, or timed-out lookups and failed deletions report unconfirmed cleanup with the provider name; they never return a usable lease. A missing name lookup after an uncertain create is not proof that a delayed provider request cannot create a resource.
 - For plugin-backed sandbox lease acquisition, unresolved creation cleanup crosses the worker RPC as a validated ownership envelope. The host records `pending_cleanup` before retrying deletion. If only a creation name is known, the provider first returns an ownership-verified sandbox ID without deleting it. The host saves that observation before authorizing deletion. Its existing cleanup sweep and durable spool preserve retries across controller restarts and environment deletion. A missing observed ID confirms cleanup after a lost deletion reply or database update; a name that has never been observed remains unresolved. Provider exceptions and resolved credentials are excluded from that envelope. This requires the matching host and plugin SDK update; probe and custom-image interactive-setup calls still report unconfirmed immediate cleanup without this lease-recovery path.
 - Reusable leases map to Daytona stop/start semantics. Non-reusable leases are deleted on release. A provider-resolved `target` does not change the identity of an existing sandbox. Release closes the same scoped lease that a later sentinel-verified resume reopens.
-- A command that has already exited returns its final separated log snapshot without opening a follow socket. The initial status check is bounded to 500ms and the caller's command deadline. An unfinished command, failed status read, or slow status read keeps the existing streaming and recovery path; the command is dispatched only once.
+- A command that has already exited returns its final separated log snapshot without opening a follow socket. The initial status check is bounded to 500ms and the caller's command deadline. Each follow connection also makes up to four early status reads, with 25/50/100/200ms delays and the same observation/deadline bounds. A confirmed exit still requires the final separated snapshot, but does not wait for the follow socket to close. Unconfirmed or unavailable status keeps streaming; the early reads stop after their bounded attempt window. The command is dispatched only once.
 - A session log socket can close while its command still runs. The driver requires a recorded command exit before returning a completed execution. It reconnects the log stream once, then polls status and log snapshots at most once per second. Replayed output is removed by byte offset, new output continues to reach the host, and the command is never dispatched again. A socket that delivers no output for 15 seconds switches directly to polling, even if the socket stays open or its initial connection never completes. Late callbacks from that socket are ignored. This recovers pending permission requests and cancellation results from saved output without replaying the command. Healthy initial and reconnected streams retain their existing lifetime under the caller’s RPC/run guard. After a clean close or an idle socket, each status or snapshot read gets a fresh operation timeout; successful recovery polling does not impose a new command lifetime. When both stream attempts fail without a clean close, fallback retains its original timeout budget starting after the stream attempts.
 - The SDK returns full log snapshots, so fallback bandwidth grows with retained output; it has no paged snapshot or stream cancellation API. A recovery-read timeout keeps partial output and explicitly reports whether command exit remains unconfirmed; it does not prove the remote command stopped. Provider/session teardown remains responsible for closing an outstanding socket. A final snapshot reconciles bytes missed by the stream before returning a recorded exit.
 - A sandbox record can survive the loss of its underlying container. Resume treats it as expired only when a fresh provider read confirms the exact missing-container error for that sandbox and marks it unrecoverable. Unknown errors and failed confirmation reads preserve the lease. The host still requires a verified native-runner backup before replacement.

packages/plugins/sandbox-providers/daytona/src/plugin.test.ts

@@ -2958,6 +2958,7 @@ describe("Daytona sandbox provider plugin", () => {
         process.env.DAYTONA_API_KEY = "host-key";
         const sandbox = createStreamingSandbox();
         let streams = 0;
+        let exited = false;
         sandbox.process.getSessionCommandLogs.mockImplementation(
           async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => {
             if (!onStdout) return { stdout: "first;last;", stderr: "" };
@@ -2965,10 +2966,10 @@ describe("Daytona sandbox provider plugin", () => {
             onStdout("first;");
             await streamFor(3_600_000, onStderr);
             if (streams === 1 && ending === "reject") throw new Error("socket error");
-            if (streams === 2) onStdout("last;");
+            if (streams === 2) { onStdout("last;"); exited = true; }
           },
         );
-        sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: streams === 2 ? 7 : undefined }));
+        sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: exited ? 7 : undefined }));
         mockGet.mockResolvedValue(sandbox);
         vi.useFakeTimers();
         try {

packages/plugins/sandbox-providers/daytona/src/plugin.ts

@@ -48,6 +48,7 @@ import type {
 } from "@paperclipai/plugin-sdk";
 import { performSyncIn, performSyncOut, withProviderSpan } from "./file-sync.js";
 import { DEFAULT_DAYTONA_OPERATION_TIMEOUT_MS } from "./manifest.js";
+import { observeSessionStreamExit } from "./session-stream-exit.js";
 
 // The Claude `setup-token` login pseudo-terminal (PTY) session for this provider.
 // The session runs the login command on a real pseudo-terminal, streams the
@@ -2050,6 +2051,7 @@ async function runSessionLogStream(
   while (true) {
     let streamClosed = false;
     let acceptingOutput = true;
+    let exitObservation: ReturnType<typeof observeSessionStreamExit> | undefined;
     let idleTimer: ReturnType<typeof setTimeout> | undefined;
     let signalIdle!: () => void;
     const idle = new Promise<"idle">((resolve) => { signalIdle = () => resolve("idle"); });
@@ -2065,15 +2067,24 @@ async function runSessionLogStream(
     };
     try {
       armIdleTimer();
-      const outcome = await withinCommandDeadline(commandDeadline, () => Promise.race([
-        sandbox.process.getSessionCommandLogs(
-          sessionId,
-          commandId,
-          (chunk) => receive("stdout", chunk),
-          (chunk) => receive("stderr", chunk),
-        ).then(() => "closed" as const),
-        idle,
-      ]));
+      const observed = await withinCommandDeadline(commandDeadline, () => {
+        exitObservation = observeSessionStreamExit(Promise.race([
+          sandbox.process.getSessionCommandLogs(
+            sessionId,
+            commandId,
+            (chunk) => receive("stdout", chunk),
+            (chunk) => receive("stderr", chunk),
+          ).then(() => "closed" as const),
+          idle,
+        ]), async () => {
+          const status = await observeSessionCommand(Math.min(timeoutMs, SESSION_INITIAL_STATUS_TIMEOUT_MS),
+            () => sandbox.process.getSessionCommand(sessionId, commandId), commandDeadline);
+          return typeof status.exitCode === "number" ? status.exitCode : null;
+        });
+        return exitObservation.result;
+      });
+      if (observed.kind === "exit") return { exitCode: observed.exitCode, pollUntilExit: false, buffer };
+      const outcome = observed.value;
       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
@@ -2089,6 +2100,7 @@ async function runSessionLogStream(
       // The SDK exposes no cancellation handle for this socket. Session
       // teardown owns closing it; late data cannot change a settled execution.
       acceptingOutput = false;
+      exitObservation?.stop();
       clearTimeout(idleTimer);
     }
     if (streamClosed) {

packages/plugins/sandbox-providers/daytona/src/session-command-completion.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { PluginContext } from "@paperclipai/plugin-sdk";
 
 const getSandbox = vi.hoisted(() => vi.fn());
 vi.mock("@daytonaio/sdk", () => ({
@@ -7,7 +8,7 @@ vi.mock("@daytonaio/sdk", () => ({
   DaytonaTimeoutError: class extends Error {},
 }));
 
-import plugin, { __resetDaytonaSandboxHandleCacheForTest } from "./plugin.js";
+import plugin, { __resetDaytonaSandboxHandleCacheForTest, __setDaytonaPluginContextForTest } from "./plugin.js";
 
 function fixture() {
   const process = {
@@ -120,4 +121,123 @@ describe("Daytona completed session commands", () => {
     expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
     expect(process.getSessionCommandLogs).not.toHaveBeenCalled();
   });
+
+  it.each([0, 7])("settles exit %s from its final snapshot before the follow socket closes", async (exitCode) => {
+    const { process, execute } = fixture();
+    process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined }).mockResolvedValue({ exitCode });
+    let lateOutput: ((chunk: string) => void) | undefined;
+    process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => {
+      if (stdout) {
+        lateOutput = stdout;
+        stdout("complete ");
+        return new Promise(() => {});
+      }
+      return { stdout: "complete stdout\n", stderr: "complete stderr\n" };
+    });
+    let settled = false;
+    const pending = execute(10_000).then(result => { settled = true; return result; });
+    await vi.advanceTimersByTimeAsync(250);
+    const settledBeforeSocketClose = settled;
+    if (!settled) await vi.advanceTimersByTimeAsync(10_000);
+    const result = await pending;
+    expect(settledBeforeSocketClose).toBe(true);
+    expect(result).toMatchObject({ exitCode, timedOut: false, stdout: "complete stdout\n", stderr: "complete stderr\n" });
+    lateOutput?.("late bytes must be ignored");
+    expect(result.stdout).toBe("complete stdout\n");
+    expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+  });
+
+  it("bounds the early status reads while a longer command keeps streaming", async () => {
+    const { process, execute } = fixture();
+    process.getSessionCommand.mockResolvedValue({ exitCode: undefined });
+    let closeStream!: () => void;
+    process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => {
+      if (stdout) { stdout("complete "); return new Promise<void>(resolve => { closeStream = resolve; }); }
+      return { stdout: "complete stdout\n", stderr: "complete stderr\n" };
+    });
+    let settled = false;
+    const pending = execute(10_000).then(result => { settled = true; return result; });
+    await vi.advanceTimersByTimeAsync(2_500);
+    expect(settled).toBe(false);
+    expect(process.getSessionCommand).toHaveBeenCalledTimes(5);
+    process.getSessionCommand.mockResolvedValue({ exitCode: 0 });
+    closeStream();
+    expect(await pending).toMatchObject({ exitCode: 0, timedOut: false, stdout: "complete stdout\n" });
+    expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it.each(["rejected", "unavailable"])("retains the healthy stream after a %s early exit read", async (failure) => {
+    const { process, execute } = fixture();
+    process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined });
+    if (failure === "rejected") process.getSessionCommand.mockRejectedValueOnce(new Error("early status unavailable"));
+    else process.getSessionCommand.mockImplementationOnce(() => new Promise(() => {}));
+    let closeStream!: () => void;
+    process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout) => {
+      if (stdout) return new Promise<void>(resolve => { closeStream = resolve; });
+      return { stdout: "complete stdout\n", stderr: "complete stderr\n" };
+    });
+    let settled = false;
+    const pending = execute(10_000).then(result => { settled = true; return result; });
+    await vi.advanceTimersByTimeAsync(600);
+    expect(settled).toBe(false);
+    expect(process.getSessionCommand).toHaveBeenCalledTimes(2);
+    closeStream();
+    expect(await pending).toMatchObject({ exitCode: 0, timedOut: false, stdout: "complete stdout\n", stderr: "complete stderr\n" });
+    expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("keeps the final snapshot deadline after observing an early exit", async () => {
+    const { process, execute } = fixture();
+    process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined });
+    process.getSessionCommandLogs.mockImplementation(() => new Promise(() => {}));
+    const pending = execute(100);
+    await vi.advanceTimersByTimeAsync(101);
+    expect(await pending).toMatchObject({ exitCode: null, timedOut: true,
+      metadata: { timeoutScope: "session_log_observation", commandExitConfirmed: true, observedExitCode: 0 } });
+    expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("stops early exit reads at the original command deadline", async () => {
+    const { process, execute } = fixture();
+    process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined }).mockImplementation(() => new Promise(() => {}));
+    const pending = execute(100);
+    await vi.advanceTimersByTimeAsync(101);
+    expect(await pending).toMatchObject({ exitCode: null, timedOut: true });
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(process.getSessionCommand).toHaveBeenCalledTimes(2);
+    expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("reconciles UTF-8 output and fences late follow callbacks after an early exit", async () => {
+    const { process, execute } = fixture();
+    const log = vi.fn();
+    const restore = __setDaytonaPluginContextForTest({ execution: { log } } as unknown as PluginContext);
+    process.getSessionCommand.mockResolvedValueOnce({ exitCode: undefined });
+    let lateStdout!: (chunk: string) => void;
+    let lateStderr!: (chunk: string) => void;
+    process.getSessionCommandLogs.mockImplementation(async (_session, _command, stdout, stderr) => {
+      if (stdout) {
+        lateStdout = stdout;
+        lateStderr = stderr!;
+        stdout("α");
+        return new Promise(() => {});
+      }
+      return { stdout: "αβ🚀\n", stderr: "γδ\n" };
+    });
+    try {
+      const pending = execute();
+      await vi.advanceTimersByTimeAsync(100);
+      expect(await pending).toMatchObject({ exitCode: 0, timedOut: false, stdout: "αβ🚀\n", stderr: "γδ\n" });
+      const deliveredCount = log.mock.calls.length;
+      lateStdout("late stdout");
+      lateStderr("late stderr");
+      expect(log).toHaveBeenCalledTimes(deliveredCount);
+      expect(process.executeSessionCommand).toHaveBeenCalledTimes(1);
+      expect(vi.getTimerCount()).toBe(0);
+    } finally { restore(); }
+  });
 });

packages/plugins/sandbox-providers/daytona/src/session-stream-exit.ts

@@ -0,0 +1,48 @@
+type SessionStreamResult<T> =
+  | { kind: "stream"; value: T }
+  | { kind: "exit"; exitCode: number };
+
+export function observeSessionStreamExit<T>(
+  stream: Promise<T>,
+  readExit: () => Promise<number | null>,
+) {
+  let stopped = false;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let releaseDelay: (() => void) | undefined;
+  let rejectStopped!: (error: Error) => void;
+  const cancelled = new Promise<never>((_resolve, reject) => { rejectStopped = reject; });
+  const streamResult: Promise<SessionStreamResult<T>> = stream.then(value => ({ kind: "stream", value }));
+  const probe = async (): Promise<SessionStreamResult<T>> => {
+    try {
+      for (const delayMs of [25, 50, 100, 200]) {
+        if (stopped) return streamResult;
+        await new Promise<void>(resolve => {
+          releaseDelay = resolve;
+          timer = setTimeout(resolve, delayMs);
+        });
+        if (stopped) return streamResult;
+        const exitCode = await readExit();
+        if (stopped) return streamResult;
+        if (exitCode !== null) return { kind: "exit", exitCode };
+      }
+    } catch {}
+    return streamResult;
+  };
+  const clearDelay = () => {
+    clearTimeout(timer);
+    releaseDelay?.();
+  };
+  const result = Promise.race([streamResult, probe(), cancelled]).finally(() => {
+    stopped = true;
+    clearDelay();
+  });
+  return {
+    result,
+    stop() {
+      if (stopped) return;
+      stopped = true;
+      clearDelay();
+      rejectStopped(new Error("Session stream exit observation stopped"));
+    },
+  };
+}