mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 01:41:42 +02:00
fix(daytona): keep commands alive after log stream closure (#14799)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Sandbox providers run agent processes and send their output to the host. > - The Daytona SDK can close a log socket while the remote command still runs. > - The driver treated a clean socket close as completion before it had a command exit code. > - This pull request recovers log observation for the same command and waits for a recorded exit. > - The host keeps receiving new output without a second command dispatch. ## Linked Issues or Issue Description **What happened?** A clean close of the Daytona session log WebSocket resolves the SDK callback promise. If the command still runs, the driver returned `exitCode: null` with `timedOut: false` after a short status check. A streamed ACP bridge can then report a process disconnect. **Expected behavior** A log socket close must not complete a running command. Recovery must preserve new output, the caller's lifetime controls, and one command dispatch. **Steps to reproduce** Use the callback form of `getSessionCommandLogs`. Let that promise resolve while `getSessionCommand` still has no exit code. Keep the command running, then expose its final logs and exit code. The new regressions exercise this sequence, including streams that run longer than the provider operation timeout. Related: #11021, #11049. #14485 covers input delivery retries, which are a separate transport path. ## What Changed - Require a recorded command exit after a clean log-stream close. Reconnect once, then read status and full log snapshots at most once per second. - Forward new output during recovery. Remove replayed prefixes and reconcile a final snapshot so bytes written after socket close are retained. - Hold a trailing UTF-8 replacement suffix until replay or completion resolves it. This handles the SDK decoder flush when a socket closes in the middle of a character. - Preserve healthy initial and reconnected stream lifetimes. Bound each recovery read. Preserve the existing fallback timeout budget after rejected stream attempts. - Retain partial output on timeout. A log-observation timeout reports an unconfirmed result and preserves any observed exit code in metadata; it does not synthesize successful completion. ## Verification - `pnpm exec vitest run --config packages/plugins/sandbox-providers/daytona/vitest.config.ts`: 335 passed, 14 gated tests skipped. - `pnpm exec vitest run --project @paperclipai/plugin-daytona`: 335 passed, 14 gated tests skipped. - `pnpm exec tsc --noEmit -p packages/plugins/sandbox-providers/daytona/tsconfig.json`: passed. - `pnpm exec tsc -p packages/plugins/sandbox-providers/daytona/tsconfig.json`: passed. - `pnpm --workspace-concurrency=1 -r typecheck`: passed. - `CARGO_BUILD_JOBS=2 pnpm --workspace-concurrency=1 -r build`: passed. - `pnpm test:run`: exited with failure after 845.77 seconds. The server phase reported 43 failed files, 529 passed, and 163 skipped; 14 failed tests, 9,122 passed, and 5,599 skipped. All failure entries were traced to local PostgreSQL startup/cleanup errors or ten macOS skill-cache rename errors. The package helper restored 17 missing PostgreSQL library links; the four sequencing/migration tests and fourteen native-workspace-finalizer tests then passed. The ten cache failures match clean-base evidence with identical source/test blobs. This is not a full-suite pass; later local test groups did not run. PR CI supplies the complete check result. - Regressions cover clean and rejected stream recovery, two hour-long streams with a five-minute operation budget, live fallback output, one dispatch, delayed final output, stale snapshots, SDK UTF-8 decoding, bounded observation, and timer cleanup. - `git diff --check` and a local diff scan for secrets and private identifiers passed. - Greptile reviewed head `293c4dbe67` at 5/5. Both prior findings are fixed, both threads are resolved, and no new actionable findings remain. ## Risks - The SDK returns full snapshots with no offset API. After both stream attempts end, polling bandwidth grows with retained output. The one-second cadence limits request frequency. - The SDK callback stream has no cancellation handle. Existing caller stop logic and provider/session teardown still own its lifetime. Late callbacks from a settled stream are ignored. - A successful status read with no exit code keeps recovery active under the existing caller guard. A failed read stops recovery; it is not retried indefinitely. - No command replay, provider API change, schema migration, or runtime timeout policy change is included. ## Model Used OpenAI GPT-6 through Codex, with tool use and independent code review. The exact serving model identifier is not exposed in 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 - [ ] 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 - [ ] 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:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
0dc8d80eea
commit
0d3e7bf6ac
3 files changed
+598
-82
No files matched your search
@@ -30,6 +30,8 @@ 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 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. Healthy initial and reconnected streams retain their existing lifetime under the caller’s RPC/run guard. After a clean close, 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.
|
||||
|
||||
## Local development
|
||||
|
||||
@@ -2460,8 +2460,8 @@ describe("Daytona sandbox provider plugin", () => {
|
||||
|
||||
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);
|
||||
// One callback stream plus a final snapshot to reconcile unseen bytes.
|
||||
expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(2);
|
||||
const streamCall = sandbox.process.getSessionCommandLogs.mock.calls[0]!;
|
||||
expect(typeof streamCall[2]).toBe("function");
|
||||
expect(typeof streamCall[3]).toBe("function");
|
||||
@@ -2492,6 +2492,391 @@ describe("Daytona sandbox provider plugin", () => {
|
||||
expect(result).toMatchObject({ exitCode: 5, timedOut: false, stdout: "done" });
|
||||
});
|
||||
|
||||
it("keeps a running command alive after a clean log-stream close", async () => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const executionLog = vi.fn();
|
||||
const restore = __setDaytonaPluginContextForTest(
|
||||
{ execution: { log: executionLog } } as unknown as PluginContext,
|
||||
);
|
||||
try {
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (!onStdout) return { stdout: "first;last;", stderr: "" };
|
||||
streams += 1;
|
||||
onStdout?.("first;");
|
||||
if (streams === 2) onStdout?.("last;");
|
||||
// The SDK resolves on WebSocket close, including an idle close.
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => ({
|
||||
id: "cmd-1", command: "", exitCode: streams === 1 ? undefined : 7,
|
||||
}));
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 }));
|
||||
|
||||
expect(result).toMatchObject({ exitCode: 7, timedOut: false, stdout: "first;last;" });
|
||||
expect(executionLog.mock.calls).toEqual([["stdout", "first;"], ["stdout", "last;"]]);
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(streams).toBe(2);
|
||||
} finally {
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["first;live;done;", "first;"])("polls live output after repeated clean stream closes with terminal snapshot %j", async (finalSnapshot) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const executionLog = vi.fn();
|
||||
const restore = __setDaytonaPluginContextForTest(
|
||||
{ execution: { log: executionLog } } as unknown as PluginContext,
|
||||
);
|
||||
try {
|
||||
const sandbox = createMockSandbox();
|
||||
let snapshots = 0;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (onStdout) {
|
||||
onStdout("first;");
|
||||
return;
|
||||
}
|
||||
snapshots += 1;
|
||||
return { stdout: snapshots === 1 ? "first;live;" : finalSnapshot, stderr: "" };
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => {
|
||||
if (snapshots > 0) {
|
||||
// New output must reach the bridge while the command still runs.
|
||||
expect(executionLog).toHaveBeenCalledWith("stdout", "live;");
|
||||
}
|
||||
return { id: "cmd-1", command: "", exitCode: snapshots > 0 ? 9 : undefined };
|
||||
});
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 }));
|
||||
|
||||
expect(result).toMatchObject({
|
||||
exitCode: 9, timedOut: false, stdout: finalSnapshot.length > 11 ? finalSnapshot : "first;live;",
|
||||
});
|
||||
expect(executionLog.mock.calls).toEqual([
|
||||
["stdout", "first;"], ["stdout", "live;"],
|
||||
...(finalSnapshot.endsWith("done;") ? [["stdout", "done;"]] : []),
|
||||
]);
|
||||
expect(sandbox.process.getSessionCommandLogs.mock.calls.filter((call) => call[2]).length).toBe(2);
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["close", "reject"])("preserves healthy stream lifetimes after an aged %s", async (ending) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (!onStdout) return { stdout: "first;last;", stderr: "" };
|
||||
streams += 1;
|
||||
onStdout("first;");
|
||||
await new Promise((resolve) => setTimeout(resolve, 3_600_000));
|
||||
if (streams === 1 && ending === "reject") throw new Error("socket error");
|
||||
if (streams === 2) onStdout("last;");
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: streams === 2 ? 7 : undefined }));
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 300_000 }));
|
||||
let settled = false;
|
||||
void promise?.then(() => { settled = true; });
|
||||
await vi.advanceTimersByTimeAsync(3_600_000);
|
||||
expect(settled).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(350);
|
||||
expect(streams).toBe(2);
|
||||
await vi.advanceTimersByTimeAsync(300_001);
|
||||
expect(settled).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(3_300_000);
|
||||
expect(await promise).toMatchObject({ exitCode: 7, timedOut: false, stdout: "first;last;" });
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("polls an aged running command after repeated EOF without inventing an exit or lifetime cap", async () => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
let snapshots = 0;
|
||||
let exited = false;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (!onStdout) { snapshots += 1; return { stdout: "partial;", stderr: "" }; }
|
||||
onStdout("partial;");
|
||||
if (++streams === 1) await new Promise((resolve) => setTimeout(resolve, 3_600_000));
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: exited ? 0 : undefined }));
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 }));
|
||||
let settled = false;
|
||||
void promise?.then(() => { settled = true; });
|
||||
await vi.advanceTimersByTimeAsync(3_602_700);
|
||||
expect(settled).toBe(false);
|
||||
expect(snapshots).toBe(3);
|
||||
exited = true;
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await promise).toMatchObject({ exitCode: 0, timedOut: false, stdout: "partial;" });
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it.each([0, 3_600_000])("retains the fallback budget and partial output after a failed stream lasting %i ms", async (streamLifetime) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (onStdout) {
|
||||
onStdout("partial;");
|
||||
if (++streams === 1 && streamLifetime > 0) await new Promise((resolve) => setTimeout(resolve, streamLifetime));
|
||||
throw new Error("socket error");
|
||||
}
|
||||
return { stdout: "partial;", stderr: "" };
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockResolvedValue({ exitCode: undefined });
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 }));
|
||||
let settled = false;
|
||||
void promise?.then(() => { settled = true; });
|
||||
await vi.advanceTimersByTimeAsync(streamLifetime + 399);
|
||||
expect(settled).toBe(false);
|
||||
expect(streams).toBe(2);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(await promise).toMatchObject({ exitCode: null, timedOut: true, stdout: "partial;" });
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves provider status errors after a clean stream close", async () => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
sandbox.process.getSessionCommandLogs.mockResolvedValue(undefined);
|
||||
sandbox.process.getSessionCommand.mockRejectedValue(new Error("session no longer exists"));
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
await expect(plugin.definition.onEnvironmentExecute?.(streamExecParams()))
|
||||
.rejects.toThrow("session no longer exists");
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("reads bytes written after socket close before the exit code becomes visible", async () => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void) => {
|
||||
if (onStdout) onStdout("prefix;");
|
||||
else return { stdout: "prefix;tail;", stderr: "final error output" };
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand
|
||||
.mockResolvedValueOnce({ exitCode: undefined })
|
||||
.mockResolvedValue({ exitCode: 3 });
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
|
||||
|
||||
expect(result).toMatchObject({ exitCode: 3, stdout: "prefix;tail;", stderr: "final error output" });
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each(["status", "snapshot", "final snapshot"])("bounds a stalled %s observation after clean close and ignores late callbacks", async (stall) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const executionLog = vi.fn();
|
||||
const restore = __setDaytonaPluginContextForTest(
|
||||
{ execution: { log: executionLog } } as unknown as PluginContext,
|
||||
);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
let lateOutput: ((chunk: string) => void) | undefined;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => {
|
||||
if (onStdout) {
|
||||
streams += 1;
|
||||
lateOutput = onStdout;
|
||||
onStdout("partial;");
|
||||
onStderr?.("partial error;");
|
||||
return;
|
||||
}
|
||||
return new Promise(() => {});
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => {
|
||||
if (stall === "status") return new Promise(() => {});
|
||||
return { exitCode: stall === "final snapshot" ? 0 : undefined };
|
||||
});
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 800 }));
|
||||
let settled = false;
|
||||
void promise?.then(() => { settled = true; });
|
||||
await vi.advanceTimersByTimeAsync(799);
|
||||
expect(settled).toBe(false);
|
||||
await vi.advanceTimersByTimeAsync(701);
|
||||
const result = await promise;
|
||||
|
||||
expect(result).toMatchObject({
|
||||
exitCode: null,
|
||||
timedOut: true,
|
||||
stdout: "partial;",
|
||||
metadata: {
|
||||
timeoutScope: "session_log_observation",
|
||||
commandExitConfirmed: stall === "final snapshot",
|
||||
...(stall === "final snapshot" ? { observedExitCode: 0 } : {}),
|
||||
},
|
||||
});
|
||||
expect(result!.stderr).toMatch(/^partial error;Session log observation timed out/);
|
||||
const received = [...executionLog.mock.calls];
|
||||
lateOutput?.("late output must be ignored");
|
||||
expect(executionLog.mock.calls).toEqual(received);
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["status", "final snapshot", "fallback snapshot"])("preserves observation state when the SDK %s times out", async (stage) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const sandbox = createMockSandbox();
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => {
|
||||
if (!onStdout) throw new MockDaytonaTimeoutError("provider timeout");
|
||||
onStdout("partial;");
|
||||
onStderr?.("partial error;");
|
||||
if (stage === "fallback snapshot") throw new Error("socket error");
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => {
|
||||
if (stage === "status") throw new MockDaytonaTimeoutError("provider timeout");
|
||||
return { exitCode: stage === "final snapshot" ? 0 : undefined };
|
||||
});
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams());
|
||||
|
||||
expect(result).toMatchObject({
|
||||
exitCode: null, timedOut: true, stdout: "partial;",
|
||||
metadata: {
|
||||
timeoutScope: "session_log_observation",
|
||||
commandExitConfirmed: stage === "final snapshot",
|
||||
...(stage === "final snapshot" ? { observedExitCode: 0 } : {}),
|
||||
},
|
||||
});
|
||||
expect(result!.stderr).toMatch(/^partial error;Session log observation timed out/);
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each(["observation", "legacy fallback", "SDK"])("retains pending replacement characters in a %s timeout result only", async (stage) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const executionLog = vi.fn();
|
||||
const restore = __setDaytonaPluginContextForTest(
|
||||
{ execution: { log: executionLog } } as unknown as PluginContext,
|
||||
);
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const sandbox = createMockSandbox();
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => {
|
||||
onStdout?.("A\uFFFD");
|
||||
onStderr?.("E\uFFFD\uFFFD");
|
||||
if (stage === "legacy fallback") throw new Error("socket error");
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => {
|
||||
if (stage === "SDK") throw new MockDaytonaTimeoutError("provider timeout");
|
||||
return new Promise(() => {});
|
||||
});
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
const promise = plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 400 }));
|
||||
await vi.advanceTimersByTimeAsync(400);
|
||||
const result = await promise;
|
||||
expect(result).toMatchObject({ exitCode: null, timedOut: true, stdout: "A\uFFFD" });
|
||||
expect(result!.stderr).toMatch(/^E\uFFFD\uFFFD/);
|
||||
expect(executionLog.mock.calls).toEqual([["stdout", "A"], ["stderr", "E"]]);
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ text: "A🙂Z", snapshot: "A🙂Z" },
|
||||
{ text: "A\uFFFD", snapshot: "" },
|
||||
{ text: "A\uFFFD\uFFFD", snapshot: "A\uFFFD" },
|
||||
{ text: "A\uFFFD\uFFFD", snapshot: "" },
|
||||
])("preserves UTF-8 when the SDK decoder flushes on clean close: $text", async ({ text, snapshot }) => {
|
||||
process.env.DAYTONA_API_KEY = "host-key";
|
||||
const executionLog = vi.fn();
|
||||
const restore = __setDaytonaPluginContextForTest(
|
||||
{ execution: { log: executionLog } } as unknown as PluginContext,
|
||||
);
|
||||
try {
|
||||
const sandbox = createMockSandbox();
|
||||
let streams = 0;
|
||||
sandbox.process.getSessionCommandLogs.mockImplementation(
|
||||
async (_sid: string, _cmdId: string, onStdout?: (chunk: string) => void, onStderr?: (chunk: string) => void) => {
|
||||
if (!onStdout) return { stdout: snapshot, stderr: "E🙂Z" };
|
||||
streams += 1;
|
||||
// The pinned SDK streams bytes through TextDecoder and flushes
|
||||
// it on clean socket close. Model that native decoder boundary
|
||||
// without importing the optional provider package in root CI.
|
||||
const emitClosedStream = (bytes: Buffer, emit: (chunk: string) => void) => {
|
||||
const decoder = new TextDecoder();
|
||||
const text = decoder.decode(bytes, { stream: true });
|
||||
if (text) emit(text);
|
||||
const tail = decoder.decode();
|
||||
if (tail) emit(tail);
|
||||
};
|
||||
const stdout = Buffer.from(text);
|
||||
const stderr = Buffer.from("E🙂Z");
|
||||
emitClosedStream(streams === 1 && text.includes("🙂") ? stdout.subarray(0, 2) : stdout, onStdout);
|
||||
emitClosedStream(streams === 1 ? stderr.subarray(0, 2) : stderr, onStderr!);
|
||||
},
|
||||
);
|
||||
sandbox.process.getSessionCommand.mockImplementation(async () => ({ exitCode: streams === 1 ? undefined : 0 }));
|
||||
mockGet.mockResolvedValue(sandbox);
|
||||
|
||||
const result = await plugin.definition.onEnvironmentExecute?.(streamExecParams({ timeoutMs: 2000 }));
|
||||
|
||||
expect(result).toMatchObject({ exitCode: 0, stdout: text, stderr: "E🙂Z" });
|
||||
for (const [stream, expected] of [["stdout", text], ["stderr", "E🙂Z"]]) {
|
||||
expect(executionLog.mock.calls.filter(([name]) => name === stream).map(([, chunk]) => chunk).join("")).toBe(expected);
|
||||
}
|
||||
expect(sandbox.process.executeSessionCommand).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
restore();
|
||||
}
|
||||
});
|
||||
|
||||
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();
|
||||
@@ -2507,7 +2892,7 @@ describe("Daytona sandbox provider plugin", () => {
|
||||
onStdout("partial");
|
||||
throw new Error("socket error");
|
||||
}
|
||||
return { stdout: "poll-out", stderr: "poll-err" };
|
||||
return { stdout: "partialpoll-out", stderr: "poll-err" };
|
||||
},
|
||||
);
|
||||
// The command still runs to its exit on the server, so the poll reads it.
|
||||
@@ -2518,7 +2903,7 @@ describe("Daytona sandbox provider plugin", () => {
|
||||
|
||||
// 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" });
|
||||
expect(result).toMatchObject({ exitCode: 9, timedOut: false, stdout: "partialpoll-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,
|
||||
@@ -2559,7 +2944,7 @@ describe("Daytona sandbox provider plugin", () => {
|
||||
|
||||
// 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);
|
||||
expect(sandbox.process.getSessionCommandLogs.mock.calls.filter((call) => call[2])).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("emits each new chunk to the host and drops a replayed prefix (test_log_stream_emits_execute_log_per_chunk)", async () => {
|
||||
|
||||
@@ -1772,10 +1772,9 @@ async function executeOneShot(
|
||||
}
|
||||
}
|
||||
|
||||
// Poll interval for a session command's exit code. The live spike measured a
|
||||
// session command resolving in about 260-300 ms, so a short interval keeps the
|
||||
// poll responsive without a busy loop.
|
||||
const SESSION_POLL_INTERVAL_MS = 50;
|
||||
// Fallback reads full log snapshots as well as status. Limit the heavier
|
||||
// polling path while keeping output available to interactive commands.
|
||||
const SESSION_LOG_POLL_INTERVAL_MS = 1000;
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
@@ -1787,10 +1786,10 @@ function sleep(ms: number): Promise<void> {
|
||||
// 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
|
||||
// A bounded reconnect for the log stream. The SDK resolves on WebSocket close,
|
||||
// which does not prove that the command exited. A 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;
|
||||
|
||||
@@ -1812,19 +1811,39 @@ const MAX_SESSION_STREAM_RECONNECTS = 1;
|
||||
function createSessionStreamBuffer(
|
||||
onNewTail?: (stream: "stdout" | "stderr", text: string) => void,
|
||||
) {
|
||||
type Stream = {
|
||||
chunks: Buffer[];
|
||||
length: number;
|
||||
connectionBytes: number;
|
||||
pendingReplacement: string;
|
||||
previousReplacement: { offset: number; text: string } | null;
|
||||
};
|
||||
const createStream = (): Stream => ({
|
||||
chunks: [], length: 0, connectionBytes: 0, pendingReplacement: "", previousReplacement: null,
|
||||
});
|
||||
const streams = {
|
||||
stdout: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 },
|
||||
stderr: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 },
|
||||
stdout: createStream(),
|
||||
stderr: createStream(),
|
||||
};
|
||||
|
||||
function append(
|
||||
streamName: "stdout" | "stderr",
|
||||
stream: { chunks: Buffer[]; length: number; connectionBytes: number },
|
||||
stream: Stream,
|
||||
chunk: string,
|
||||
final = false,
|
||||
publish = true,
|
||||
): void {
|
||||
const buf = Buffer.from(chunk, "utf8");
|
||||
// The SDK flushes an incomplete UTF-8 character as U+FFFD on socket close.
|
||||
// Do not publish that suffix or count its replacement bytes until a replay
|
||||
// confirms it. A real final U+FFFD is flushed at confirmed command exit.
|
||||
const text = stream.pendingReplacement + chunk;
|
||||
stream.pendingReplacement = final ? "" : (text.match(/\uFFFD+$/u)?.[0] ?? "");
|
||||
const buf = Buffer.from(text.slice(0, text.length - stream.pendingReplacement.length), "utf8");
|
||||
const start = stream.connectionBytes;
|
||||
stream.connectionBytes = start + buf.length;
|
||||
if (stream.previousReplacement && stream.connectionBytes > stream.previousReplacement.offset) {
|
||||
stream.previousReplacement = null;
|
||||
}
|
||||
// The whole chunk falls before the delivered byte count, so it is a replay.
|
||||
if (start + buf.length <= stream.length) {
|
||||
return;
|
||||
@@ -1837,7 +1856,7 @@ function createSessionStreamBuffer(
|
||||
stream.length += tail.length;
|
||||
// Deliver only the genuinely new tail to the live sink, so a replayed
|
||||
// prefix on a reconnect never reaches the host twice.
|
||||
if (onNewTail && tail.length > 0) {
|
||||
if (publish && onNewTail && tail.length > 0) {
|
||||
onNewTail(streamName, tail.toString("utf8"));
|
||||
}
|
||||
}
|
||||
@@ -1848,8 +1867,34 @@ function createSessionStreamBuffer(
|
||||
// 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;
|
||||
for (const stream of Object.values(streams)) {
|
||||
if (stream.pendingReplacement &&
|
||||
(!stream.previousReplacement || stream.connectionBytes > stream.previousReplacement.offset ||
|
||||
(stream.connectionBytes === stream.previousReplacement.offset &&
|
||||
stream.pendingReplacement.length > stream.previousReplacement.text.length))) {
|
||||
stream.previousReplacement = { offset: stream.connectionBytes, text: stream.pendingReplacement };
|
||||
}
|
||||
stream.pendingReplacement = "";
|
||||
stream.connectionBytes = 0;
|
||||
}
|
||||
},
|
||||
finish(publish = true): void {
|
||||
for (const name of ["stdout", "stderr"] as const) {
|
||||
const stream = streams[name];
|
||||
if (stream.previousReplacement?.offset === stream.connectionBytes &&
|
||||
stream.previousReplacement.text.length > stream.pendingReplacement.length) {
|
||||
stream.pendingReplacement = stream.previousReplacement.text;
|
||||
}
|
||||
append(name, stream, "", true, publish);
|
||||
// A short terminal snapshot must not discard a genuine trailing
|
||||
// replacement character already received from a longer stream.
|
||||
const previous = stream.previousReplacement;
|
||||
if (previous && previous.offset >= stream.length) {
|
||||
stream.connectionBytes = previous.offset;
|
||||
append(name, stream, previous.text, true, publish);
|
||||
}
|
||||
stream.previousReplacement = null;
|
||||
}
|
||||
},
|
||||
get stdout(): string {
|
||||
return Buffer.concat(streams.stdout.chunks).toString("utf8");
|
||||
@@ -1860,34 +1905,85 @@ function createSessionStreamBuffer(
|
||||
};
|
||||
}
|
||||
|
||||
type SessionLogStreamResult =
|
||||
| { ok: true; stdout: string; stderr: string }
|
||||
| { ok: false };
|
||||
type SessionLogStreamResult = {
|
||||
exitCode: number | null;
|
||||
closedStream: boolean;
|
||||
buffer: ReturnType<typeof createSessionStreamBuffer>;
|
||||
};
|
||||
|
||||
class SessionCommandDeadlineError extends Error {}
|
||||
class SessionObservationTimeoutError extends Error {}
|
||||
|
||||
async function beforeSessionDeadline<T>(deadlineMs: number, action: () => Promise<T>): Promise<T> {
|
||||
const remainingMs = deadlineMs - Date.now();
|
||||
if (remainingMs <= 0) throw new SessionCommandDeadlineError();
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
action(),
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => reject(new SessionCommandDeadlineError()), remainingMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer !== undefined) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
// 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> {
|
||||
try {
|
||||
return await beforeSessionDeadline(Date.now() + timeoutMs, action);
|
||||
} catch (error) {
|
||||
if (error instanceof SessionCommandDeadlineError || error instanceof DaytonaTimeoutError) {
|
||||
throw new SessionObservationTimeoutError();
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
// The stream buffer drops a replayed prefix by byte offset on a reconnect.
|
||||
// Both a rejected stream and a clean close without a confirmed process exit
|
||||
// use bounded reconnects, then polling.
|
||||
async function runSessionLogStream(
|
||||
sandbox: Sandbox,
|
||||
sessionId: string,
|
||||
commandId: string,
|
||||
onNewTail?: (stream: "stdout" | "stderr", text: string) => void,
|
||||
timeoutMs: number,
|
||||
buffer: ReturnType<typeof createSessionStreamBuffer>,
|
||||
): Promise<SessionLogStreamResult> {
|
||||
const buffer = createSessionStreamBuffer(onNewTail);
|
||||
let reconnects = 0;
|
||||
let closedStream = false;
|
||||
while (true) {
|
||||
let streamClosed = false;
|
||||
let acceptingOutput = true;
|
||||
try {
|
||||
await sandbox.process.getSessionCommandLogs(sessionId, commandId, buffer.onStdout, buffer.onStderr);
|
||||
return { ok: true, stdout: buffer.stdout, stderr: buffer.stderr };
|
||||
await sandbox.process.getSessionCommandLogs(
|
||||
sessionId,
|
||||
commandId,
|
||||
(chunk) => { if (acceptingOutput) buffer.onStdout(chunk); },
|
||||
(chunk) => { if (acceptingOutput) buffer.onStderr(chunk); },
|
||||
);
|
||||
streamClosed = true;
|
||||
closedStream = true;
|
||||
} catch {
|
||||
if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) {
|
||||
return { ok: false };
|
||||
}
|
||||
reconnects += 1;
|
||||
buffer.resetConnectionCursors();
|
||||
// The command can still be running after a log transport failure.
|
||||
} finally {
|
||||
// The SDK exposes no cancellation handle for this socket. Session
|
||||
// teardown owns closing it; late data cannot change a settled execution.
|
||||
acceptingOutput = false;
|
||||
}
|
||||
if (streamClosed) {
|
||||
const exitCode = await readSessionExitCode(sandbox, sessionId, commandId, timeoutMs);
|
||||
if (exitCode !== null) return { exitCode, closedStream, buffer };
|
||||
}
|
||||
if (reconnects >= MAX_SESSION_STREAM_RECONNECTS) {
|
||||
return { exitCode: null, closedStream, buffer };
|
||||
}
|
||||
reconnects += 1;
|
||||
buffer.resetConnectionCursors();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1899,14 +1995,15 @@ async function readSessionExitCode(
|
||||
sandbox: Sandbox,
|
||||
sessionId: string,
|
||||
commandId: string,
|
||||
timeoutMs: number,
|
||||
): Promise<number | null> {
|
||||
const first = await sandbox.process.getSessionCommand(sessionId, commandId);
|
||||
const first = await observeSessionCommand(timeoutMs, () => 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);
|
||||
const status = await observeSessionCommand(timeoutMs, () => sandbox.process.getSessionCommand(sessionId, commandId));
|
||||
if (typeof status.exitCode === "number") {
|
||||
return status.exitCode;
|
||||
}
|
||||
@@ -1948,6 +2045,8 @@ async function executeInSession(
|
||||
// the exec wall-time spent before the abort, so a slow command is still
|
||||
// attributed to the provider boundary.
|
||||
let execStart: number | null = null;
|
||||
let confirmedExitCode: number | null = null;
|
||||
const streamBuffer = createSessionStreamBuffer((stream, text) => pluginContext?.execution.log(stream, text));
|
||||
|
||||
try {
|
||||
if (stdinPath) {
|
||||
@@ -1975,8 +2074,8 @@ async function executeInSession(
|
||||
|
||||
// Log-stream path. A session command always tries the stream first: it
|
||||
// streams stdout and stderr from the callback log form, then reads 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.
|
||||
// code. A log socket closing without a recorded exit leaves the command
|
||||
// running, so reconnect and then poll the same command without replaying it.
|
||||
//
|
||||
// Emit each genuinely new output chunk to the host during the active execute
|
||||
// call. The host routes it to the runner log sink by the host-issued
|
||||
@@ -1986,71 +2085,101 @@ async function executeInSession(
|
||||
sandbox,
|
||||
sessionId,
|
||||
commandId,
|
||||
(stream, text) => pluginContext?.execution.log(stream, text),
|
||||
effectiveTimeoutMs,
|
||||
streamBuffer,
|
||||
);
|
||||
if (streamResult.ok) {
|
||||
const exitCode = await readSessionExitCode(sandbox, sessionId, commandId);
|
||||
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));
|
||||
streamBuffer.resetConnectionCursors();
|
||||
streamBuffer.onStdout(logs?.stdout ?? "");
|
||||
streamBuffer.onStderr(logs?.stderr ?? "");
|
||||
streamBuffer.finish();
|
||||
const durationMs = timingNow() - execStart;
|
||||
return {
|
||||
exitCode,
|
||||
exitCode: streamResult.exitCode,
|
||||
timedOut: false,
|
||||
stdout: streamResult.stdout,
|
||||
stderr: streamResult.stderr,
|
||||
stdout: streamResult.buffer.stdout,
|
||||
stderr: streamResult.buffer.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`. The poll path is the fallback when the log stream
|
||||
// fails.
|
||||
// Keep forwarding output while waiting for the actual command exit. ACP
|
||||
// peers may need a tool response before they can finish the command. Waiting
|
||||
// for exit before reading logs would strand those peers after a stream loss.
|
||||
// An EOF after a live stream does not end the command's lifetime. Bound
|
||||
// each recovery read independently; the caller still owns stop/teardown.
|
||||
// Preserve the legacy budget starting at fallback entry when both stream
|
||||
// attempts fail, including a stream that fails after running for hours.
|
||||
const deadlineMs = Date.now() + effectiveTimeoutMs;
|
||||
let exitCode: number | null = null;
|
||||
while (true) {
|
||||
const status = await sandbox.process.getSessionCommand(sessionId, commandId);
|
||||
if (typeof status.exitCode === "number") {
|
||||
exitCode = status.exitCode;
|
||||
break;
|
||||
const observe = async <T>(action: () => Promise<T>): Promise<T> => {
|
||||
try {
|
||||
return await (streamResult.closedStream
|
||||
? observeSessionCommand(effectiveTimeoutMs, action)
|
||||
: beforeSessionDeadline(deadlineMs, action));
|
||||
} catch (error) {
|
||||
if (error instanceof DaytonaTimeoutError) throw new SessionObservationTimeoutError();
|
||||
throw error;
|
||||
}
|
||||
if (Date.now() >= deadlineMs) {
|
||||
};
|
||||
while (true) {
|
||||
const status = await observe(() => sandbox.process.getSessionCommand(sessionId, commandId));
|
||||
if (typeof status.exitCode === "number") confirmedExitCode = status.exitCode;
|
||||
const logs = await observe(() => sandbox.process.getSessionCommandLogs(sessionId, commandId));
|
||||
streamResult.buffer.resetConnectionCursors();
|
||||
streamResult.buffer.onStdout(logs.stdout ?? "");
|
||||
streamResult.buffer.onStderr(logs.stderr ?? "");
|
||||
if (confirmedExitCode !== null) {
|
||||
streamResult.buffer.finish();
|
||||
const durationMs = timingNow() - execStart;
|
||||
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.`
|
||||
: `Command timed out after ${Math.round(effectiveTimeoutMs / 1000)} s.`;
|
||||
return {
|
||||
exitCode: null,
|
||||
timedOut: true,
|
||||
stdout: "",
|
||||
stderr: `${timeoutMessage}\n`,
|
||||
exitCode: confirmedExitCode,
|
||||
timedOut: false,
|
||||
stdout: streamResult.buffer.stdout,
|
||||
stderr: streamResult.buffer.stderr,
|
||||
metadata: { durationMs },
|
||||
};
|
||||
}
|
||||
await sleep(SESSION_POLL_INTERVAL_MS);
|
||||
// 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.closedStream ? SESSION_LOG_POLL_INTERVAL_MS
|
||||
: Math.max(0, Math.min(SESSION_LOG_POLL_INTERVAL_MS, deadlineMs - Date.now())));
|
||||
}
|
||||
|
||||
// Read true, separated stdout and stderr from the logs endpoint. The
|
||||
// synchronous dispatch response fields are optional, so the logs endpoint is
|
||||
// the source of truth.
|
||||
const logs = await sandbox.process.getSessionCommandLogs(sessionId, commandId);
|
||||
const durationMs = timingNow() - execStart;
|
||||
return {
|
||||
exitCode,
|
||||
timedOut: false,
|
||||
stdout: logs.stdout ?? "",
|
||||
stderr: logs.stderr ?? "",
|
||||
metadata: { durationMs },
|
||||
};
|
||||
} catch (error) {
|
||||
if (error instanceof DaytonaTimeoutError) {
|
||||
if (error instanceof SessionObservationTimeoutError) {
|
||||
// Preserve undecided decoder suffixes in the partial result, without
|
||||
// injecting a possibly incomplete character into the live protocol.
|
||||
streamBuffer.finish(false);
|
||||
return {
|
||||
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`,
|
||||
metadata: {
|
||||
...(execStart != null ? { durationMs: timingNow() - execStart } : {}),
|
||||
timeoutScope: "session_log_observation",
|
||||
commandExitConfirmed: confirmedExitCode !== null,
|
||||
...(confirmedExitCode !== null ? { observedExitCode: confirmedExitCode } : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
if (error instanceof DaytonaTimeoutError || error instanceof SessionCommandDeadlineError) {
|
||||
streamBuffer.finish(false);
|
||||
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.message.trim();
|
||||
: error instanceof SessionCommandDeadlineError
|
||||
? `Command timed out after ${Math.round(effectiveTimeoutMs / 1000)} s.`
|
||||
: error.message.trim();
|
||||
const durationMs = execStart != null ? timingNow() - execStart : undefined;
|
||||
return {
|
||||
exitCode: null,
|
||||
timedOut: true,
|
||||
stdout: "",
|
||||
stderr: `${timeoutMessage}\n`,
|
||||
stdout: streamBuffer.stdout,
|
||||
stderr: `${streamBuffer.stderr}${timeoutMessage}\n`,
|
||||
...(durationMs != null ? { metadata: { durationMs } } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user