diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index a43c2543e1..9a5a613a63 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -1692,6 +1692,13 @@ async function buildRuntime(input: { executionTarget.transport === "sandbox" && Boolean(executionTarget.runner) && Boolean(agentCommandShell); + // Stream the agent output through the persistent session log stream instead of + // the host output-file poll. Default OFF; an operator opts a sandbox + // environment in through the environment config. + const streamAgentSessionOutput = + executionTarget?.kind === "remote" && + executionTarget.transport === "sandbox" && + executionTarget.streamAgentSessionOutput === true; // The ACP `session/new` cwd and every cwd-keyed session-state site // (fingerprint, compat, persist, ensureSession, error) bind to THIS single // value so a warm/resumable session created with the in-sandbox cwd is reused @@ -2004,6 +2011,7 @@ async function buildRuntime(input: { onLog: input.ctx.onLog, getRuntimeParentContext: input.getRuntimeParentContext, runtimeSpan: input.runtimeSpan, + streamOutputViaSession: streamAgentSessionOutput, }), concurrentBridgeStepMetrics, ); diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 823766ee13..5688b4f194 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -34,6 +34,27 @@ export interface CommandManagedRuntimeRunner { timeoutMs?: number; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + /** + * Run this command through the lease's persistent session even when no run + * step is active. A sandbox provider opens the session on the first + * non-bypassed command; the ACP process session bridge sets this so the + * long-lived agent command streams its output through the session log + * stream. The default keeps the context-based session selection. + */ + useSession?: boolean; + /** + * Run this command outside the lease's persistent session even when a run + * step is active. The persistent session is a single serialized shell. In + * streamed mode the agent runs as one long-lived foreground command that + * holds the session for the whole run. The bridge control-plane execs + * (input delivery, output read, callback relay, and the queue/setup + * bookkeeping) must run concurrently with the agent, so they run as + * independent one-shot commands. On the session they queue behind the agent + * command that never returns — a permanent deadlock. An explicit bypass + * always wins over the context-based session selection and over + * `useSession`. The default keeps the context-based session selection. + */ + bypassSession?: boolean; }): Promise; /** * Optional native inbound file transfer. Present only when the sandbox diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index d4425a6ddf..db7eeeda20 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -901,6 +901,344 @@ describe("sandbox adapter execution targets", () => { } }); + describe("streamed output (streamOutputViaSession)", () => { + it("bridges bidirectional sessions when the wrapper streams output to stdout", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-stream-echo-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "echo-acp-child.mjs"); + await writeFile( + childPath, + [ + "process.stdin.on('data', (chunk) => {", + " process.stdout.write('out:' + chunk.toString());", + " process.stderr.write('err:' + chunk.toString());", + "});", + ].join("\n"), + "utf8", + ); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner: createLocalSandboxRunner(), + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stream-echo", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + streamOutputViaSession: true, + }); + expect(bridge).not.toBeNull(); + + try { + const result = await runProxyWithInput(bridge!.agentCommand, "hello\n"); + expect(result.code).toBe(0); + expect(result.stdout).toBe("out:hello\n"); + expect(result.stderr).toBe("err:hello\n"); + } finally { + await bridge?.stop(); + } + }); + + it("buffers streamed output until the local proxy connects", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-stream-buffer-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "fast-stream-child.mjs"); + await writeFile( + childPath, + [ + "process.stdout.write('early-out\\n');", + "process.stderr.write('early-err\\n');", + "setTimeout(() => process.exit(0), 20);", + ].join("\n"), + "utf8", + ); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner: createLocalSandboxRunner(), + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stream-buffer", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + streamOutputViaSession: true, + }); + expect(bridge).not.toBeNull(); + + try { + await new Promise((resolve) => setTimeout(resolve, 300)); + const result = await runProxyWithInput(bridge!.agentCommand, ""); + expect(result.code).toBe(0); + // The seq guard delivers the early output exactly once even though the + // live stream and the terminal result both carry it. + expect(result.stdout).toBe("early-out\n"); + expect(result.stderr).toBe("early-err\n"); + } finally { + await bridge?.stop(); + } + }); + + it("delivers full streamed output when the sandbox child exits immediately", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-stream-fast-exit-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "instant-stream-child.mjs"); + await writeFile( + childPath, + [ + "process.stdout.write('final-out\\n');", + "process.stderr.write('final-err\\n');", + ].join("\n"), + "utf8", + ); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner: createLocalSandboxRunner(), + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stream-fast-exit", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + streamOutputViaSession: true, + }); + expect(bridge).not.toBeNull(); + + try { + const result = await runProxyWithInput(bridge!.agentCommand, ""); + expect(result.code).toBe(0); + expect(result.stdout).toBe("final-out\n"); + expect(result.stderr).toBe("final-err\n"); + } finally { + await bridge?.stop(); + } + }); + + it("streams live output before the child exits and never writes output event files", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-stream-live-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "live-stream-child.mjs"); + await writeFile( + childPath, + [ + "process.stdin.setEncoding('utf8');", + "process.stdin.on('data', (chunk) => {", + " if (chunk.includes('ping')) {", + " process.stdout.write('delta:ping\\n');", + " process.stderr.write('trace:ping\\n');", + " }", + " if (chunk.includes('finish')) process.exit(0);", + "});", + "process.stdin.resume();", + ].join("\n"), + "utf8", + ); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner: createLocalSandboxRunner(), + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stream-live", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + streamOutputViaSession: true, + }); + expect(bridge).not.toBeNull(); + + const child = spawn(bridge!.agentCommand, [], { stdio: ["pipe", "pipe", "pipe"] }); + let stdout = ""; + let stderr = ""; + let exited = false; + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + stdout += chunk; + }); + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + const exitPromise = new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + child.kill("SIGKILL"); + reject(new Error("Timed out waiting for streamed process session proxy.")); + }, 5000); + child.on("error", (error) => { + clearTimeout(timeout); + reject(error); + }); + child.on("exit", (exitCode) => { + exited = true; + clearTimeout(timeout); + resolve(exitCode); + }); + }); + + try { + child.stdin.write("ping\n"); + await waitForCondition( + () => stdout.includes("delta:ping\n") && stderr.includes("trace:ping\n"), + "Timed out waiting for live streamed process session output.", + 3000, + ); + expect(exited).toBe(false); + + child.stdin.end("finish\n"); + await expect(exitPromise).resolves.toBe(0); + + // The streamed path uses the stdout wrapper, not the output-file poll, so + // no `events` directory is ever created under the session runtime tree. + const hasEventsDir = await readdir( + path.posix.join(rootDir, ".paperclip-runtime", "acpx", "process-sessions"), + { withFileTypes: true, recursive: true }, + ) + .then((entries) => entries.some((entry) => entry.isDirectory() && entry.name === "events")) + .catch(() => false); + expect(hasEventsDir).toBe(false); + } finally { + if (!exited) { + child.kill("SIGKILL"); + await exitPromise.catch(() => undefined); + } + await bridge?.stop(); + } + }); + + it("keeps the agent command on the persistent session and forces bridge control execs off it", async () => { + // Regression guard for the streamed-mode startup deadlock. The persistent + // session is one serialized shell. In streamed mode the agent runs as a + // long-lived foreground session command that holds the session for the + // whole run. The bridge control-plane execs (script sync, stdin delivery, + // teardown) must run concurrently with the agent, so each must force + // itself off the session. On the session they queue behind the agent + // command that never returns, and the first handshake write never drains. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-stream-isolation-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "echo-acp-child.mjs"); + await writeFile( + childPath, + [ + "process.stdin.on('data', (chunk) => {", + " process.stdout.write('out:' + chunk.toString());", + "});", + ].join("\n"), + "utf8", + ); + + const delegate = createLocalSandboxRunner(); + const execs: Array<{ useSession?: boolean; bypassSession?: boolean; script: string }> = []; + const runner = { + execute: vi.fn( + async ( + input: Parameters[0] & { + useSession?: boolean; + bypassSession?: boolean; + }, + ) => { + execs.push({ + useSession: input.useSession, + bypassSession: input.bypassSession, + script: input.args?.[1] ?? "", + }); + return delegate.execute(input); + }, + ), + }; + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner, + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stream-isolation", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + streamOutputViaSession: true, + }); + expect(bridge).not.toBeNull(); + + try { + // Round-trip one input so a stdin-delivery control exec runs and gets + // recorded before the assertions below. + const result = await runProxyWithInput(bridge!.agentCommand, "hello\n"); + expect(result.stdout).toBe("out:hello\n"); + + // Exactly one exec runs on the persistent session: the long-lived agent + // command. It streams its output through the session log stream, so it + // must not also bypass the session. + const sessionExecs = execs.filter((exec) => exec.useSession === true); + expect(sessionExecs).toHaveLength(1); + expect(sessionExecs[0]!.bypassSession).not.toBe(true); + expect(sessionExecs[0]!.script).toContain("node "); + + // Every other exec is bridge control-plane plumbing. Each must force + // itself off the persistent session so it never queues behind the agent + // command that holds it. + const controlExecs = execs.filter((exec) => exec.useSession !== true); + expect(controlExecs.length).toBeGreaterThan(0); + for (const exec of controlExecs) { + expect(exec.bypassSession).toBe(true); + } + } finally { + await bridge?.stop(); + } + }); + }); + it("applies the remote sandbox fallback when adapter timeoutSec is unset", () => { const sandboxTarget: AdapterSandboxExecutionTarget = { kind: "remote", diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 5e7aa5719f..90989c6f9f 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -106,6 +106,13 @@ export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWor * set to `false` to explicitly opt out back to batch-at-end delivery. */ streamRunLogs?: boolean | null; + /** + * Stream the interactive ACP agent output through the persistent session log + * stream instead of the host-side output-file poll. The process session + * bridge runs the agent as one long-lived session command and reads its + * output frames from the stream. Default OFF: the bridge keeps the poll path. + */ + streamAgentSessionOutput?: boolean | null; } export type AdapterExecutionTarget = @@ -1278,6 +1285,10 @@ async function readBridgeForwardResponseBody(response: Response, maxBodyBytes: n const PROCESS_SESSION_PROXY_SCRIPT = "paperclip-process-session-proxy.mjs"; const PROCESS_SESSION_REMOTE_SCRIPT = "paperclip-process-session-remote.mjs"; +// The streamed variant writes its output frames to stdout, so it rides a +// separate remote path. A sandbox can hold both scripts without the content +// hash-skip gate thrashing when a run switches output mode. +const PROCESS_SESSION_REMOTE_STREAM_SCRIPT = "paperclip-process-session-remote-stream.mjs"; const PROCESS_SESSION_AUTH_TIMEOUT_MS = 5_000; function jsonLine(value: unknown): string { @@ -1309,13 +1320,14 @@ async function syncProcessSessionRemoteScript(input: { remoteScriptPath: string; timeoutMs?: number | null; shellCommand?: "bash" | "sh" | null; + outputToStdout?: boolean; }): Promise<{ uploaded: boolean }> { const { uploaded } = await syncRemoteTextFileWithHashSkip({ runner: input.runner, remoteCwd: input.remoteCwd, remoteDir: input.remoteScriptDir, remotePath: input.remoteScriptPath, - body: getProcessSessionRemoteSource(), + body: getProcessSessionRemoteSource({ outputToStdout: input.outputToStdout === true }), label: "Process session remote script", action: "sync process session remote script", lockDir: path.posix.join(input.remoteScriptDir, ".paperclip-process-session-script.lock"), @@ -1391,6 +1403,11 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { // group under one wrapper span. When it is absent, the work runs under the run // parent with no wrapper span, exactly like the earlier behavior. runtimeSpan?: RuntimeSpanRunner; + // Stream the agent output through the persistent session log stream instead of + // the host output-file poll. When true, the bridge runs the wrapper as one + // long-lived session command and reads its stdout frames from the stream, and + // it does not start the 100 ms poll. Default OFF: the bridge keeps the poll. + streamOutputViaSession?: boolean; }): Promise { if (!input.target || input.target.kind !== "remote" || input.target.transport !== "sandbox") { return null; @@ -1420,7 +1437,14 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { const sessionDir = path.posix.join(bridgeRuntimeDir, sessionId); const stdinDir = path.posix.join(sessionDir, "stdin"); const eventsDir = path.posix.join(sessionDir, "events"); - const remoteScriptPath = path.posix.join(bridgeRuntimeDir, PROCESS_SESSION_REMOTE_SCRIPT); + // The streamed wrapper writes its frames to stdout and rides a separate remote + // path, so a warm sandbox can hold both wrapper scripts without the content + // hash-skip gate thrashing when a run switches output mode. + const streamOutput = input.streamOutputViaSession === true; + const remoteScriptPath = path.posix.join( + bridgeRuntimeDir, + streamOutput ? PROCESS_SESSION_REMOTE_STREAM_SCRIPT : PROCESS_SESSION_REMOTE_SCRIPT, + ); const client = createCommandManagedSandboxCallbackBridgeQueueClient({ runner, remoteCwd: target.remoteCwd, @@ -1438,6 +1462,7 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { remoteScriptPath, timeoutMs, shellCommand, + outputToStdout: streamOutput, }); // Resolve the launch env AFTER the env-independent setup above, so a caller @@ -1451,26 +1476,34 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { env: sanitizeRemoteExecutionEnv(launchEnv), }), "utf8").toString("base64"); - await onLog("stdout", `[paperclip] Starting ACP process session bridge in sandbox (${target.providerKey ?? "provider"}).\n`); - const startResult = await runner.execute({ - command: shellCommand, - args: shellCommandArgs( - [ - `mkdir -p ${shellQuote(stdinDir)} ${shellQuote(eventsDir)}`, - `PAPERCLIP_PROCESS_SESSION_DIR=${shellQuote(sessionDir)} ` + - `PAPERCLIP_PROCESS_SESSION_COMMAND_B64=${shellQuote(commandPayload)} ` + - `nohup node ${shellQuote(remoteScriptPath)} >/dev/null 2>&1 < /dev/null &`, - "printf '%s\\n' \"$!\"", - ].join("\n"), - ), - cwd: target.remoteCwd, - env: { - PAPERCLIP_SANDBOX_EXEC_CHANNEL: "bridge", - }, - timeoutMs, - }); - if (startResult.timedOut || (startResult.exitCode ?? 1) !== 0) { - throw new Error(`Failed to start sandbox ACP process session bridge: ${startResult.stderr || startResult.stdout}`); + // Legacy poll path: background the wrapper with `nohup` and read its output + // event files with the host poll below. The streamed path launches the wrapper + // as one foreground session command further down instead, so skip this. + if (!streamOutput) { + await onLog("stdout", `[paperclip] Starting ACP process session bridge in sandbox (${target.providerKey ?? "provider"}).\n`); + const startResult = await runner.execute({ + command: shellCommand, + args: shellCommandArgs( + [ + `mkdir -p ${shellQuote(stdinDir)} ${shellQuote(eventsDir)}`, + `PAPERCLIP_PROCESS_SESSION_DIR=${shellQuote(sessionDir)} ` + + `PAPERCLIP_PROCESS_SESSION_COMMAND_B64=${shellQuote(commandPayload)} ` + + `nohup node ${shellQuote(remoteScriptPath)} >/dev/null 2>&1 < /dev/null &`, + "printf '%s\\n' \"$!\"", + ].join("\n"), + ), + cwd: target.remoteCwd, + env: { + PAPERCLIP_SANDBOX_EXEC_CHANNEL: "bridge", + }, + timeoutMs, + // The wrapper launch is bridge plumbing. Keep it off the persistent + // session so it never queues behind an in-run session command. + bypassSession: true, + }); + if (startResult.timedOut || (startResult.exitCode ?? 1) !== 0) { + throw new Error(`Failed to start sandbox ACP process session bridge: ${startResult.stderr || startResult.stdout}`); + } } let socket: net.Socket | null = null; @@ -1640,7 +1673,105 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { const port = await waitForLocalServerListen(server); const agentCommand = await writeProcessSessionProxyScript(proxyDir, port, token); - schedulePoll(); + + if (streamOutput) { + // Streamed output path. Run the wrapper as one long-lived session command; + // its stdout carries newline-delimited JSON frames that reach the host + // through the provider session log stream. Deliver each frame exactly once + // by its monotonic `seq`, so a frame that arrives both live and in the final + // result is not repeated. There is no host output-file poll here. + let streamBuffer = ""; + let lastSeq = 0; + let sawTerminal = false; + const deliverFrame = (frame: (typeof pendingRemoteEvents)[number] & { seq?: number }) => { + if (typeof frame.seq === "number") { + if (frame.seq <= lastSeq) return; + lastSeq = frame.seq; + } + if (frame.type === "exit" || frame.type === "error") sawTerminal = true; + deliverRemoteEvent(frame); + }; + const parseFrameLine = (line: string) => { + if (!line.trim()) return; + let frame: (typeof pendingRemoteEvents)[number] & { seq?: number }; + try { + frame = JSON.parse(line) as typeof frame; + } catch { + return; + } + deliverFrame(frame); + }; + // Live delivery: buffer partial lines across stream chunks, deliver each + // complete frame line as it arrives. + const ingestStreamChunk = (text: string) => { + streamBuffer += text; + const split = splitJsonLines(streamBuffer); + streamBuffer = split.rest; + for (const line of split.lines) parseFrameLine(line); + }; + // Terminal delivery (the defined fallback to the poll): the resolved result + // carries the full wrapper stdout even when the live stream degraded to the + // provider session-log poll. The text is complete and self-contained, so + // re-parse it on its own; the `seq` guard drops every frame the live stream + // already delivered. Drop any partial live line — its complete form is in the + // full text. + const ingestFinalText = (text: string) => { + streamBuffer = ""; + for (const line of text.split(/\n/)) parseFrameLine(line); + }; + + const launchEnvForStream = + typeof input.env === "function" ? await input.env() : input.env; + const streamCommandPayload = Buffer.from(JSON.stringify({ + command: input.command, + args: input.args, + cwd: input.cwd || target.remoteCwd, + env: sanitizeRemoteExecutionEnv(launchEnvForStream), + }), "utf8").toString("base64"); + await onLog( + "stdout", + `[paperclip] Starting streamed ACP process session bridge in sandbox (${target.providerKey ?? "provider"}).\n`, + ); + // Fire the long-lived command; do NOT await it here. `useSession` forces the + // persistent session so the provider streams the wrapper stdout back through + // `onLog`. On resolve, the terminal re-parse fills any frames the live stream + // missed; on reject, deliver one error frame so the local proxy fails loud. + void runner + .execute({ + command: shellCommand, + args: shellCommandArgs(`node ${shellQuote(remoteScriptPath)}`), + cwd: target.remoteCwd, + env: { + PAPERCLIP_PROCESS_SESSION_DIR: sessionDir, + PAPERCLIP_PROCESS_SESSION_COMMAND_B64: streamCommandPayload, + PAPERCLIP_SANDBOX_EXEC_CHANNEL: "bridge", + }, + timeoutMs, + useSession: true, + onLog: async (stream, chunk) => { + if (stream === "stdout") ingestStreamChunk(chunk); + }, + }) + .then((result) => { + ingestFinalText(result.stdout); + if (!sawTerminal && !stopping) { + deliverRemoteEvent({ + type: "exit", + code: typeof result.exitCode === "number" ? result.exitCode : null, + }); + } + }) + .catch((error) => { + if (!stopping) { + deliverRemoteEvent({ + type: "error", + message: error instanceof Error ? error.message : String(error), + }); + } + }); + } else { + schedulePoll(); + } return { agentCommand, @@ -1706,7 +1837,94 @@ socket.on("close", () => { `; } -function getProcessSessionRemoteSource(): string { +function getProcessSessionRemoteSource(input?: { outputToStdout?: boolean }): string { + return input?.outputToStdout === true + ? getProcessSessionRemoteStreamSource() + : getProcessSessionRemoteEventFileSource(); +} + +// The shared stdin drain. Both wrappers read newline-delimited stdin messages +// from the stdin file queue and write them to the child, then end the child +// stdin on `stdinEnd`. A write to a closed child stdin only emits an `error` +// event, so the wrapper installs a no-op handler at the call site. +const PROCESS_SESSION_STDIN_POLL_TAIL = `child.stdin.on("error", () => {}); + +async function pollStdin() { + while (!stdinClosed) { + const entries = (await fs.readdir(stdinDir).catch(() => [])).filter((name) => name.endsWith(".json")).sort(); + for (const name of entries) { + const file = path.posix.join(stdinDir, name); + const raw = await fs.readFile(file, "utf8").catch(() => null); + await fs.rm(file, { force: true }).catch(() => undefined); + if (!raw) continue; + const message = JSON.parse(raw); + if (message.type === "stdin" && typeof message.data === "string") { + if (!stdinClosed) child.stdin.write(Buffer.from(message.data, "base64")); + } else if (message.type === "stdinEnd") { + stdinClosed = true; + child.stdin.end(); + break; + } + } + if (!stdinClosed) await new Promise((resolve) => setTimeout(resolve, 50)); + } +} + +void pollStdin().catch((error) => void writeEvent({ type: "error", message: error instanceof Error ? error.message : String(error) })); +`; + +// Streamed variant: the wrapper writes each output frame as one newline- +// delimited JSON line to its stdout. The host runs this wrapper as one +// long-lived session command and reads the frames from the session log stream, +// so there is no host output-file poll. Each frame carries a monotonic `seq`, +// so the host delivers every frame exactly once whether it arrives live or in +// the final result. The wrapper exits when the child closes, so the session +// command settles and the session shell (the subshell wrap around it) survives. +function getProcessSessionRemoteStreamSource(): string { + return `import { spawn } from "node:child_process"; +import { promises as fs } from "node:fs"; +import path from "node:path"; + +const sessionDir = process.env.PAPERCLIP_PROCESS_SESSION_DIR; +const commandPayload = process.env.PAPERCLIP_PROCESS_SESSION_COMMAND_B64; +if (!sessionDir || !commandPayload) throw new Error("Missing process session bridge env."); + +const stdinDir = path.posix.join(sessionDir, "stdin"); +let seq = 0; +let stdinClosed = false; + +const config = JSON.parse(Buffer.from(commandPayload, "base64").toString("utf8")); +await fs.mkdir(stdinDir, { recursive: true }); + +// One newline-delimited JSON frame per event. Node keeps process.stdout writes +// ordered, and the base64 payload holds no newline, so each frame is one line. +function writeEvent(event) { + seq += 1; + process.stdout.write(JSON.stringify({ seq, ...event }) + "\\n"); +} + +const child = spawn(config.command, Array.isArray(config.args) ? config.args : [], { + cwd: config.cwd || process.cwd(), + env: { ...process.env, ...(config.env || {}) }, + stdio: ["pipe", "pipe", "pipe"], +}); + +child.stdout.on("data", (chunk) => writeEvent({ type: "data", stream: "stdout", data: Buffer.from(chunk).toString("base64") })); +child.stderr.on("data", (chunk) => writeEvent({ type: "data", stream: "stderr", data: Buffer.from(chunk).toString("base64") })); +child.on("error", (error) => writeEvent({ type: "error", message: error.message })); +// "close" (not "exit") so stdout/stderr fully drain before the exit frame. +// Stop the stdin poll and set the exit code, then let the event loop drain: a +// natural exit flushes the stdout pipe, so the exit frame always lands. +child.on("close", (code, signal) => { + writeEvent({ type: "exit", code, signal }); + stdinClosed = true; + process.exitCode = typeof code === "number" ? code : 1; +}); + +${PROCESS_SESSION_STDIN_POLL_TAIL}`; +} + +function getProcessSessionRemoteEventFileSource(): string { return `import { spawn } from "node:child_process"; import { promises as fs } from "node:fs"; import path from "node:path"; @@ -1750,29 +1968,7 @@ child.on("error", (error) => void writeEvent({ type: "error", message: error.mes // the write chain then guarantees the exit file lands after every data file. child.on("close", (code, signal) => void writeEvent({ type: "exit", code, signal })); -async function pollStdin() { - while (!stdinClosed) { - const entries = (await fs.readdir(stdinDir).catch(() => [])).filter((name) => name.endsWith(".json")).sort(); - for (const name of entries) { - const file = path.posix.join(stdinDir, name); - const raw = await fs.readFile(file, "utf8").catch(() => null); - await fs.rm(file, { force: true }).catch(() => undefined); - if (!raw) continue; - const message = JSON.parse(raw); - if (message.type === "stdin" && typeof message.data === "string") { - child.stdin.write(Buffer.from(message.data, "base64")); - } else if (message.type === "stdinEnd") { - stdinClosed = true; - child.stdin.end(); - break; - } - } - if (!stdinClosed) await new Promise((resolve) => setTimeout(resolve, 50)); - } -} - -void pollStdin().catch((error) => void writeEvent({ type: "error", message: error instanceof Error ? error.message : String(error) })); -`; +${PROCESS_SESSION_STDIN_POLL_TAIL}`; } export async function startAdapterExecutionTargetPaperclipBridge(input: { diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index bd5a83ab5f..35ee47db70 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -236,6 +236,13 @@ async function runShell( }, timeoutMs, stdin, + // Every command that rides this helper is bridge control-plane plumbing: + // input delivery, output read, callback relay, and queue/setup bookkeeping. + // It must run concurrently with the agent, so force it off the persistent + // session. In streamed mode the agent holds that single serialized session + // for the whole run; a control write on the same session queues behind the + // agent command that never returns — a permanent deadlock. + bypassSession: true, }); } diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 7b2eb020c2..61d064961a 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -1668,6 +1668,53 @@ describe("Daytona sandbox provider plugin", () => { expect(result).toMatchObject({ exitCode: 0, timedOut: false, stdout: "AAABBB", stderr: "EEEFFF" }); expect(sandbox.process.getSessionCommandLogs).toHaveBeenCalledTimes(2); }); + + it("emits each new chunk to the host and drops a replayed prefix (test_log_stream_emits_execute_log_per_chunk)", 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 attempt = 0; + sandbox.process.getSessionCommandLogs.mockImplementation( + async ( + _sid: string, + _cmdId: string, + onStdout?: (chunk: string) => void, + onStderr?: (chunk: string) => void, + ) => { + attempt += 1; + if (attempt === 1) { + onStdout?.("AAA"); + onStderr?.("EEE"); + throw new Error("socket error"); + } + // Reconnect replays the whole log from byte 0, then the new tail. + onStdout?.("AAA"); + onStdout?.("BBB"); + onStderr?.("EEE"); + onStderr?.("FFF"); + }, + ); + sandbox.process.getSessionCommand.mockResolvedValue({ id: "cmd-1", command: "", exitCode: 0 }); + mockGet.mockResolvedValue(sandbox); + + await plugin.definition.onEnvironmentExecute?.(streamExecParams()); + + // Each genuinely new chunk reaches the host exactly once. The replayed + // prefix ("AAA"/"EEE") on the reconnect is not re-emitted. + expect(executionLog.mock.calls).toEqual([ + ["stdout", "AAA"], + ["stderr", "EEE"], + ["stdout", "BBB"], + ["stderr", "FFF"], + ]); + } finally { + restore(); + } + }); }); }); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index b760f79b34..9b8e160898 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -1508,13 +1508,16 @@ const MAX_SESSION_STREAM_RECONNECTS = 1; // 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() { +function createSessionStreamBuffer( + onNewTail?: (stream: "stdout" | "stderr", text: string) => void, +) { const streams = { stdout: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 }, stderr: { chunks: [] as Buffer[], length: 0, connectionBytes: 0 }, }; function append( + streamName: "stdout" | "stderr", stream: { chunks: Buffer[]; length: number; connectionBytes: number }, chunk: string, ): void { @@ -1531,11 +1534,16 @@ function createSessionStreamBuffer() { const tail = start >= stream.length ? buf : buf.subarray(stream.length - start); stream.chunks.push(tail); 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) { + onNewTail(streamName, tail.toString("utf8")); + } } return { - onStdout: (chunk: string) => append(streams.stdout, chunk), - onStderr: (chunk: string) => append(streams.stderr, chunk), + onStdout: (chunk: string) => append("stdout", streams.stdout, chunk), + onStderr: (chunk: string) => append("stderr", 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 { @@ -1564,8 +1572,9 @@ async function runSessionLogStream( sandbox: Sandbox, sessionId: string, commandId: string, + onNewTail?: (stream: "stdout" | "stderr", text: string) => void, ): Promise { - const buffer = createSessionStreamBuffer(); + const buffer = createSessionStreamBuffer(onNewTail); let reconnects = 0; while (true) { try { @@ -1668,7 +1677,16 @@ async function executeInSession( // 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); + // 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 invocation id. This is a no-op when no plugin context is + // set (a direct test call) or when the host has no active execute route. + const streamResult = await runSessionLogStream( + sandbox, + sessionId, + commandId, + (stream, text) => pluginContext?.execution.log(stream, text), + ); if (streamResult.ok) { const exitCode = await readSessionExitCode(sandbox, sessionId, commandId); const durationMs = timingNow() - execStart; diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index e14ba5fa26..3fb8e8035a 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1927,6 +1927,29 @@ export interface WorkerToHostNotifications { channel: string; companyId: string; }; + + /** + * Deliver one incremental output chunk of the active `environmentExecute` + * call to the host runner log sink. + * + * The worker emits this notification for each new `stdout` or `stderr` chunk + * while one execute call runs. The host reads the active invocation id from + * the envelope field `paperclipInvocationId`, which the worker RPC host stamps + * from the active invocation context. The host correlates the chunk to the + * host-owned execute route for that id and delivers it to that route's + * `onLog` callback. + * + * Security: the notification carries no company id on purpose. The + * invocation-to-company binding on the host execute route is authoritative. + * The host never reads a company id from this payload to select the route or + * to grant access. The `chunk` is a text string, because JSON-RPC cannot + * carry raw bytes; the host drops a chunk that is not a bounded non-empty + * string or whose stream name is not exactly `stdout` or `stderr`. + */ + "execute.log": { + stream: "stdout" | "stderr"; + chunk: string; + }; } /** Union of all worker→host notification method names. */ diff --git a/packages/plugins/sdk/src/testing.ts b/packages/plugins/sdk/src/testing.ts index 225fbecc43..af12e9a8c8 100644 --- a/packages/plugins/sdk/src/testing.ts +++ b/packages/plugins/sdk/src/testing.ts @@ -2473,6 +2473,11 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { }, }; })(), + execution: { + log(_stream: "stdout" | "stderr", _chunk: string) { + // No-op in test harness — the host runner log sink is not wired here. + }, + }, tools: { register(name, _decl, fn) { requireCapability(manifest, capabilitySet, "agent.tools.register"); diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index ed1daedbf4..adfad498d9 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -1984,6 +1984,30 @@ export interface PluginStreamsClient { close(channel: string): void; } +/** + * `ctx.execution` — deliver incremental command output from an environment + * driver's active `execute` call to the host runner log sink. + * + * A sandbox provider that streams a long-lived command's output calls + * `ctx.execution.log(stream, chunk)` for each new chunk while the execute call + * runs. The host correlates the chunk to the active execute invocation by the + * host-issued invocation id on the message envelope, and delivers it to that + * call's log callback before the final result. The default is a no-op that + * never throws, so a provider that does not stream keeps its current behavior. + * + * The `chunk` is a text string, not raw bytes. The host drops a chunk with an + * unknown stream name or a chunk that is empty or too large. + */ +export interface PluginExecutionClient { + /** + * Deliver one incremental output chunk of the active execute call. + * + * @param stream - Either `"stdout"` or `"stderr"`. + * @param chunk - The new output text for that stream. + */ + log(stream: "stdout" | "stderr", chunk: string): void; +} + // --------------------------------------------------------------------------- // Full plugin context // --------------------------------------------------------------------------- @@ -2094,6 +2118,11 @@ export interface PluginContext { /** Push real-time events from the worker to the plugin UI via SSE. */ streams: PluginStreamsClient; + /** Deliver incremental command output from the active execute call to the + * host runner log sink. The default is a no-op for a provider that does not + * stream. */ + execution: PluginExecutionClient; + /** Register agent tool handlers. Requires `agent.tools.register`. */ tools: PluginToolsClient; diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 8b56a48972..ccad211040 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -1353,6 +1353,19 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }; })(), + execution: { + log(stream: "stdout" | "stderr", chunk: string): void { + // Emit one incremental output chunk of the active execute call. + // `notifyHost` stamps the active invocation id from the invocation + // context, so the host correlates the chunk to the host-owned execute + // route for that call. The notification carries no company id; the + // host binds the company from its own execute route. A chunk sent with + // no active invocation carries no id and the host drops it. + if (typeof chunk !== "string" || chunk.length === 0) return; + notifyHost("execute.log", { stream, chunk }); + }, + }, + tools: { register( name: string, diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index e88833671b..c178a2fe2e 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -630,3 +630,128 @@ describe("worker provider tracer", () => { expect(spanRecords).toHaveLength(0); }); }); + +describe("worker execute.log emitter", () => { + // Run one data handler that calls `ctx.execution.log`, and capture the + // `execute.log` notifications the worker sends to the host. + async function runExecuteLogProbe( + invocation: PluginInvocationContext | undefined, + entries: Array<{ stream: "stdout" | "stderr"; chunk: string }>, + ) { + const hostToWorker = new PassThrough(); + const workerToHost = new PassThrough(); + const hostReadline = createInterface({ input: workerToHost }); + const pending = new Map void>(); + const logRecords: Array<{ params: unknown; invocationId?: string }> = []; + let nextRequestId = 1; + + const plugin = definePlugin({ + async setup(ctx) { + ctx.data.register("emit-logs", async () => { + for (const entry of entries) { + ctx.execution.log(entry.stream, entry.chunk); + } + return { ok: true }; + }); + }, + }); + + const worker = startWorkerRpcHost({ plugin, stdin: hostToWorker, stdout: workerToHost }); + + function callWorker(method: string, params: unknown, inv?: PluginInvocationContext) { + const id = `host-${nextRequestId++}`; + const request = { + ...createRequest(method, params, id), + ...(inv ? { paperclipInvocation: inv } : {}), + }; + const result = new Promise((resolve, reject) => { + pending.set(id, (response) => { + if ("error" in response && response.error) { + reject(new Error(response.error.message)); + return; + } + resolve((response as { result?: unknown }).result); + }); + }); + hostToWorker.write(serializeMessage(request)); + return result; + } + + hostReadline.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { + pending.get(String(message.id))?.(message); + pending.delete(String(message.id)); + return; + } + // `execute.log` is a fire-and-forget notification (no id), so it is not a + // JSON-RPC request. Match on the method name directly. + if ((message as { method?: string }).method === "execute.log") { + logRecords.push({ + params: (message as { params?: unknown }).params, + invocationId: (message as { paperclipInvocationId?: string }).paperclipInvocationId, + }); + } + }); + + try { + await callWorker("initialize", { + manifest: { + id: "paperclip.execute-log-test", + apiVersion: 1, + version: "1.0.0", + displayName: "Execute log test", + description: "Execute log test", + author: "Paperclip", + categories: ["automation"], + capabilities: ["environment.drivers.register"], + entrypoints: { worker: "dist/worker.js" }, + }, + config: {}, + instanceInfo: { instanceId: "test", hostVersion: "0.0.0" }, + apiVersion: 1, + }); + await callWorker("getData", { key: "emit-logs", companyId: "company-a", params: {} }, invocation); + // Let the fire-and-forget execute.log notifications flush. + await new Promise((resolve) => setTimeout(resolve, 20)); + return logRecords; + } finally { + worker.stop(); + hostReadline.close(); + hostToWorker.destroy(); + workerToHost.destroy(); + } + } + + it("stamps the active invocation id on each execute.log notification", async () => { + const records = await runExecuteLogProbe( + { id: "invocation-a", scope: { companyId: "company-a" } }, + [ + { stream: "stdout", chunk: "one" }, + { stream: "stderr", chunk: "two" }, + ], + ); + expect(records).toHaveLength(2); + expect(records[0]).toEqual({ + params: { stream: "stdout", chunk: "one" }, + invocationId: "invocation-a", + }); + expect(records[1]).toEqual({ + params: { stream: "stderr", chunk: "two" }, + invocationId: "invocation-a", + }); + }); + + it("drops an empty chunk before it reaches the host", async () => { + const records = await runExecuteLogProbe( + { id: "invocation-a", scope: { companyId: "company-a" } }, + [ + { stream: "stdout", chunk: "" }, + { stream: "stdout", chunk: "kept" }, + ], + ); + expect(records).toEqual([ + { params: { stream: "stdout", chunk: "kept" }, invocationId: "invocation-a" }, + ]); + }); +}); diff --git a/packages/shared/src/types/environment.ts b/packages/shared/src/types/environment.ts index 2a2867ffc1..aa08f82ab7 100644 --- a/packages/shared/src/types/environment.ts +++ b/packages/shared/src/types/environment.ts @@ -30,6 +30,11 @@ export interface FakeSandboxEnvironmentConfig { reuseLease: boolean; /** Stream agent CLI stdout/stderr during sandbox runs (bridge log-tail loop). */ streamRunLogs?: boolean; + /** + * Stream the interactive ACP agent output through the persistent session log + * stream instead of the host output-file poll. Default OFF. + */ + streamAgentSessionOutput?: boolean; /** * Archive the sandbox on lease release instead of deleting it, so operators * can inspect it from the provider dashboard. Injected by test/probe paths; @@ -44,6 +49,11 @@ export interface PluginSandboxEnvironmentConfig { timeoutMs?: number; /** Stream agent CLI stdout/stderr during sandbox runs (bridge log-tail loop). */ streamRunLogs?: boolean; + /** + * Stream the interactive ACP agent output through the persistent session log + * stream instead of the host output-file poll. Default OFF. + */ + streamAgentSessionOutput?: boolean; /** * Archive the sandbox on lease release instead of deleting it, so operators * can inspect it from the provider dashboard. Injected by test/probe paths; diff --git a/server/src/__tests__/environment-execution-target.test.ts b/server/src/__tests__/environment-execution-target.test.ts index 558621fb9a..662cf4c071 100644 --- a/server/src/__tests__/environment-execution-target.test.ts +++ b/server/src/__tests__/environment-execution-target.test.ts @@ -453,6 +453,70 @@ describe("resolveEnvironmentExecutionTarget", () => { expect(environmentRuntime.execute).toHaveBeenCalledTimes(2); }); + it("forwards the session flags to the environment runtime execute", async () => { + mockResolveEnvironmentDriverConfigForRuntime.mockResolvedValue({ + driver: "sandbox", + config: { + provider: "fake-plugin", + reuseLease: false, + timeoutMs: 30_000, + }, + }); + + const environmentRuntime = { + execute: vi.fn().mockResolvedValue({ + exitCode: 0, + signal: null, + timedOut: false, + stdout: "ok", + stderr: "", + metadata: { durationMs: 600, getDurationMs: 15 }, + }), + supportsSync: vi.fn().mockReturnValue(false), + }; + + const target = await resolveEnvironmentExecutionTarget({ + db: {} as never, + companyId: "company-1", + adapterType: "codex_local", + environment: { id: "env-1", driver: "sandbox", config: { provider: "fake-plugin" } }, + leaseId: "lease-1", + leaseMetadata: { remoteCwd: "/workspace" }, + lease: { id: "lease-1" } as never, + environmentRuntime: environmentRuntime as never, + }); + + const runner = (target as { runner?: { + execute(input: { + command: string; + args?: string[]; + useSession?: boolean; + bypassSession?: boolean; + }): Promise; + } }).runner!; + + // The agent command opts onto the persistent session with `useSession`, + // which the seam maps to `forceSession`. It never bypasses the session. + await runner.execute({ command: "node", args: ["script.js"], useSession: true }); + // A bridge control-plane exec opts off the persistent session with + // `bypassSession`, which the seam forwards unchanged. + await runner.execute({ command: "sh", args: ["-c", "cat"], bypassSession: true }); + + const first = environmentRuntime.execute.mock.calls[0]![0] as { + forceSession?: boolean; + bypassSession?: boolean; + }; + expect(first.forceSession).toBe(true); + expect(first.bypassSession).toBeUndefined(); + + const second = environmentRuntime.execute.mock.calls[1]![0] as { + forceSession?: boolean; + bypassSession?: boolean; + }; + expect(second.forceSession).toBeUndefined(); + expect(second.bypassSession).toBe(true); + }); + // A recording tracer that captures each provider-exec span's name, attribute // map, and end. It satisfies the structural tracer the seam calls. function createRecordingExecTracer() { @@ -549,6 +613,122 @@ describe("resolveEnvironmentExecutionTarget", () => { } }).runner!; } + // Run the sandbox runner's execute with an incremental log sink and collect + // the ordered deliveries. The runner's execute accepts an `onLog`, so this + // casts past the narrowed helper return type. + async function runExecuteCollectingLogs( + runner: { execute(input: unknown): Promise }, + ): Promise> { + const delivered: Array<[string, string]> = []; + await runner.execute({ + command: "echo", + onLog: async (stream: "stdout" | "stderr", chunk: string) => { + delivered.push([stream, chunk]); + }, + }); + return delivered; + } + + it("delivers only the un-streamed suffix after a provider streams a prefix then polls the complete result", async () => { + // The provider streams a prefix through the incremental sink, then its + // stream fails and it polls the complete output as the final result. The + // reconciler must deliver the remaining tail once, so no output byte is lost + // or repeated. + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: async (input: unknown) => { + const typed = input as { onLog?: (s: "stdout" | "stderr", c: string) => Promise }; + await typed.onLog?.("stdout", "hello "); + await typed.onLog?.("stderr", "warn:"); + return { exitCode: 0, signal: null, timedOut: false, stdout: "hello world", stderr: "warn:done" }; + }, + }); + const delivered = await runExecuteCollectingLogs( + runner as { execute(input: unknown): Promise }, + ); + expect(delivered).toEqual([ + ["stdout", "hello "], + ["stderr", "warn:"], + ["stdout", "world"], + ["stderr", "done"], + ]); + }); + + it("does not repeat output when the provider already streamed the complete result", async () => { + // The provider streams the whole output through the incremental sink and + // returns the same complete result. The suffix is empty, so the reconciler + // never re-delivers the streamed bytes. + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: async (input: unknown) => { + const typed = input as { onLog?: (s: "stdout" | "stderr", c: string) => Promise }; + await typed.onLog?.("stdout", "full"); + return { exitCode: 0, signal: null, timedOut: false, stdout: "full", stderr: "" }; + }, + }); + const delivered = await runExecuteCollectingLogs( + runner as { execute(input: unknown): Promise }, + ); + expect(delivered).toEqual([["stdout", "full"]]); + }); + + it("delivers the full captured output when the provider streams nothing incrementally", async () => { + // The provider streams no incremental chunk, so the whole final result is + // the suffix and reaches the sink once. + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + stdout: "batch-out", + stderr: "batch-err", + }), + }); + const delivered = await runExecuteCollectingLogs( + runner as { execute(input: unknown): Promise }, + ); + expect(delivered).toEqual([ + ["stdout", "batch-out"], + ["stderr", "batch-err"], + ]); + }); + + it("delivers the whole final output when the poll fallback buffer does not continue the streamed prefix", async () => { + // The provider streams a prefix, then its stream fails and it polls a + // buffer that does NOT start with that prefix. A length slice would drop + // the leading bytes of the poll buffer and corrupt the durable log, so the + // reconciler delivers the whole final output instead. The streamed prefix + // repeats, but no output byte is lost or truncated. + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: async (input: unknown) => { + const typed = input as { onLog?: (s: "stdout" | "stderr", c: string) => Promise }; + await typed.onLog?.("stdout", "hello "); + await typed.onLog?.("stderr", "warn:"); + // The poll buffer starts with different leading text on both streams. + return { exitCode: 0, signal: null, timedOut: false, stdout: "RESYNCED output", stderr: "RESET err" }; + }, + }); + const delivered = await runExecuteCollectingLogs( + runner as { execute(input: unknown): Promise }, + ); + expect(delivered).toEqual([ + ["stdout", "hello "], + ["stderr", "warn:"], + ["stdout", "RESYNCED output"], + ["stderr", "RESET err"], + ]); + }); + it("sets the provider duration attributes from finite Daytona-shaped metadata", async () => { const { tracer, spans } = createRecordingExecTracer(); const runner = await runnerFor({ @@ -947,6 +1127,138 @@ describe("resolveEnvironmentExecutionTarget", () => { expect(onLog).toHaveBeenCalledTimes(1); }); + it("delivers incremental logs before the final result and does not duplicate them", async () => { + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: vi.fn(async (input: { onLog?: (s: string, c: string) => Promise }) => { + // The provider streams the output while the command runs. + await input.onLog?.("stdout", "chunk-1"); + await input.onLog?.("stderr", "chunk-2"); + await input.onLog?.("stdout", "chunk-3"); + return { + exitCode: 0, + signal: null, + timedOut: false, + stdout: "chunk-1chunk-3", + stderr: "chunk-2", + }; + }), + }); + + const onLog = vi.fn(); + const result = await (runner as { + execute(input: unknown): Promise<{ stdout: string; stderr: string; exitCode: number }>; + }).execute({ command: "echo", onLog }); + + // The runner receives the incremental chunks in order, and NOT a repeated + // delivery of the final stdout/stderr. + expect(onLog.mock.calls).toEqual([ + ["stdout", "chunk-1"], + ["stderr", "chunk-2"], + ["stdout", "chunk-3"], + ]); + // The final result stays available to the caller for parsing and fallback. + expect(result).toMatchObject({ exitCode: 0, stdout: "chunk-1chunk-3", stderr: "chunk-2" }); + }); + + it("still delivers the final result to the runner when the provider does not stream", async () => { + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + // A provider that never calls onLog returns only the final result. + execute: vi.fn(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + stdout: "final-out", + stderr: "final-err", + })), + }); + + const onLog = vi.fn(); + const result = await (runner as { + execute(input: unknown): Promise<{ stdout: string; stderr: string }>; + }).execute({ command: "echo", onLog }); + + expect(onLog.mock.calls).toEqual([ + ["stdout", "final-out"], + ["stderr", "final-err"], + ]); + expect(result).toMatchObject({ stdout: "final-out", stderr: "final-err" }); + }); + + it("creates exactly one sandbox.exec span for one streamed provider call", async () => { + const { tracer, spans } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: vi.fn(async (input: { onLog?: (s: string, c: string) => Promise }) => { + await input.onLog?.("stdout", "chunk-a"); + await input.onLog?.("stdout", "chunk-b"); + await input.onLog?.("stderr", "chunk-c"); + return { + exitCode: 0, + signal: null, + timedOut: false, + stdout: "chunk-achunk-b", + stderr: "chunk-c", + }; + }), + }); + + const onLog = vi.fn(); + await (runner as { execute(input: unknown): Promise }).execute({ + command: "echo", + onLog, + }); + + // One long-lived provider call opens one span; each stream chunk opens none. + expect(spans).toHaveLength(1); + expect(spans[0]!.name).toBe("sandbox.exec"); + expect(spans[0]!.ended).toBe(true); + }); + + it("keeps streamed log text and secret values out of span attributes", async () => { + const { tracer, spans } = createRecordingExecTracer(); + const secret = "sk-super-secret-value"; + const runner = await runnerWithExecute({ + provider: "daytona", + tracer, + execute: vi.fn(async (input: { onLog?: (s: string, c: string) => Promise }) => { + await input.onLog?.("stdout", `token=${secret}\n`); + return { + exitCode: 0, + signal: null, + timedOut: false, + stdout: `token=${secret}\n`, + stderr: "", + }; + }), + }); + + const onLog = vi.fn(); + // The secret rides the env and the streamed chunk, never the command label. + await (runner as { execute(input: unknown): Promise }).execute({ + command: "run-agent", + env: { API_KEY: secret }, + onLog, + }); + + expect(spans).toHaveLength(1); + const span = spans[0]!; + // Attributes carry only the closed allowlist — never log text or secrets. + for (const key of Object.keys(span.attributes)) { + expect(ALLOWED_EXEC_SPAN_ATTRIBUTE_KEYS.has(key), `non-allowlisted key "${key}"`).toBe(true); + } + const serialized = JSON.stringify(span.attributes); + expect(serialized).not.toContain(secret); + expect(serialized).not.toContain("token="); + expect(serialized).not.toContain("chunk"); + }); + // Fire one run-time exec from a bridge continuation that runs after the step // span ended. Each bridge step (`bridge.paperclip`, `bridge.process-session`) // starts long-lived work with `criticalPath: false`. The bridge boundary wraps diff --git a/server/src/__tests__/environment-routes.test.ts b/server/src/__tests__/environment-routes.test.ts index 06c04d2d94..b838aed1a7 100644 --- a/server/src/__tests__/environment-routes.test.ts +++ b/server/src/__tests__/environment-routes.test.ts @@ -1587,6 +1587,64 @@ describe("environment routes", () => { expect(mockSecretService.create).not.toHaveBeenCalled(); }); + it("keeps host-owned stream flags when the provider plugin drops them from its normalized config", async () => { + // The host owns `streamRunLogs` and `streamAgentSessionOutput`. It reads + // them to select the run-log stream and the ACP session output stream. A + // provider plugin normalizes only its own driver fields, so it drops these + // host flags from its normalized config. The host must re-apply them, or the + // saved environment loses the operator opt-in and the streams never start. + const environment = { + ...createEnvironment(), + id: "env-sandbox-fake-plugin", + name: "Streamed Sandbox", + driver: "sandbox" as const, + config: { provider: "fake-plugin", image: "fake:test" }, + }; + mockEnvironmentService.create.mockResolvedValue(environment); + mockValidatePluginSandboxProviderConfig.mockImplementation(async ({ provider, config }) => { + // Drop the host flags to reproduce a plugin that allowlists driver fields. + const { streamRunLogs, streamAgentSessionOutput, ...driverConfig } = + config as Record; + void streamRunLogs; + void streamAgentSessionOutput; + return { + normalizedConfig: driverConfig, + pluginId: `plugin-${provider}`, + pluginKey: `plugin.${provider}`, + driver: { + driverKey: provider, + kind: "sandbox_provider", + displayName: provider, + configSchema: { type: "object" }, + }, + }; + }); + const pluginWorkerManager = {}; + const app = createApp({ + type: "board", + userId: "user-1", + source: "local_implicit", + }, { pluginWorkerManager }); + + const res = await request(app) + .post("/api/companies/company-1/environments") + .send({ + name: "Streamed Sandbox", + driver: "sandbox", + config: { + provider: "fake-plugin", + image: "fake:test", + streamRunLogs: false, + streamAgentSessionOutput: true, + }, + }); + + expect(res.status).toBe(201); + const persisted = mockEnvironmentService.create.mock.calls[0][0].config as Record; + expect(persisted.streamAgentSessionOutput).toBe(true); + expect(persisted.streamRunLogs).toBe(false); + }); + it("creates a schema-driven sandbox environment with secret-ref fields persisted as secrets", async () => { const environment = { ...createEnvironment(), diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 7397b99732..01210c1bda 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -702,7 +702,9 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(executed.stdout).toBe("ok\n"); expect(released).toHaveLength(1); expect(released[0]?.lease.status).toBe("released"); - expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentExecute", expect.anything(), 31000); + // The execute call carries the optional log sink as the fifth argument; it + // is undefined when the caller passes no sink. + expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentExecute", expect.anything(), 31000, undefined); expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentReleaseLease", expect.anything(), 31234); }); @@ -1385,11 +1387,13 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(executed.stdout).toBe("ok\n"); expect(released).toHaveLength(1); expect(released[0]?.lease.status).toBe("released"); + // The execute call carries the optional log sink as the fifth argument; it + // is undefined when the caller passes no sink. expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentExecute", expect.objectContaining({ config: expect.objectContaining({ apiKey: "resolved-provider-key", }), - }), 31234); + }), 31234, undefined); expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentReleaseLease", expect.objectContaining({ config: expect.objectContaining({ apiKey: "resolved-provider-key", diff --git a/server/src/__tests__/fixtures/plugin-worker-execute-log.cjs b/server/src/__tests__/fixtures/plugin-worker-execute-log.cjs new file mode 100644 index 0000000000..7d7feacb66 --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-execute-log.cjs @@ -0,0 +1,117 @@ +// Test worker fixture for the `execute.log` worker→host notification route. +// +// On `environmentExecute` the fixture emits the `execute.log` notifications +// listed in `params.logs`, then returns the final result. Each log entry sets +// the envelope invocation id by its `tag`: +// - "echo" → the real host-issued id (the normal streaming path) +// - "unknown" → a forged id with no active route (must be dropped) +// - "none" → no id at all (must be dropped) +// - "forge-previous" → the id of the PREVIOUS execute call the same worker +// process handled. One worker process sees every active +// invocation id, so this reproduces a worker that runs +// company A and forges company B's active id. +// The default tag is "echo". The entry `stream` and `chunk` pass through +// verbatim, so a test can send an invalid stream name or an empty chunk to +// prove the host drops invalid input. +// +// When `params.oversizedLogChunkChars > 0` the fixture emits one VALID +// `execute.log` note (own id) whose chunk holds that many characters BEFORE the +// normal logs, so a test can prove the host drops an over-length line before it +// parses the JSON. +const readline = require("node:readline"); + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +// The id of the previous execute call this worker process handled. A single +// worker process serves every company, so it can name another company's active +// invocation. The fixture uses this to forge a peer id. +let previousInvocationId = null; + +const rl = readline.createInterface({ + input: process.stdin, + crlfDelay: Infinity, +}); + +rl.on("line", (line) => { + if (!line.trim()) return; + const message = JSON.parse(line); + const method = message && typeof message.method === "string" ? message.method : null; + + if (method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { ok: true, supportedMethods: ["environmentExecute"] }, + }); + return; + } + + if (method === "environmentExecute") { + const invocationId = message.paperclipInvocation && message.paperclipInvocation.id; + const forgeableId = previousInvocationId; + previousInvocationId = invocationId; + const params = message.params ?? {}; + const oversizedChars = Number(params.oversizedLogChunkChars ?? 0); + if (oversizedChars > 0) { + send({ + jsonrpc: "2.0", + method: "execute.log", + paperclipInvocationId: invocationId, + params: { stream: "stdout", chunk: "d".repeat(oversizedChars) }, + }); + } + const logs = Array.isArray(params.logs) ? params.logs : []; + for (const entry of logs) { + const note = { + jsonrpc: "2.0", + method: "execute.log", + params: { stream: entry.stream, chunk: entry.chunk }, + }; + const tag = entry.tag ?? "echo"; + if (tag === "echo") { + note.paperclipInvocationId = invocationId; + } else if (tag === "unknown") { + note.paperclipInvocationId = "unknown-invocation"; + } else if (tag === "forge-previous") { + note.paperclipInvocationId = forgeableId; + } + // tag === "none" → omit the invocation id entirely. + send(note); + } + + const finish = () => { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + exitCode: 0, + signal: null, + timedOut: false, + stdout: params.finalStdout ?? "", + stderr: params.finalStderr ?? "", + }, + }); + }; + const delayMs = Number(params.delayMs ?? 0); + if (delayMs > 0) { + setTimeout(finish, delayMs); + } else { + finish(); + } + return; + } + + if (method === "shutdown") { + send({ jsonrpc: "2.0", id: message.id, result: {} }); + setImmediate(() => process.exit(0)); + return; + } + + send({ + jsonrpc: "2.0", + id: message.id, + error: { code: -32601, message: `Unhandled method: ${method}` }, + }); +}); diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index 451e7d1c60..7d837a47dd 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -23,6 +23,7 @@ const INVOCATION_SCOPE_WORKER_ENTRYPOINT = path.join( "plugin-worker-invocation-scope.cjs", ); const TERMINATED_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-terminated.cjs"); +const EXECUTE_LOG_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-execute-log.cjs"); const TEST_MANIFEST: PaperclipPluginManifestV1 = { id: "test.plugin", @@ -779,3 +780,277 @@ describe("plugin proactive events.subscribe: options-seeded scope + filter parit } }); }); + +// --------------------------------------------------------------------------- +// execute.log worker→host notification route +// --------------------------------------------------------------------------- + +function makeExecuteLogHandle(extra?: Record) { + return createPluginWorkerHandle("test.plugin", { + entrypointPath: EXECUTE_LOG_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { instanceId: "instance-1", hostVersion: "1.0.0" }, + apiVersion: 1, + hostHandlers: {}, + ...extra, + }); +} + +function executeParams( + overrides: Record, +): HostToWorkerMethods["environmentExecute"][0] { + return { + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + config: {}, + lease: { providerLeaseId: "lease-1" }, + command: "echo", + ...overrides, + } as unknown as HostToWorkerMethods["environmentExecute"][0]; +} + +describe("plugin worker manager execute.log route", () => { + it("delivers ordered execute.log chunks to the execute log sink", async () => { + const handle = makeExecuteLogHandle(); + const sink = vi.fn(); + try { + await handle.start(); + const result = await handle.call( + "environmentExecute", + executeParams({ + logs: [ + { stream: "stdout", chunk: "one" }, + { stream: "stderr", chunk: "two" }, + { stream: "stdout", chunk: "three" }, + ], + finalStdout: "onethree", + finalStderr: "two", + }), + undefined, + sink, + ); + expect(result).toMatchObject({ exitCode: 0 }); + expect(sink.mock.calls).toEqual([ + ["stdout", "one"], + ["stderr", "two"], + ["stdout", "three"], + ]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops an execute.log chunk with a forged or missing invocation id", async () => { + const handle = makeExecuteLogHandle(); + const sink = vi.fn(); + try { + await handle.start(); + await handle.call( + "environmentExecute", + executeParams({ + logs: [ + { stream: "stdout", chunk: "valid", tag: "echo" }, + { stream: "stdout", chunk: "forged", tag: "unknown" }, + { stream: "stdout", chunk: "orphan", tag: "none" }, + ], + }), + undefined, + sink, + ); + // Only the chunk that carries this call's own host-issued id is delivered. + expect(sink.mock.calls).toEqual([["stdout", "valid"]]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops an execute.log chunk with an invalid stream name or an empty chunk", async () => { + const handle = makeExecuteLogHandle(); + const sink = vi.fn(); + try { + await handle.start(); + await handle.call( + "environmentExecute", + executeParams({ + logs: [ + { stream: "stdout", chunk: "keep" }, + { stream: "bogus", chunk: "dropped-stream" }, + { stream: "stdout", chunk: "" }, + ], + }), + undefined, + sink, + ); + expect(sink.mock.calls).toEqual([["stdout", "keep"]]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("routes two concurrent same-company execute calls to their own sink only", async () => { + const handle = makeExecuteLogHandle(); + const sinkA = vi.fn(); + const sinkB = vi.fn(); + try { + await handle.start(); + const callA = handle.call( + "environmentExecute", + executeParams({ + companyId: "company-1", + logs: [{ stream: "stdout", chunk: "a1" }], + delayMs: 40, + }), + undefined, + sinkA, + ); + const callB = handle.call( + "environmentExecute", + executeParams({ + companyId: "company-1", + logs: [{ stream: "stdout", chunk: "b1" }], + delayMs: 40, + }), + undefined, + sinkB, + ); + await Promise.all([callA, callB]); + // Both calls belong to one company, so the shared pipe stays + // single-company and each chunk reaches only its own call's sink. + expect(sinkA.mock.calls).toEqual([["stdout", "a1"]]); + expect(sinkB.mock.calls).toEqual([["stdout", "b1"]]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("fails closed and never delivers execute.log across companies, even with a forged peer id", async () => { + // A single worker process serves every company, so it knows both companies' + // active invocation ids. While company B's execute stays active, company A's + // execute forges B's known, valid id and aims a chunk at B's route. The host + // must not deliver it to B. Before the exact-company-scope validation, the + // route lookup by the worker-supplied id delivered the forged chunk to B. + const handle = makeExecuteLogHandle(); + const sinkA = vi.fn(); + const sinkB = vi.fn(); + try { + await handle.start(); + // Company B opens first and stays active (delayed finish), so its route is + // registered and known to the worker when company A runs. + const callB = handle.call( + "environmentExecute", + executeParams({ companyId: "company-b", logs: [], delayMs: 200 }), + undefined, + sinkB, + ); + // Let the worker process B's execute, so it records B's id as the peer id. + await new Promise((resolve) => setTimeout(resolve, 40)); + const callA = handle.call( + "environmentExecute", + executeParams({ + companyId: "company-a", + logs: [{ stream: "stdout", chunk: "forged-into-b", tag: "forge-previous" }], + }), + undefined, + sinkA, + ); + await Promise.all([callA, callB]); + expect(sinkB).not.toHaveBeenCalled(); + expect(sinkA).not.toHaveBeenCalled(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops execute.log chunks once one execute call exceeds its output budget", async () => { + // Bound the total streamed output for one execute call. Past the ceiling the + // host drops further chunks, so one runaway or hostile execution cannot flood + // the host without limit. + const handle = makeExecuteLogHandle({ + executeLogLimits: { maxTotalCharsPerExecute: 10 }, + }); + const sink = vi.fn(); + try { + await handle.start(); + await handle.call( + "environmentExecute", + executeParams({ + logs: [ + { stream: "stdout", chunk: "aaaaa" }, // total 5 → delivered + { stream: "stdout", chunk: "bbbbb" }, // total 10 → delivered + { stream: "stdout", chunk: "c" }, // total 11 > 10 → dropped + ], + }), + undefined, + sink, + ); + expect(sink.mock.calls).toEqual([ + ["stdout", "aaaaa"], + ["stdout", "bbbbb"], + ]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops an over-length worker line before parsing it and keeps serving the call", async () => { + // Enforce the framing bound before the JSON parse. The oversized note is a + // valid execute.log line for this call's own id, so without the pre-parse + // guard the host would parse and deliver it. The normal note stays under the + // limit and reaches the sink, and the call still completes. + const handle = makeExecuteLogHandle({ + executeLogLimits: { maxIncomingMessageChars: 400 }, + }); + const sink = vi.fn(); + try { + await handle.start(); + const result = await handle.call( + "environmentExecute", + executeParams({ + oversizedLogChunkChars: 1_000, + logs: [{ stream: "stdout", chunk: "kept" }], + finalStdout: "kept", + }), + undefined, + sink, + ); + expect(result).toMatchObject({ exitCode: 0 }); + expect(sink.mock.calls).toEqual([["stdout", "kept"]]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("completes an execute call that sends no execute.log notification", async () => { + const handle = makeExecuteLogHandle(); + const sink = vi.fn(); + try { + await handle.start(); + const result = await handle.call( + "environmentExecute", + executeParams({ logs: [], finalStdout: "done" }), + undefined, + sink, + ); + expect(result).toMatchObject({ exitCode: 0, stdout: "done" }); + expect(sink).not.toHaveBeenCalled(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("does not throw when execute.log arrives but no sink is registered", async () => { + const handle = makeExecuteLogHandle(); + try { + await handle.start(); + const result = await handle.call( + "environmentExecute", + executeParams({ logs: [{ stream: "stdout", chunk: "no-sink" }] }), + ); + expect(result).toMatchObject({ exitCode: 0 }); + } finally { + await handle.stop().catch(() => undefined); + } + }); +}); diff --git a/server/src/services/environment-config.ts b/server/src/services/environment-config.ts index efd33c818e..433839de82 100644 --- a/server/src/services/environment-config.ts +++ b/server/src/services/environment-config.ts @@ -77,6 +77,7 @@ const fakeSandboxEnvironmentConfigSchema = z.object({ .default("ubuntu:24.04"), reuseLease: z.boolean().optional().default(false), streamRunLogs: z.boolean().optional(), + streamAgentSessionOutput: z.boolean().optional(), archiveOnRelease: z.boolean().optional(), }).strict(); @@ -93,6 +94,7 @@ const pluginSandboxEnvironmentConfigSchema = z.object({ timeoutMs: z.coerce.number().int().min(1).max(86_400_000).optional(), reuseLease: z.boolean().optional().default(false), streamRunLogs: z.boolean().optional(), + streamAgentSessionOutput: z.boolean().optional(), archiveOnRelease: z.boolean().optional(), }).catchall(z.unknown()); @@ -371,6 +373,30 @@ export function stripSandboxProviderEnvelope(config: SandboxEnvironmentConfig): return driverConfig; } +// The host owns these sandbox run-behavior flags, not the provider plugin. The +// host reads them to select the run-log stream and the ACP session output +// stream. The host passes the whole config to the plugin, so a plugin that +// allowlists its own driver fields drops these flags from its normalized +// config. Re-apply them from the parsed envelope after the plugin normalizes, +// or a saved environment loses the operator opt-in and the stream never starts. +const HOST_OWNED_SANDBOX_STREAM_FLAGS = [ + "streamRunLogs", + "streamAgentSessionOutput", +] as const; + +function applyHostOwnedSandboxStreamFlags( + normalizedConfig: Record, + envelope: Record, +): Record { + const merged: Record = { ...normalizedConfig }; + for (const key of HOST_OWNED_SANDBOX_STREAM_FLAGS) { + if (envelope[key] !== undefined) { + merged[key] = envelope[key]; + } + } + return merged; +} + export function normalizeEnvironmentConfig(input: { driver: EnvironmentDriver; config: Record | null | undefined; @@ -458,7 +484,7 @@ export function normalizeEnvironmentConfigForProbe(input: { ...(await resolveConfigSecretRefsForProbe({ db: input.db, companyId: input.companyId, - config: validated.normalizedConfig, + config: applyHostOwnedSandboxStreamFlags(validated.normalizedConfig, parsed.data), accessContext: input.accessContext, schema: validated.driver.configSchema && @@ -550,7 +576,7 @@ export async function normalizeEnvironmentConfigForPersistence(input: { secretProvider: input.secretProvider, config: { provider: parsed.data.provider, - ...validated.normalizedConfig, + ...applyHostOwnedSandboxStreamFlags(validated.normalizedConfig, parsed.data), }, schema: validated.driver.configSchema && typeof validated.driver.configSchema === "object" && !Array.isArray(validated.driver.configSchema) diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index c18b27564b..0cfb88806d 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -69,6 +69,28 @@ function toBoolean(value: unknown): boolean | undefined { return typeof value === "boolean" ? value : undefined; } +/** + * Compute the tail of `final` that the provider did NOT already stream. + * + * The provider streams output chunks in order. Those chunks form `delivered`. + * The final result is `final`. In the normal path `final` continues + * `delivered`, so the tail is `final` past the delivered length. + * + * A provider can stream a prefix and then fall back to a poll that returns a + * different buffer. When `final` does not start with `delivered`, a length + * slice would drop unrelated leading output or cut a chunk mid-text, so the + * durable log would hold truncated or corrupt output. In that case this + * function returns the whole `final` instead. That can repeat the streamed + * prefix in the log, but the complete final output always reaches the log. + * Repetition is safer than a silent loss of output. + */ +function undeliveredSuffix(delivered: string, final: string): string { + if (!final) return ""; + if (delivered.length === 0) return final; + if (final.startsWith(delivered)) return final.slice(delivered.length); + return final; +} + /** * The closed input for one `sandbox.exec` span. The seam builds it from the * exec result and the active step context. Every field is already bounded or @@ -234,6 +256,10 @@ export async function resolveEnvironmentExecutionTarget(input: { // output reaches the UI mid-run; `streamRunLogs: false` is an explicit // opt-out back to batch-at-end delivery. streamRunLogs: parsed.config.streamRunLogs !== false, + // Interactive ACP output streaming through the persistent session log + // stream. Default OFF: the process session bridge keeps the output-file + // poll unless an operator opts a sandbox environment in. + streamAgentSessionOutput: parsed.config.streamAgentSessionOutput === true, runner: input.environmentRuntime && input.lease ? { // Provider-backed sandbox RPCs do not surface bounded mid-stream @@ -266,6 +292,27 @@ export async function resolveEnvironmentExecutionTarget(input: { // provider execution marks the span failed. A later log-callback // rejection sits outside this block and never flips a successful // execution to failed. + // Incremental log sink. The provider streams each output chunk + // through the execute.log notification while the command runs. + // Serialize the delivery per execute call so the runner sees the + // chunks in order, and keep the delivered text per stream, so the + // final-result delivery below emits only the un-streamed suffix + // and can detect a provider poll fallback that returns a + // different buffer. + let incrementalLogChain: Promise = Promise.resolve(); + let deliveredStdout = ""; + let deliveredStderr = ""; + const onIncrementalLog = ( + stream: "stdout" | "stderr", + chunk: string, + ): Promise => { + if (stream === "stdout") deliveredStdout += chunk; + else deliveredStderr += chunk; + incrementalLogChain = incrementalLogChain.then(() => + commandInput.onLog?.(stream, chunk), + ); + return incrementalLogChain; + }; let result; try { result = await input.environmentRuntime!.execute({ @@ -277,6 +324,16 @@ export async function resolveEnvironmentExecutionTarget(input: { env: commandInput.env, stdin: commandInput.stdin, timeoutMs: commandInput.timeoutMs, + onLog: commandInput.onLog ? onIncrementalLog : undefined, + // The ACP process session bridge sets `useSession` so its + // long-lived agent command opens the persistent session and + // streams output, even though it runs with no active step. + forceSession: commandInput.useSession, + // The bridge control-plane execs set `bypassSession` so they + // run one-shot and never queue behind the long-lived agent + // command on the persistent session. An explicit bypass wins + // over `forceSession` and over the active-step selection. + bypassSession: commandInput.bypassSession, }); } catch (error) { // The provider execution threw. Mark the span failed with the @@ -322,12 +379,28 @@ export async function resolveEnvironmentExecutionTarget(input: { // Observability must not change execution control flow. } } - // Deliver the captured output. A rejected `onLog` still - // propagates to the caller (control flow is unchanged), but the - // span already carries the successful outcome, so a log failure - // never marks the execution failed. - if (result.stdout) await commandInput.onLog?.("stdout", result.stdout); - if (result.stderr) await commandInput.onLog?.("stderr", result.stderr); + // Drain the ordered incremental delivery before the final + // result. The provider streamed chunks arrive as execute.log + // notifications while the command runs; awaiting the chain keeps + // the runner order and surfaces a log-sink rejection. + await incrementalLogChain; + // Deliver only the suffix the provider did NOT already stream. + // The streamed chunks usually form an in-order prefix of the + // final result, so the remaining output is the final text past + // the delivered text. When the provider streamed nothing, the + // whole output is the suffix. When it streamed the complete + // output, the suffix is empty and nothing repeats. When it + // streamed a prefix and then fell back to a poll whose buffer + // does not continue that prefix, `undeliveredSuffix` returns the + // whole final output, so the durable log keeps the complete + // result and never holds a truncated slice. A rejected `onLog` + // still propagates to the caller (control flow is unchanged), + // but the span already carries the successful outcome, so a log + // failure never marks the execution failed. + const stdoutSuffix = undeliveredSuffix(deliveredStdout, result.stdout ?? ""); + if (stdoutSuffix) await commandInput.onLog?.("stdout", stdoutSuffix); + const stderrSuffix = undeliveredSuffix(deliveredStderr, result.stderr ?? ""); + if (stderrSuffix) await commandInput.onLog?.("stderr", stderrSuffix); return { exitCode: result.exitCode, signal: result.signal ?? null, diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 0e381340a9..7bcd679739 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -48,7 +48,7 @@ import { sandboxConfigFromLeaseMetadataLoose, } from "./sandbox-provider-runtime.js"; import { pluginRegistryService } from "./plugin-registry.js"; -import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import type { ExecuteLogSink, PluginWorkerManager } from "./plugin-worker-manager.js"; import { destroyPluginEnvironmentLease, executePluginEnvironmentCommand, @@ -202,6 +202,23 @@ export interface EnvironmentDriverExecuteInput extends EnvironmentDriverLeaseInp * span parents to the run trace. The default keeps the session path. */ bypassSession?: boolean; + /** + * Force the command onto the lease's persistent session even when no run step + * is active. The ACP process session bridge sets this so the long-lived agent + * command opens the session and streams its output through the session log + * stream. `bypassSession: true` still wins, so an explicit bypass is never + * overridden. The default keeps the context-based session selection. + */ + forceSession?: boolean; + /** + * Incremental log sink for one execute call. When set, the plugin worker + * delivers each `stdout` and `stderr` chunk to this sink through the + * `execute.log` notification while the command runs, before the final result. + * The runtime forwards it to the plugin worker manager, which routes each + * chunk to this sink by the host-issued invocation id. A driver that does not + * stream ignores it and returns only the final result. + */ + onLog?: ExecuteLogSink; } export interface EnvironmentDriverSyncInput extends EnvironmentDriverLeaseInput { @@ -1337,7 +1354,12 @@ function createSandboxEnvironmentDriver( // the first in-run command that carries a run parent (an agent tool // command runs under the run trace), whose setup span parents to the run // trace. A command that sets `bypassSession` explicitly always bypasses. - const bypassSession = input.bypassSession === true || activeStep === null; + // A command that sets `forceSession` keeps the session even with no + // active step: the ACP process session bridge runs the long-lived agent + // command this way, so the session opens and streams its output through + // the session log stream. An explicit `bypassSession` still wins. + const bypassSession = + input.bypassSession === true || (activeStep === null && input.forceSession !== true); const pluginId = readString(input.lease.metadata?.pluginId); const providerKey = readString(input.lease.metadata?.provider); if (pluginId && providerKey) { @@ -1374,7 +1396,7 @@ function createSandboxEnvironmentDriver( }, resolvePluginExecuteRpcTimeoutMs({ requestedTimeoutMs: input.timeoutMs, config: sanitizedConfig, - })); + }), input.onLog); } } throw new Error("Sandbox driver does not support direct command execution for built-in providers."); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index e0a5cbd0d8..28d152541e 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -107,6 +107,37 @@ const CRASH_WINDOW_MS = 10 * 60 * 1_000; /** Maximum number of stderr characters retained for worker failure context. */ const MAX_STDERR_EXCERPT_CHARS = 8_000; +/** Maximum characters accepted for one `execute.log` chunk. A larger chunk is + * dropped, so a faulty or hostile worker cannot flood the host with one + * unbounded notification. */ +const MAX_EXECUTE_LOG_CHUNK_CHARS = 1_000_000; + +/** + * Maximum characters accepted for one incoming worker stdout line before the + * host parses it as JSON. The host drops a longer line without a parse, so a + * faulty or hostile worker cannot force the host to parse an unbounded document + * and exhaust memory. The bound sits far above the largest legitimate framed + * message, so a real large command result still passes. A worker can override + * it through `WorkerStartOptions.executeLogLimits`. + */ +const MAX_WORKER_MESSAGE_CHARS = 128 * 1024 * 1024; + +/** + * Default ceiling for the total characters one execute call may stream through + * `execute.log`. The host counts the delivered characters for each active + * execute route and drops further chunks past this bound, so one runaway or + * hostile execution cannot flood the host and the run-log sink without limit. + * The final command result still delivers the complete output through its own + * capture path. A worker can override it through + * `WorkerStartOptions.executeLogLimits`. + */ +const MAX_EXECUTE_LOG_TOTAL_CHARS = 128 * 1024 * 1024; + +/** Minimum time between two dropped-`execute.log` debug records. The router + * rate-limits the record so a flood of dropped chunks writes at most one line + * per window with a running count. */ +const EXECUTE_LOG_DROP_LOG_INTERVAL_MS = 1_000; + // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- @@ -239,6 +270,18 @@ export interface WorkerStartOptions { * The host wires this to the PluginStreamBus to fan out events to SSE clients. */ onStreamNotification?: (method: string, params: Record) => void; + /** + * Framing and flood limits for the `execute.log` route. The defaults bound + * one incoming line before the JSON parse and the total streamed output for + * one execute call. A test overrides them to exercise the drop paths without + * huge inputs. + */ + executeLogLimits?: { + /** Max characters for one incoming worker line before the JSON parse. */ + maxIncomingMessageChars?: number; + /** Max total characters one execute call may stream through `execute.log`. */ + maxTotalCharsPerExecute?: number; + }; } /** @@ -268,6 +311,42 @@ interface ActiveInvocation { traceparent?: string; } +/** + * Sink for one incremental output chunk of an active `environmentExecute` call. + * The host runner passes it to `call` for the execute method, and the manager + * delivers each `execute.log` chunk to it. The sink may return a promise; the + * caller owns the ordering. + */ +export type ExecuteLogSink = ( + stream: "stdout" | "stderr", + chunk: string, +) => void | Promise; + +/** + * Host-owned route for one active execute call. The host mints the invocation + * id and stores the exact company id and log sink here. A worker never selects + * this record; the host looks it up by the host-issued invocation id on the + * message envelope. The company id is the single authority for the delivery + * target, so an `execute.log` notification never carries a company id. + */ +interface ExecuteLogRoute { + companyId: string; + onLog: ExecuteLogSink; + /** + * The count of characters delivered through this route. The router bounds the + * per-execute total and drops chunks past the configured ceiling. + */ + deliveredChars: number; + /** + * Latched when the router cannot bind the shared worker pipe to a single + * company, because a second company's execute overlapped this one. After the + * latch the router drops every further chunk for this route and lets the final + * command result deliver the complete output. The latch keeps the delivered + * prefix contiguous, so the run log never shows a gap. + */ + crossCompanyBlocked: boolean; +} + // --------------------------------------------------------------------------- // PluginWorkerHandle — manages a single worker process // --------------------------------------------------------------------------- @@ -316,6 +395,7 @@ export interface PluginWorkerHandle { method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, + executeLogSink?: ExecuteLogSink, ): Promise; /** @@ -424,6 +504,7 @@ export interface PluginWorkerManager { method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, + executeLogSink?: ExecuteLogSink, ): Promise; } @@ -460,6 +541,28 @@ export function createPluginWorkerHandle( const pendingRequests = new Map(); let nextRequestId = 1; const activeInvocations = new Map(); + // Host-owned execute routes, keyed by the host-issued invocation id. Only an + // `environmentExecute` call with a log sink registers a route here. The + // `execute.log` router delivers only through this map — never through the + // generic `activeInvocations` record — so a non-execute call can never become + // a log target. + const activeExecuteRoutes = new Map(); + // Rate-limit state for dropped `execute.log` notifications. The debug record + // never carries chunk bytes. + let executeLogDropCount = 0; + let executeLogDropLoggedAtMs = 0; + // Rate-limit state for dropped oversized worker lines. The warn record carries + // only the length, never the line bytes. + let oversizedLineDropCount = 0; + let oversizedLineLoggedAtMs = 0; + + // Framing and flood limits for the `execute.log` route. The defaults bound one + // incoming line before the JSON parse and the total streamed output for one + // execute call. A caller (a test) can lower them. + const maxIncomingMessageChars = + options.executeLogLimits?.maxIncomingMessageChars ?? MAX_WORKER_MESSAGE_CHARS; + const maxExecuteLogTotalChars = + options.executeLogLimits?.maxTotalCharsPerExecute ?? MAX_EXECUTE_LOG_TOTAL_CHARS; // ------------------------------------------------------------------ // Proactive company scopes (LOOA-629) @@ -547,6 +650,14 @@ export function createPluginWorkerHandle( function handleLine(line: string): void { if (!line.trim()) return; + // Enforce the framing bound BEFORE the JSON parse. A line longer than the + // limit is dropped without a parse, so a faulty or hostile worker cannot + // force the host to parse an unbounded document and exhaust memory. + if (line.length > maxIncomingMessageChars) { + dropOversizedLine(line.length); + return; + } + let message: unknown; try { message = parseMessage(line); @@ -658,6 +769,157 @@ export function createPluginWorkerHandle( activeInvocations.delete(invocation.id); } + // Store the host-owned execute route for one active execute call. The host + // holds the exact company id and log sink; the worker never supplies them. + function registerExecuteRoute( + invocationId: string, + companyId: string, + onLog: ExecuteLogSink, + ): void { + activeExecuteRoutes.set(invocationId, { + companyId, + onLog, + deliveredChars: 0, + crossCompanyBlocked: false, + }); + } + + function clearExecuteRoute(invocationId: string | undefined): void { + if (invocationId) activeExecuteRoutes.delete(invocationId); + } + + // Drop an oversized incoming worker line before the JSON parse. Write a + // rate-limited warn record with the length and a running drop count. The + // record never carries the line bytes. + function dropOversizedLine(lineLength: number): void { + oversizedLineDropCount += 1; + const nowMs = Date.now(); + if (nowMs - oversizedLineLoggedAtMs >= EXECUTE_LOG_DROP_LOG_INTERVAL_MS) { + log.warn( + { lineLength, maxIncomingMessageChars, droppedSinceLastLog: oversizedLineDropCount }, + "dropping oversized worker line before JSON parse", + ); + oversizedLineLoggedAtMs = nowMs; + oversizedLineDropCount = 0; + } + } + + // Drop an `execute.log` notification. Write a rate-limited debug record with + // the reason and a running drop count. The record never carries the chunk + // bytes, the company id, or command data. + function dropExecuteLogNotification(reason: string): void { + executeLogDropCount += 1; + const nowMs = Date.now(); + if (nowMs - executeLogDropLoggedAtMs >= EXECUTE_LOG_DROP_LOG_INTERVAL_MS) { + log.debug( + { reason, droppedSinceLastLog: executeLogDropCount }, + "dropping execute.log notification", + ); + executeLogDropLoggedAtMs = nowMs; + executeLogDropCount = 0; + } + } + + // Route one `execute.log` notification to its host-owned execute route. The + // route is the single authority for the delivery target and the company + // binding. This never reads a company id from the notification and never + // routes through the generic active-invocation record. + // + // Complete mediation: the host and the worker share one stdio pipe, and the + // worker process sees every active invocation id. So the host cannot prove + // which concurrent invocation produced a notification, and it must NOT treat + // the worker-supplied `paperclipInvocationId` alone as proof of origin. The + // host validates the exact company scope instead: it delivers only while every + // active execute route on this worker belongs to ONE company. When a second + // company's execute overlaps, the host fails closed — it latches the active + // routes and drops the chunk — so a worker that runs company A can never forge + // company B's active id and inject output into B's route. The final command + // result still delivers the complete output, so no byte is lost; only the live + // stream pauses while two companies overlap. + function routeExecuteLogNotification(notification: JsonRpcNotification): void { + const invocationId = readNonEmptyString( + (notification as { paperclipInvocationId?: unknown }).paperclipInvocationId, + ); + const params = isRecord(notification.params) ? notification.params : {}; + const stream = params.stream; + const chunk = params.chunk; + // Runtime-validate the payload. Drop invalid input without a throw. + if (stream !== "stdout" && stream !== "stderr") { + dropExecuteLogNotification("invalid-stream"); + return; + } + if ( + typeof chunk !== "string" || + chunk.length === 0 || + chunk.length > MAX_EXECUTE_LOG_CHUNK_CHARS + ) { + dropExecuteLogNotification("invalid-chunk"); + return; + } + if (!invocationId) { + dropExecuteLogNotification("missing-invocation"); + return; + } + const route = activeExecuteRoutes.get(invocationId); + if (!route) { + // No active execute route for this id: a late chunk after settlement or + // timeout, a non-execute invocation, or an unknown id. Drop it. + dropExecuteLogNotification("no-active-route"); + return; + } + // The route already lost single-company attribution earlier in its life, so + // it stays closed for the rest of the call. + if (route.crossCompanyBlocked) { + dropExecuteLogNotification("cross-company-scope"); + return; + } + // Validate the exact company scope. Deliver only while every active execute + // route on this worker belongs to one company. A second company's active + // route makes the shared pipe ambiguous, so the host fails closed: it + // latches every active route and drops the chunk. + let onlyCompanyId: string | null = null; + let crossCompany = false; + for (const active of activeExecuteRoutes.values()) { + if (onlyCompanyId === null) { + onlyCompanyId = active.companyId; + } else if (onlyCompanyId !== active.companyId) { + crossCompany = true; + break; + } + } + if (crossCompany) { + for (const active of activeExecuteRoutes.values()) { + active.crossCompanyBlocked = true; + } + dropExecuteLogNotification("cross-company-scope"); + return; + } + // Bound the total characters one execute call may stream. Past the ceiling + // the host drops further chunks, so one runaway or hostile execution cannot + // flood the host and the run-log sink without limit. + if (route.deliveredChars + chunk.length > maxExecuteLogTotalChars) { + dropExecuteLogNotification("execute-output-cap"); + return; + } + route.deliveredChars += chunk.length; + try { + const delivery = route.onLog(stream, chunk); + if (delivery && typeof (delivery as Promise).then === "function") { + void (delivery as Promise).catch((err) => { + log.error( + { err: err instanceof Error ? err.message : String(err) }, + "execute.log delivery failed", + ); + }); + } + } catch (err) { + log.error( + { err: err instanceof Error ? err.message : String(err) }, + "execute.log delivery threw", + ); + } + } + /** * Extract the single company a worker→host call references, mirroring the SDK * governed-access gate's own derivation (host-client-factory.ts @@ -811,6 +1073,13 @@ export function createPluginWorkerHandle( return; } + // Execute-log notifications: deliver one incremental output chunk to the + // host-owned execute route for the active execute call. + if (notification.method === "execute.log") { + routeExecuteLogNotification(notification); + return; + } + // Stream notifications: forward to the stream bus via callback if ( notification.method === "streams.open" || @@ -1273,6 +1542,7 @@ export function createPluginWorkerHandle( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, + executeLogSink?: ExecuteLogSink, ): Promise { const rpcPromise = new Promise((resolve, reject) => { if (!childProcess?.stdin?.writable) { @@ -1288,6 +1558,13 @@ export function createPluginWorkerHandle( const timeout = resolveRpcCallTimeoutMs(timeoutMs, rpcTimeoutMs); const invocationScope = deriveInvocationScope(method, params); const invocation = invocationScope ? registerInvocation(invocationScope) : null; + // Register the host-owned execute route only for an execute call that + // carries a log sink. The company id comes from the host-derived + // invocation scope, never from the worker. This binds the sink to the + // exact company for the life of the call. + if (invocation && invocationScope && executeLogSink && method === "environmentExecute") { + registerExecuteRoute(invocation.id, invocationScope.companyId, executeLogSink); + } // Guard against double-settlement. When a process exits all pending // requests are rejected via rejectAllPending(), but the timeout timer @@ -1301,6 +1578,7 @@ export function createPluginWorkerHandle( clearTimeout(timer); pendingRequests.delete(id); clearInvocation(invocation); + clearExecuteRoute(invocation?.id); fn(value); }; @@ -1343,6 +1621,7 @@ export function createPluginWorkerHandle( clearTimeout(timer); pendingRequests.delete(id); clearInvocation(invocation); + clearExecuteRoute(invocation?.id); reject( new Error( `Failed to send "${method}" to worker: ${ @@ -1396,6 +1675,7 @@ export function createPluginWorkerHandle( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, + executeLogSink?: ExecuteLogSink, ): Promise { if (status !== "running" && status !== "starting") { return Promise.reject( @@ -1404,7 +1684,7 @@ export function createPluginWorkerHandle( ), ); } - return callInternal(method, params, timeoutMs); + return callInternal(method, params, timeoutMs, executeLogSink); }, notify(method: string, params: unknown) { @@ -1635,6 +1915,7 @@ export function createPluginWorkerManager( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, + executeLogSink?: ExecuteLogSink, ): Promise { const handle = workers.get(pluginId); if (!handle) { @@ -1642,7 +1923,7 @@ export function createPluginWorkerManager( new Error(`No worker registered for plugin "${pluginId}"`), ); } - return handle.call(method, params, timeoutMs); + return handle.call(method, params, timeoutMs, executeLogSink); }, }; }