RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(plugin-daytona): stream session command logs behind useLogStream (#11021) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Sandbox provider plugins let agents run commands in remote
environments.
> - The Daytona provider polls the exit code while it waits for command
logs.
> - Polling delays log delivery and does not support long-lived streamed
commands.
> - This pull request adds an opt-in Daytona log stream with one
reconnect and a poll fallback.
> - The benefit is faster log delivery while the existing default path
stays unchanged.

## Linked Issues or Issue Description

Refs: #10941

**Subsystem affected**

packages/plugins — sandbox provider plugins.

**Problem or motivation**

The Daytona provider polls the command exit code every 50 milliseconds
while it waits for logs. This delays output and does not support a
long-lived streamed command.

**Proposed solution**

Add the `useLogStream` provider option. Stream stdout and stderr from
the Daytona callback log form, read the exit code after the stream ends,
retry the read with bounded backoff, and fall back to the existing poll
path after a disconnect.

**Alternatives considered**

Keep polling for all commands. This keeps the current behavior but does
not provide timely logs or a path for long-lived commands.

**Roadmap alignment**

This supports the roadmap item for cloud and sandbox agents, including
Daytona.

## What Changed

- Add the opt-in `useLogStream` option with a default of `false`.
- Stream stdout and stderr from the Daytona callback log form.
- Drop replayed log prefixes by delivered byte offset after reconnect.
- Retry once after disconnect, then use the existing poll path.
- Read the exit code once after a successful stream and retry when the
code is not ready.
- Add tests for ordered output, exit-code reads, disconnect fallback,
and reconnect replay handling.

## Verification

- Run `node_modules/.bin/vitest run --config
packages/plugins/sandbox-providers/daytona/vitest.config.ts
packages/plugins/sandbox-providers/daytona/src/plugin.test.ts`.
- Confirm that the full Daytona plugin test project passes with 127
tests.
- Confirm that the changed files type-check against Daytona SDK 0.203.0
types.
- Review the PR against parent PR #10941 before it reaches `master`.

## Risks

- The stream path changes behavior only when `useLogStream` is `true`.
- A stream failure can add one reconnect attempt before the existing
poll fallback.
- The stream path has no command deadline because it supports long-lived
commands.
- No endpoint, stored data, telemetry shape, authentication rule, or
result shape changes.

## Model Used

OpenAI Codex, GPT-5, tool use and code execution. The context window and
reasoning mode are not exposed by this run.

## 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
- [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>
This commit is contained in:
Nicky Leach 2026-08-06 22:56:21 -07:00 committed by GitHub
parent 9485ffea70
commit 4e76227f12
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 294 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/plugins/sandbox-providers/daytona/src/manifest.ts
View File

@ -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,
},
},
},
},

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

@ -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 () => {

153
packages/plugins/sandbox-providers/daytona/src/plugin.ts
View File

@ -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) {