mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(runner): record pending calls at sidecar shutdown (#20272)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The ACPX sidecar carries task tools between a provider and its controller. > - A closed controller pipe rejects all pending tools with `sidecar stdin closed`. > - Current logs do not record the first shutdown trigger or the pending calls at that point. > - The sampled runs record a server SIGTERM before the 300-second tool idle timeouts and later pipe failures. The exact pipe-close cause remains unresolved. > - This pull request adds one bounded shutdown record before cleanup removes the evidence. > - Operators can use it to compare controller shutdown with pending tool calls. ## Linked Issues or Issue Description **What happened?** Several task document reads failed together with `sidecar stdin closed`. In two sampled staging traces, earlier task tools reached a 300-second idle timeout. The sidecar source maps the closure error to pending-call rejection during shutdown. It does not record the trigger or the pending call population. Pattern: [Task reads lose their tool connection](https://www.braintrust.dev/app/Paperclip/p/Paperclip%20Staging/patterns/5544e6f3-9e33-4661-95c3-5849c9a8f2b6). **Expected behavior** Record the local shutdown reason and a bounded snapshot of pending calls before cleanup removes them. Keep tool cancellation and authorization unchanged. **Steps to reproduce** 1. Start the sidecar with pending task calls. 2. Close controller stdin, or send SIGTERM or SIGINT. 3. Inspect the sidecar event stream and stderr. **Paperclip version or commit** Baseline: `33a23e024697552a4a639e49957a3443c75a5bd5`. ### Hypothesis and thesis for a fix The provisional hypothesis is that controller teardown causes the batch failures after an earlier stall. The source proves how EOF rejects pending tools. The traces establish ordering. They do not prove why the controller closed stdin or why the earlier calls stalled. Run logs now confirm an earlier server shutdown in both examples. One later recovery could not find the provider; the other later ended at a durable receipt limit. The first failed settlement and exact pipe-close trigger remain unresolved. The later example already used a provider revision containing #19365 and #19290, so those earlier fixes do not alone explain all occurrences. This is a **diagnostic-only change**. The sidecar owns the local shutdown signal and its pending waiters. Record that state once at the first shutdown request. Do not add retries or change deadlines without evidence. This supports each provider that uses this sidecar. It preserves company/task/run authority, approval checks, tool admission and cleanup ownership. It does not resolve or claim recovery of the original read stalls. ## What Changed - Emit `sidecar_shutdown_requested` through the existing diagnostic stream and stderr before shutdown can remove waiters. - Record pending tool, input, permission and completion-normalization counts. Include at most 16 oldest tool calls with elapsed time and a SHA-256 call ID. Exclude arguments, results, operation names and raw call IDs. - Add real sidecar process tests for EOF and both shutdown signals. Check successful resolution, cancelled calls, pending-call rejection, provider cleanup and sidecar exit with code 0 and no signal before fallback cleanup. - Document the record and its limits. Saved session formats and previous-release behavior stay unchanged. ## Verification - Candidate: `85a929eef87d5f775bbed9f2c98409432c86c702`. The baseline replay uses an isolated detached checkout at the baseline SHA with the exact candidate regression file. - Baseline red: the three process regressions failed because the expected shutdown record was absent. The sidecar initialized, handled calls and shut down before each assertion. - Unchanged process regressions pass with the implementation. The focused sidecar and snapshot suites passed: 117 tests, one unrelated macOS candidate-admission test skipped on Linux x64. - `cd packages/paperclip-runner && pnpm exec vitest run src/cli/acpx-runtime-sidecar.test.ts src/cli/acpx-sidecar-shutdown.test.ts` - `cd packages/paperclip-runner && node --test test/acpx-sidecar-contract.test.mjs`: 16 passed. - `cd packages/paperclip-runner && pnpm exec tsc -p tsconfig.json --noEmit`: passed after building the local eval-kernel dependency. - `git diff --check`: passed. - These tests use the real sidecar pipe and process boundary with a controlled provider host. Independent outcome and cleanup files prove rejection and process cleanup. They do not exercise a live authenticated Claude/staging run or prove deployed recovery. - The local preflight could not build Rust because this host has no Cargo. CI must supply the applicable Rust/build checks. At head `85a929eef87d5f775bbed9f2c98409432c86c702`, all 61 check runs are complete: 58 passed and three conditional jobs skipped. The legacy Snyk status passed. Greptile is 5/5 with both threads resolved. [CI result](https://github.com/paperclipai/paperclip/actions/runs/42630335496). ## Risks - The controller can stop reading before it receives the diagnostic frame. Stderr is an additional path, not a delivery guarantee. - EOF records a local pipe closure. It cannot identify the upstream controller cause by itself. - Hashed IDs require offline hashing of the existing tool-call journal for correlation. At most 16 samples are retained; counts include the full pending population. - No persistent state, retry, cancellation or authorization contract changes. ## Model Used OpenAI Codex, GPT-6, with reasoning, tool use and code execution. The runtime does not expose a more specific model ID or context window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
6 files changed
+326
-7
No files matched your search
@@ -5,6 +5,21 @@ Run-log events write to the `heartbeat_run_events` table
|
||||
Paperclip Telemetry events, and they are not OpenTelemetry exports. A run-log
|
||||
event needs no operator endpoint.
|
||||
|
||||
## ACPX sidecar shutdown
|
||||
|
||||
The ACPX sidecar emits `runtime.diagnostic` with code
|
||||
`sidecar_shutdown_requested` once at the first local shutdown request, before
|
||||
waiters are removed. Its JSON message records the local reason, pending tool,
|
||||
input and permission counts, completion-normalization count, and at most 16
|
||||
oldest pending tools. Each tool sample contains elapsed milliseconds and a
|
||||
SHA-256 call ID. Hash the corresponding `runtime.tool_called` call ID to match
|
||||
it. Arguments, results, operation names and raw call IDs are excluded.
|
||||
The existing diagnostic path also writes this message to sidecar stderr.
|
||||
A controller that has already stopped may not ingest the frame. Stdin EOF
|
||||
identifies a closed controller pipe, not the upstream reason for that closure.
|
||||
This is local run evidence; no first-party Telemetry event or OpenTelemetry
|
||||
span is added. Retry, cancellation and authorization behavior stay unchanged.
|
||||
|
||||
## Startup reuse evidence
|
||||
|
||||
`run.startup.reuse` uses schema `paperclip.startup-reuse.v1`. Each receipt names
|
||||
|
||||
@@ -193,6 +193,18 @@ and terminal settlement at the process boundary. Copilot and Pi remain gated.
|
||||
Verified distributions are build-owned; no provider accepts an arbitrary executable.
|
||||
See [the rich ACP capability report](../../doc/architecture/runner-rich-acp-capabilities.md).
|
||||
|
||||
On stdin EOF, SIGTERM, SIGINT, or an internal shutdown request, the sidecar emits
|
||||
`sidecar_shutdown_requested` once through `runtime.diagnostic` and stderr before
|
||||
cleanup removes pending calls. The JSON message gives the shutdown reason,
|
||||
pending tool/input/permission counts, completion-normalization count, and at most
|
||||
16 oldest tool calls with elapsed milliseconds and SHA-256 call IDs. Hash a
|
||||
`runtime.tool_called` call ID to correlate it with this snapshot. Arguments,
|
||||
results, operation names, and raw call IDs are excluded. This diagnostic records
|
||||
the local trigger; stdin EOF alone does not identify why the controller stopped.
|
||||
A broken output pipe can prevent event delivery; consult retained stderr too.
|
||||
It does not change retries, cancellation, or tool authorization. No saved session
|
||||
shape changes; sessions from the previous release retain their existing behavior.
|
||||
|
||||
If the controller sends assigned MCP tools in the PRP catalog, the sidecar uses
|
||||
its authenticated tool bridge.
|
||||
Isolated provider admission accepts this route without a separate native MCP endpoint.
|
||||
|
||||
@@ -1054,6 +1054,172 @@ describe("qualified ACPX runtime sidecar", () => {
|
||||
expect(sidecar.stderr()).not.toContain("ERR_USE_AFTER_CLOSE");
|
||||
});
|
||||
|
||||
it.each(["stdin", "SIGTERM", "SIGINT"] as const)("records pending calls before %s shutdown rejects them", async trigger => {
|
||||
const root = mkdtempSync(join(tmpdir(), "acpx-shutdown-"));
|
||||
const loader = join(root, "loader.mjs"), fixture = join(root, "host.mjs");
|
||||
const base = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href;
|
||||
writeFileSync(fixture, `import { AcpxRuntimeHost as Base } from ${JSON.stringify(base)};
|
||||
import { appendFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
export class AcpxRuntimeHost extends Base {
|
||||
static async open(options) { const host = await Base.open(options); Object.setPrototypeOf(host, this.prototype); return host; }
|
||||
startTurn({ requestId }) {
|
||||
for (const callId of ["resolved", "cancelled", "pending-1", "pending-2"]) {
|
||||
const controller = new AbortController();
|
||||
this.options.semanticTools.handler({callId, tool:"call_api", arguments:{secret:"PRIVATE_REQUEST_BODY"}, signal:controller.signal})
|
||||
.then(() => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":resolved\\n"),
|
||||
error => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":"+error.message+"\\n"));
|
||||
if (callId === "cancelled") controller.abort();
|
||||
}
|
||||
return {requestId, promptStarted:Promise.resolve(), result:this.result, events:(async function*(){})()};
|
||||
}
|
||||
}`);
|
||||
writeFileSync(loader, `import { registerHooks } from "node:module"; import { pathToFileURL } from "node:url";
|
||||
registerHooks({resolve(specifier,context,next){
|
||||
if(context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js"))
|
||||
return {url:pathToFileURL(${JSON.stringify(fixture)}).href,shortCircuit:true};
|
||||
return next(specifier,context);
|
||||
}});`);
|
||||
const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}});
|
||||
children.add(sidecar);
|
||||
const send = async (id:number, command:string, params:Record<string,unknown>) => {
|
||||
sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params});
|
||||
expect(await sidecar.next(frame=>frame.id === id)).toMatchObject({ok:true});
|
||||
};
|
||||
try {
|
||||
sidecar.write(initializeRequest(1,"codex"));
|
||||
expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true});
|
||||
await send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root,
|
||||
normalizedSessionId:"shutdown-session",permissionMode:"deny-all",systemInstructions:"Test shutdown.",tools:[]});
|
||||
await send(3,"run.attach",{runId:"shutdown-run",catalogRevision:1,tools:[]});
|
||||
await send(4,"turn.start",{turnId:"shutdown-turn",message:"Test pending calls."});
|
||||
await sidecar.next(frame=>frame.eventType === "runtime.tool_called" && (frame.payload as Record<string,unknown>).callId === "pending-2");
|
||||
await send(5,"tool.resolve",{turnId:"shutdown-turn",callId:"resolved",result:{private:"PRIVATE_RESPONSE_BODY"}});
|
||||
await expect(sidecar.shutdown(trigger)).resolves.toEqual({code:0, signal:null});
|
||||
expect(sidecar.stderr()).toContain("sidecar_shutdown_requested");
|
||||
const observation = await sidecar.next(frame=>frame.eventType === "runtime.diagnostic" && (frame.payload as Record<string,unknown>).code === "sidecar_shutdown_requested");
|
||||
expect(observation).toMatchObject({runId:"shutdown-run",turnId:"shutdown-turn"});
|
||||
const snapshot = JSON.parse((observation.payload as {message:string}).message);
|
||||
expect(snapshot).toMatchObject({reason:trigger === "stdin" ? "sidecar stdin closed" : `sidecar received ${trigger}`,
|
||||
pendingTools:2,pendingInputs:0,pendingPermissions:0,normalizingTools:0,truncatedTools:0});
|
||||
expect(snapshot.tools.map((tool:{callIdSha256:string})=>tool.callIdSha256)).toEqual(
|
||||
["pending-1","pending-2"].map(id=>createHash("sha256").update(id).digest("hex")));
|
||||
expect(snapshot.tools.every((tool:{elapsedMs:number})=>Number.isFinite(tool.elapsedMs) && tool.elapsedMs >= 0)).toBe(true);
|
||||
expect(sidecar.stderr()).not.toContain("PRIVATE_");
|
||||
expect(sidecar.pending(frame=>frame.eventType === "runtime.diagnostic" && (frame.payload as Record<string,unknown>).code === "sidecar_shutdown_requested")).toBe(0);
|
||||
const outcomes = readFileSync(join(root,"outcomes"),"utf8");
|
||||
expect(outcomes).toContain("resolved:resolved");
|
||||
expect(outcomes).toContain("cancelled:ACPX tool call was cancelled");
|
||||
expect(outcomes).toContain(`pending-1:${snapshot.reason}`);
|
||||
expect(outcomes).toContain(`pending-2:${snapshot.reason}`);
|
||||
expect(JSON.parse(readFileSync(join(root,"provider-cleanup.json"),"utf8"))).toMatchObject({closeCount:1,code:0});
|
||||
} finally { await sidecar.close(); rmSync(root,{recursive:true,force:true}); }
|
||||
}, 15000);
|
||||
|
||||
it("finishes cleanup when the controller closed the output pipe before shutdown", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "acpx-shutdown-"));
|
||||
const loader = join(root, "loader.mjs"), fixture = join(root, "host.mjs");
|
||||
const base = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href;
|
||||
writeFileSync(fixture, `import { AcpxRuntimeHost as Base } from ${JSON.stringify(base)};
|
||||
import { appendFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
export class AcpxRuntimeHost extends Base {
|
||||
static async open(options) { const host = await Base.open(options); Object.setPrototypeOf(host, this.prototype); return host; }
|
||||
startTurn({ requestId }) {
|
||||
for (const callId of ["resolved", "cancelled", "pending-1", "pending-2"]) {
|
||||
const controller = new AbortController();
|
||||
this.options.semanticTools.handler({callId, tool:"call_api", arguments:{secret:"PRIVATE_REQUEST_BODY"}, signal:controller.signal})
|
||||
.then(() => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":resolved\\n"),
|
||||
error => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":"+error.message+"\\n"));
|
||||
if (callId === "cancelled") controller.abort();
|
||||
}
|
||||
return {requestId, promptStarted:Promise.resolve(), result:this.result, events:(async function*(){})()};
|
||||
}
|
||||
}`);
|
||||
writeFileSync(loader, `import { registerHooks } from "node:module"; import { pathToFileURL } from "node:url";
|
||||
registerHooks({resolve(specifier,context,next){
|
||||
if(context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js"))
|
||||
return {url:pathToFileURL(${JSON.stringify(fixture)}).href,shortCircuit:true};
|
||||
return next(specifier,context);
|
||||
}});`);
|
||||
const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}});
|
||||
children.add(sidecar);
|
||||
const send = async (id:number, command:string, params:Record<string,unknown>) => {
|
||||
sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params});
|
||||
expect(await sidecar.next(frame=>frame.id === id)).toMatchObject({ok:true});
|
||||
};
|
||||
try {
|
||||
sidecar.write(initializeRequest(1,"codex"));
|
||||
expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true});
|
||||
await send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root,
|
||||
normalizedSessionId:"shutdown-session",permissionMode:"deny-all",systemInstructions:"Test shutdown.",tools:[]});
|
||||
await send(3,"run.attach",{runId:"shutdown-run",catalogRevision:1,tools:[]});
|
||||
await send(4,"turn.start",{turnId:"shutdown-turn",message:"Test pending calls."});
|
||||
await sidecar.next(frame=>frame.eventType === "runtime.tool_called" && (frame.payload as Record<string,unknown>).callId === "pending-2");
|
||||
sidecar.closeOutput();
|
||||
await expect(sidecar.shutdown("stdin")).resolves.toEqual({code:1, signal:null});
|
||||
expect(sidecar.stderr()).toContain("sidecar_shutdown_requested");
|
||||
expect(sidecar.stderr()).not.toMatch(/Unhandled 'error' event|EPIPE/);
|
||||
const outcomes = readFileSync(join(root,"outcomes"),"utf8");
|
||||
expect(outcomes).toContain("pending-1:sidecar stdin closed");
|
||||
expect(outcomes).toContain("pending-2:sidecar stdin closed");
|
||||
expect(JSON.parse(readFileSync(join(root,"provider-cleanup.json"),"utf8"))).toMatchObject({closeCount:1,code:0});
|
||||
} finally { await sidecar.close(); rmSync(root,{recursive:true,force:true}); }
|
||||
}, 15000);
|
||||
|
||||
it("shuts down and finishes cleanup when the controller output pipe closes mid-turn", async () => {
|
||||
const root = mkdtempSync(join(tmpdir(), "acpx-shutdown-"));
|
||||
const loader = join(root, "loader.mjs"), fixture = join(root, "host.mjs");
|
||||
const base = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href;
|
||||
writeFileSync(fixture, `import { AcpxRuntimeHost as Base } from ${JSON.stringify(base)};
|
||||
import { appendFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
export class AcpxRuntimeHost extends Base {
|
||||
static async open(options) { const host = await Base.open(options); Object.setPrototypeOf(host, this.prototype); return host; }
|
||||
startTurn({ requestId }) {
|
||||
for (const callId of ["resolved", "cancelled", "pending-1", "pending-2"]) {
|
||||
const controller = new AbortController();
|
||||
this.options.semanticTools.handler({callId, tool:"call_api", arguments:{secret:"PRIVATE_REQUEST_BODY"}, signal:controller.signal})
|
||||
.then(() => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":resolved\\n"),
|
||||
error => appendFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT,"outcomes"), callId+":"+error.message+"\\n"));
|
||||
if (callId === "cancelled") controller.abort();
|
||||
}
|
||||
return {requestId, promptStarted:Promise.resolve(), result:this.result, events:(async function*(){})()};
|
||||
}
|
||||
}`);
|
||||
writeFileSync(loader, `import { registerHooks } from "node:module"; import { pathToFileURL } from "node:url";
|
||||
registerHooks({resolve(specifier,context,next){
|
||||
if(context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js"))
|
||||
return {url:pathToFileURL(${JSON.stringify(fixture)}).href,shortCircuit:true};
|
||||
return next(specifier,context);
|
||||
}});`);
|
||||
const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}});
|
||||
children.add(sidecar);
|
||||
const send = async (id:number, command:string, params:Record<string,unknown>) => {
|
||||
sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params});
|
||||
expect(await sidecar.next(frame=>frame.id === id)).toMatchObject({ok:true});
|
||||
};
|
||||
try {
|
||||
sidecar.write(initializeRequest(1,"codex"));
|
||||
expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true});
|
||||
await send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root,
|
||||
normalizedSessionId:"shutdown-session",permissionMode:"deny-all",systemInstructions:"Test shutdown.",tools:[]});
|
||||
await send(3,"run.attach",{runId:"shutdown-run",catalogRevision:1,tools:[]});
|
||||
await send(4,"turn.start",{turnId:"shutdown-turn",message:"Test pending calls."});
|
||||
await sidecar.next(frame=>frame.eventType === "runtime.tool_called" && (frame.payload as Record<string,unknown>).callId === "pending-2");
|
||||
sidecar.closeOutput();
|
||||
sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id:5,command:"tool.resolve",
|
||||
params:{turnId:"shutdown-turn",callId:"resolved",result:{private:"PRIVATE_RESPONSE_BODY"}}});
|
||||
await expect(sidecar.exited()).resolves.toEqual({code:1, signal:null});
|
||||
expect(sidecar.stderr()).toContain("sidecar_shutdown_requested");
|
||||
expect(sidecar.stderr()).not.toMatch(/Unhandled 'error' event|EPIPE/);
|
||||
const outcomes = readFileSync(join(root,"outcomes"),"utf8");
|
||||
expect(outcomes).toContain("pending-1:ACPX sidecar output failed");
|
||||
expect(outcomes).toContain("pending-2:ACPX sidecar output failed");
|
||||
expect(JSON.parse(readFileSync(join(root,"provider-cleanup.json"),"utf8"))).toMatchObject({closeCount:1,code:0});
|
||||
} finally { await sidecar.close(); rmSync(root,{recursive:true,force:true}); }
|
||||
}, 15000);
|
||||
|
||||
it("keeps session admission closed while any cleanup owner remains", () => {
|
||||
const cleanup = Promise.resolve();
|
||||
|
||||
@@ -1729,6 +1895,7 @@ class SidecarProcess {
|
||||
readonly #signals: Array<() => void> = [];
|
||||
#stderr = "";
|
||||
#closed = false;
|
||||
readonly #exit: Promise<{code:number|null;signal:NodeJS.Signals|null}>;
|
||||
|
||||
constructor(options: {loader?:string;environment?:Record<string,string>} = {}) {
|
||||
this.#child = spawn(
|
||||
@@ -1736,6 +1903,7 @@ class SidecarProcess {
|
||||
[...(options.loader ? ["--import","tsx","--import",options.loader] : []), fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url))],
|
||||
{ stdio: ["pipe", "pipe", "pipe"], env: { PATH: process.env.PATH, LANG: "C.UTF-8", ...options.environment } },
|
||||
);
|
||||
this.#exit = new Promise(resolve => this.#child.once("exit", (code, signal) => resolve({code,signal})));
|
||||
let stdout = "";
|
||||
this.#child.stdout.setEncoding("utf8");
|
||||
this.#child.stdout.on("data", (chunk: string) => {
|
||||
@@ -1760,6 +1928,28 @@ class SidecarProcess {
|
||||
this.#child.stdin.write(`${JSON.stringify(value)}\n`);
|
||||
}
|
||||
|
||||
closeOutput(): void {
|
||||
this.#child.stdout.destroy();
|
||||
}
|
||||
|
||||
async shutdown(trigger: "stdin" | "SIGTERM" | "SIGINT") {
|
||||
if (trigger === "stdin") this.#child.stdin.end();
|
||||
else this.#child.kill(trigger);
|
||||
return this.exited();
|
||||
}
|
||||
|
||||
async exited() {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
this.#exit,
|
||||
new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("Sidecar did not exit without forced cleanup")), 5000);
|
||||
}),
|
||||
]);
|
||||
} finally { clearTimeout(timer); }
|
||||
}
|
||||
|
||||
stderr(): string {
|
||||
return this.#stderr;
|
||||
}
|
||||
@@ -1797,7 +1987,7 @@ class SidecarProcess {
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.#closed) return;
|
||||
if (this.#closed || this.#child.exitCode !== null || this.#child.signalCode !== null) return;
|
||||
this.#closed = true;
|
||||
this.#child.stdin.end();
|
||||
const exit = new Promise<void>((resolve) => {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
#!/usr/bin/env node
|
||||
import { recordSidecarShutdownThenQueue, sidecarShutdownSnapshot } from "./acpx-sidecar-shutdown.js";
|
||||
import { classifyClaudeStatusBanner } from "../drivers/acpx/claude-status-banners.js";
|
||||
import { resolveGrokDisableWebSearch } from "../drivers/acpx/grok-web-search.js";
|
||||
import { toolFileChangesFromAcpContent, workspaceRootsFor, type ToolFileChange } from "../tool-file-changes.js";
|
||||
@@ -120,6 +121,11 @@ const MAX_PENDING_INPUTS = 16;
|
||||
const MAX_BUFFERED_OUTPUT_BYTES = 4 * ACPX_SIDECAR_MAX_FRAME_BYTES;
|
||||
const MAX_EVENT_PAYLOAD_BYTES = 256 * 1024;
|
||||
let outputFailed = false;
|
||||
process.stdout.on("error", () => {
|
||||
outputFailed = true;
|
||||
requestShutdown("ACPX sidecar output failed");
|
||||
});
|
||||
process.stderr.on("error", () => undefined);
|
||||
let goalSourceRevision = 0;
|
||||
function observedGoalProjection(...args: Parameters<typeof acpxGoalProjection>) {
|
||||
return { ...acpxGoalProjection(...args), providerRevision: ++goalSourceRevision };
|
||||
@@ -145,6 +151,7 @@ function reportRetainedAcpxCleanupFailure(
|
||||
|
||||
interface PendingTool {
|
||||
turnId: string;
|
||||
startedAt: number;
|
||||
settle(value: unknown): void;
|
||||
reject(error: Error): void;
|
||||
cleanup(): void;
|
||||
@@ -225,12 +232,19 @@ process.once("SIGINT", () => {
|
||||
function requestShutdown(reason: string): void {
|
||||
if (shutdownRequested) return;
|
||||
shutdownRequested = true;
|
||||
if (!inputClosed) lines.pause();
|
||||
pendingInput = enqueueAcpxSidecarInput(
|
||||
pendingInput,
|
||||
() => shutdown(reason),
|
||||
(error) => diagnostic("sidecar_shutdown_failed", safeMessage(error)),
|
||||
);
|
||||
if (!recordSidecarShutdownThenQueue(
|
||||
() => diagnostic("sidecar_shutdown_requested", JSON.stringify(sidecarShutdownSnapshot(
|
||||
reason, tools, inputs.size, permissions.size, normalizingTools.size,
|
||||
))),
|
||||
() => {
|
||||
if (!inputClosed) lines.pause();
|
||||
pendingInput = enqueueAcpxSidecarInput(
|
||||
pendingInput,
|
||||
() => shutdown(reason),
|
||||
(error) => diagnostic("sidecar_shutdown_failed", safeMessage(error)),
|
||||
);
|
||||
},
|
||||
)) outputFailed = true;
|
||||
}
|
||||
|
||||
async function receiveLine(line: string): Promise<void> {
|
||||
@@ -907,6 +921,7 @@ async function waitForTool(call: RunnerToolCall): Promise<unknown> {
|
||||
call.signal.addEventListener("abort", abort, { once: true });
|
||||
tools.set(callId, {
|
||||
turnId: activeTurnId,
|
||||
startedAt: performance.now(),
|
||||
settle: (result) => {
|
||||
// A pipe write alone does not prove receiver admission. Only this
|
||||
// call's turn-bound tool.resolve success confirms runnerd accepted
|
||||
@@ -939,6 +954,7 @@ async function waitForTool(call: RunnerToolCall): Promise<unknown> {
|
||||
call.signal.addEventListener("abort", abort, { once: true });
|
||||
tools.set(callId, {
|
||||
turnId: activeTurnId,
|
||||
startedAt: performance.now(),
|
||||
settle,
|
||||
reject,
|
||||
cleanup: () => call.signal.removeEventListener("abort", abort),
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { recordSidecarShutdownThenQueue, sidecarShutdownSnapshot } from "./acpx-sidecar-shutdown.js";
|
||||
|
||||
describe("sidecar shutdown snapshot", () => {
|
||||
it("bounds samples, keeps the oldest calls and omits private data", () => {
|
||||
const tools = new Map(Array.from({ length: 512 }, (_, index) => [
|
||||
`private-call-${index}`,
|
||||
{ startedAt: 1000 - index, arguments: "PRIVATE_INPUT", result: "PRIVATE_OUTPUT" },
|
||||
]));
|
||||
const snapshot = sidecarShutdownSnapshot("sidecar stdin closed", tools, 3, 2, 1, 2000);
|
||||
expect(snapshot).toMatchObject({ pendingTools: 512, pendingInputs: 3, pendingPermissions: 2,
|
||||
normalizingTools: 1, truncatedTools: 496 });
|
||||
expect(snapshot.tools).toHaveLength(16);
|
||||
expect(snapshot.tools[0]).toEqual({
|
||||
callIdSha256: createHash("sha256").update("private-call-511").digest("hex"), elapsedMs: 1511,
|
||||
});
|
||||
expect(snapshot.tools[15]?.elapsedMs).toBe(1496);
|
||||
const encoded = JSON.stringify(snapshot);
|
||||
expect(Buffer.byteLength(encoded)).toBeLessThan(4096);
|
||||
expect(encoded).not.toMatch(/PRIVATE|private-call/);
|
||||
expect(tools.size).toBe(512);
|
||||
});
|
||||
|
||||
it("records an idle shutdown without retaining completed calls", () => {
|
||||
expect(sidecarShutdownSnapshot("sidecar received SIGTERM", new Map(), 0, 0, 0)).toEqual({
|
||||
reason: "sidecar received SIGTERM", pendingTools: 0, pendingInputs: 0,
|
||||
pendingPermissions: 0, normalizingTools: 0, tools: [], truncatedTools: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("records the snapshot before queueing shutdown", () => {
|
||||
const order: string[] = [];
|
||||
expect(recordSidecarShutdownThenQueue(() => order.push("record"), () => order.push("queue"))).toBe(true);
|
||||
expect(order).toEqual(["record", "queue"]);
|
||||
});
|
||||
|
||||
it("still queues shutdown when the snapshot write throws", () => {
|
||||
const order: string[] = [];
|
||||
expect(recordSidecarShutdownThenQueue(() => {
|
||||
order.push("record");
|
||||
throw new Error("EPIPE: stdout closed");
|
||||
}, () => order.push("queue"))).toBe(false);
|
||||
expect(order).toEqual(["record", "queue"]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,40 @@
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
const MAX_SHUTDOWN_TOOL_SAMPLES = 16;
|
||||
|
||||
export function sidecarShutdownSnapshot(
|
||||
reason: string,
|
||||
tools: ReadonlyMap<string, { startedAt: number }>,
|
||||
pendingInputs: number,
|
||||
pendingPermissions: number,
|
||||
normalizingTools: number,
|
||||
now = performance.now(),
|
||||
) {
|
||||
const samples = [...tools]
|
||||
.sort((left, right) => left[1].startedAt - right[1].startedAt)
|
||||
.slice(0, MAX_SHUTDOWN_TOOL_SAMPLES)
|
||||
.map(([callId, pending]) => ({
|
||||
callIdSha256: createHash("sha256").update(callId).digest("hex"),
|
||||
elapsedMs: Math.max(0, Math.floor(now - pending.startedAt)),
|
||||
}));
|
||||
return {
|
||||
reason,
|
||||
pendingTools: tools.size,
|
||||
pendingInputs,
|
||||
pendingPermissions,
|
||||
normalizingTools,
|
||||
tools: samples,
|
||||
truncatedTools: tools.size - samples.length,
|
||||
};
|
||||
}
|
||||
|
||||
export function recordSidecarShutdownThenQueue(record: () => void, queueShutdown: () => void): boolean {
|
||||
let recorded = true;
|
||||
try {
|
||||
record();
|
||||
} catch {
|
||||
recorded = false;
|
||||
}
|
||||
queueShutdown();
|
||||
return recorded;
|
||||
}
|
||||
Reference in new issue
Block a user