From 066b4f0160883d01dfa433f034c0d91047ed7964 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 10 Sep 2026 01:51:04 -0500 Subject: [PATCH] Stop legacy sandbox execution before completing cancellation Bind remote CLI and ACP processes to host-owned cancellation scopes, wait for teardown and final saves, and distinguish deliberate bridge shutdown from transport loss. Co-Authored-By: Paperclip --- doc/sandbox-work-folders.md | 12 ++ .../src/adapter-run-cancellation.test.ts | 45 +++++ .../src/adapter-run-cancellation.ts | 67 ++++++++ .../src/cancellable-sandbox-command.test.ts | 78 +++++++++ .../src/cancellable-sandbox-command.ts | 82 +++++++++ .../src/execution-target-sandbox.test.ts | 93 ++++++++++ .../adapter-utils/src/execution-target.ts | 162 +++++++++--------- .../heartbeat-process-metadata.test.ts | 54 ++++++ server/src/services/heartbeat.ts | 34 +++- 9 files changed, 544 insertions(+), 83 deletions(-) create mode 100644 packages/adapter-utils/src/adapter-run-cancellation.test.ts create mode 100644 packages/adapter-utils/src/adapter-run-cancellation.ts create mode 100644 packages/adapter-utils/src/cancellable-sandbox-command.test.ts create mode 100644 packages/adapter-utils/src/cancellable-sandbox-command.ts diff --git a/doc/sandbox-work-folders.md b/doc/sandbox-work-folders.md index e7a524ddd0..20c3a0f7f4 100644 --- a/doc/sandbox-work-folders.md +++ b/doc/sandbox-work-folders.md @@ -417,6 +417,18 @@ Only credential acquisition is retried, before starting Git or `gh`; repository operations are never replayed. Acceptance must exercise both transport paths and record which one was actually selected. +Controller-requested bridge shutdown marks the transport complete before closing +its provider channel. A connection loss observed earlier remains latched; closing +the native Git bridge after execution must not invent a transport failure. + +Legacy sandbox cancellation stops the owned remote CLI process group or ACP +process session before waiting for run teardown and the final file flush. The +host sends a command-scoped cancellation marker for CLI execution; remote PIDs +are never passed to the host process killer. Cancellation is persisted before +stopping execution so its exit cannot admit an automatic retry. Failed stop +requests remain visible and can be retried explicitly. Local and native runner +cancellation retain their existing authorities. + Automated tests do not qualify a deployed runner image. Before merging, use a new pinned staging stack with the branch's Cloud image and matching migrator. The deployed harness must target that tenant URL without launching a local diff --git a/packages/adapter-utils/src/adapter-run-cancellation.test.ts b/packages/adapter-utils/src/adapter-run-cancellation.test.ts new file mode 100644 index 0000000000..75d1bf3bb8 --- /dev/null +++ b/packages/adapter-utils/src/adapter-run-cancellation.test.ts @@ -0,0 +1,45 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + beginAdapterRunCancellation, bindAdapterRunStop, cancelAdapterRunExecution, + finishAdapterRunCancellation, hasAdapterRunCancellation, throwIfAdapterRunCancelled, +} from "./adapter-run-cancellation.js"; + +afterEach(() => finishAdapterRunCancellation("run")); +describe("sandbox adapter cancellation ownership", () => { + it("waits for both remote stop and final save before completing cancellation", async () => { + beginAdapterRunCancellation("run"); + let finishStop!: () => void; + const stop = vi.fn(() => new Promise((resolve) => { finishStop = resolve; })); + const cleanup = await bindAdapterRunStop("run", stop); + let completed = false; + const cancellation = cancelAdapterRunExecution("run").then(() => { completed = true; }); + expect(() => throwIfAdapterRunCancelled("run")).toThrow("cancelled"); + finishStop(); + await cleanup(); + expect(completed).toBe(false); + finishAdapterRunCancellation("run"); + await cancellation; + expect(stop).toHaveBeenCalledOnce(); + expect(hasAdapterRunCancellation("run")).toBe(false); + }); + + it("stops a resource acquired after cancellation instead of exposing it", async () => { + beginAdapterRunCancellation("run"); + const cancellation = cancelAdapterRunExecution("run"); + const stop = vi.fn(async () => {}); + await expect(bindAdapterRunStop("run", stop)).rejects.toThrow("cancelled"); + expect(stop).toHaveBeenCalledOnce(); + finishAdapterRunCancellation("run"); + await cancellation; + }); + + it("does not acknowledge a failed stop as completed or affect another run", async () => { + beginAdapterRunCancellation("run"); + const failure = new Error("provider unavailable"); + await bindAdapterRunStop("run", async () => { throw failure; }); + await expect(cancelAdapterRunExecution("run")).rejects.toBe(failure); + expect(hasAdapterRunCancellation("run")).toBe(true); + await cancelAdapterRunExecution("unrelated"); + expect(() => throwIfAdapterRunCancelled("unrelated")).not.toThrow(); + }); +}); diff --git a/packages/adapter-utils/src/adapter-run-cancellation.ts b/packages/adapter-utils/src/adapter-run-cancellation.ts new file mode 100644 index 0000000000..c5cd0d833a --- /dev/null +++ b/packages/adapter-utils/src/adapter-run-cancellation.ts @@ -0,0 +1,67 @@ +/** Host-owned cancellation scopes. Neither provider IDs nor host PIDs cross this seam. */ +type Stop = () => Promise; +interface Scope { + cancelled: boolean; + stops: Set; + finished: Promise; + finish(): void; +} +const scopes = new Map(); + +export function beginAdapterRunCancellation(runId: string): void { + if (scopes.has(runId)) throw new Error("Adapter cancellation scope already exists"); + let finish!: () => void; + const finished = new Promise((resolve) => { finish = resolve; }); + scopes.set(runId, { cancelled: false, stops: new Set(), finished, finish }); +} + +export function hasAdapterRunCancellation(runId: string): boolean { + return scopes.has(runId); +} + +export function throwIfAdapterRunCancelled(runId: string): void { + if (scopes.get(runId)?.cancelled) { + throw Object.assign(new Error("Adapter run cancelled by control plane"), { code: "ADAPTER_RUN_CANCELLED" }); + } +} + +export function registerAdapterRunStop(runId: string, stop: Stop): () => void { + const scope = scopes.get(runId); + scope?.stops.add(stop); + return () => { scope?.stops.delete(stop); }; +} + +/** Call only after the run's final save and resource teardown have settled. */ +export function finishAdapterRunCancellation(runId: string): void { + const scope = scopes.get(runId); + if (!scope) return; + scopes.delete(runId); + scope.finish(); +} + +/** The caller persists cancellation before stopping execution, preventing retry admission. */ +export async function cancelAdapterRunExecution(runId: string): Promise { + const scope = scopes.get(runId); + if (!scope) return; + scope.cancelled = true; + // Keep failed stops available for an explicit retry. Never mistake a failed + // control request for proof that execution or its final save has completed. + await Promise.all([...scope.stops].map((stop) => stop())); + await scope.finished; +} + +/** A resource acquired during cancellation must be stopped before it is exposed. */ +export async function bindAdapterRunStop(runId: string, stop: Stop): Promise { + let stopped: Promise | undefined; + const once = () => stopped ??= stop(); + const unregister = registerAdapterRunStop(runId, once); + try { throwIfAdapterRunCancelled(runId); } + catch (error) { + unregister(); + await once(); + throw error; + } + return async () => { + try { await once(); } finally { unregister(); } + }; +} diff --git a/packages/adapter-utils/src/cancellable-sandbox-command.test.ts b/packages/adapter-utils/src/cancellable-sandbox-command.test.ts new file mode 100644 index 0000000000..3c39cd5448 --- /dev/null +++ b/packages/adapter-utils/src/cancellable-sandbox-command.test.ts @@ -0,0 +1,78 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { executeCancellableSandboxCommand } from "./cancellable-sandbox-command.js"; +import type { CommandManagedRuntimeRunner } from "./command-managed-runtime.js"; +import { runAdapterExecutionTargetProcess } from "./execution-target.js"; +import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation } from "./adapter-run-cancellation.js"; + +describe("sandbox CLI cancellation through the execution target", () => { + it("retains a failed cancellation for an explicit retry", async () => { + const runId = randomUUID(); + beginAdapterRunCancellation(runId); + const result = { exitCode: 143, signal: null, stdout: "", stderr: "", timedOut: false, pid: null, startedAt: null }; + let complete!: (value: typeof result) => void; + const running = new Promise((resolve) => { complete = resolve; }); + let requests = 0; + const runner: CommandManagedRuntimeRunner = { execute: async (input) => { + if (input.bypassSession) { + requests++; + if (requests === 1) return { ...result, exitCode: 1 }; + complete(result); + return { ...result, exitCode: 0 }; + } + return running; + } }; + const execution = executeCancellableSandboxCommand(runId, runner, { command: "node", args: [] }, 100) + .finally(() => finishAdapterRunCancellation(runId)); + try { + await expect(cancelAdapterRunExecution(runId)).rejects.toThrow("execution may still be active"); + await cancelAdapterRunExecution(runId); + expect((await execution).exitCode).toBe(143); + expect(requests).toBe(2); + } finally { + complete(result); + await execution; + } + }); + + it("interrupts a live remote command before its own timeout and leaves the sandbox usable", async () => { + const runId = randomUUID(); + beginAdapterRunCancellation(runId); + let ready!: () => void; + const readiness = new Promise((resolve) => { ready = resolve; }); + const controls: boolean[] = []; + const runner: CommandManagedRuntimeRunner = { + execute: async (input) => { + if (input.env?.PAPERCLIP_SANDBOX_EXEC_CHANNEL === "bridge") controls.push(input.bypassSession === true); + const startedAt = new Date().toISOString(); + return new Promise((resolve, reject) => { + const child = spawn(input.command === "node" ? process.execPath : input.command, input.args, { + cwd: input.cwd, env: { ...process.env, ...input.env }, stdio: ["pipe", "pipe", "pipe"], + }); + let stdout = "", stderr = ""; + child.stdout.on("data", (chunk) => { stdout += chunk; if (stdout.includes("READY")) ready(); }); + child.stderr.on("data", (chunk) => { stderr += chunk; }); + child.on("error", reject); + child.on("close", (exitCode, signal) => resolve({ exitCode, signal, stdout, stderr, timedOut: false, pid: child.pid ?? null, startedAt, finishedAt: new Date().toISOString() })); + child.stdin.end(input.stdin); + }); + }, + }; + const execution = runAdapterExecutionTargetProcess(runId, { + kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId: "environment", + leaseId: "lease", remoteCwd: process.cwd(), runner, + }, process.execPath, ["-e", "const {spawn}=require('node:child_process');spawn(process.execPath,['-e',\"process.on('SIGTERM',()=>{});console.log('READY');setTimeout(()=>process.exit(0),1500)\"],{stdio:'inherit'});process.on('SIGTERM',()=>process.exit(0));setTimeout(()=>process.exit(0),1500)"], { + cwd: process.cwd(), env: {}, timeoutSec: 10, graceSec: 0.1, onLog: async () => {}, + }).finally(() => finishAdapterRunCancellation(runId)); + await Promise.race([readiness, execution.then(() => { throw new Error("Command ended before readiness"); })]); + const started = performance.now(); + await cancelAdapterRunExecution(runId); + const result = await execution; + expect(result.exitCode).toBe(143); + expect(performance.now() - started).toBeLessThan(1000); + expect(controls).toEqual([true]); + const after = await runner.execute({ command: "node", args: ["-e", "console.log('SAVE_AVAILABLE')"] }); + expect(after.stdout.trim()).toBe("SAVE_AVAILABLE"); + }); +}); diff --git a/packages/adapter-utils/src/cancellable-sandbox-command.ts b/packages/adapter-utils/src/cancellable-sandbox-command.ts new file mode 100644 index 0000000000..8eec966646 --- /dev/null +++ b/packages/adapter-utils/src/cancellable-sandbox-command.ts @@ -0,0 +1,82 @@ +import { randomUUID } from "node:crypto"; +import type { CommandManagedRuntimeRunner } from "./command-managed-runtime.js"; +import { + hasAdapterRunCancellation, + registerAdapterRunStop, + throwIfAdapterRunCancelled, +} from "./adapter-run-cancellation.js"; + +// The supervisor owns the child and its process group inside the sandbox. +// Cancellation transfers only a random command-scoped marker, never a PID. +const supervisor = String.raw` +const fs = require('node:fs'); +const { spawn } = require('node:child_process'); +const { command, args, marker, graceMs } = JSON.parse(process.argv[1]); +const cancelled = () => { + try { const s = fs.lstatSync(marker); if (!s.isFile() || s.isSymbolicLink()) throw Error('Invalid cancellation marker'); return true; } + catch (e) { if (e.code === 'ENOENT') return false; throw e; } +}; +if (cancelled()) { fs.unlinkSync(marker); process.exit(143); } +const child = spawn(command, args, { stdio: 'inherit', detached: true }); +let stopping = false, exited = false, escalation; +const signal = value => { + if (!child.pid) return; + try { process.kill(-child.pid, value); } catch (e) { if (e.code !== 'ESRCH') throw e; } +}; +const stop = () => { + if (stopping || exited) return; + stopping = true; signal('SIGTERM'); + escalation = setTimeout(() => { if (!exited) signal('SIGKILL'); }, graceMs); +}; +const poll = setInterval(() => { if (cancelled()) stop(); }, 100); +for (const name of ['SIGTERM', 'SIGINT', 'SIGHUP']) process.on(name, stop); +const cleanup = () => { clearInterval(poll); clearTimeout(escalation); try { fs.unlinkSync(marker); } catch (e) { if (e.code !== 'ENOENT') throw e; } }; +child.once('error', () => { exited = true; cleanup(); process.exitCode = 127; }); +child.once('exit', code => { + // Do not leave a shell/tool child holding the provider's output pipe after + // the CLI exits in response to cancellation. Never retain a PID kill timer. + if (stopping) signal('SIGKILL'); + exited = true; cleanup(); process.exitCode = stopping ? 143 : (code ?? 128); +}); +process.once('exit', () => { if (!exited) signal('SIGKILL'); }); +`; + +const requestStop = String.raw` +const fs = require('node:fs'), marker = process.argv[1]; +try { fs.closeSync(fs.openSync(marker, 'wx', 0o600)); } +catch (e) { if (e.code !== 'EEXIST') throw e; const s = fs.lstatSync(marker); if (!s.isFile() || s.isSymbolicLink()) throw Error('Invalid cancellation marker'); } +`; + +export async function executeCancellableSandboxCommand( + runId: string, + runner: CommandManagedRuntimeRunner, + input: Parameters[0], + graceMs: number, +) { + if (!hasAdapterRunCancellation(runId)) return runner.execute(input); + throwIfAdapterRunCancelled(runId); + const marker = `/tmp/paperclip-command-cancel-${randomUUID()}`; + let execution: ReturnType | undefined; + const unregister = registerAdapterRunStop(runId, async () => { + if (!execution) return; + const result = await runner.execute({ + command: "node", args: ["-e", requestStop, marker], cwd: input.cwd, + env: { PAPERCLIP_SANDBOX_EXEC_CHANNEL: "bridge" }, + timeoutMs: 10_000, bypassSession: true, + }); + if (result.timedOut || result.exitCode !== 0) throw new Error("Sandbox cancellation request failed; execution may still be active"); + await execution; + }); + try { + throwIfAdapterRunCancelled(runId); + execution = runner.execute({ + ...input, + command: "node", + args: ["-e", supervisor, JSON.stringify({ + command: input.command, args: input.args, marker, + graceMs: Math.max(1, Math.min(30_000, Math.trunc(graceMs) || 1)), + })], + }); + return await execution; + } finally { unregister(); } +} diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 5b02244a1f..542dc16594 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -1,3 +1,5 @@ +import { existsSync } from "node:fs"; +import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation } from "./adapter-run-cancellation.js"; import { createServer } from "node:http"; import http2 from "node:http2"; import net from "node:net"; @@ -874,6 +876,40 @@ describe("sandbox adapter execution targets", () => { } }); + it.each([false, true])("cancels the actual ACP child through its existing shutdown protocol (streamed=%s)", async (streamOutputViaSession) => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-acp-cancel-")); + cleanupDirs.push(rootDir); + const runId = `cancel-${rootDir}`; + const readyPath = path.join(rootDir, "ready"), stoppedPath = path.join(rootDir, "stopped"); + const childPath = path.join(rootDir, "child.cjs"); + await writeFile(childPath, `const fs=require('node:fs'); +process.on('SIGTERM',()=>{fs.writeFileSync(${JSON.stringify(stoppedPath)},'stopped');process.exit(0)}); +fs.writeFileSync(${JSON.stringify(readyPath)},'ready'); +process.stdin.resume();setTimeout(()=>process.exit(2),20000);`); + beginAdapterRunCancellation(runId); + let bridge: Awaited> = null; + let cancelling: Promise | undefined; + try { + bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId, target: { kind: "remote", transport: "sandbox", providerKey: "local-test", remoteCwd: rootDir, + timeoutMs: 30_000, runner: createLocalSandboxRunner() }, + runtimeRootDir: path.join(rootDir, ".paperclip-runtime"), adapterKey: "acpx", + command: process.execPath, args: [childPath], cwd: rootDir, env: {}, timeoutSec: 10, + streamOutputViaSession, + }); + await waitForCondition(() => existsSync(readyPath), "ACP child never started", 5_000); + cancelling = cancelAdapterRunExecution(runId); + await waitForCondition(() => existsSync(stoppedPath), "Cancellation did not reach the ACP child", 5_000); + finishAdapterRunCancellation(runId); + await cancelling; + expect(await readFile(stoppedPath, "utf8")).toBe("stopped"); + } finally { + await bridge?.stop(); + finishAdapterRunCancellation(runId); + await cancelling; + } + }, 15_000); + it("bridges bidirectional sandbox process sessions through a local ACPX-spawnable proxy", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-")); cleanupDirs.push(rootDir); @@ -5315,6 +5351,63 @@ describe("sandbox adapter execution targets", () => { } }, 20000); + it.each([false, true])("controller bridge shutdown preserves prior loss without inventing one (priorLoss=%s)", async (priorLoss) => { + // A loss ordered after a host-observed orderly completion is a normal + // teardown, not a failure: the run already completed. The disposition + // latch must keep the success and emit no loss event for it. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-orderly-close-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const api = await startRecordingApiServer(); + let emitExit: (() => void) | null = null; + const { runner } = makeHttp2SelectionRunner((ctx) => { + emitExit = ctx.emitExit; + ctx.emitReady(); + ctx.connectHttp2(); + }); + const open = runner.openDuplexChannel; + runner.openDuplexChannel = async (input) => { + const channel = await open(input); + return { ...channel, close: async () => { emitExit!(); await channel.close(); } }; + }; + const { recorder, events, counters } = createRecordingDuplexRecorder(); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-orderly-close", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + duplexObservabilityRecorder: recorder, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + if (priorLoss) { + emitExit!(); + await waitForCondition(() => events.some((event) => event.dimensions.loss_reason === "provider_exit"), "Expected real loss before shutdown", 4_000); + } + await bridge!.stop(); + await new Promise((resolve) => setImmediate(resolve)); + expect(bridge?.readRunDisposition?.()).toEqual({ failed: priorLoss, lossReason: priorLoss ? "provider_exit" : null }); + expect(events.some((event) => event.dimensions.loss_reason !== undefined)).toBe(priorLoss); + expect(counters.some((counter) => counter.metric === DUPLEX_COUNTER_LOSS_TOTAL)).toBe(priorLoss); + } finally { + await api.close(); + } + }, 20000); + it("test_safe_request_during_channel_loss_stays_retryable", async () => { // A GET never changes host state, so a response-body read failure stays // retryable: the host answers 502 with no indeterminate marker, the same diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 6f7628e324..b45cacb172 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1,3 +1,5 @@ +import { executeCancellableSandboxCommand } from "./cancellable-sandbox-command.js"; +import { bindAdapterRunStop, throwIfAdapterRunCancelled } from "./adapter-run-cancellation.js"; import fs from "node:fs/promises"; import net from "node:net"; import os from "node:os"; @@ -835,6 +837,7 @@ export async function runAdapterExecutionTargetProcess( options: AdapterExecutionTargetProcessOptions, ): Promise { if (target?.kind === "remote" && target.transport === "sandbox") { + throwIfAdapterRunCancelled(runId); const runner = requireSandboxRunner(target); const env = sanitizeRemoteExecutionEnv(options.env); await options.onRuntimeProgress?.({ @@ -849,7 +852,7 @@ export async function runAdapterExecutionTargetProcess( runLogTail.start(options.onLog); } try { - const result = await runner.execute({ + const result = await executeCancellableSandboxCommand(runId, runner, { command: execCommand, args: execArgs, cwd: target.workFolderHome ?? target.remoteCwd, @@ -862,7 +865,7 @@ export async function runAdapterExecutionTargetProcess( onSpawn: options.onSpawn ? async (meta) => options.onSpawn?.({ ...meta, processGroupId: null }) : undefined, - }); + }, options.graceSec * 1000); // Settle the duplex run disposition synchronously at the clean-completion // boundary, before the run-log tail finishes. The atomic settle marks the // host-observed orderly completion in one broker step, so a gateway exit @@ -1780,6 +1783,7 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { return null; } + throwIfAdapterRunCancelled(input.runId); const target = input.target; const onLog = input.onLog ?? (async () => {}); const runner = requireSandboxRunner(target); @@ -2255,82 +2259,80 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { })(); }; - return { - agentCommand, - stop: async () => { - stopping = true; - // End the `sandbox.agentProcess` span now, before the caller ends the run - // root span, even if the remote command has not resolved yet. - signalStopped(); - if (pollTimer) clearTimeout(pollTimer); - for (const liveSocket of liveSockets) liveSocket.destroy(); - await new Promise((resolve) => server.close(() => resolve())).catch(() => undefined); - // Wait for every accepted stdin write before `stdinEnd`. The socket handler - // fires each chunk write un-awaited through `stdinWriteChain`, so an earlier - // chunk can still be pending here. Chain the `stdinEnd` write onto the same - // per-session chain, so its file rename never finishes before an earlier - // chunk. `stdinSeq` is stable now, because the sockets are destroyed and the - // server is closed, so no new message can increment it. - const stdinEndPath = path.posix.join( - stdinDir, - `${String(stdinSeq + 1).padStart(12, "0")}.json`, - ); - const stdinEndWrite = stdinWriteChain.then(() => - client.writeTextFile(stdinEndPath, jsonLine({ type: "stdinEnd" })), - ); - stdinWriteChain = stdinEndWrite.then(() => undefined, () => undefined); - await stdinEndWrite.catch(() => undefined); - // The `shutdown` control message tells the wrapper to terminate itself - // and its own child (I3: no operating-system signal and no process - // identifier cross this boundary — only a file-queue message does). - // Chain it onto the same per-session write order as `stdinEnd`, so its - // file never lands before the earlier one. - const shutdownPath = path.posix.join( - stdinDir, - `${String(stdinSeq + 2).padStart(12, "0")}.json`, - ); - const shutdownWrite = stdinWriteChain.then(() => - client.writeTextFile(shutdownPath, jsonLine({ type: "shutdown" })), - ); - stdinWriteChain = shutdownWrite.then(() => undefined, () => undefined); - await shutdownWrite.catch(() => undefined); - // Wait a bounded budget for a hint that the wrapper stopped: only the - // `shutdownAck` event counts; an `exit` or `error` event is untrusted - // telemetry from inside the sandbox and never shortens this wait or - // suppresses the warning below. `shutdownAck` itself is ALSO an - // untrusted hint, not proof: any process that shares the sandbox can - // write the same event under this session's event directory. It can - // only shorten this wait and suppress the warning below; it never - // gates, shortens, or replaces the unconditional removal further down. - // What actually makes the wrapper's own termination deterministic is - // the wrapper-side session-identity latch, not this event. - let acknowledgedInTime = false; - readShutdownAckUntil(Date.now() + DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS); - await Promise.race([ - shutdownAcknowledged.then(() => { - acknowledgedInTime = true; - }), - new Promise((resolve) => { - const budgetTimer = setTimeout(resolve, DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS); - budgetTimer.unref?.(); - }), - ]); - stopReadingForShutdownAck = true; - if (!acknowledgedInTime) { - await onLog( - "stderr", - `[paperclip] ACP process session wrapper did not acknowledge shutdown within ${DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS}ms; removing the session directory anyway.\n`, - ).catch(() => undefined); - } - // Unconditional: this removal runs whether or not the wrapper - // acknowledged, and whether or not any event (real or forged) arrived - // under `sessionDir`. `stop()` runs during run teardown and must stay - // non-fatal, so every step above is best-effort and this step never - // throws. - await client.remove(sessionDir).catch(() => undefined); - await fs.rm(proxyDir, { recursive: true, force: true }).catch(() => undefined); - }, - }; + const stop = await bindAdapterRunStop(input.runId, async () => { + stopping = true; + // End the `sandbox.agentProcess` span now, before the caller ends the run + // root span, even if the remote command has not resolved yet. + signalStopped(); + if (pollTimer) clearTimeout(pollTimer); + for (const liveSocket of liveSockets) liveSocket.destroy(); + await new Promise((resolve) => server.close(() => resolve())).catch(() => undefined); + // Wait for every accepted stdin write before `stdinEnd`. The socket handler + // fires each chunk write un-awaited through `stdinWriteChain`, so an earlier + // chunk can still be pending here. Chain the `stdinEnd` write onto the same + // per-session chain, so its file rename never finishes before an earlier + // chunk. `stdinSeq` is stable now, because the sockets are destroyed and the + // server is closed, so no new message can increment it. + const stdinEndPath = path.posix.join( + stdinDir, + `${String(stdinSeq + 1).padStart(12, "0")}.json`, + ); + const stdinEndWrite = stdinWriteChain.then(() => + client.writeTextFile(stdinEndPath, jsonLine({ type: "stdinEnd" })), + ); + stdinWriteChain = stdinEndWrite.then(() => undefined, () => undefined); + await stdinEndWrite.catch(() => undefined); + // The `shutdown` control message tells the wrapper to terminate itself + // and its own child (I3: no operating-system signal and no process + // identifier cross this boundary — only a file-queue message does). + // Chain it onto the same per-session write order as `stdinEnd`, so its + // file never lands before the earlier one. + const shutdownPath = path.posix.join( + stdinDir, + `${String(stdinSeq + 2).padStart(12, "0")}.json`, + ); + const shutdownWrite = stdinWriteChain.then(() => + client.writeTextFile(shutdownPath, jsonLine({ type: "shutdown" })), + ); + stdinWriteChain = shutdownWrite.then(() => undefined, () => undefined); + await shutdownWrite.catch(() => undefined); + // Wait a bounded budget for a hint that the wrapper stopped: only the + // `shutdownAck` event counts; an `exit` or `error` event is untrusted + // telemetry from inside the sandbox and never shortens this wait or + // suppresses the warning below. `shutdownAck` itself is ALSO an + // untrusted hint, not proof: any process that shares the sandbox can + // write the same event under this session's event directory. It can + // only shorten this wait and suppress the warning below; it never + // gates, shortens, or replaces the unconditional removal further down. + // What actually makes the wrapper's own termination deterministic is + // the wrapper-side session-identity latch, not this event. + let acknowledgedInTime = false; + readShutdownAckUntil(Date.now() + DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS); + await Promise.race([ + shutdownAcknowledged.then(() => { + acknowledgedInTime = true; + }), + new Promise((resolve) => { + const budgetTimer = setTimeout(resolve, DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS); + budgetTimer.unref?.(); + }), + ]); + stopReadingForShutdownAck = true; + if (!acknowledgedInTime) { + await onLog( + "stderr", + `[paperclip] ACP process session wrapper did not acknowledge shutdown within ${DEFAULT_PROCESS_SESSION_SHUTDOWN_WAIT_MS}ms; removing the session directory anyway.\n`, + ).catch(() => undefined); + } + // Unconditional: this removal runs whether or not the wrapper + // acknowledged, and whether or not any event (real or forged) arrived + // under `sessionDir`. `stop()` runs during run teardown and must stay + // non-fatal, so every step above is best-effort and this step never + // throws. + await client.remove(sessionDir).catch(() => undefined); + await fs.rm(proxyDir, { recursive: true, force: true }).catch(() => undefined); + }); + return { agentCommand, stop }; } function getProcessSessionProxySource(input: { port: number; token: string }): string { @@ -4465,6 +4467,10 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { settleRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.settleRunDisposition(), markOrderlyCompletion: (): void => dispositionLatch.markOrderlyCompletion(), stop: async () => { + // A controller-requested shutdown is not a provider failure. The + // native Git bridge also stops here without the CLI/ACP result + // seam. Preserve any earlier loss before closing the channel. + dispositionLatch.markOrderlyCompletion(); // Close the HTTP/2 server's sessions, then the channel, before // lease release, so no live provider session remains when the // caller releases the lease. diff --git a/server/src/__tests__/heartbeat-process-metadata.test.ts b/server/src/__tests__/heartbeat-process-metadata.test.ts index 4565a91b60..477e513c4d 100644 --- a/server/src/__tests__/heartbeat-process-metadata.test.ts +++ b/server/src/__tests__/heartbeat-process-metadata.test.ts @@ -5,6 +5,8 @@ import { agents, companies, createDb, heartbeatRuns, startEmbeddedPostgresTestDa import * as processes from "../services/hot-restart.js"; import * as adapters from "../adapters/index.js"; import * as orchestration from "../services/environment-run-orchestrator.js"; +import * as compatibility from "../services/legacy-sandbox-workspace.js"; +import { bindAdapterRunStop, hasAdapterRunCancellation } from "@paperclipai/adapter-utils/adapter-run-cancellation"; import * as executionTargets from "@paperclipai/adapter-utils/execution-target"; import { heartbeatService, persistHeartbeatRunProcessMetadata } from "../services/heartbeat.js"; @@ -66,6 +68,58 @@ describe("heartbeat process identity persistence", () => { } }, 30_000); + it("cancels the remote adapter and waits for teardown before acknowledging", async () => { + const originalOrchestrator = orchestration.environmentRunOrchestrator; + vi.spyOn(orchestration, "environmentRunOrchestrator").mockImplementation((...args) => { + const actual = originalOrchestrator(...args); + return { ...actual, realizeForRun: async (input) => ({ + ...await actual.realizeForRun(input), + executionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/remote/task", shellCommand: "sh" } as never, + }) }; + }); + vi.spyOn(compatibility, "hasLegacySandboxWorkspace").mockReturnValue(true); + vi.spyOn(executionTargets, "prepareGitHubOperationLaunchers").mockImplementation(async (input) => input.env); + vi.spyOn(executionTargets, "cleanupGitHubOperationLaunchers").mockResolvedValue(undefined); + let ready!: () => void, stopped!: () => void, release!: () => void; + const started = new Promise((resolve) => { ready = resolve; }); + const interrupted = new Promise((resolve) => { stopped = resolve; }); + const teardown = new Promise((resolve) => { release = resolve; }); + const heartbeat = heartbeatService(db); + vi.spyOn(adapters, "getServerAdapter").mockReturnValue({ + supportsLocalAgentJwt: false, + execute: async (input) => { + expect(hasAdapterRunCancellation(input.runId)).toBe(true); + const cleanup = await bindAdapterRunStop(input.runId, async () => { + expect((await heartbeat.getRun(input.runId))?.status).toBe("cancelled"); + stopped(); + }); + ready(); + await interrupted; + await teardown; + await cleanup(); + return { exitCode: 143, signal: "SIGTERM", timedOut: false }; + }, + } as ReturnType); + let pending: Promise | undefined; + try { + const queued = await heartbeat.invoke(agentId, "on_demand", {}, "manual"); + expect(queued).not.toBeNull(); + await started; + let acknowledged = false; + pending = heartbeat.cancelRun(queued!.id).then((result) => { acknowledged = true; return result; }); + await interrupted; + expect(acknowledged).toBe(false); + release(); + await pending; + expect((await heartbeat.getRun(queued!.id))?.status).toBe("cancelled"); + expect(hasAdapterRunCancellation(queued!.id)).toBe(false); + } finally { + stopped(); release(); + await pending; + await heartbeat.drainActiveRunExecutions(); + } + }, 30_000); + it("uses the remote marker even when its PID exists on the host", async () => { const run = await running(); const host = vi.spyOn(processes, "readProcessStartedAt"); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ddf0cc5f7f..9f88ef5d13 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { beginAdapterRunCancellation, cancelAdapterRunExecution, finishAdapterRunCancellation, hasAdapterRunCancellation } from "@paperclipai/adapter-utils/adapter-run-cancellation"; import { measureSandboxOperation, runWithSandboxPerformanceTrace, setSandboxPerformanceRunAttributes } from "./sandbox-performance.js"; import { initializeRunIdentity } from "./run-identity.js"; import { startNativeGitHubCallbackBridge } from "./native-github-bridge.js"; @@ -13410,6 +13411,7 @@ export function heartbeatService( }, ); if (!interruptedStatus.updated || !interruptedStatus.run) continue; + if (run.runtimeMode !== "native") await cancelAdapterRunExecution(run.id); let interrupted = interruptedStatus.run; await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: now, @@ -17990,10 +17992,14 @@ export function heartbeatService( // The diagnostic path is opt-in and persists bounded batches after execution. async function executeRun(runId: string, runOptions: Parameters[1] = {}) { - return runWithSandboxPerformanceTrace({ runId, onBatch: async (batch) => { - const observed = await getRun(runId); - if (observed) await appendRunEvent(observed, { eventType: "sandbox.performance.batch", stream: "system", level: "info", payload: batch }); - } }, () => executeRunMeasured(runId, runOptions)); + try { + return await runWithSandboxPerformanceTrace({ runId, onBatch: async (batch) => { + const observed = await getRun(runId); + if (observed) await appendRunEvent(observed, { eventType: "sandbox.performance.batch", stream: "system", level: "info", payload: batch }); + } }, () => executeRunMeasured(runId, runOptions)); + } finally { + finishAdapterRunCancellation(runId); + } } async function executeRunMeasured( @@ -21142,6 +21148,10 @@ export function heartbeatService( .where(eq(heartbeatRuns.id, run.id)))); } setSandboxPerformanceRunAttributes({ runtime: nativeRuntimeResolution.kind }); + if (nativeRuntimeResolution.kind === "legacy" + && executionTarget?.kind === "remote" && executionTarget.transport === "sandbox") { + beginAdapterRunCancellation(run.id); + } const localAgentJwtScope = issueRef?.workMode === "skill_test" ? { kind: "skill_test" as const, issueId: issueRef.id } @@ -26034,8 +26044,13 @@ export function heartbeatService( !CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes( run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number], ) - ) + ) { + if (run.status === "cancelled" && hasAdapterRunCancellation(run.id)) { + await cancelAdapterRunExecution(run.id); + return await getRun(run.id); + } return run; + } const agent = await getAgent(run.agentId); const errorCode = options.errorCode ?? "cancelled"; const resultJson = agent @@ -26049,6 +26064,14 @@ export function heartbeatService( } : options.resultJson; + if (run.runtimeMode !== "native" && hasAdapterRunCancellation(run.id)) { + await setRunStatus(run.id, "cancelled", { + finishedAt: new Date(), error: reason, errorCode, + ...(resultJson ? { resultJson } : {}), + }); + await cancelAdapterRunExecution(run.id); + } + const running = runningProcesses.get(run.id); try { await cancelHeartbeatNativeRun({ @@ -26186,6 +26209,7 @@ export function heartbeatService( graceMs: Math.max(1, running.graceSec) * 1000, }); } + if (run.runtimeMode !== "native") await cancelAdapterRunExecution(run.id); runningProcesses.delete(run.id); await releaseIssueExecutionAndPromote(run); }