diff --git a/packages/plugins/sandbox-providers/daytona/src/manifest.ts b/packages/plugins/sandbox-providers/daytona/src/manifest.ts
index 38eedf238a..7b52c685ee 100644
--- a/packages/plugins/sandbox-providers/daytona/src/manifest.ts
+++ b/packages/plugins/sandbox-providers/daytona/src/manifest.ts
@@ -120,6 +120,12 @@ const manifest: PaperclipPluginManifestV1 = {
               "Whether to stop and later resume the sandbox across runs instead of deleting it on release.",
             default: false,
           },
+          useLogStream: {
+            type: "boolean",
+            description:
+              "When true, a session command streams stdout and stderr from the Daytona callback log form and reads the exit code one time after the stream ends. When false, the command polls the exit code and reads the logs one time. Defaults to false.",
+            default: false,
+          },
         },
       },
     },
diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
index e37d2b348f..7b2eb020c2 100644
--- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
+++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
@@ -169,6 +169,7 @@ describe("Daytona sandbox provider plugin", () => {
         reuseLease: true,
         archiveOnRelease: false,
         useSessions: false,
+        useLogStream: false,
       },
     });
   });
@@ -1533,6 +1534,141 @@ describe("Daytona sandbox provider plugin", () => {
       expect(result).toMatchObject({ exitCode: null, timedOut: true });
       expect(result!.stderr).toMatch(/timed out/);
     });
+
+    describe("log stream (useLogStream)", () => {
+      // A session exec params helper with the log-stream flag on. It streams
+      // stdout and stderr from the callback log form instead of the 50-ms poll.
+      const streamExecParams = (overrides: Record<string, unknown> = {}) =>
+        sessionExecParams({
+          config: { timeoutMs: 300000, reuseLease: false, useSessions: true, useLogStream: true },
+          ...overrides,
+        });
+
+      it("streams ordered stdout and stderr from the callback log form (test_log_stream_delivers_ordered_chunks)", async () => {
+        process.env.DAYTONA_API_KEY = "host-key";
+        const sandbox = createMockSandbox();
+        sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 0 });
+        // The callback form emits stdout and stderr chunks in order. The plugin
+        // keeps each stream in its own arrival order.
+        sandbox.process.getSessionCommandLogs.mockImplementation(
+          async (
+            _sid: string,
+            _cmdId: string,
+            onStdout?: (chunk: string) => void,
+            onStderr?: (chunk: string) => void,
+          ) => {
+            onStdout?.("out-1;");
+            onStderr?.("err-1;");
+            onStdout?.("out-2;");
+            onStderr?.("err-2;");
+          },
+        );
+        mockGet.mockResolvedValue(sandbox);
+
+        const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
+
+        // The callback stream form ran (four args), not the 50-ms snapshot poll.
+        expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(1);
+        const streamCall = sandbox.process.getSessionCommandLogs.mock.calls[0]!;
+        expect(typeof streamCall[2]).toBe("function");
+        expect(typeof streamCall[3]).toBe("function");
+        expect(result).toMatchObject({
+          exitCode: 0,
+          timedOut: false,
+          stdout: "out-1;out-2;",
+          stderr: "err-1;err-2;",
+        });
+        expect(typeof (result!.metadata as Record<string, unknown>)?.durationMs).toBe("number");
+      });
+
+      it("reads the exit code once after the stream ends (test_log_stream_reads_exit_code_once_after_stream_end)", async () => {
+        process.env.DAYTONA_API_KEY = "host-key";
+        const sandbox = createMockSandbox();
+        sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 5 });
+        sandbox.process.getSessionCommandLogs.mockImplementation(
+          async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
+            onStdout?.("done");
+          },
+        );
+        mockGet.mockResolvedValue(sandbox);
+
+        const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
+
+        // The exit code read runs one time after the stream promise resolves.
+        expect(sandbox.process.getSessionCommand).toHaveBeenCalledTimes(1);
+        expect(result).toMatchObject({ exitCode: 5, timedOut: false, stdout: "done" });
+      });
+
+      it("falls back to the poll path when the stream promise rejects (test_log_stream_disconnect_rejects_and_falls_back_to_poll)", async () => {
+        process.env.DAYTONA_API_KEY = "host-key";
+        const sandbox = createMockSandbox();
+        // The callback form disconnects and rejects. The snapshot form (no
+        // callbacks) serves the poll fallback read.
+        sandbox.process.getSessionCommandLogs.mockImplementation(
+          async (
+            _sid: string,
+            _cmdId: string,
+            onStdout?: (chunk: string) => void,
+          ) => {
+            if (onStdout) {
+              onStdout("partial");
+              throw new Error("socket error");
+            }
+            return { stdout: "poll-out", stderr: "poll-err" };
+          },
+        );
+        // The command still runs to its exit on the server, so the poll reads it.
+        sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 9 });
+        mockGet.mockResolvedValue(sandbox);
+
+        const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
+
+        // The poll fallback served the final result, and the command still
+        // yielded its exit code.
+        expect(result).toMatchObject({ exitCode: 9, timedOut: false, stdout: "poll-out", stderr: "poll-err" });
+        // The snapshot form (two args) ran for the fallback read.
+        const snapshotCalls = sandbox.process.getSessionCommandLogs.mock.calls.filter(
+          (call) => call[2] === undefined,
+        );
+        expect(snapshotCalls.length).toBeGreaterThanOrEqual(1);
+      });
+
+      it("drops the replayed prefix by byte offset on a reconnect (test_log_stream_reconnect_drops_replayed_prefix_by_byte_offset)", async () => {
+        process.env.DAYTONA_API_KEY = "host-key";
+        const sandbox = createMockSandbox();
+        let attempt = 0;
+        sandbox.process.getSessionCommandLogs.mockImplementation(
+          async (
+            _sid: string,
+            _cmdId: string,
+            onStdout?: (chunk: string) => void,
+            onStderr?: (chunk: string) => void,
+          ) => {
+            attempt += 1;
+            if (attempt === 1) {
+              // First connection: deliver a prefix, then the socket drops.
+              onStdout?.("AAA");
+              onStderr?.("EEE");
+              throw new Error("socket error");
+            }
+            // Reconnect: Daytona replays the whole log from byte 0, then the new
+            // tail. The plugin must drop the replayed prefix.
+            onStdout?.("AAA");
+            onStdout?.("BBB");
+            onStderr?.("EEE");
+            onStderr?.("FFF");
+          },
+        );
+        sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 0 });
+        mockGet.mockResolvedValue(sandbox);
+
+        const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
+
+        // The pre-disconnect bytes appear one time, not two.
+        expect(result).toMatchObject({ exitCode: 0, timedOut: false, stdout: "AAABBB", stderr: "EEEFFF" });
+        expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(2);
+      });
+    });
   });
 
   it("executes commands one-shot and returns combined output via stdout", async () => {
diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts
index 18306045b3..b760f79b34 100644
--- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts
+++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts
@@ -123,6 +123,7 @@ interface DaytonaDriverConfig {
   reuseLease: boolean;
   archiveOnRelease: boolean;
   useSessions: boolean;
+  useLogStream: boolean;
 }
 
 type WorkspaceSentinelResult = {
@@ -229,6 +230,12 @@ function parseDriverConfig(raw: Record<string, unknown>): DaytonaDriverConfig {
     // Daytona session per lease and dispatches every command into it. The flag
     // stays default off until a live leak soak passes.
     useSessions: raw.useSessions === true,
+    // Log-stream opt-in. Default OFF. When off, the session dispatch polls the
+    // exit code every 50 ms and then reads the logs one time. When on, the
+    // dispatch streams stdout and stderr from the callback log form and reads
+    // the exit code one time after the stream ends. The flag stays default off
+    // until a live soak passes.
+    useLogStream: raw.useLogStream === true,
   };
 }
 
@@ -1473,6 +1480,130 @@ function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+// Backoff delays for the exit-code read after the log stream ends. The live
+// spike measured the exit code available within one poll (91-202 ms), so the
+// first read almost always holds the code. These delays cover the rare case
+// where the first read has no code yet.
+const SESSION_EXIT_CODE_RETRY_DELAYS_MS = [50, 100, 200];
+
+// A bounded reconnect for the log stream. A disconnect settles the stream
+// promise as a rejection while the command still runs on the server. One
+// reconnect replays the log from byte 0; the stream buffer drops the replayed
+// prefix by byte offset. After this many reconnects the dispatch falls back to
+// the poll path.
+const MAX_SESSION_STREAM_RECONNECTS = 1;
+
+// Buffers the stdout and stderr of one session command from the callback log
+// stream, and drops a replayed prefix by byte offset.
+//
+// The Daytona callback stream replays the whole log from byte 0 after a
+// reconnect (it does not resume from an offset and does not omit earlier
+// bytes). So the buffer tracks the byte count it already holds per stream and
+// drops any replayed bytes that fall before that count. The dedupe runs at the
+// byte level, because Daytona replays the log byte-for-byte. The SDK keeps each
+// multibyte UTF-8 character whole per chunk and per stream, so the delivered
+// byte count always lands on a character boundary and the byte-offset split is
+// safe.
+//
+// The buffer stores each new tail as a separate chunk and joins the chunks one
+// time at read. It does not copy the earlier output on each append, so total
+// buffering work stays linear in the output size, not quadratic.
+function createSessionStreamBuffer() {
+  const streams = {
+    stdout: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 },
+    stderr: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 },
+  };
+
+  function append(
+    stream: { chunks: Buffer[]; length: number; connectionBytes: number },
+    chunk: string,
+  ): void {
+    const buf = Buffer.from(chunk, "utf8");
+    const start = stream.connectionBytes;
+    stream.connectionBytes = start + buf.length;
+    // The whole chunk falls before the delivered byte count, so it is a replay.
+    if (start + buf.length <= stream.length) {
+      return;
+    }
+    // Keep only the new tail. When the whole chunk is new, `start >=
+    // stream.length` and the tail is the whole chunk. When the chunk straddles
+    // the delivered byte count, the tail starts after the replayed prefix.
+    const tail = start >= stream.length ? buf : buf.subarray(stream.length - start);
+    stream.chunks.push(tail);
+    stream.length += tail.length;
+  }
+
+  return {
+    onStdout: (chunk: string) => append(streams.stdout, chunk),
+    onStderr: (chunk: string) => append(streams.stderr, chunk),
+    // Reset the per-connection read cursors after a reconnect, so the replayed
+    // prefix drops against the already-delivered byte count.
+    resetConnectionCursors(): void {
+      streams.stdout.connectionBytes = 0;
+      streams.stderr.connectionBytes = 0;
+    },
+    get stdout(): string {
+      return Buffer.concat(streams.stdout.chunks).toString("utf8");
+    },
+    get stderr(): string {
+      return Buffer.concat(streams.stderr.chunks).toString("utf8");
+    },
+  };
+}
+
+type SessionLogStreamResult =
+  | { ok: true; stdout: string; stderr: string }
+  | { ok: false };
+
+// Stream stdout and stderr of one session command from the callback log form.
+// The stream buffer drops a replayed prefix by byte offset on a reconnect. A
+// disconnect rejects the stream promise; the dispatch reconnects a bounded
+// number of times, then reports failure so the caller falls back to the poll
+// path.
+async function runSessionLogStream(
+  sandbox: Sandbox,
+  sessionId: string,
+  commandId: string,
+): Promise<SessionLogStreamResult> {
+  const buffer = createSessionStreamBuffer();
+  let reconnects = 0;
+  while (true) {
+    try {
+      await sandbox.process.getSessionCommandLogs(sessionId, commandId, buffer.onStdout, buffer.onStderr);
+      return { ok: true, stdout: buffer.stdout, stderr: buffer.stderr };
+    } catch {
+      if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) {
+        return { ok: false };
+      }
+      reconnects += 1;
+      buffer.resetConnectionCursors();
+    }
+  }
+}
+
+// Read the exit code one time after the log stream ends. The exit code is
+// available within one poll, so the first read almost always holds it. Add a
+// small bounded retry with backoff only for the rare case where the first read
+// has no code yet. Return null when no read holds a numeric code.
+async function readSessionExitCode(
+  sandbox: Sandbox,
+  sessionId: string,
+  commandId: string,
+): Promise<number | null> {
+  const first = await sandbox.process.getSessionCommand(sessionId, commandId);
+  if (typeof first.exitCode === "number") {
+    return first.exitCode;
+  }
+  for (const delayMs of SESSION_EXIT_CODE_RETRY_DELAYS_MS) {
+    await sleep(delayMs);
+    const status = await sandbox.process.getSessionCommand(sessionId, commandId);
+    if (typeof status.exitCode === "number") {
+      return status.exitCode;
+    }
+  }
+  return null;
+}
+
 // Dispatch one user command into the persistent session and return its true
 // stdout and stderr.
 //
@@ -1532,9 +1663,29 @@ async function executeInSession(
     );
     const commandId = dispatched.cmdId;
 
+    // Log-stream path (opt-in). Stream stdout and stderr from the callback log
+    // form, then read the exit code one time. On a stream failure, fall through
+    // to the poll path below, because the command still runs to its exit on the
+    // server.
+    if (config.useLogStream) {
+      const streamResult = await runSessionLogStream(sandbox, sessionId, commandId);
+      if (streamResult.ok) {
+        const exitCode = await readSessionExitCode(sandbox, sessionId, commandId);
+        const durationMs = timingNow() - execStart;
+        return {
+          exitCode,
+          timedOut: false,
+          stdout: streamResult.stdout,
+          stderr: streamResult.stderr,
+          metadata: { durationMs },
+        };
+      }
+    }
+
     // Poll for the exit code; the SDK has no wait method. The poll deadline uses
     // the wall clock, separate from the injected timing clock that measures the
-    // reported `durationMs`.
+    // reported `durationMs`. The poll path is the default when the log stream is
+    // off, and the fallback when the log stream fails.
     const deadlineMs = Date.now() + effectiveTimeoutMs;
     let exitCode: number | null = null;
     while (true) {