feat(acp): stream Daytona sandbox agent output and remove the host output poll (#11049)
## Thinking Path > - Paperclip is the open source app that manages AI agents for work > - Sandbox providers let agents run in remote and isolated environments > - Daytona session commands need a path that sends agent output to the host without host polling > - Host polling adds delay and repeats provider output work > - This pull request adds typed execute.log notifications and a log sink for incremental output > - This pull request adds an optional ACP session stream with final-result replay protection > - The benefit is lower output delay while the default flags keep current behavior unchanged ## Linked Issues or Issue Description **Subsystem affected** Cross-cutting. The change spans the plugin SDK, Daytona provider, adapter utilities, and server execution services. **Problem or motivation** The Daytona ACP bridge polls a host output file while an agent command runs. This adds delay and can repeat work. The host also needs a safe route for provider output chunks. **Proposed solution** Add a typed `execute.log` notification with host-issued invocation correlation. Add an ordered log sink to the environment execute path. Add an optional ACP session-log path that parses newline-delimited JSON frames and removes the host output poll for that path. **Alternatives considered** Keep the output-file poll as the only path. This keeps the current behavior but does not provide timely output. The new path stays behind flags, so the existing path remains the default fallback. **Roadmap alignment** This change supports the shipped Cloud / Sandbox agents milestone in `ROADMAP.md`, including Daytona support. ## What Changed - Add the typed `execute.log` worker-to-host notification and company-scoped host route. - Add ordered `stdout` and `stderr` chunk delivery before the final execute result. - Add the Daytona session log sink and the optional ACP streamed session path. - Add monotonic frame handling so live and final output reach the host once. - Keep `useLogStream` and `streamAgentSessionOutput` off by default. - Add unit and integration coverage for the notification, execution target, runtime, and Daytona paths. ## Verification - Run adapter-utils tests: 445 tests pass locally. - Run server environment tests: 73 tests pass locally. - Run Daytona plugin tests: 131 tests pass locally. - Run TypeScript checks for shared, adapter-utils, and server. - Review the pull request checks after GitHub completes them. - All required GitHub checks pass on the current head. ## Risks The new paths change output delivery only when a feature flag enables them. The final execute result remains available for parsing and fallback. The main risk is a provider stream or frame-order error; the final-result parser limits that risk. ## Model Used OpenAI Codex, GPT-5, tool use and code execution. The runtime did not supply a context-window value. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes No operator documentation change applies because both new flags remain disabled by default. - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
0a511ed1b0
commit
6b7e0814a0
|
|
@ -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,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -34,6 +34,27 @@ export interface CommandManagedRuntimeRunner {
|
|||
timeoutMs?: number;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
onSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
|
||||
/**
|
||||
* 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<RunProcessResult>;
|
||||
/**
|
||||
* Optional native inbound file transfer. Present only when the sandbox
|
||||
|
|
|
|||
|
|
@ -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<number | null>((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<typeof delegate.execute>[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",
|
||||
|
|
|
|||
|
|
@ -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<AdapterExecutionTargetProcessSessionBridgeHandle | null> {
|
||||
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: {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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<SessionLogStreamResult> {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, (response: JsonRpcResponse) => 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<unknown>((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" },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<unknown>;
|
||||
} }).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<unknown> },
|
||||
): Promise<Array<[string, string]>> {
|
||||
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<void> };
|
||||
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<unknown> },
|
||||
);
|
||||
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<void> };
|
||||
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<unknown> },
|
||||
);
|
||||
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<unknown> },
|
||||
);
|
||||
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<void> };
|
||||
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<unknown> },
|
||||
);
|
||||
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<void> }) => {
|
||||
// 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<void> }) => {
|
||||
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<unknown> }).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<void> }) => {
|
||||
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<unknown> }).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
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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(),
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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}` },
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string, unknown>) {
|
||||
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<string, unknown>,
|
||||
): 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);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>,
|
||||
envelope: Record<string, unknown>,
|
||||
): Record<string, unknown> {
|
||||
const merged: Record<string, unknown> = { ...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<string, unknown> | 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)
|
||||
|
|
|
|||
|
|
@ -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<void> = Promise.resolve();
|
||||
let deliveredStdout = "";
|
||||
let deliveredStderr = "";
|
||||
const onIncrementalLog = (
|
||||
stream: "stdout" | "stderr",
|
||||
chunk: string,
|
||||
): Promise<void> => {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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.");
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>) => 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<void>;
|
||||
|
||||
/**
|
||||
* 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<HostToWorkerMethods[M][1]>;
|
||||
|
||||
/**
|
||||
|
|
@ -424,6 +504,7 @@ export interface PluginWorkerManager {
|
|||
method: M,
|
||||
params: HostToWorkerMethods[M][0],
|
||||
timeoutMs?: number,
|
||||
executeLogSink?: ExecuteLogSink,
|
||||
): Promise<HostToWorkerMethods[M][1]>;
|
||||
}
|
||||
|
||||
|
|
@ -460,6 +541,28 @@ export function createPluginWorkerHandle(
|
|||
const pendingRequests = new Map<string | number, PendingRequest>();
|
||||
let nextRequestId = 1;
|
||||
const activeInvocations = new Map<string, ActiveInvocation>();
|
||||
// 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<string, ExecuteLogRoute>();
|
||||
// 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<void>).then === "function") {
|
||||
void (delivery as Promise<void>).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<HostToWorkerMethods[M][1]> {
|
||||
const rpcPromise = new Promise<HostToWorkerMethods[M][1]>((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<HostToWorkerMethods[M][1]> {
|
||||
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<HostToWorkerMethods[M][1]> {
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue