diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index d2786120d1..35d89fccda 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -3051,6 +3051,83 @@ describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / re expect(disposedRunIds).toContain("run-a"); }); + // F5 leak window 1: a seam that throws AFTER a successful stage() must not leave + // the fresh in-sandbox runtime behind. The engine records the staged result in + // the stage wrapper and disposes it when the seam throws. + it("test_seam_throw_after_stage_disposes_fresh_staged_runtime", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const stagedRuntimes = new Map(); + let capturedRuntimeRoot: string | null = null; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes, + createRuntime: () => recordingRuntime({ ensureInputs: [] }) as never, + prepareRemoteManagedHome: async (input) => { + const staged = await input.stage([]); + capturedRuntimeRoot = staged.runtimeRootDir; + // The seam fails AFTER a successful stage (e.g. an in-sandbox config + // materialize step throws). The fresh in-sandbox runtime is already shipped. + throw new Error("seam failed after stage"); + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + await expect( + execute({ runId: "run-a", runtime: {}, ...base } as never), + ).rejects.toThrow("seam failed after stage"); + + // The stage shipped a real in-sandbox runtime root (local runner = host FS)... + expect(typeof capturedRuntimeRoot).toBe("string"); + // ...and the seam throw disposed it, so the abandoned managed home is gone. + expect(await pathExists(capturedRuntimeRoot!)).toBe(false); + // The failed fresh stage was never cached for reuse. + expect(stagedRuntimes.size).toBe(0); + }); + + // F5 leak window 2: when a bridge fails during bring-up on a resume that BORROWED + // the cached staged runtime, the rollback must remove the borrowed cache entry + // through the same identity guard `discardStagedRuntime` uses — else a later + // resume reuses an entry whose host staged-temp was already disposed. + it("test_partial_bridge_rollback_removes_borrowed_staged_entry", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const stagedRuntimes = new Map(); + let disposeCalls = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes, + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => ({ + stagedRuntime: await input.stage([]), + disposeStaged: async () => { + disposeCalls += 1; + }, + }), + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + expect(first.exitCode).toBe(0); + // The clean turn cached the staged runtime for reuse; nothing disposed yet. + expect(stagedRuntimes.size).toBe(1); + expect(disposeCalls).toBe(0); + + // Run B resumes the same session and borrows the cached staged runtime, but a + // bridge fails during bring-up. + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(async () => { + throw new Error("paperclip bridge boom"); + }); + await expect( + execute({ runId: "run-b", runtime: { sessionParams: first.sessionParams }, ...base } as never), + ).rejects.toThrow("paperclip bridge boom"); + + // The rollback removed the borrowed cache entry through the identity guard, so a + // later resume can never reuse an entry whose host staged-temp was disposed... + expect(stagedRuntimes.size).toBe(0); + // ...and it fired that entry's shared idempotent dispose exactly once. + expect(disposeCalls).toBe(1); + }); + // Greptile P1 "Concurrent Runs Corrupt Cache Ownership": two overlapping runs // of the same session key must not ship into the same remote workspace at once. // The per-key staging lock serializes the stage-or-reuse section, so their @@ -4377,3 +4454,785 @@ describe("ACPX engine per-step startup timing (run.startup.step events)", () => expect(emitted.has("bridge.process-session")).toBe(false); }); }); + +describe("ACPX engine run lifecycle corrections (F1: settle every failure after buildRuntime)", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + // A remote sandbox setup that stages the host worktree through the real local + // runner, so a run reaches the post-build window with live bridges and a held + // staging lease. + async function setupRemoteSandbox() { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + return { root, stateDir, localCwd, remoteCwd, executionTarget }; + } + + it("test_runtime_create_failure_returns_error_result_with_phase_create_runtime", async () => { + const root = await makeTempRoot(); + const execute = createAcpxEngineExecutor({ + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }); + const result = await execute({ + runId: "run-create-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + // The post-build runtime-creation window failure returns a settled error + // result instead of throwing. + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("create_runtime"); + expect(result.summary).toContain("createRuntime boom"); + }); + + it("test_runtime_create_failure_stops_bridges_and_releases_staging_lease", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const paperclipStop = vi.fn(async () => {}); + const processStop = vi.fn(async () => {}); + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce( + async () => ({ env: {}, stop: paperclipStop }) as never, + ); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce( + async () => ({ agentCommand: null, stop: processStop }) as never, + ); + const stagingLocks = new Map>(); + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }); + + const result = await execute({ + runId: "run-create-fail-remote", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("create_runtime"); + // Both live bridges stop exactly once. + expect(paperclipStop).toHaveBeenCalledTimes(1); + expect(processStop).toHaveBeenCalledTimes(1); + // The per-session staging lease released, so the lock map does not strand the + // next same-session run. + expect(stagingLocks.size).toBe(0); + }); + + it("test_prompt_build_failure_returns_error_result_with_phase_prepare_turn", async () => { + const root = await makeTempRoot(); + const close = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close, + }) as never, + }); + + // A throwing accessor on a field only `buildPrompt` reads makes the prompt + // build fail after the session handshake succeeds. + const context: Record = {}; + Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { + enumerable: false, + get() { + throw new Error("prompt build boom"); + }, + }); + + const result = await execute({ + runId: "run-prepare-fail", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("prepare_turn"); + // The established runtime is closed on the pre-turn failure, so no session + // leaks. + expect(close).toHaveBeenCalledTimes(1); + }); + + it("test_sandbox_startup_span_ends_exactly_once_on_every_exit_path", async () => { + // Each scenario runs a fresh remote bring-up with a recording trace and asserts + // the one sandbox.startup span ends exactly once, whatever the exit path. + const failingEnsure = () => + ({ + ensureSession: async () => { + throw new Error("ensureSession boom"); + }, + startTurn: () => ({ + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: async () => {}, + }) as never; + const scenarios: Array<{ name: string; createRuntime: AcpxEngineExecutorOptions["createRuntime"] }> = [ + { name: "success", createRuntime: () => buildRuntime() as never }, + { + name: "create_runtime failure", + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + }, + { name: "ensure_session failure", createRuntime: () => failingEnsure() }, + ]; + + for (const scenario of scenarios) { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { traceContext, spans } = createRecordingStartupTrace(); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: scenario.createRuntime, + }); + + await execute({ + runId: `run-span-${scenario.name.replace(/\s+/g, "-")}`, + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + startupTraceContext: traceContext, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + } as never); + + const startupSpans = spans.filter((span) => span.name === "sandbox.startup"); + expect(startupSpans, `scenario "${scenario.name}" must open one sandbox.startup span`).toHaveLength(1); + expect( + startupSpans[0]!.endCalls, + `scenario "${scenario.name}" must end the sandbox.startup span exactly once`, + ).toBe(1); + } + }); +}); + +describe("ACPX engine run lifecycle corrections (F2: close the runtime for every pre-turn failure)", () => { + const startedAt = "2026-01-01T00:00:00.000Z"; + const handle = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + + it("test_handshake_failure_closes_runtime_and_removes_warm_entry", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const closeSpy = vi.fn(async () => {}); + let created = 0; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: (options) => { + created += 1; + const opts = options as AcpRuntimeOptions & { + onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }; + return { + ensureSession: async () => { + await opts.onAgentSpawn?.({ pid: 4242, startedAt }); + return handle; + }, + startTurn: () => ({ + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: closeSpy, + } as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + + const first = await execute({ + runId: "warm-handshake-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + expect(warmHandles.size).toBe(1); + expect(created).toBe(1); + + // The warm-hit reuses runtime #1 and fails while persisting process identity. + const second = await execute({ + runId: "warm-handshake-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => { + throw new Error("onSpawn boom"); + }, + } as never); + + expect(created).toBe(1); + expect(second.exitCode).toBe(1); + expect(second.resultJson?.phase).toBe("ensure_session"); + // The reused runtime is closed and the warm entry removed. + expect(closeSpy).toHaveBeenCalledTimes(1); + expect(warmHandles.size).toBe(0); + }); + + it("test_missing_session_handle_closes_runtime", async () => { + const root = await makeTempRoot(); + const closeSpy = vi.fn(async () => {}); + const execute = createAcpxEngineExecutor({ + createRuntime: () => + ({ + // A runtime that returns no session handle drives the missing-handle path. + ensureSession: async () => undefined, + startTurn: () => ({ + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: closeSpy, + }) as never, + }); + + const result = await execute({ + runId: "missing-handle", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("ensure_session"); + expect(result.errorCode).toBe("acpx_runtime_error"); + // The constructed runtime is closed so its child process cannot leak. + expect(closeSpy).toHaveBeenCalledTimes(1); + }); + + it("test_warm_hit_failure_closes_runtime_and_does_not_rearm_idle_timer", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + let created = 0; + const closes: number[] = []; + const warmHandles = new Map(); + const execute = createAcpxEngineExecutor({ + warmHandles, + createRuntime: (options) => { + const id = (created += 1); + const opts = options as AcpRuntimeOptions & { + onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + }; + return { + ensureSession: async () => { + await opts.onAgentSpawn?.({ pid: 1000 + id, startedAt }); + return handle; + }, + startTurn: () => ({ + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: async () => { + closes.push(id); + }, + } as never; + }, + }); + const config = { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir, + mode: "persistent", + warmHandleIdleMs: 60_000, + }; + + const first = await execute({ + runId: "warm-timer-1", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + expect(first.exitCode).toBe(0); + expect(warmHandles.size).toBe(1); + + // A warm-hit whose identity-persist step fails reuses runtime #1 (no new create). + const second = await execute({ + runId: "warm-timer-2", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => { + throw new Error("onSpawn boom"); + }, + } as never); + expect(created).toBe(1); + expect(second.exitCode).toBe(1); + // Runtime #1 is closed and its warm entry removed. The warm-hit already cleared + // the idle timer, and the failure does not re-arm it, so the map is empty. + expect(closes).toEqual([1]); + expect(warmHandles.size).toBe(0); + + // A later heartbeat finds no warm entry and constructs a fresh runtime, proving + // the failed warm-hit did not leave a re-armed entry behind. + const third = await execute({ + runId: "warm-timer-3", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: { sessionParams: first.sessionParams }, + config, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onSpawn: async () => {}, + } as never); + expect(third.exitCode).toBe(0); + expect(created).toBe(2); + }); +}); + +describe("ACPX engine run lifecycle corrections (F3: one teardown error policy)", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + const okHandle = { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + + async function setupRemoteSandbox() { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + return { root, stateDir, localCwd, remoteCwd, executionTarget }; + } + + // Stub both sandbox bridges with stop spies collected per start, so a test can + // assert the bridges stopped without running the real bridge transport. + function stubBridges() { + const paperclipStops: Array> = []; + const processStops: Array> = []; + vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + paperclipStops.push(stop); + return { env: {}, stop } as never; + }); + vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementation(async () => { + const stop = vi.fn(async () => {}); + processStops.push(stop); + return { agentCommand: null, stop } as never; + }); + const anyStopped = (stops: Array>) => + stops.some((stop) => stop.mock.calls.length > 0); + const stoppedCount = (stops: Array>) => + stops.filter((stop) => stop.mock.calls.length > 0).length; + return { paperclipStops, processStops, anyStopped, stoppedCount }; + } + + function throwingHandoffContext(): Record { + const context: Record = {}; + Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { + enumerable: false, + get() { + throw new Error("prompt build boom"); + }, + }); + return context; + } + + function completedTurn() { + return { + events: (async function* () {})(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; + } + + function throwingTurn() { + return { + events: (async function* () { + throw new Error("turn upstream boom"); + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; + } + + function remoteArgs( + stateDir: string, + localCwd: string, + executionTarget: unknown, + overrides: Record = {}, + ) { + return { + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + context: {}, + authToken: "real-run-jwt", + executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + onEvent: async () => {}, + ...overrides, + }; + } + + it("test_teardown_continues_after_one_teardown_step_fails", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, anyStopped } = stubBridges(); + const stagingLocks = new Map>(); + const logs: Array<{ stream: string; text: string }> = []; + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + // A failing teardown step must not stop the later teardown steps. + close: async () => { + throw new Error("close boom"); + }, + }) as never, + }); + + const result = await execute({ + runId: "td-continue", + ...remoteArgs(stateDir, localCwd, executionTarget, { + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + }), + } as never); + + expect(result.exitCode).toBe(1); + // The close failure did not stop the bridge stops or the lease release. + expect(anyStopped(paperclipStops)).toBe(true); + expect(anyStopped(processStops)).toBe(true); + expect(stagingLocks.size).toBe(0); + // The close failure was recorded, not silently dropped. + expect( + logs.some((entry) => entry.stream === "stderr" && entry.text.includes('teardown step "runtime-close" failed')), + ).toBe(true); + }); + + it("test_staging_lease_releases_in_finally_on_every_exit_path", async () => { + const scenarios: Array<{ + name: string; + createRuntime: AcpxEngineExecutorOptions["createRuntime"]; + context: Record; + }> = [ + { + name: "create_runtime", + createRuntime: () => { + throw new Error("createRuntime boom"); + }, + context: {}, + }, + { + name: "ensure_session", + createRuntime: () => + ({ + ensureSession: async () => { + throw new Error("ensure boom"); + }, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + context: {}, + }, + { + name: "prepare_turn", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + context: throwingHandoffContext(), + }, + { + name: "turn", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => {}, + }) as never, + context: {}, + }, + { + name: "success", + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => completedTurn(), + close: async () => {}, + }) as never, + context: {}, + }, + ]; + + for (const scenario of scenarios) { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + stubBridges(); + const stagingLocks = new Map>(); + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: scenario.createRuntime, + }); + + await execute({ + runId: `lease-${scenario.name}`, + ...remoteArgs(stateDir, localCwd, executionTarget, { context: scenario.context }), + } as never).catch(() => {}); + + // The staging lease is released in a finally on every exit path, so the lock + // map never strands the next same-session run. + expect(stagingLocks.size, `lease must release on the ${scenario.name} path`).toBe(0); + } + }); + + it("test_result_emission_failure_does_not_skip_teardown", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, anyStopped } = stubBridges(); + const stagingLocks = new Map>(); + const execute = createAcpxEngineExecutor({ + stagingLocks, + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => {}, + }) as never, + }); + + await execute({ + runId: "emit-fail", + ...remoteArgs(stateDir, localCwd, executionTarget, { + // Fail the acpx.error emission only, so teardown must still run. + onLog: async (_stream: "stdout" | "stderr", text: string) => { + if (text.includes('"type":"acpx.error"')) throw new Error("onLog boom"); + }, + }), + } as never).catch(() => {}); + + expect(anyStopped(paperclipStops)).toBe(true); + expect(anyStopped(processStops)).toBe(true); + expect(stagingLocks.size).toBe(0); + }); + + it("test_result_mapping_throw_after_close_does_not_rerun_teardown", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const { paperclipStops, processStops, stoppedCount } = stubBridges(); + let closeCount = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + // A completed turn whose result mapping throws after the close is read. + result: Promise.resolve({ + status: "completed", + get stopReason(): string { + throw new Error("mapping boom"); + }, + }), + cancel: async () => {}, + }), + close: async () => { + closeCount += 1; + }, + }) as never, + }); + + await execute({ + runId: "map-throw", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never).catch(() => {}); + + // The completed turn closed the runtime once; the mapping throw did not re-run + // the teardown through the turn catch. + expect(closeCount).toBe(1); + expect(stoppedCount(paperclipStops)).toBe(1); + expect(stoppedCount(processStops)).toBe(1); + }); + + it("test_teardown_failure_does_not_change_external_result", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + stubBridges(); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => + ({ + ensureSession: async () => okHandle, + startTurn: () => throwingTurn(), + close: async () => { + throw new Error("close boom"); + }, + }) as never, + }); + + const result = await execute({ + runId: "td-result", + ...remoteArgs(stateDir, localCwd, executionTarget), + } as never); + + // The exit cause (the turn failure) determines the external result; the failing + // teardown step never leaks into it. + expect(result.exitCode).toBe(1); + expect(result.resultJson?.phase).toBe("turn"); + expect(result.errorCode).toBe("acpx_turn_failed"); + expect(result.errorMessage).toContain("turn upstream boom"); + expect(result.errorMessage).not.toContain("close boom"); + }); + + it("test_flush_child_stderr_runs_on_every_exit_path", async () => { + const writes: string[] = []; + const spy = vi.spyOn(process.stderr, "write").mockImplementation(((chunk: unknown) => { + writes.push(String(chunk)); + return true; + }) as never); + try { + const scenarios: Array<{ name: string; mode: string; context: Record }> = [ + { name: "ensure_session", mode: "ensure_fail", context: {} }, + { name: "prepare_turn", mode: "success", context: throwingHandoffContext() }, + { name: "turn", mode: "turn_fail", context: {} }, + { name: "success", mode: "success", context: {} }, + ]; + + for (const scenario of scenarios) { + const root = await makeTempRoot(); + const marker = `paperclip-flush-probe-${scenario.name}`; + const execute = createAcpxEngineExecutor({ + createRuntime: (options) => { + const opts = options as { + onAgentStderr?: (chunk: string) => void; + }; + // Emit a partial (newline-free) stderr chunk so it stays pending until a + // teardown flush writes it to process.stderr. + const emit = () => opts.onAgentStderr?.(marker); + return { + ensureSession: async () => { + emit(); + if (scenario.mode === "ensure_fail") throw new Error("ensure boom"); + return okHandle; + }, + startTurn: () => + scenario.mode === "turn_fail" ? throwingTurn() : completedTurn(), + close: async () => {}, + } as never; + }, + }); + + await execute({ + runId: `flush-${scenario.name}`, + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: scenario.context, + onLog: async () => {}, + onMeta: async () => {}, + } as never).catch(() => {}); + + expect( + writes.some((write) => write.includes(marker)), + `flushChildStderr must run on the ${scenario.name} path`, + ).toBe(true); + } + } finally { + spy.mockRestore(); + } + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index 3dc9fa7260..cc31a0c412 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -20,6 +20,7 @@ import { prepareAdapterExecutionTargetRuntime, readAdapterExecutionTarget, resolveAdapterExecutionTargetTimeout, + runAdapterExecutionTargetShellCommand, startAdapterExecutionTargetPaperclipBridge, startAdapterExecutionTargetProcessSessionBridge, type AdapterExecutionTarget, @@ -1360,6 +1361,46 @@ async function stageAcpRemoteRuntime(input: { }); } +// Dispose a freshly staged in-sandbox runtime after a managed-home seam threw. The +// seam shipped the workspace/home into the sandbox through `stage()` but threw +// before it could return its `disposeStaged`, so the engine removes the managed +// runtime root here — otherwise the abandoned in-sandbox managed home (with its +// staged credentials) leaks in a persistent sandbox. Fail-open: a cleanup miss on +// an already-failing run is logged and never re-thrown, so it cannot mask the seam +// error. +async function disposeFreshStagedRuntime(input: { + runId: string; + target: AdapterExecutionTarget; + stagedRuntime: PreparedAdapterExecutionTargetRuntime; + cwd: string; + timeoutSec: number; + onLog: AdapterExecutionContext["onLog"]; +}): Promise { + const runtimeRootDir = input.stagedRuntime.runtimeRootDir; + if (!runtimeRootDir) return; + try { + await runAdapterExecutionTargetShellCommand( + input.runId, + input.target, + `rm -rf ${shellQuote(runtimeRootDir)}`, + { + cwd: input.cwd, + env: {}, + timeoutSec: Math.max(input.timeoutSec, 15), + graceSec: 20, + onLog: input.onLog, + }, + ); + } catch (err) { + await input.onLog( + "stderr", + `[paperclip] Failed to dispose the fresh staged runtime after a managed-home seam error: ${ + err instanceof Error ? err.message : String(err) + }\n`, + ); + } +} + async function buildRuntime(input: { ctx: AdapterExecutionContext; engine: AcpxEngineSettings; @@ -1866,8 +1907,12 @@ async function buildRuntime(input: { stagedRuntimes.delete(sessionKey); if (stale.dispose) await stale.dispose().catch(() => {}); } - const stage = (assets: AdapterManagedRuntimeAsset[]) => - stageAcpRemoteRuntime({ + // Record each successful `stage()` result before the seam can fail. A seam + // that throws AFTER a successful stage never returns its `disposeStaged`, so + // the engine disposes the fresh staged runtime here (see the catch below). + let freshlyStagedRuntime: PreparedAdapterExecutionTargetRuntime | null = null; + const stage = async (assets: AdapterManagedRuntimeAsset[]) => { + const staged = await stageAcpRemoteRuntime({ runId, target: remoteTarget, adapterKey: input.engine.adapterType, @@ -1880,6 +1925,9 @@ async function buildRuntime(input: { onRuntimeProgress: input.ctx.onRuntimeProgress, runtimeSpan: input.stageRuntimeSpan, }); + freshlyStagedRuntime = staged; + return staged; + }; // Snapshot env before the seam so we can capture exactly which keys it // repointed onto the in-sandbox home (e.g. `CODEX_HOME`) and replay them // verbatim on a later compatible resume. Add/change only — every seam sets @@ -1900,19 +1948,37 @@ async function buildRuntime(input: { dispose: (() => Promise) | null; }> => { if (input.deps.prepareRemoteManagedHome) { - const seeded = await input.deps.prepareRemoteManagedHome({ - acpxAgent, - companyId: agent.companyId, - runId, - config, - executionTarget: remoteTarget, - workspaceLocalDir: cwd, - timeoutSec, - env, - onLog: input.ctx.onLog, - onRuntimeProgress: input.ctx.onRuntimeProgress, - stage, - }); + let seeded: AcpxRemoteManagedHomeResult; + try { + seeded = await input.deps.prepareRemoteManagedHome({ + acpxAgent, + companyId: agent.companyId, + runId, + config, + executionTarget: remoteTarget, + workspaceLocalDir: cwd, + timeoutSec, + env, + onLog: input.ctx.onLog, + onRuntimeProgress: input.ctx.onRuntimeProgress, + stage, + }); + } catch (seamErr) { + // The seam failed after a possible successful stage. Dispose the fresh + // staged runtime so the abandoned in-sandbox managed home does not leak, + // then rethrow the seam error. + if (freshlyStagedRuntime) { + await disposeFreshStagedRuntime({ + runId, + target: remoteTarget, + stagedRuntime: freshlyStagedRuntime, + cwd: sessionCwd, + timeoutSec, + onLog: input.ctx.onLog, + }); + } + throw seamErr; + } return { stagedRuntime: seeded.stagedRuntime, teardown: seeded.teardown ?? null, @@ -2058,8 +2124,21 @@ async function buildRuntime(input: { // abandoned, so its staged temp must be released — and release the per-session // staging lease so the abandoned run does not strand the next same-session run // (cleanupRemoteBridges, which normally releases it, is never reached here). + // + // Route the dispose through the same ownership-guarded removal + // `discardStagedRuntime` uses. When this run BORROWED the cached staged runtime + // (a compatible resume), the dispose releases the shared host staged-temp, so + // the map entry must go too — otherwise a later resume reuses an entry whose + // temp is already gone. The identity guard removes the entry only when the map + // still holds this exact staged runtime, so a fresh stage (no cache entry) and + // a concurrent run's entry are both left untouched. await remoteManagedHomeTeardown?.().catch(() => {}); - await remoteStagingDispose?.().catch(() => {}); + await releaseStagedRuntimeEntry({ + handles: stagedRuntimes, + sessionKey, + stagedRuntime, + dispose: remoteStagingDispose, + }); sessionStagingLeaseRelease?.(); throw err; } @@ -2225,20 +2304,25 @@ async function settleRemoteBridgeStarts( } async function cleanupRemoteBridges(prepared: AcpxPreparedRuntime): Promise { - await Promise.allSettled([ - prepared.processSessionBridge?.stop(), - prepared.paperclipBridge?.stop(), - ]); - // Runs AFTER the bridges stop (mirrors the CLI finally: stop bridge → restore - // workspace). Fires the codex auth copy-back via `restoreWorkspace()` and - // removes staged temp dirs. The seam logs and swallows its own failures — an - // unclean-teardown copy-back miss is the accepted, loud `refresh_token_reused` - // residual on the next host Codex use, never silent HOST-credential corruption - // — so a teardown fault never masks or fails the run result here. - if (prepared.remoteManagedHomeTeardown) { - await prepared.remoteManagedHomeTeardown().catch(() => {}); + try { + await Promise.allSettled([ + prepared.processSessionBridge?.stop(), + prepared.paperclipBridge?.stop(), + ]); + // Runs AFTER the bridges stop (mirrors the CLI finally: stop bridge → restore + // workspace). Fires the codex auth copy-back via `restoreWorkspace()` and + // removes staged temp dirs. The seam logs and swallows its own failures — an + // unclean-teardown copy-back miss is the accepted, loud `refresh_token_reused` + // residual on the next host Codex use, never silent HOST-credential corruption + // — so a teardown fault never masks or fails the run result here. + if (prepared.remoteManagedHomeTeardown) { + await prepared.remoteManagedHomeTeardown().catch(() => {}); + } + } finally { + // Release the per-session staging lease in a finally, so an earlier teardown + // fault never strands it and deadlocks the next run of this session. + prepared.sessionStagingLeaseRelease?.(); } - prepared.sessionStagingLeaseRelease?.(); } function renderPaperclipEnvNote(env: Record): string { @@ -2555,7 +2639,12 @@ export function summarizeAcpxTurnUsage(input: { return { usage, usageDetail, costUsd, cumulativeCostUsd }; } -type AcpxExecutionPhase = "ensure_session" | "configure_session" | "turn"; +type AcpxExecutionPhase = + | "create_runtime" + | "ensure_session" + | "configure_session" + | "prepare_turn" + | "turn"; function describeErrorDiagnostics(err: unknown): { errorName: string; @@ -2792,11 +2881,35 @@ async function discardStagedRuntime(input: { prepared: AcpxPreparedRuntime; }): Promise { const { handles, prepared } = input; - const existing = handles.get(prepared.sessionKey); - if (existing && prepared.stagedRuntime && existing.stagedRuntime === prepared.stagedRuntime) { - handles.delete(prepared.sessionKey); + await releaseStagedRuntimeEntry({ + handles, + sessionKey: prepared.sessionKey, + stagedRuntime: prepared.stagedRuntime, + dispose: prepared.remoteStagingDispose, + }); +} + +// Ownership-guarded removal + dispose for a staged-runtime cache entry a run owns. +// Shared by `discardStagedRuntime` (clean-run drop) and the partial-bring-up +// rollback so both release a borrowed entry the same way: +// 1. Delete the map entry ONLY when it still holds the exact staged runtime this +// run installed or reused (object identity). A concurrent run that replaced +// the entry with its own keeps it. +// 2. Fire this run's own one-time host staged-temp dispose regardless. The +// dispose closure is idempotent, so a shared closure re-fired across a reuse +// chain is safe. +async function releaseStagedRuntimeEntry(input: { + handles: Map; + sessionKey: string; + stagedRuntime: PreparedAdapterExecutionTargetRuntime | null; + dispose: (() => Promise) | null; +}): Promise { + const { handles, sessionKey, stagedRuntime, dispose } = input; + const existing = handles.get(sessionKey); + if (existing && stagedRuntime && existing.stagedRuntime === stagedRuntime) { + handles.delete(sessionKey); } - if (prepared.remoteStagingDispose) await prepared.remoteStagingDispose().catch(() => {}); + if (dispose) await dispose().catch(() => {}); } // Per-`sessionKey` async lease: chains each caller after the previous one so @@ -3228,7 +3341,56 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { stepWallSumMs += wallMs; }, }; - let prepared: AcpxPreparedRuntime; + // `startupFailed` drives the sandbox.startup span end status. It stays true + // until the bring-up establishes a session handle. The one `finally` below + // ends the span exactly once, on every return and on every throw. + let startupFailed = true; + // `buildRuntimeSettled` marks that buildRuntime returned. A failure before it + // settled has nothing to settle here; a failure after it settled routes + // through the create_runtime handler below. + let buildRuntimeSettled = false; + // The bring-up assigns these. The turn and teardown paths read them after + // the sandbox.startup span closes, so they are declared here. + let prepared!: AcpxPreparedRuntime; + let runtime!: AcpRuntime; + let sessionHandle!: AcpRuntimeHandle; + let childStderrState!: ChildStderrState; + let processIdentitySink!: AcpxProcessIdentitySink; + let resumedSession = false; + let clearSession = false; + let referencedProjectStagingFailuresField: + | { referencedProjectStagingFailures: Array<{ projectId: string }> } + | Record = {}; + // One teardown policy for the run. `settleRunResources` stops the bridges, + // runs the managed-home copy-back, releases the staging lease (in a finally + // inside cleanupRemoteBridges), and flushes the child stderr. It runs at most + // once, records each step error, and never throws, so a result-emission + // failure or an earlier teardown fault never skips a later step. + let runResourcesSettled = false; + // `turnTeardownDone` marks that the turn path already closed the runtime and + // settled the staged runtime, so a result-mapping throw after the close does + // not run the turn teardown a second time. + let turnTeardownDone = false; + const recordTeardownError = async (step: string, teardownErr: unknown) => { + const reason = teardownErr instanceof Error ? teardownErr.message : String(teardownErr); + await ctx + .onLog("stderr", `[paperclip] ACPX teardown step "${step}" failed: ${reason}\n`) + .catch(() => {}); + }; + const settleRunResources = async () => { + if (runResourcesSettled) return; + runResourcesSettled = true; + try { + await cleanupRemoteBridges(prepared); + } catch (teardownErr) { + await recordTeardownError("cleanup-remote-bridges", teardownErr); + } + try { + if (childStderrState) flushChildStderr(childStderrState); + } catch (teardownErr) { + await recordTeardownError("flush-child-stderr", teardownErr); + } + }; try { // Publish the `sandbox.startup` context to the runtime-parent store for // the whole bring-up. A startup-body exec that runs outside a measured @@ -3240,237 +3402,305 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { prepared = await runWithRuntimeParent(spanParent.parentContext, () => buildRuntime({ ctx, engine, deps, spanParent, getRuntimeParentContext, runtimeSpan: runRuntimeSpan, stageRuntimeSpan: runStageSpan }), ); - } catch (err) { - rootSpan.end(true); - throw err; - } - // Per-project staging outcomes for the referenced (mentioned) projects, surfaced back to the - // server on the run result. A referenced project that failed to stage into the sandbox is a - // first-class, counted failure in the requested-vs-synced observability, not only a warning. The - // list is empty on a local target, on a transport that does not stage referenced projects, or - // when every staged referenced project succeeded, so the spread adds the field only when there - // is a failure to report. - const referencedProjectStagingFailures = ( - prepared.stagedRuntime?.additionalSourceFailures ?? [] - ).map((failure) => ({ projectId: failure.projectId })); - const referencedProjectStagingFailuresField = - referencedProjectStagingFailures.length > 0 ? { referencedProjectStagingFailures } : {}; - // State the effective wall-clock timeout and its source up front so a - // later timeout is diagnosable from the run log alone. Goes to stderr: - // the acpx stdout log stream carries JSON acpx.* event payloads and must - // stay machine-parseable line by line. - await ctx.onLog( - "stderr", - `[paperclip] ${formatAdapterExecutionTimeoutStartLogLine(prepared.timeoutResolution)}\n`, - ); - await cleanupIdleHandles({ handles: warmHandles, now: now(), idleMs: warmIdleMs }); - - const previousParams = parseObject(ctx.runtime.sessionParams); - const canResume = isCompatibleSession(previousParams, prepared); - const resumeSessionId = canResume ? asString(previousParams.acpSessionId, "") || undefined : undefined; - const cached = canResume ? warmHandles.get(prepared.sessionKey) : undefined; - const childStderrState = cached?.childStderrState ?? { logPath: null, pendingLiveLine: "" }; - const processIdentitySink = cached?.processIdentitySink ?? { - current: ctx.onSpawn, - latest: null, - }; - // ACPX runtimes can stay warm across heartbeat runs. Keep the callback - // target mutable so a later agent respawn records identity on the current - // heartbeat instead of the run that originally created the runtime. - processIdentitySink.current = ctx.onSpawn; - flushChildStderr(childStderrState); - childStderrState.logPath = prepared.childStderrLogPath; - const runtimeOptions: PaperclipAcpRuntimeOptions = { - cwd: prepared.cwd, - // Host-only spawn cwd for the relay proxy on the remote process-session - // lane; `undefined` elsewhere so acpx falls back to `cwd` (byte-identical). - // The advertised `session/new` cwd (`prepared.cwd` = `remoteCwd`) and the - // fingerprint / compat key are unaffected — this redirects ONLY the host - // `spawn()` `chdir`, not the in-sandbox data path. - spawnCwd: prepared.hostSpawnCwd, - sessionStore: createRuntimeStore({ stateDir: prepared.stateDir }), - agentRegistry: prepared.agentRegistry, - permissionMode: prepared.permissionMode, - nonInteractivePermissions: prepared.nonInteractivePermissions, - mcpServers: prepared.mcpServers, - timeoutMs: prepared.timeoutSec > 0 ? prepared.timeoutSec * 1000 : undefined, - // Scope ACPX runtime verbose logs to the claude agent only. Codex - // and custom agents already emit their own per-tool output and don't - // benefit from doubling the log volume. - verbose: prepared.acpxAgent === "claude", - onAgentStderr: prepared.childStderrLogPath - ? (chunk) => routeChildStderr(childStderrState, chunk) - : undefined, - onAgentSpawn: async (meta) => { - processIdentitySink.latest = meta; - await processIdentitySink.current?.({ - pid: meta.pid, - processGroupId: null, - startedAt: meta.startedAt, - }); - }, - getRuntimeParentContext, - }; - // Open Q2: split the ~7s `acp.handshake` into the two in-repo-observable - // sub-phases — the ACP runtime construction (`createRuntime`) vs the session - // establishment envelope (`ensureSession`). The patched spawn lifecycle - // hook records process identity, but the finer spawn/`initialize`/ - // `session/new` timing split still lives inside external `acpx`. - // `createRuntime` runs once and only on a cold start; a warm-handle hit - // reuses `cached.runtime`, so `createRuntimeMs` stays undefined and the - // split reports nothing for it. - let createRuntimeMs: number | undefined; - let runtime: AcpRuntime; - // A warm handle reuses the running ACP runtime; a miss constructs one. The - // root span records this as `cold_start`. - coldStart = !cached?.runtime; - if (cached?.runtime) { - runtime = cached.runtime; - } else { - const createRuntimeStart = now(); - runtime = createRuntime(runtimeOptions); - createRuntimeMs = now() - createRuntimeStart; - } - if (cached) clearWarmHandleTimer(cached); - if (!canResume && asString(previousParams.runtimeSessionName, "")) { + buildRuntimeSettled = true; + // Per-project staging outcomes for the referenced (mentioned) projects, surfaced back to the + // server on the run result. A referenced project that failed to stage into the sandbox is a + // first-class, counted failure in the requested-vs-synced observability, not only a warning. The + // list is empty on a local target, on a transport that does not stage referenced projects, or + // when every staged referenced project succeeded, so the spread adds the field only when there + // is a failure to report. + const referencedProjectStagingFailures = ( + prepared.stagedRuntime?.additionalSourceFailures ?? [] + ).map((failure) => ({ projectId: failure.projectId })); + referencedProjectStagingFailuresField = + referencedProjectStagingFailures.length > 0 ? { referencedProjectStagingFailures } : {}; + // State the effective wall-clock timeout and its source up front so a + // later timeout is diagnosable from the run log alone. Goes to stderr: + // the acpx stdout log stream carries JSON acpx.* event payloads and must + // stay machine-parseable line by line. await ctx.onLog( - "stdout", - `[paperclip] ACPX session "${asString(previousParams.runtimeSessionName, "")}" does not match the current agent/cwd/mode/runtime identity; starting fresh in "${prepared.cwd}".\n`, + "stderr", + `[paperclip] ${formatAdapterExecutionTimeoutStartLogLine(prepared.timeoutResolution)}\n`, ); - } + await cleanupIdleHandles({ handles: warmHandles, now: now(), idleMs: warmIdleMs }); - let handle = cached?.handle ?? null; - let resumedSession = Boolean(handle ?? resumeSessionId); - let clearSession = false; - - try { - if (!handle) { - try { - // Step 7 — acp.handshake: ACP session establishment (session/new or - // resume). A throwing handshake still reports its duration before the - // resume-retry path below runs. The createRuntime/ensureSession - // sub-split rides the step span as fixed, closed keys (Open Q2). - let ensureSessionMs: number | undefined; - handle = await measureStartupStep(ctx, now, "acp.handshake", async () => { - const ensureSessionStart = now(); - const established = await runtime.ensureSession({ - sessionKey: prepared.sessionKey, - agent: prepared.acpxAgent, - mode: prepared.mode, - cwd: prepared.cwd, - resumeSessionId, - sessionOptions: { env: prepared.env }, - }); - ensureSessionMs = now() - ensureSessionStart; - return established; - }, { - ...prepared.stepMetrics, - // The two sub-times ride the span as fixed, closed keys. - spanWallTimes: () => ({ - createRuntime: createRuntimeMs, - ensureSession: ensureSessionMs, - }), + const previousParams = parseObject(ctx.runtime.sessionParams); + const canResume = isCompatibleSession(previousParams, prepared); + const resumeSessionId = canResume ? asString(previousParams.acpSessionId, "") || undefined : undefined; + const cached = canResume ? warmHandles.get(prepared.sessionKey) : undefined; + childStderrState = cached?.childStderrState ?? { logPath: null, pendingLiveLine: "" }; + processIdentitySink = cached?.processIdentitySink ?? { + current: ctx.onSpawn, + latest: null, + }; + // ACPX runtimes can stay warm across heartbeat runs. Keep the callback + // target mutable so a later agent respawn records identity on the current + // heartbeat instead of the run that originally created the runtime. + processIdentitySink.current = ctx.onSpawn; + flushChildStderr(childStderrState); + childStderrState.logPath = prepared.childStderrLogPath; + const runtimeOptions: PaperclipAcpRuntimeOptions = { + cwd: prepared.cwd, + // Host-only spawn cwd for the relay proxy on the remote process-session + // lane; `undefined` elsewhere so acpx falls back to `cwd` (byte-identical). + // The advertised `session/new` cwd (`prepared.cwd` = `remoteCwd`) and the + // fingerprint / compat key are unaffected — this redirects ONLY the host + // `spawn()` `chdir`, not the in-sandbox data path. + spawnCwd: prepared.hostSpawnCwd, + sessionStore: createRuntimeStore({ stateDir: prepared.stateDir }), + agentRegistry: prepared.agentRegistry, + permissionMode: prepared.permissionMode, + nonInteractivePermissions: prepared.nonInteractivePermissions, + mcpServers: prepared.mcpServers, + timeoutMs: prepared.timeoutSec > 0 ? prepared.timeoutSec * 1000 : undefined, + // Scope ACPX runtime verbose logs to the claude agent only. Codex + // and custom agents already emit their own per-tool output and don't + // benefit from doubling the log volume. + verbose: prepared.acpxAgent === "claude", + onAgentStderr: prepared.childStderrLogPath + ? (chunk) => routeChildStderr(childStderrState, chunk) + : undefined, + onAgentSpawn: async (meta) => { + processIdentitySink.latest = meta; + await processIdentitySink.current?.({ + pid: meta.pid, + processGroupId: null, + startedAt: meta.startedAt, }); - } catch (err) { - if (!resumeSessionId || !isResumeFailure(err)) throw err; - clearSession = true; - resumedSession = false; - await ctx.onLog( - "stdout", - `[paperclip] ACPX resume session "${resumeSessionId}" is unavailable; retrying with a fresh session.\n`, - ); - // Fresh-session retry: the runtime was already constructed on the - // first attempt (never re-created), so this event reports only its - // own `ensureSessionMs` — no `createRuntimeMs`. - let retryEnsureSessionMs: number | undefined; - handle = await measureStartupStep(ctx, now, "acp.handshake", async () => { - const ensureSessionStart = now(); - const established = await runtime.ensureSession({ - sessionKey: prepared.sessionKey, - agent: prepared.acpxAgent, - mode: prepared.mode, - cwd: prepared.cwd, - sessionOptions: { env: prepared.env }, + }, + getRuntimeParentContext, + }; + // Open Q2: split the ~7s `acp.handshake` into the two in-repo-observable + // sub-phases — the ACP runtime construction (`createRuntime`) vs the session + // establishment envelope (`ensureSession`). The patched spawn lifecycle + // hook records process identity, but the finer spawn/`initialize`/ + // `session/new` timing split still lives inside external `acpx`. + // `createRuntime` runs once and only on a cold start; a warm-handle hit + // reuses `cached.runtime`, so `createRuntimeMs` stays undefined and the + // split reports nothing for it. + let createRuntimeMs: number | undefined; + // A warm handle reuses the running ACP runtime; a miss constructs one. The + // root span records this as `cold_start`. + coldStart = !cached?.runtime; + if (cached?.runtime) { + runtime = cached.runtime; + } else { + const createRuntimeStart = now(); + runtime = createRuntime(runtimeOptions); + createRuntimeMs = now() - createRuntimeStart; + } + if (cached) clearWarmHandleTimer(cached); + if (!canResume && asString(previousParams.runtimeSessionName, "")) { + await ctx.onLog( + "stdout", + `[paperclip] ACPX session "${asString(previousParams.runtimeSessionName, "")}" does not match the current agent/cwd/mode/runtime identity; starting fresh in "${prepared.cwd}".\n`, + ); + } + + let handle = cached?.handle ?? null; + resumedSession = Boolean(handle ?? resumeSessionId); + + try { + if (!handle) { + try { + // Step 7 — acp.handshake: ACP session establishment (session/new or + // resume). A throwing handshake still reports its duration before the + // resume-retry path below runs. The createRuntime/ensureSession + // sub-split rides the step span as fixed, closed keys (Open Q2). + let ensureSessionMs: number | undefined; + handle = await measureStartupStep(ctx, now, "acp.handshake", async () => { + const ensureSessionStart = now(); + const established = await runtime.ensureSession({ + sessionKey: prepared.sessionKey, + agent: prepared.acpxAgent, + mode: prepared.mode, + cwd: prepared.cwd, + resumeSessionId, + sessionOptions: { env: prepared.env }, + }); + ensureSessionMs = now() - ensureSessionStart; + return established; + }, { + ...prepared.stepMetrics, + // The two sub-times ride the span as fixed, closed keys. + spanWallTimes: () => ({ + createRuntime: createRuntimeMs, + ensureSession: ensureSessionMs, + }), }); - retryEnsureSessionMs = now() - ensureSessionStart; - return established; - }, { - ...prepared.stepMetrics, - // The retry reuses the runtime from the first attempt, so it reports - // only its own ensure-session sub-time on the span. - spanWallTimes: () => ({ ensureSession: retryEnsureSessionMs }), + } catch (err) { + if (!resumeSessionId || !isResumeFailure(err)) throw err; + clearSession = true; + resumedSession = false; + await ctx.onLog( + "stdout", + `[paperclip] ACPX resume session "${resumeSessionId}" is unavailable; retrying with a fresh session.\n`, + ); + // Fresh-session retry: the runtime was already constructed on the + // first attempt (never re-created), so this event reports only its + // own `ensureSessionMs` — no `createRuntimeMs`. + let retryEnsureSessionMs: number | undefined; + handle = await measureStartupStep(ctx, now, "acp.handshake", async () => { + const ensureSessionStart = now(); + const established = await runtime.ensureSession({ + sessionKey: prepared.sessionKey, + agent: prepared.acpxAgent, + mode: prepared.mode, + cwd: prepared.cwd, + sessionOptions: { env: prepared.env }, + }); + retryEnsureSessionMs = now() - ensureSessionStart; + return established; + }, { + ...prepared.stepMetrics, + // The retry reuses the runtime from the first attempt, so it reports + // only its own ensure-session sub-time on the span. + spanWallTimes: () => ({ ensureSession: retryEnsureSessionMs }), + }); + } + } else { + // Warm-handle hit: a compatible cached handle reuses the running ACP + // agent, so the `acp.handshake` step does no work. Emit a step span and + // event with `outcome = skipped` and a zero wall time, so the trace and + // the run log show the skip as a distinct outcome, never a misleading + // zero-work `ok` step. + await emitSkippedStartupStep(ctx, "acp.handshake", { + tracer: prepared.stepMetrics.tracer, + parentContext: prepared.stepMetrics.parentContext, }); } - } else { - // Warm-handle hit: a compatible cached handle reuses the running ACP - // agent, so the `acp.handshake` step does no work. Emit a step span and - // event with `outcome = skipped` and a zero wall time, so the trace and - // the run log show the skip as a distinct outcome, never a misleading - // zero-work `ok` step. - await emitSkippedStartupStep(ctx, "acp.handshake", { - tracer: prepared.stepMetrics.tracer, - parentContext: prepared.stepMetrics.parentContext, - }); + // A compatible warm handle reuses the already-running ACP agent and does + // not emit another spawn event. Persist its known identity on this run + // before the next prompt starts so every running heartbeat is adoptable. + if (handle && cached && processIdentitySink.latest && ctx.onSpawn) { + await ctx.onSpawn({ + pid: processIdentitySink.latest.pid, + processGroupId: null, + startedAt: processIdentitySink.latest.startedAt, + }); + } + } catch (err) { + // Close the runtime on a pre-turn handshake failure and drop the matching + // warm entry, the same as the configuration failure path. A warm-hit + // reuse already cleared the idle timer before the handshake ran, so the + // failure must close and remove the entry, never re-arm it. The close and + // the staged-runtime drop run before the result emission, so a throwing + // emission never skips them; `settleRunResources` in the finally stops the + // bridges, releases the staging lease, and flushes the child stderr. + if (handle) { + await runtime.close({ + handle, + reason: "paperclip handshake cleanup", + discardPersistentState: false, + }).catch((closeErr) => recordTeardownError("runtime-close", closeErr)); + const existing = warmHandles.get(prepared.sessionKey); + if (warmHandleMatches(existing, runtime, handle) && existing) { + clearWarmHandleTimer(existing); + warmHandles.delete(prepared.sessionKey); + } + } + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); + try { + const { classified, message } = await emitAcpxFailure({ + ctx, + prepared, + err, + phase: "ensure_session", + }); + return { + exitCode: 1, + signal: null, + timedOut: false, + errorMessage: message, + ...classified, + ...billingFields, + ...referencedProjectStagingFailuresField, + model: prepared.requestedModel || null, + clearSession, + resultJson: { phase: "ensure_session" }, + summary: message, + }; + } finally { + await settleRunResources(); + } } - // A compatible warm handle reuses the already-running ACP agent and does - // not emit another spawn event. Persist its known identity on this run - // before the next prompt starts so every running heartbeat is adoptable. - if (handle && cached && processIdentitySink.latest && ctx.onSpawn) { - await ctx.onSpawn({ - pid: processIdentitySink.latest.pid, - processGroupId: null, - startedAt: processIdentitySink.latest.startedAt, - }); - } - } catch (err) { - // Bring-up failed at the handshake — close the root span with error status. - rootSpan.end(true); - const { classified, message } = await emitAcpxFailure({ - ctx, - prepared, - err, - phase: "ensure_session", - }); - await discardStagedRuntime({ handles: stagedRuntimes, prepared }); - await cleanupRemoteBridges(prepared); - return { - exitCode: 1, - signal: null, - timedOut: false, - errorMessage: message, - ...classified, - ...billingFields, - ...referencedProjectStagingFailuresField, - model: prepared.requestedModel || null, - clearSession, - resultJson: { phase: "ensure_session" }, - summary: message, - }; - } - if (!handle) { - // Bring-up produced no session handle — close the root span with error status. - rootSpan.end(true); + if (!handle) { + // ensureSession returned no session handle. Close the runtime the run + // constructed so its child process cannot leak. There is no established + // handle, so build a minimal one from the prepared identity; the close is + // best effort and never blocks the error result. + await runtime.close({ + handle: { + sessionKey: prepared.sessionKey, + backend: prepared.acpxAgent, + runtimeSessionName: prepared.sessionKey, + }, + reason: "paperclip missing-handle cleanup", + discardPersistentState: false, + }).catch((closeErr) => recordTeardownError("runtime-close", closeErr)); + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); + try { + return { + exitCode: 1, + signal: null, + timedOut: false, + errorMessage: "ACPX did not return a runtime session handle.", + errorCode: "acpx_runtime_error", + ...billingFields, + ...referencedProjectStagingFailuresField, + model: prepared.requestedModel || null, + resultJson: { phase: "ensure_session" }, + summary: "ACPX did not return a runtime session handle.", + }; + } finally { + await settleRunResources(); + } + } + sessionHandle = handle; + startupFailed = false; + } catch (err) { + if (!buildRuntimeSettled) { + // buildRuntime failed before it staged or bridged anything, so there is + // nothing to settle here. The finally below ends the sandbox.startup + // span; let the failure propagate. + throw err; + } + // The post-build runtime-creation window failed after buildRuntime returned + // live bridges and a held staging lease. Drop the staged runtime, then emit + // and settle: `settleRunResources` in the finally stops the bridges, + // releases the staging lease, and flushes the child stderr even if the + // emission throws. await discardStagedRuntime({ handles: stagedRuntimes, prepared }); - await cleanupRemoteBridges(prepared); - return { - exitCode: 1, - signal: null, - timedOut: false, - errorMessage: "ACPX did not return a runtime session handle.", - errorCode: "acpx_runtime_error", - ...billingFields, - ...referencedProjectStagingFailuresField, - model: prepared.requestedModel || null, - resultJson: { phase: "ensure_session" }, - summary: "ACPX did not return a runtime session handle.", - }; + try { + const { classified, message } = await emitAcpxFailure({ + ctx, + prepared, + err, + phase: "create_runtime", + }); + return { + exitCode: 1, + signal: null, + timedOut: false, + errorMessage: message, + ...classified, + ...billingFields, + ...referencedProjectStagingFailuresField, + model: prepared.requestedModel || null, + clearSession, + resultJson: { phase: "create_runtime" }, + summary: message, + }; + } finally { + await settleRunResources(); + } + } finally { + // End the sandbox.startup span exactly once, on every return and on every + // throw. It covers buildRuntime through acp.handshake and no further; the + // agent turn runs after and is out of the span's scope. + rootSpan.end(startupFailed); } - // Bring-up is complete: the session handle is established. Close the root - // span here, so it covers `buildRuntime` through `acp.handshake` and no - // further. The agent turn runs after and is out of the startup root's scope. - rootSpan.end(false); - const sessionHandle = handle; try { await applySessionConfigOptions({ runtime, @@ -3479,90 +3709,51 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { onLog: ctx.onLog, }); } catch (err) { - const { classified, message } = await emitAcpxFailure({ - ctx, - prepared, - err, - phase: "configure_session", - }); + // Close the runtime and drop the matching warm entry and the staged runtime + // before the result emission, so a throwing emission never skips them. + // `settleRunResources` in the finally stops the bridges, releases the + // staging lease, and flushes the child stderr. await runtime.close({ handle: sessionHandle, reason: "paperclip config cleanup", discardPersistentState: false, - }).catch(() => {}); + }).catch((closeErr) => recordTeardownError("runtime-close", closeErr)); const existing = warmHandles.get(prepared.sessionKey); if (warmHandleMatches(existing, runtime, sessionHandle) && existing) { clearWarmHandleTimer(existing); warmHandles.delete(prepared.sessionKey); } await discardStagedRuntime({ handles: stagedRuntimes, prepared }); - await cleanupRemoteBridges(prepared); - return { - exitCode: 1, - signal: null, - timedOut: false, - errorMessage: message, - ...classified, - ...billingFields, - ...referencedProjectStagingFailuresField, - model: prepared.requestedModel || null, - clearSession, - resultJson: { + try { + const { classified, message } = await emitAcpxFailure({ + ctx, + prepared, + err, phase: "configure_session", - agent: prepared.acpxAgent, - requestedModel: prepared.requestedModel || null, - requestedThinkingEffort: prepared.requestedThinkingEffort || null, - fastMode: prepared.fastMode, - }, - summary: message, - }; + }); + return { + exitCode: 1, + signal: null, + timedOut: false, + errorMessage: message, + ...classified, + ...billingFields, + ...referencedProjectStagingFailuresField, + model: prepared.requestedModel || null, + clearSession, + resultJson: { + phase: "configure_session", + agent: prepared.acpxAgent, + requestedModel: prepared.requestedModel || null, + requestedThinkingEffort: prepared.requestedThinkingEffort || null, + fastMode: prepared.fastMode, + }, + summary: message, + }; + } finally { + await settleRunResources(); + } } - const { prompt, promptMetrics, commandNotes } = await buildPrompt(ctx, resumedSession, prepared.env); - const runPrompt = joinPromptSections([prepared.skillPromptInstructions, prompt]); - await emitAcpxLog(ctx, { - type: "acpx.session", - agent: prepared.acpxAgent, - sessionId: sessionHandle.backendSessionId, - acpSessionId: sessionHandle.backendSessionId, - agentSessionId: sessionHandle.agentSessionId, - runtimeSessionName: sessionHandle.runtimeSessionName, - mode: prepared.mode, - permissionMode: prepared.permissionMode, - model: prepared.requestedModel || null, - thinkingEffort: prepared.requestedThinkingEffort || null, - fastMode: prepared.fastMode, - }); - if (ctx.onMeta) { - await ctx.onMeta({ - adapterType: engine.adapterType, - command: prepared.agentCommand ?? prepared.acpxAgent, - cwd: prepared.cwd, - commandNotes: [ - `ACPX runtime embedded in Paperclip with ${prepared.mode} session mode.`, - `Effective ACPX permission mode: ${prepared.permissionMode}.`, - ...(prepared.requestedModel - ? [ - prepared.acpxAgent === "claude" - ? `Requested ACPX model: ${prepared.requestedModel} (set via ANTHROPIC_MODEL env at startup).` - : prepared.acpxAgent === "codex" - ? `Requested ACPX model: ${prepared.requestedModel} (set via CODEX_CONFIG at startup).` - : `Requested ACPX model: ${prepared.requestedModel}.`, - ] - : []), - ...(prepared.requestedThinkingEffort ? [`Requested ACPX thinking effort: ${prepared.requestedThinkingEffort}.`] : []), - ...(prepared.fastMode ? ["Requested ACPX Codex fast mode."] : []), - ...(Array.isArray(prepared.skillsIdentity.commandNotes) - ? prepared.skillsIdentity.commandNotes.filter((note): note is string => typeof note === "string") - : []), - ...commandNotes, - ], - env: prepared.loggedEnv, - prompt: runPrompt, - promptMetrics, - context: ctx.context, - }); - } - let cancelActiveTurn: ((reason: string) => Promise) | null = null; let controller: AbortController | null = null; let timeout: NodeJS.Timeout | null = null; @@ -3570,6 +3761,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const textParts: string[] = []; let eventBreakdown: AcpRuntimeUsageBreakdown | null = null; let eventCostUsd: number | null = null; + // `turnStarted` separates a pre-turn preparation failure (prompt build or + // metadata emit) from a failure of the running turn, so the turn catch below + // reports the right phase and settles the runtime on both. + let turnStarted = false; // Open the agent turn span as a child of the run root span. It wraps the // whole turn: the executor holds `turnSpan.parentContext` for later exec // parenting, and the `finally` below ends the span once on every path. The @@ -3580,6 +3775,54 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { // `finally` resets the holder to the `task.run` token. currentRunParentContext = turnSpan.parentContext; try { + // Build the prompt and emit the run metadata inside the turn failure + // boundary. A failure here settles the runtime through the turn catch and + // returns an error result with phase `prepare_turn`. + const { prompt, promptMetrics, commandNotes } = await buildPrompt(ctx, resumedSession, prepared.env); + const runPrompt = joinPromptSections([prepared.skillPromptInstructions, prompt]); + await emitAcpxLog(ctx, { + type: "acpx.session", + agent: prepared.acpxAgent, + sessionId: sessionHandle.backendSessionId, + acpSessionId: sessionHandle.backendSessionId, + agentSessionId: sessionHandle.agentSessionId, + runtimeSessionName: sessionHandle.runtimeSessionName, + mode: prepared.mode, + permissionMode: prepared.permissionMode, + model: prepared.requestedModel || null, + thinkingEffort: prepared.requestedThinkingEffort || null, + fastMode: prepared.fastMode, + }); + if (ctx.onMeta) { + await ctx.onMeta({ + adapterType: engine.adapterType, + command: prepared.agentCommand ?? prepared.acpxAgent, + cwd: prepared.cwd, + commandNotes: [ + `ACPX runtime embedded in Paperclip with ${prepared.mode} session mode.`, + `Effective ACPX permission mode: ${prepared.permissionMode}.`, + ...(prepared.requestedModel + ? [ + prepared.acpxAgent === "claude" + ? `Requested ACPX model: ${prepared.requestedModel} (set via ANTHROPIC_MODEL env at startup).` + : prepared.acpxAgent === "codex" + ? `Requested ACPX model: ${prepared.requestedModel} (set via CODEX_CONFIG at startup).` + : `Requested ACPX model: ${prepared.requestedModel}.`, + ] + : []), + ...(prepared.requestedThinkingEffort ? [`Requested ACPX thinking effort: ${prepared.requestedThinkingEffort}.`] : []), + ...(prepared.fastMode ? ["Requested ACPX Codex fast mode."] : []), + ...(Array.isArray(prepared.skillsIdentity.commandNotes) + ? prepared.skillsIdentity.commandNotes.filter((note): note is string => typeof note === "string") + : []), + ...commandNotes, + ], + env: prepared.loggedEnv, + prompt: runPrompt, + promptMetrics, + context: ctx.context, + }); + } // Snapshot pre-turn usage so cumulative agent-reported cost can be // attributed to this run alone. const preTurnStatus = await readRuntimeStatus(runtime, sessionHandle); @@ -3600,6 +3843,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { timeoutMs, signal: controller?.signal, }); + turnStarted = true; cancelActiveTurn = async (reason: string) => { await turn.cancel({ reason }); }; @@ -3693,6 +3937,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { } else { await discardStagedRuntime({ handles: stagedRuntimes, prepared }); } + // The turn path has closed the runtime and settled the staged runtime. Mark + // it done so a result-mapping throw below runs the turn teardown once, not a + // second time through the catch. + turnTeardownDone = true; const errorMessage = timedOut ? formatAdapterExecutionTimeoutErrorMessage(prepared.timeoutResolution) @@ -3704,6 +3952,11 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { stopReason: terminalStopReason, message: errorMessage, }); + // Settle the run resources once: stop the bridges, run the copy-back, + // release the staging lease (in a finally inside cleanupRemoteBridges), and + // flush the child stderr. Mark them settled first so a result-mapping throw + // below does not run the teardown again through the turn catch. + runResourcesSettled = true; await cleanupRemoteBridges(prepared); flushChildStderr(childStderrState); // The one clean-completion path clears the run failure flag; every other @@ -3741,47 +3994,62 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }; } catch (err) { if (timeout) clearTimeout(timeout); + // A failure before `startTurn` returned is a turn-preparation failure; a + // failure after it is a running-turn failure. The teardown is the same; + // only the reported phase differs. + const phase: AcpxExecutionPhase = turnStarted ? "turn" : "prepare_turn"; const messageOverride = timedOut ? formatAdapterExecutionTimeoutErrorMessage(prepared.timeoutResolution) : undefined; const cancel = cancelActiveTurn as ((reason: string) => Promise) | null; const preEmitMessage = messageOverride ?? (err instanceof Error ? err.message : String(err)); - if (cancel) await cancel(preEmitMessage).catch(() => {}); - await runtime.close({ - handle: sessionHandle, - reason: timedOut ? "paperclip timeout cleanup" : "paperclip error cleanup", - discardPersistentState: timedOut, - }).catch(() => {}); - const existing = warmHandles.get(prepared.sessionKey); - if (warmHandleMatches(existing, runtime, sessionHandle) && existing) { - clearWarmHandleTimer(existing); - warmHandles.delete(prepared.sessionKey); + // Skip the cancel, close, and staged-runtime drop when the success path + // already ran them: a result-mapping throw after the close reaches this + // catch, and the turn teardown must not run twice. The close records its + // error and never blocks the later teardown; `settleRunResources` in the + // finally stops the bridges, releases the staging lease, and flushes the + // child stderr even if the emission throws. + if (!turnTeardownDone) { + turnTeardownDone = true; + if (cancel) await cancel(preEmitMessage).catch(() => {}); + await runtime.close({ + handle: sessionHandle, + reason: timedOut ? "paperclip timeout cleanup" : "paperclip error cleanup", + discardPersistentState: timedOut, + }).catch((closeErr) => recordTeardownError("runtime-close", closeErr)); + const existing = warmHandles.get(prepared.sessionKey); + if (warmHandleMatches(existing, runtime, sessionHandle) && existing) { + clearWarmHandleTimer(existing); + warmHandles.delete(prepared.sessionKey); + } + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); + } + try { + const { classified, message } = await emitAcpxFailure({ + ctx, + prepared, + err, + phase, + messageOverride, + }); + return { + exitCode: 1, + signal: timedOut ? "SIGTERM" : null, + timedOut, + errorMessage: message, + errorCode: timedOut ? "acpx_timeout" : classified.errorCode, + errorMeta: classified.errorMeta, + ...billingFields, + ...referencedProjectStagingFailuresField, + model: prepared.requestedModel || null, + clearSession: clearSession || timedOut, + resultJson: { phase }, + summary: message, + }; + } finally { + await settleRunResources(); } - await discardStagedRuntime({ handles: stagedRuntimes, prepared }); - const { classified, message } = await emitAcpxFailure({ - ctx, - prepared, - err, - phase: "turn", - messageOverride, - }); - await cleanupRemoteBridges(prepared); - flushChildStderr(childStderrState); - return { - exitCode: 1, - signal: timedOut ? "SIGTERM" : null, - timedOut, - errorMessage: message, - errorCode: timedOut ? "acpx_timeout" : classified.errorCode, - errorMeta: classified.errorMeta, - ...billingFields, - ...referencedProjectStagingFailuresField, - model: prepared.requestedModel || null, - clearSession: clearSession || timedOut, - resultJson: { phase: "turn" }, - summary: message, - }; } finally { // End the agent turn span exactly once, on every return and on a throw. // `runFailed` is `false` only on a completed, non-timed-out turn, so the diff --git a/packages/adapters/claude-local/src/server/acp.test.ts b/packages/adapters/claude-local/src/server/acp.test.ts index 45acb00db0..4c54b1ac7f 100644 --- a/packages/adapters/claude-local/src/server/acp.test.ts +++ b/packages/adapters/claude-local/src/server/acp.test.ts @@ -618,6 +618,79 @@ describe("claude_local ACP lane", () => { expect(Object.keys(meta[0]?.env ?? {}).filter((key) => key.startsWith("XDG_"))).toEqual([]); }); + it("test_claude_acp_seam_registers_workspace_sync_back", async () => { + const root = await makeTempRoot("paperclip-claude-acp-syncback-"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + const sharedClaudeConfig = path.join(root, "shared-claude-config"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.mkdir(sharedClaudeConfig, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + await fs.writeFile( + path.join(sharedClaudeConfig, "settings.json"), + JSON.stringify({ permissions: { defaultMode: "acceptEdits" } }), + "utf8", + ); + await fs.writeFile(path.join(sharedClaudeConfig, "CLAUDE.md"), "# shared guidance\n", "utf8"); + process.env.PAPERCLIP_HOME = path.join(root, "paperclip-home"); + process.env.PAPERCLIP_INSTANCE_ID = "test"; + process.env.CLAUDE_CONFIG_DIR = sharedClaudeConfig; + + // The runtime writes a NEW file into the in-sandbox workspace during the turn. + // The seam must register a workspace sync-back teardown, so the file lands in + // the host worktree after the run. + const runtime = new FakeRuntime({}); + const startTurn = runtime.startTurn.bind(runtime); + runtime.startTurn = (input) => { + const turn = startTurn(input); + const remoteWorkspaceCwd = input.handle.cwd ?? remoteCwd; + return { + ...turn, + result: (async () => { + await fs.writeFile(path.join(remoteWorkspaceCwd, "from-sandbox.txt"), "synced", "utf8"); + return await turn.result; + })(), + }; + }; + + const execute = createClaudeAcpExecutor({ + createRuntime: (options: FakeRuntimeOptions) => { + Object.assign(runtime.options, options); + return runtime as never; + }, + }); + + const result = await execute( + buildContext(localCwd, { + config: { + engine: "acp", + cwd: localCwd, + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + promptTemplate: "Do the assigned work.", + }, + context: { + issueId: "issue-1", + paperclipWorkspace: { cwd: localCwd, source: "project_workspace", workspaceId: "workspace-1" }, + }, + executionTarget: { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + } as never, + authToken: "real-run-jwt", + }), + ); + + expect(result.exitCode).toBe(0); + // The teardown fired `restoreWorkspace`, so the sandbox-authored file is now + // in the host worktree. + await expect(fs.readFile(path.join(localCwd, "from-sandbox.txt"), "utf8")).resolves.toBe("synced"); + }); + it("remaps a workspace-relative explicit CLAUDE_CONFIG_DIR onto the in-sandbox workspace path", async () => { const root = await makeTempRoot("paperclip-claude-acp-explicit-inworkspace-"); const localCwd = path.join(root, "worktree"); diff --git a/packages/adapters/claude-local/src/server/acp.ts b/packages/adapters/claude-local/src/server/acp.ts index 1513b6c360..f0dbc46b96 100644 --- a/packages/adapters/claude-local/src/server/acp.ts +++ b/packages/adapters/claude-local/src/server/acp.ts @@ -181,7 +181,8 @@ export function resolveClaudeAcpBillingIdentity( * asset, materialize it into an in-sandbox config dir (copying the sandbox's own * `$HOME/.claude` credentials in), then repoint `CLAUDE_CONFIG_DIR` onto that * in-sandbox config dir. Claude has no credential copy-back (its CLI lane has - * none — mirroring the CLI is the contract), so no teardown hook. + * none — mirroring the CLI is the contract). The teardown hook therefore only + * syncs the sandbox workspace back to the host; it does not touch credentials. * * An explicit `CLAUDE_CONFIG_DIR` (user-managed) is honored only if it can reach * the remote sandbox; a host-only path cannot, so we do NOT forward it verbatim @@ -193,6 +194,25 @@ async function prepareClaudeRemoteManagedHome( input: AcpxRemoteManagedHomeContext, ): Promise { const { env, runId, onLog, executionTarget } = input; + // Fail-open workspace sync-back for every exit path (mirrors the Claude CLI + // lane's restore-hook finally and the Codex ACP seam's teardown). Claude has no + // credential copy-back, so the teardown only syncs the sandbox workspace back to + // the host. A restore miss is logged and never fails the run. + const registerWorkspaceSyncBack = ( + stagedRuntime: AcpxRemoteManagedHomeResult["stagedRuntime"], + ): AcpxRemoteManagedHomeResult["teardown"] => async () => { + try { + await onLog("stdout", "[paperclip] Restoring workspace changes from the sandbox.\n"); + await stagedRuntime.restoreWorkspace((line) => onLog("stdout", line)); + } catch (err) { + await onLog( + "stderr", + `[paperclip] Claude ACP teardown workspace restore failed: ${ + err instanceof Error ? err.message : String(err) + }\n`, + ); + } + }; const envConfig = parseObject(input.config.env); const explicitClaudeConfigDir = typeof envConfig.CLAUDE_CONFIG_DIR === "string" && envConfig.CLAUDE_CONFIG_DIR.trim().length > 0 @@ -228,7 +248,7 @@ async function prepareClaudeRemoteManagedHome( "stdout", `[paperclip] Remapped operator CLAUDE_CONFIG_DIR from host path ${explicitClaudeConfigDir} onto the in-sandbox workspace path ${remappedConfigDir} for the remote ACP run.\n`, ); - return { stagedRuntime }; + return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) }; } await onLog( "stderr", @@ -266,7 +286,7 @@ async function prepareClaudeRemoteManagedHome( }); // Repoint CLAUDE_CONFIG_DIR onto the in-sandbox config dir. env.CLAUDE_CONFIG_DIR = remoteClaudeConfigDir; - return { stagedRuntime }; + return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) }; } function withClaudeAcpDefaults(options: ClaudeAcpExecutorOptions): AcpxEngineExecutorOptions { diff --git a/packages/adapters/gemini-local/src/server/acp.test.ts b/packages/adapters/gemini-local/src/server/acp.test.ts index 8ee4001a1d..ae6ff3780b 100644 --- a/packages/adapters/gemini-local/src/server/acp.test.ts +++ b/packages/adapters/gemini-local/src/server/acp.test.ts @@ -551,6 +551,68 @@ describe("gemini_local ACP lane", () => { expect(Object.keys(meta[0]?.env ?? {}).filter((key) => key.startsWith("XDG_"))).toEqual([]); }); + it("test_gemini_acp_seam_registers_workspace_sync_back", async () => { + const root = await makeTempRoot("paperclip-gemini-acp-syncback-"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8"); + + // The runtime writes a NEW file into the in-sandbox workspace during the turn. + // The seam must register a workspace sync-back teardown, so the file lands in + // the host worktree after the run. + const runtime = new FakeRuntime({}); + const startTurn = runtime.startTurn.bind(runtime); + runtime.startTurn = (input) => { + const turn = startTurn(input); + const remoteWorkspaceCwd = input.handle.cwd ?? remoteCwd; + return { + ...turn, + result: (async () => { + await fs.writeFile(path.join(remoteWorkspaceCwd, "from-sandbox.txt"), "synced", "utf8"); + return await turn.result; + })(), + }; + }; + + const execute = createGeminiAcpExecutor({ + createRuntime: (options) => { + Object.assign(runtime.options, options); + return runtime as never; + }, + }); + + const result = await execute( + buildContext(localCwd, { + config: { + engine: "acp", + cwd: localCwd, + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + promptTemplate: "Do the assigned work.", + }, + context: { + issueId: "issue-1", + paperclipWorkspace: { cwd: localCwd, source: "project_workspace", workspaceId: "workspace-1" }, + }, + executionTarget: { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + } as never, + authToken: "real-run-jwt", + }), + ); + + expect(result.exitCode).toBe(0); + // The teardown fired `restoreWorkspace`, so the sandbox-authored file is now + // in the host worktree. + await expect(fs.readFile(path.join(localCwd, "from-sandbox.txt"), "utf8")).resolves.toBe("synced"); + }); + it("does not persist an api-key auth selector from a host-only credential", async () => { const root = await makeTempRoot("paperclip-gemini-acp-hostkey-"); const localCwd = path.join(root, "worktree"); diff --git a/packages/adapters/gemini-local/src/server/acp.ts b/packages/adapters/gemini-local/src/server/acp.ts index ea07c647e7..ca573626cd 100644 --- a/packages/adapters/gemini-local/src/server/acp.ts +++ b/packages/adapters/gemini-local/src/server/acp.ts @@ -149,12 +149,32 @@ function resolveGeminiSkillsHome(config: Record): string { * * The seed never writes key bytes: the key is only read as a boolean signal to * decide whether to persist the auth-method selector. Gemini has no credential - * copy-back, so no teardown hook. + * copy-back. The teardown hook therefore only syncs the sandbox workspace back to + * the host; it does not touch credentials. */ async function prepareGeminiRemoteManagedHome( input: AcpxRemoteManagedHomeContext, ): Promise { const { env, runId, onLog, executionTarget } = input; + // Fail-open workspace sync-back for every exit path (mirrors the Gemini CLI + // lane's restore-hook finally and the Codex ACP seam's teardown). Gemini has no + // credential copy-back, so the teardown only syncs the sandbox workspace back to + // the host. A restore miss is logged and never fails the run. + const registerWorkspaceSyncBack = ( + stagedRuntime: AcpxRemoteManagedHomeResult["stagedRuntime"], + ): AcpxRemoteManagedHomeResult["teardown"] => async () => { + try { + await onLog("stdout", "[paperclip] Restoring workspace changes from the sandbox.\n"); + await stagedRuntime.restoreWorkspace((line) => onLog("stdout", line)); + } catch (err) { + await onLog( + "stderr", + `[paperclip] Gemini ACP teardown workspace restore failed: ${ + err instanceof Error ? err.message : String(err) + }\n`, + ); + } + }; const geminiSkillsHome = resolveGeminiSkillsHome(input.config); const stagedRuntime = await input.stage( geminiSkillsHome @@ -168,8 +188,9 @@ async function prepareGeminiRemoteManagedHome( const managedRemoteHomeDir = stagedRuntime.runtimeRootDir; if (!managedRemoteHomeDir) { // No runtime root resolved — leave HOME as-is (host fallback) and skip the - // in-sandbox seed; nothing to remap onto. - return { stagedRuntime }; + // in-sandbox seed; nothing to remap onto. The workspace still staged, so the + // sync-back teardown still applies. + return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) }; } env.HOME = managedRemoteHomeDir; @@ -222,7 +243,7 @@ async function prepareGeminiRemoteManagedHome( ); } - return { stagedRuntime }; + return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) }; } function withGeminiAcpDefaults(options: GeminiAcpExecutorOptions): AcpxEngineExecutorOptions { diff --git a/packages/shared/src/environment-support.test.ts b/packages/shared/src/environment-support.test.ts index 396945b027..0a2c72b364 100644 --- a/packages/shared/src/environment-support.test.ts +++ b/packages/shared/src/environment-support.test.ts @@ -43,3 +43,26 @@ describe("isSandboxProviderSupportedForAdapter", () => { ]); }); }); + +describe("getEnvironmentCapabilities reusable leases default", () => { + it("test_absent_reusable_leases_presents_as_false", () => { + const capabilities = getEnvironmentCapabilities(["codex_local"], { + sandboxProviders: { + // The manifest does not declare supportsReusableLeases. + "fake-plugin": { displayName: "Fake Plugin" }, + }, + }); + expect(capabilities.sandboxProviders["fake-plugin"].supportsReusableLeases).toBe(false); + + // A manifest that declares true still presents as true. + const declared = getEnvironmentCapabilities(["codex_local"], { + sandboxProviders: { + "reuse-plugin": { displayName: "Reuse Plugin", supportsReusableLeases: true }, + }, + }); + expect(declared.sandboxProviders["reuse-plugin"].supportsReusableLeases).toBe(true); + + // The built-in fake provider presents as false. + expect(capabilities.sandboxProviders.fake.supportsReusableLeases).toBe(false); + }); +}); diff --git a/packages/shared/src/environment-support.ts b/packages/shared/src/environment-support.ts index a60f77c5dd..f19fb95355 100644 --- a/packages/shared/src/environment-support.ts +++ b/packages/shared/src/environment-support.ts @@ -122,7 +122,9 @@ export function getEnvironmentCapabilities( supportsSavedProbe: true, supportsUnsavedProbe: true, supportsRunExecution: false, - supportsReusableLeases: true, + // The fake provider runtime declares false. The presentation must match + // it, so execution and presentation agree. + supportsReusableLeases: false, supportsInteractiveSetup: false, interactiveSetupConnectionTypes: [], supportsTemplateCapture: false, @@ -137,7 +139,9 @@ export function getEnvironmentCapabilities( supportsSavedProbe: capability.supportsSavedProbe ?? true, supportsUnsavedProbe: capability.supportsUnsavedProbe ?? true, supportsRunExecution: capability.supportsRunExecution ?? true, - supportsReusableLeases: capability.supportsReusableLeases ?? true, + // Default absent to false. A manifest that does not declare reusable + // leases must present as false, so execution (=== true) agrees. + supportsReusableLeases: capability.supportsReusableLeases ?? false, supportsInteractiveSetup: capability.supportsInteractiveSetup ?? false, interactiveSetupConnectionTypes: capability.interactiveSetupConnectionTypes ?? [], supportsTemplateCapture: capability.supportsTemplateCapture ?? false, diff --git a/server/src/__tests__/environment-routes.test.ts b/server/src/__tests__/environment-routes.test.ts index 9f20cc21fe..b718e89ab2 100644 --- a/server/src/__tests__/environment-routes.test.ts +++ b/server/src/__tests__/environment-routes.test.ts @@ -1164,7 +1164,9 @@ describe("environment routes", () => { expect(res.body.sandboxProviders["secure-plugin"]).toMatchObject({ status: "supported", supportsRunExecution: true, - supportsReusableLeases: true, + // The mock manifest omits reusable leases, so the capability defaults to + // false and agrees with the execution guard. + supportsReusableLeases: false, supportsInteractiveSetup: true, interactiveSetupConnectionTypes: ["ssh"], supportsTemplateCapture: true, diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index b22dd69a99..e36439ab55 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -3207,4 +3207,68 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(sshRelease).not.toHaveBeenCalled(); expect(acquired.lease.metadata?.driver).toBe("local"); }); + + it("test_release_run_leases_continues_after_first_release_fails", async () => { + const { companyId, environment, runId } = await seedEnvironment(); + const environmentsSvc = environmentService(db); + + // Seed two active leases for one run. The first driver release throws; the + // second must still release. + const failingLease = await environmentsSvc.acquireLease({ + companyId, + environmentId: environment.id, + heartbeatRunId: runId, + provider: "local", + providerLeaseId: "fail-release", + metadata: { driver: "local" }, + }); + const healthyLease = await environmentsSvc.acquireLease({ + companyId, + environmentId: environment.id, + heartbeatRunId: runId, + provider: "local", + providerLeaseId: "healthy-release", + metadata: { driver: "local" }, + }); + + const runtimeWithFailingDriver = environmentRuntimeService(db, { + drivers: [ + { + driver: "local", + acquireRunLease: async () => { + throw new Error("acquire should not be called"); + }, + releaseRunLease: async ({ lease, status }) => { + if (lease.providerLeaseId === "fail-release") { + throw new Error("driver release failed"); + } + return await environmentsSvc.releaseLease(lease.id, status); + }, + }, + ], + }); + + const errors: Array<{ leaseId: string; error: unknown }> = []; + const released = await runtimeWithFailingDriver.releaseRunLeases( + runId, + "released", + (leaseId, error) => errors.push({ leaseId, error }), + ); + + // The healthy lease released even though the first release failed. + expect(released).toHaveLength(1); + expect(released[0]?.lease.id).toBe(healthyLease.id); + expect(released[0]?.lease.status).toBe("released"); + + // The failed release reported one lease-specific error. + expect(errors).toHaveLength(1); + expect(errors[0]?.leaseId).toBe(failingLease.id); + expect((errors[0]?.error as Error).message).toBe("driver release failed"); + + // The database confirms the isolation. The healthy lease released; the + // failing lease stayed active. + const rows = await db.select().from(environmentLeases); + expect(rows.find((row) => row.id === healthyLease.id)?.status).toBe("released"); + expect(rows.find((row) => row.id === failingLease.id)?.status).toBe("active"); + }); }); diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts new file mode 100644 index 0000000000..f139d6e7f0 --- /dev/null +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -0,0 +1,941 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { + agents, + companies, + createDb, + environmentLeases, + environments, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() })); +vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient })); + +vi.mock("../middleware/logger.js", () => ({ + logger: { + child: vi.fn(function child() { + return this; + }), + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + }, + httpLogger: vi.fn(), +})); + +import { logger } from "../middleware/logger.ts"; +import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres pending_cleanup sweep tests on this host: ${ + embeddedPostgresSupport.reason ?? "unsupported environment" + }`, + ); +} + +// The sweep reads and writes its retry state under these lease metadata keys. +const ATTEMPTS_KEY = "pendingCleanupRetryAttempts"; +const CAP_WARNED_KEY = "pendingCleanupRetryCapWarned"; +const ATTEMPT_CAP = 5; + +describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pending-cleanup-sweep-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + beforeEach(() => { + vi.mocked(logger.warn).mockClear(); + vi.mocked(logger.error).mockClear(); + }); + + afterEach(async () => { + await db.delete(environmentLeases); + await db.delete(environments); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedCompanyAndEnvironment() { + const companyId = randomUUID(); + const environmentId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(environments).values({ + id: environmentId, + name: "Fake Sandbox", + driver: "sandbox", + status: "active", + config: { provider: "fake", image: "ubuntu:24.04" }, + createdAt: new Date(), + updatedAt: new Date(), + }); + return { companyId, environmentId }; + } + + async function insertPendingCleanupLease(input: { + companyId: string; + environmentId: string; + updatedAt: Date; + metadata?: Record; + }): Promise { + const id = randomUUID(); + await db.insert(environmentLeases).values({ + id, + companyId: input.companyId, + environmentId: input.environmentId, + status: "pending_cleanup", + leasePolicy: "reuse_by_environment", + provider: "fake", + providerLeaseId: `sandbox://fake/${id}`, + cleanupStatus: "failed", + metadata: { driver: "sandbox", ...(input.metadata ?? {}) }, + acquiredAt: input.updatedAt, + lastUsedAt: input.updatedAt, + releasedAt: input.updatedAt, + createdAt: input.updatedAt, + updatedAt: input.updatedAt, + }); + return id; + } + + function fakeRuntime( + destroyRunLease: HeartbeatEnvironmentRuntime["destroyRunLease"], + ): HeartbeatEnvironmentRuntime { + return { destroyRunLease } as unknown as HeartbeatEnvironmentRuntime; + } + + it("test_pending_cleanup_sweep_retries_and_destroys_lease", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy succeeds on retry and moves the lease out of pending_cleanup. + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + const now = new Date(); + const row = await db + .update(environmentLeases) + .set({ status: "expired", cleanupStatus: "success", updatedAt: now }) + .where(eq(environmentLeases.id, lease.id)) + .returning() + .then((rows) => rows[0] ?? null); + return row ? { ...row, status: "expired" as const } : null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]), + }); + + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + + expect(result).toEqual({ swept: 1, destroyed: 1, capped: 0 }); + expect(destroyRunLease).toHaveBeenCalledTimes(1); + expect(destroyRunLease.mock.calls[0]?.[0]).toMatchObject({ + failureReason: "pending_cleanup_retry", + lease: expect.objectContaining({ id: leaseId }), + }); + + const status = await db + .select({ status: environmentLeases.status }) + .from(environmentLeases) + .where(eq(environmentLeases.id, leaseId)) + .then((rows) => rows[0]?.status); + expect(status).toBe("expired"); + }); + + it("test_pending_cleanup_sweep_respects_backoff_and_page_size", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // A lease touched inside the backoff window is not eligible yet. + await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(), + }); + + const destroyRunLease = vi.fn(async () => null); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]), + }); + + const backoffResult = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60 * 1000 }); + expect(backoffResult.swept).toBe(0); + expect(destroyRunLease).not.toHaveBeenCalled(); + + // Seed more eligible leases than one page holds. The sweep processes one + // page only. + const eligibleCount = 22; + for (let index = 0; index < eligibleCount; index += 1) { + await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + } + + const pageResult = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60 * 1000 }); + expect(pageResult.swept).toBe(20); + expect(destroyRunLease).toHaveBeenCalledTimes(20); + }); + + it("test_pending_cleanup_sweep_stops_at_attempt_cap_and_warns_once", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP }, + }); + + const destroyRunLease = vi.fn(async () => null); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]), + }); + + const first = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(first).toEqual({ swept: 1, destroyed: 0, capped: 1 }); + // A capped lease is not retried. + expect(destroyRunLease).not.toHaveBeenCalled(); + + const metadataAfterFirst = await db + .select({ metadata: environmentLeases.metadata }) + .from(environmentLeases) + .where(eq(environmentLeases.id, leaseId)) + .then((rows) => rows[0]?.metadata as Record | null); + expect(metadataAfterFirst?.[CAP_WARNED_KEY]).toBe(true); + expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1); + + // A second sweep warns no more; the lease keeps its warned flag. + const second = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(second).toEqual({ swept: 1, destroyed: 0, capped: 1 }); + expect(destroyRunLease).not.toHaveBeenCalled(); + expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1); + }); + + // Two sweep ticks can overlap. Without an atomic claim, both read the same + // attempt count, both destroy the same lease, and the retry cap counts one + // attempt for two destroys. The atomic claim must let only one sweep destroy + // the lease per attempt. + it("test_pending_cleanup_overlapping_sweeps_destroy_lease_once", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy fails, so the lease stays in pending_cleanup. The counter + // records how many sweeps reached the destroy for this lease. + let destroyCount = 0; + const destroyRunLease = vi.fn(async () => { + destroyCount += 1; + return null; + }); + + // Two sweeps run on two separate database clients, so they truly overlap. + // A single client serializes the queries on one connection and hides the + // race. The second client connects to the same embedded database. + const dbB = createDb(tempDb!.connectionString); + try { + // Warm up the second connection first. A cold connection would add setup + // latency and let the first sweep finish before the second sweep reads. + await dbB.select({ id: environmentLeases.id }).from(environmentLeases).limit(1); + + const heartbeatA = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + const heartbeatB = heartbeatService(dbB, { + environmentRuntime: fakeRuntime( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // Both sweeps use the production backoff and run at the same time. + const backoffMs = 5 * 60 * 1000; + const [first, second] = await Promise.all([ + heartbeatA.sweepPendingCleanupLeases({ backoffMs }), + heartbeatB.sweepPendingCleanupLeases({ backoffMs }), + ]); + + // Only one sweep claimed the attempt and destroyed the lease. + expect(destroyCount).toBe(1); + expect(first.destroyed + second.destroyed).toBe(0); + } finally { + await (dbB as unknown as { $client: { end: () => Promise } }).$client.end(); + } + + // The attempt count advanced by exactly one; the cap is not exceeded. + const metadata = await db + .select({ metadata: environmentLeases.metadata, status: environmentLeases.status }) + .from(environmentLeases) + .where(eq(environmentLeases.id, leaseId)) + .then((rows) => rows[0]); + expect((metadata?.metadata as Record | null)?.[ATTEMPTS_KEY]).toBe(1); + expect(metadata?.status).toBe("pending_cleanup"); + }); + + // Read the current retry attempt count from a lease's metadata. + async function readAttempts(leaseId: string): Promise { + const metadata = await db + .select({ metadata: environmentLeases.metadata }) + .from(environmentLeases) + .where(eq(environmentLeases.id, leaseId)) + .then((rows) => rows[0]?.metadata as Record | null); + const value = metadata?.[ATTEMPTS_KEY]; + return typeof value === "number" ? value : 0; + } + + // Run two sweeps at the same time on two separate database clients, so they + // truly overlap. A single client serializes the queries on one connection and + // hides the race. The second client connects to the same embedded database. + // The caller reuses the returned destroy spy to count the real destroys. + async function runOverlappingSweeps( + destroyRunLease: HeartbeatEnvironmentRuntime["destroyRunLease"], + backoffMs: number, + ): Promise { + const dbB = createDb(tempDb!.connectionString); + try { + // Warm up the second connection first. A cold connection would add setup + // latency and let the first sweep finish before the second sweep reads. + await dbB.select({ id: environmentLeases.id }).from(environmentLeases).limit(1); + + const heartbeatA = heartbeatService(db, { + environmentRuntime: fakeRuntime(destroyRunLease), + }); + const heartbeatB = heartbeatService(dbB, { + environmentRuntime: fakeRuntime(destroyRunLease), + }); + + await Promise.all([ + heartbeatA.sweepPendingCleanupLeases({ backoffMs }), + heartbeatB.sweepPendingCleanupLeases({ backoffMs }), + ]); + } finally { + await (dbB as unknown as { $client: { end: () => Promise } }).$client.end(); + } + } + + // Two overlapping sweeps read the same pending_cleanup lease. The atomic claim + // must let only one sweep destroy the sandbox. This test runs the two sweeps + // on two clients, then asserts a single destroy and a single attempt + // increment. + it("test_concurrent_sweeps_destroy_a_lease_at_most_once", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy fails, so the lease stays in pending_cleanup and the test can + // read the final attempt count. + const destroyRunLease = vi.fn(async () => null); + + await runOverlappingSweeps( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + 5 * 60 * 1000, + ); + + // Only one sweep wins the claim, so the destroy runs once and the attempt + // count advances by one. + expect(destroyRunLease).toHaveBeenCalledTimes(1); + expect(await readAttempts(leaseId)).toBe(1); + }); + + // A lost increment lets the real attempt count pass the cap: two destroys run + // but the counter advances by one. The atomic claim prevents the lost + // increment. This test starts a lease at cap - 1, overlaps two sweeps, then + // asserts one destroy, one increment to the cap, and at most one cap warning. + it("test_concurrent_sweeps_never_exceed_attempt_cap", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 }, + }); + + // The destroy fails, so the lease stays in pending_cleanup and the test can + // read the final attempt count. + const destroyRunLease = vi.fn(async () => null); + + await runOverlappingSweeps( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + 5 * 60 * 1000, + ); + + // One real retry ran, so the count advances by one to the cap and never + // passes it. + expect(destroyRunLease).toHaveBeenCalledTimes(1); + const attempts = await readAttempts(leaseId); + expect(attempts).toBe(ATTEMPT_CAP); + expect(attempts).toBeLessThanOrEqual(ATTEMPT_CAP); + + // The cap warning logs at most once across the overlap. + const capWarnings = vi + .mocked(logger.warn) + .mock.calls.filter( + (call) => + call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup", + ); + expect(capWarnings.length).toBeLessThanOrEqual(1); + }); + + // Run `inject` exactly once, right before the sweep's first UPDATE on the + // environmentLeases table runs. The sweep's first update is always its claim + // (the retry claim or the cap-warning claim). This lets a test place a + // concurrent writer between the sweep's read and its claim. + function injectBeforeFirstLeaseClaim(inject: () => Promise): { restore: () => void } { + const realUpdate = db.update.bind(db); + let injected = false; + const spy = vi.spyOn(db, "update").mockImplementation(((table: unknown) => { + const builder = (realUpdate as (t: unknown) => unknown)(table) as { + set: (values: unknown) => unknown; + }; + if (table === environmentLeases && !injected) { + const realSet = builder.set.bind(builder); + builder.set = (values: unknown) => { + const afterSet = realSet(values) as { + returning: (...args: unknown[]) => unknown; + }; + if (!injected) { + injected = true; + const realReturning = afterSet.returning.bind(afterSet); + afterSet.returning = (...args: unknown[]) => + (async () => { + await inject(); + return realReturning(...args); + })(); + } + return afterSet; + }; + } + return builder; + }) as unknown as typeof db.update); + return { restore: () => spy.mockRestore() }; + } + + // Read the full metadata object of a lease. + async function readMetadata(leaseId: string): Promise | null> { + return db + .select({ metadata: environmentLeases.metadata }) + .from(environmentLeases) + .where(eq(environmentLeases.id, leaseId)) + .then((rows) => (rows[0]?.metadata as Record | null) ?? null); + } + + // A provider or plugin destroy rejection can carry a credential in its `name` + // or `code`, not only its message. The per-lease retry log must never + // serialize any error-derived string. + it("test_pending_cleanup_retry_log_omits_sentinel_in_error_name_and_code", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy rejects with a credential-shaped sentinel in both the error + // name and the error code. + const sentinel = "Bearer sk-SENTINEL-a1b2c3"; + const rejection = new Error("provider destroy failed"); + rejection.name = `ProviderDestroyError ${sentinel}`; + (rejection as { code?: string }).code = `EPROVIDER ${sentinel}`; + const destroyRunLease = vi.fn(async () => { + throw rejection; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result).toEqual({ swept: 1, destroyed: 0, capped: 0 }); + expect(destroyRunLease).toHaveBeenCalledTimes(1); + + const retryCall = vi + .mocked(logger.warn) + .mock.calls.find((call) => call[1] === "pending_cleanup lease retry failed"); + expect(retryCall).toBeDefined(); + const record = retryCall?.[0] as Record; + + // No error-derived string reaches the log. The sentinel never appears, and + // no error field is present. + expect(JSON.stringify(record)).not.toContain(sentinel); + expect(record).not.toHaveProperty("err"); + expect(record).not.toHaveProperty("errorName"); + expect(record).not.toHaveProperty("errorCode"); + expect(record).not.toHaveProperty("message"); + expect(record).not.toHaveProperty("stack"); + expect(record).not.toHaveProperty("cause"); + + // The log carries only fixed, locally generated fields. + expect(record).toMatchObject({ + leaseId, + environmentId, + attempts: 1, + errorKind: "destroy_failed", + }); + }); + + // The outer sweep catch logs a failure of the sweep itself. It must log a + // constant errorKind only, never an error-derived string in the name or code. + it("test_pending_cleanup_sweep_failure_log_omits_sentinel_in_error_name_and_code", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // Make the sweep itself throw. The lease query rejects with an error that + // carries a credential-shaped sentinel in both the name and the message. The + // reaper's own queries pass through, so only the sweep fails. + const sentinel = "Bearer sk-SENTINEL-d4e5f6"; + const realSelect = db.select.bind(db); + const selectSpy = vi + .spyOn(db, "select") + .mockImplementation((...args: Parameters) => { + const builder = realSelect(...args); + const realFrom = builder.from.bind(builder); + (builder as { from: unknown }).from = (table: unknown) => { + if (table === environmentLeases) { + const failure = new Error(`sweep query failed: ${sentinel}`); + failure.name = `SweepQueryError ${sentinel}`; + (failure as { code?: string }).code = `ESWEEP ${sentinel}`; + throw failure; + } + return realFrom(table as Parameters[0]); + }; + return builder; + }); + + try { + const destroyRunLease = vi.fn(async () => null); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The reaper isolates the sweep, so the reaper itself resolves. + await expect(heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 })).resolves.toBeDefined(); + + const sweepCall = vi + .mocked(logger.error) + .mock.calls.find((call) => call[1] === "pending_cleanup lease sweep failed"); + expect(sweepCall).toBeDefined(); + const record = sweepCall?.[0] as Record; + + expect(JSON.stringify(record)).not.toContain(sentinel); + expect(record).not.toHaveProperty("err"); + expect(record).not.toHaveProperty("errorName"); + expect(record).not.toHaveProperty("errorCode"); + expect(record).not.toHaveProperty("message"); + expect(record).not.toHaveProperty("stack"); + expect(record).toMatchObject({ errorKind: "sweep_failed" }); + } finally { + selectSpy.mockRestore(); + } + }); + + // The claim writes only its own attempts key with `jsonb_set`. A concurrent + // writer that sets an unrelated metadata key between the sweep's read and its + // claim must keep that key. A full copied-metadata write would lose it. + it("test_pending_cleanup_claim_preserves_unrelated_concurrent_metadata_key", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy fails, so the lease stays pending_cleanup and the test can + // read the final metadata. + const destroyRunLease = vi.fn(async () => null); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // A concurrent writer sets an unrelated key right before the claim runs. + const hook = injectBeforeFirstLeaseClaim(async () => { + await db + .update(environmentLeases) + .set({ + metadata: sql`jsonb_set(coalesce(${environmentLeases.metadata}, '{}'::jsonb), array['concurrentKey'], to_jsonb('keep-me'::text), true)`, + }) + .where(eq(environmentLeases.id, leaseId)); + }); + + try { + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result).toEqual({ swept: 1, destroyed: 0, capped: 0 }); + } finally { + hook.restore(); + } + + // The claim advanced its own attempts key and kept the concurrent key. + const metadata = await readMetadata(leaseId); + expect(metadata?.[ATTEMPTS_KEY]).toBe(1); + expect(metadata?.concurrentKey).toBe("keep-me"); + }); + + // A provider can write a malformed value under the attempts key. The guarded + // cast must read it as zero, not throw. One malformed lease must never abort + // the page sweep, so the sweep still processes the other leases on the page. + it("test_pending_cleanup_malformed_retry_metadata_does_not_abort_sweep", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // The malformed lease sorts first, so a throw would abort before the sweep + // reaches the healthy lease. + const malformedLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 120 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: "corrupt" }, + }); + const healthyLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy fails, so both leases stay pending_cleanup and the test can + // count the retries. + const destroyedLeaseIds: string[] = []; + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + destroyedLeaseIds.push(lease.id); + return null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The sweep resolves; the malformed cast does not throw. + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result.swept).toBe(2); + + // The sweep still processed the healthy lease. + expect(destroyedLeaseIds).toContain(healthyLeaseId); + + // The malformed attempts value normalized to a real number on the claim. + const malformedMetadata = await readMetadata(malformedLeaseId); + expect(malformedMetadata?.[ATTEMPTS_KEY]).toBe(1); + }); + + // The cap warning fires only for a pending_cleanup lease at or above the cap. + // A lease that leaves pending_cleanup between the read and the claim must get + // no warn-flag write, because the claim carries a status predicate. + it("test_pending_cleanup_cap_warning_requires_pending_cleanup_status_and_cap", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // A lease at the cap. It leaves pending_cleanup right before the warn claim. + const cappedLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP }, + }); + + const destroyRunLease = vi.fn(async () => null); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // A concurrent path moves the lease out of pending_cleanup right before the + // warn claim runs. + const hook = injectBeforeFirstLeaseClaim(async () => { + await db + .update(environmentLeases) + .set({ status: "expired" }) + .where(eq(environmentLeases.id, cappedLeaseId)); + }); + + try { + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + } finally { + hook.restore(); + } + + // The lease left pending_cleanup, so the warn claim wrote no flag. + const metadata = await readMetadata(cappedLeaseId); + expect(metadata?.[CAP_WARNED_KEY]).not.toBe(true); + + // The cap warning did not log. + const capWarnings = vi + .mocked(logger.warn) + .mock.calls.filter( + (call) => + call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup", + ); + expect(capWarnings.length).toBe(0); + + // A lease below the cap gets no warn flag either. The retry path never + // reaches the cap branch. + const belowCapLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 }, + }); + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + const belowCapMetadata = await readMetadata(belowCapLeaseId); + expect(belowCapMetadata?.[CAP_WARNED_KEY]).not.toBe(true); + }); + + // Override a lease metadata root with a raw JSON value. The insert helper + // always writes an object root, so a scalar or array root test needs a raw + // update. The value binds as text and casts to jsonb, so `"5"` writes a number + // scalar root and `"[1,2,3]"` writes an array root. + async function setRawMetadataRoot(leaseId: string, rawJson: string): Promise { + await db + .update(environmentLeases) + .set({ metadata: sql`${rawJson}::jsonb` }) + .where(eq(environmentLeases.id, leaseId)); + } + + // A provider can write a negative attempt count. The SQL reader and the + // TypeScript reader both clamp a negative value to zero, so the claim predicate + // matches. The lease still claims one attempt and the destroy still runs. + it("test_pending_cleanup_negative_attempts_metadata_normalizes_and_lease_still_destroys", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // The negative-attempts lease sorts first. A thrown or mismatched claim would + // strand it before the sweep reaches the healthy lease. + const negativeLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 120 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: -3 }, + }); + const healthyLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy succeeds and moves each lease out of pending_cleanup. + const destroyedLeaseIds: string[] = []; + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + destroyedLeaseIds.push(lease.id); + const now = new Date(); + const row = await db + .update(environmentLeases) + .set({ status: "expired", cleanupStatus: "success", updatedAt: now }) + .where(eq(environmentLeases.id, lease.id)) + .returning() + .then((rows) => rows[0] ?? null); + return row ? { ...row, status: "expired" as const } : null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The sweep resolves; the negative value does not strand the lease. + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result.swept).toBe(2); + + // The negative-attempts lease claimed one attempt and destroyed. + expect(destroyedLeaseIds).toContain(negativeLeaseId); + expect(destroyedLeaseIds).toContain(healthyLeaseId); + expect(result.destroyed).toBe(2); + + // The negative value normalized to zero, so the claim advanced it to one. + const negativeMetadata = await readMetadata(negativeLeaseId); + expect(negativeMetadata?.[ATTEMPTS_KEY]).toBe(1); + + const negativeStatus = await db + .select({ status: environmentLeases.status }) + .from(environmentLeases) + .where(eq(environmentLeases.id, negativeLeaseId)) + .then((rows) => rows[0]?.status); + expect(negativeStatus).toBe("expired"); + }); + + // A provider can write a finite number outside the 32-bit integer range. The + // SQL reader computes as numeric and clamps to the cap, so the `::int` cast + // never runs on the stored value. The sweep does not throw, and the malformed + // lease normalizes into the cap behavior. + it("test_pending_cleanup_out_of_range_attempts_metadata_does_not_abort_sweep", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // The out-of-range lease sorts first. A thrown cast would abort before the + // sweep reaches the healthy lease. + const outOfRangeLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 120 * 60 * 1000), + metadata: { [ATTEMPTS_KEY]: 1e300 }, + }); + const healthyLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy fails, so the healthy lease stays pending_cleanup and the test + // can confirm the sweep reached it. + const destroyedLeaseIds: string[] = []; + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + destroyedLeaseIds.push(lease.id); + return null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The sweep resolves; the out-of-range value does not abort the page. + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result.swept).toBe(2); + + // The value clamped to the cap, so the lease took the cap branch, not a + // destroy. The warn-once flag is set exactly once. + expect(result.capped).toBe(1); + expect(destroyedLeaseIds).not.toContain(outOfRangeLeaseId); + const outOfRangeMetadata = await readMetadata(outOfRangeLeaseId); + expect(outOfRangeMetadata?.[CAP_WARNED_KEY]).toBe(true); + + // The sweep still processed the healthy lease. + expect(destroyedLeaseIds).toContain(healthyLeaseId); + }); + + // A provider can write a scalar metadata root. `jsonb_set` fails on a + // non-object root, so the reader falls back to an empty object. The sweep does + // not throw, and the lease still claims one attempt and destroys. + it("test_pending_cleanup_scalar_metadata_root_does_not_abort_sweep", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // The scalar-root lease sorts first. A failed `jsonb_set` would abort before + // the sweep reaches the healthy lease. + const scalarLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 120 * 60 * 1000), + }); + await setRawMetadataRoot(scalarLeaseId, "5"); + const healthyLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy succeeds and moves each lease out of pending_cleanup. + const destroyedLeaseIds: string[] = []; + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + destroyedLeaseIds.push(lease.id); + const now = new Date(); + const row = await db + .update(environmentLeases) + .set({ status: "expired", cleanupStatus: "success", updatedAt: now }) + .where(eq(environmentLeases.id, lease.id)) + .returning() + .then((rows) => rows[0] ?? null); + return row ? { ...row, status: "expired" as const } : null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The sweep resolves; the scalar root does not break the claim write. + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result.swept).toBe(2); + expect(result.destroyed).toBe(2); + expect(destroyedLeaseIds).toContain(scalarLeaseId); + expect(destroyedLeaseIds).toContain(healthyLeaseId); + + // The claim replaced the scalar root with an object that holds one attempt. + const scalarMetadata = await readMetadata(scalarLeaseId); + expect(scalarMetadata?.[ATTEMPTS_KEY]).toBe(1); + }); + + // A provider can write an array metadata root. `jsonb_set` fails on a + // non-object root, so the reader falls back to an empty object. The sweep does + // not throw, and the lease still claims one attempt and destroys. + it("test_pending_cleanup_array_metadata_root_does_not_abort_sweep", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + + // The array-root lease sorts first. A failed `jsonb_set` would abort before + // the sweep reaches the healthy lease. + const arrayLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 120 * 60 * 1000), + }); + await setRawMetadataRoot(arrayLeaseId, "[1, 2, 3]"); + const healthyLeaseId = await insertPendingCleanupLease({ + companyId, + environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000), + }); + + // The destroy succeeds and moves each lease out of pending_cleanup. + const destroyedLeaseIds: string[] = []; + const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => { + destroyedLeaseIds.push(lease.id); + const now = new Date(); + const row = await db + .update(environmentLeases) + .set({ status: "expired", cleanupStatus: "success", updatedAt: now }) + .where(eq(environmentLeases.id, lease.id)) + .returning() + .then((rows) => rows[0] ?? null); + return row ? { ...row, status: "expired" as const } : null; + }); + const heartbeat = heartbeatService(db, { + environmentRuntime: fakeRuntime( + destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"], + ), + }); + + // The sweep resolves; the array root does not break the claim write. + const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(result.swept).toBe(2); + expect(result.destroyed).toBe(2); + expect(destroyedLeaseIds).toContain(arrayLeaseId); + expect(destroyedLeaseIds).toContain(healthyLeaseId); + + // The claim replaced the array root with an object that holds one attempt. + const arrayMetadata = await readMetadata(arrayLeaseId); + expect(arrayMetadata?.[ATTEMPTS_KEY]).toBe(1); + }); +}); diff --git a/server/src/__tests__/reusable-leases-default.test.ts b/server/src/__tests__/reusable-leases-default.test.ts new file mode 100644 index 0000000000..a77ee06cad --- /dev/null +++ b/server/src/__tests__/reusable-leases-default.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { getEnvironmentCapabilities } from "@paperclipai/shared"; +import { getSandboxProvider } from "../services/sandbox-provider-runtime.ts"; + +// The execution guard in environment-runtime.ts admits a reusable lease only +// when the provider declares supportsReusableLeases === true. This helper +// mirrors that guard, so the tests can compare the presentation default with +// the execution decision. +function executionSupportsReusableLeases(declared: boolean | undefined): boolean { + return declared === true; +} + +describe("reusable leases default", () => { + it("test_fake_provider_reusable_leases_matches_runtime_declaration", () => { + const fakeRuntime = getSandboxProvider("fake"); + expect(fakeRuntime).not.toBeNull(); + expect(fakeRuntime?.supportsReusableLeases).toBe(false); + + const capabilities = getEnvironmentCapabilities(["codex_local"]); + expect(capabilities.sandboxProviders.fake.supportsReusableLeases).toBe( + fakeRuntime?.supportsReusableLeases ?? false, + ); + }); + + it("test_api_presentation_agrees_with_runtime_execution_for_reusable_leases", () => { + // Built-in fake provider: the runtime declares false, so execution and + // presentation both resolve to false. + const fakeRuntime = getSandboxProvider("fake"); + const builtin = getEnvironmentCapabilities(["codex_local"]); + expect(builtin.sandboxProviders.fake.supportsReusableLeases).toBe( + executionSupportsReusableLeases(fakeRuntime?.supportsReusableLeases), + ); + + // Plug-in provider whose manifest does not declare reusable leases. + const absent = getEnvironmentCapabilities(["codex_local"], { + sandboxProviders: { "no-reuse-plugin": { displayName: "No Reuse" } }, + }); + expect(absent.sandboxProviders["no-reuse-plugin"].supportsReusableLeases).toBe( + executionSupportsReusableLeases(undefined), + ); + + // Plug-in provider whose manifest declares reusable leases. + const declared = getEnvironmentCapabilities(["codex_local"], { + sandboxProviders: { + "reuse-plugin": { displayName: "Reuse", supportsReusableLeases: true }, + }, + }); + expect(declared.sandboxProviders["reuse-plugin"].supportsReusableLeases).toBe( + executionSupportsReusableLeases(true), + ); + }); +}); diff --git a/server/src/routes/environments.ts b/server/src/routes/environments.ts index fbe07dc9f8..d7f157c1c3 100644 --- a/server/src/routes/environments.ts +++ b/server/src/routes/environments.ts @@ -715,7 +715,9 @@ export function environmentRoutes( supportsSavedProbe: true, supportsUnsavedProbe: true, supportsRunExecution: true, - supportsReusableLeases: driver.supportsReusableLeases ?? true, + // Default absent to false, so the presentation agrees with the + // execution guard (=== true). + supportsReusableLeases: driver.supportsReusableLeases ?? false, supportsInteractiveSetup: driver.supportsInteractiveSetup, interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes, supportsTemplateCapture: driver.supportsTemplateCapture, diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index 8d60e77d8e..975358ec82 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -561,7 +561,11 @@ export function environmentRunOrchestrator( let releasedLeases: EnvironmentRuntimeLeaseRecord[]; try { - releasedLeases = await environmentRuntime.releaseRunLeases(input.heartbeatRunId, status); + releasedLeases = await environmentRuntime.releaseRunLeases( + input.heartbeatRunId, + status, + (leaseId, error) => result.errors.push({ leaseId, error }), + ); } catch (err) { result.errors.push({ leaseId: "*", error: err }); return result; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 7bcd679739..3035819281 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1968,6 +1968,7 @@ export function environmentRuntimeService( async releaseRunLeases( heartbeatRunId: string, status: Extract = "released", + onLeaseReleaseError?: (leaseId: string, error: unknown) => void, ): Promise { const leaseRows = await db .select() @@ -1982,31 +1983,39 @@ export function environmentRuntimeService( return []; } + // Release each lease in its own try/catch. One driver error must not stop + // the release of the later leases. The caller records each lease-specific + // error through `onLeaseReleaseError` for its log path. Keep the order + // serial. const released: EnvironmentRuntimeLeaseRecord[] = []; for (const leaseRow of leaseRows) { - const environment = await environmentsSvc.getById(leaseRow.environmentId); - if (!environment) continue; + try { + const environment = await environmentsSvc.getById(leaseRow.environmentId); + if (!environment) continue; - const leaseSnapshot = toEnvironmentLeaseSnapshot(leaseRow); - const driver = getDriver(getLeaseDriverKey(leaseSnapshot, environment)); - const lease = driver - ? await driver.releaseRunLease({ - environment, - lease: leaseSnapshot, - status, - }) - : await environmentsSvc.releaseLease(leaseRow.id, status); - if (!lease) continue; + const leaseSnapshot = toEnvironmentLeaseSnapshot(leaseRow); + const driver = getDriver(getLeaseDriverKey(leaseSnapshot, environment)); + const lease = driver + ? await driver.releaseRunLease({ + environment, + lease: leaseSnapshot, + status, + }) + : await environmentsSvc.releaseLease(leaseRow.id, status); + if (!lease) continue; - released.push({ - environment, - lease, - leaseContext: { - executionWorkspaceId: lease.executionWorkspaceId, - executionWorkspaceMode: - (lease.metadata?.executionWorkspaceMode as ExecutionWorkspace["mode"] | null | undefined) ?? null, - }, - }); + released.push({ + environment, + lease, + leaseContext: { + executionWorkspaceId: lease.executionWorkspaceId, + executionWorkspaceMode: + (lease.metadata?.executionWorkspaceMode as ExecutionWorkspace["mode"] | null | undefined) ?? null, + }, + }); + } catch (error) { + onLeaseReleaseError?.(leaseRow.id, error); + } } return released; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a63d435601..a096530402 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -44,6 +44,7 @@ import { documentAnnotationComments, documentAnnotationThreads, documentRevisions, + environmentLeases, issueDocuments, executionWorkspaces, heartbeatRunEvents, @@ -356,6 +357,67 @@ const PAPERCLIP_WAKE_PAYLOAD_KEY = "paperclipWake"; const PAPERCLIP_AGENT_MESSAGE_KEY = "paperclipAgentMessage"; const PAPERCLIP_HARNESS_CHECKOUT_KEY = "paperclipHarnessCheckedOut"; const DETACHED_PROCESS_ERROR_CODE = "process_detached"; +// The reaper sweeps at most this many pending_cleanup leases per tick. +const PENDING_CLEANUP_SWEEP_PAGE_SIZE = 20; +// The reaper stops retrying a pending_cleanup lease after this many attempts. +const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5; +// The reaper stores its retry state under these keys in the lease metadata. +const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts"; +const PENDING_CLEANUP_CAP_WARNED_METADATA_KEY = "pendingCleanupRetryCapWarned"; + +// A provider or plugin destroy rejection can carry a bearer credential, a +// signed URL, or provider response detail in its name, code, message, cause, or +// stack. The exception fields cross the server boundary, so they are not a +// trusted enum. The pending_cleanup sweep logs never read the exception. Each +// catch site logs a constant, locally generated `errorKind` instead. +const PENDING_CLEANUP_RETRY_ERROR_KIND = "destroy_failed"; +const PENDING_CLEANUP_SWEEP_ERROR_KIND = "sweep_failed"; + +// Read the stored retry attempt count as a safe value, directly in SQL. A +// provider can write a malformed value under the attempts key. The type guard +// makes any non-number value read as zero. The reader computes as numeric and +// never casts to int, so a finite number outside the 32-bit range (for example +// 1e300) never throws. The reader clamps a negative value to zero and a positive +// value to the attempt cap. One malformed lease therefore never aborts the page +// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`, +// which clamps to the same range. The claim predicate compares the two readers, +// so both must yield the same value for every input. +function pendingCleanupAttemptsSql() { + return sql` + case + when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number' + then least( + greatest( + floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric), + 0 + ), + ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP} + ) + else 0 + end`; +} + +// Choose the `jsonb_set` target root. A provider can write a scalar or array +// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the +// stored metadata only when its root is an object. A NULL, scalar, or array root +// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also +// covers a NULL root. +function pendingCleanupMetadataObjectSql() { + return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`; +} + +// Read the stored cap-warned flag as a safe boolean, directly in SQL. A +// malformed value reads as false, so the boolean cast never throws. +function pendingCleanupCapWarnedSql() { + return sql`coalesce( + case + when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY}) = 'boolean' + then (${environmentLeases.metadata} ->> ${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY})::boolean + else false + end, + false + )`; +} const REPO_ONLY_CWD_SENTINEL = "/__paperclip_repo_only__"; const MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS = 10 * 60 * 1000; const MAX_INLINE_WAKE_COMMENTS = 8; @@ -13123,6 +13185,170 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); } + // Clamp the stored attempt count to the range [0, cap]. The SQL reader + // `pendingCleanupAttemptsSql` clamps to the same range, so both readers yield + // the same value for every input. The claim predicate compares the two values, + // so this alignment lets the claim match for a malformed lease. + function readPendingCleanupRetryAttempts(metadata: Record): number { + const value = metadata[PENDING_CLEANUP_ATTEMPTS_METADATA_KEY]; + if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) return 0; + return Math.min(Math.floor(value), PENDING_CLEANUP_SWEEP_ATTEMPT_CAP); + } + + // Atomically claim one retry attempt on a pending_cleanup lease. The update + // only matches when the lease is still pending_cleanup and its stored attempt + // count still equals `expectedAttempts`. Two concurrent sweeps read the same + // count, but Postgres serializes the two updates on the row and only the first + // matches the guard. The loser gets zero rows and skips the lease. This bounds + // the retries to the cap and stops a second destroy of the same lease. + // Returns true only for the sweep that won the claim. + // + // The update writes only the attempts key with `jsonb_set`. It never writes a + // copied metadata object, so a concurrent write to an unrelated metadata key + // survives. The guard reads the stored count through the safe SQL reader, so a + // malformed value never throws. + async function claimPendingCleanupRetryAttempt( + leaseId: string, + expectedAttempts: number, + ): Promise { + const now = new Date(); + const claimed = await db + .update(environmentLeases) + .set({ + metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], to_jsonb(${expectedAttempts + 1}::int), true)`, + lastUsedAt: now, + updatedAt: now, + }) + .where( + and( + eq(environmentLeases.id, leaseId), + eq(environmentLeases.status, "pending_cleanup"), + sql`${pendingCleanupAttemptsSql()} = ${expectedAttempts}`, + ), + ) + .returning({ id: environmentLeases.id }); + return claimed.length > 0; + } + + // Atomically claim the one-time cap warning for a lease. The update only + // matches when the lease is still pending_cleanup, its stored attempt count is + // at or above the cap, and it has not yet carried the warned flag. Two + // concurrent sweeps that both reach the cap race here, but only one update + // sets the flag and returns a row. The loser skips the warning. This keeps the + // warning to one log line per lease. + // + // The status and cap predicates are the last line of defense. They stop a warn + // flag write to a lease that left pending_cleanup or dropped below the cap + // between the read and this claim. The update writes only the warned key with + // `jsonb_set`, so a concurrent write to an unrelated metadata key survives. + async function claimPendingCleanupCapWarning(leaseId: string): Promise { + const now = new Date(); + const claimed = await db + .update(environmentLeases) + .set({ + metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY}], to_jsonb(true), true)`, + updatedAt: now, + }) + .where( + and( + eq(environmentLeases.id, leaseId), + eq(environmentLeases.status, "pending_cleanup"), + sql`${pendingCleanupAttemptsSql()} >= ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}`, + sql`${pendingCleanupCapWarnedSql()} = false`, + ), + ) + .returning({ id: environmentLeases.id }); + return claimed.length > 0; + } + + // Retry the leases stranded in "pending_cleanup". A failed destroy leaves a + // lease in that state forever without this sweep. The reaper tick runs the + // sweep. The backoff equals the reaper staleness threshold, so a lease waits + // for that period between attempts. The sweep reads and writes the attempt + // count in the lease metadata. It warns once when a lease reaches the attempt + // cap and then stops the retries for that lease. + async function sweepPendingCleanupLeases(opts?: { backoffMs?: number }): Promise<{ + swept: number; + destroyed: number; + capped: number; + }> { + const backoffMs = opts?.backoffMs ?? 0; + const now = new Date(); + const cutoff = new Date(now.getTime() - backoffMs); + + const rows = await db + .select() + .from(environmentLeases) + .where( + and( + eq(environmentLeases.status, "pending_cleanup"), + backoffMs > 0 ? lte(environmentLeases.updatedAt, cutoff) : undefined, + ), + ) + .orderBy(asc(environmentLeases.updatedAt)) + .limit(PENDING_CLEANUP_SWEEP_PAGE_SIZE); + + let destroyed = 0; + let capped = 0; + for (const row of rows) { + const metadata = { ...(row.metadata ?? {}) } as Record; + const attempts = readPendingCleanupRetryAttempts(metadata); + + if (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) { + capped += 1; + // Warn once, then leave the lease for manual cleanup. The atomic claim + // keeps the warning to one log line even when two sweeps overlap. + if (metadata[PENDING_CLEANUP_CAP_WARNED_METADATA_KEY] !== true) { + const warned = await claimPendingCleanupCapWarning(row.id); + if (warned) { + logger.warn( + { leaseId: row.id, environmentId: row.environmentId, attempts }, + "environment lease reached the pending_cleanup retry cap; left for manual cleanup", + ); + } + } + continue; + } + + const environment = await environmentsSvc.getById(row.environmentId); + const lease = await environmentsSvc.getLeaseById(row.id); + if (!environment || !lease) continue; + + // Atomically claim the attempt before the retry. Only the winning sweep + // increments the count and destroys the lease, so an overlapping sweep + // never destroys the same lease twice or exceeds the attempt cap. The + // claim records the attempt before the retry, so a thrown driver error + // still counts against the cap. + const claimed = await claimPendingCleanupRetryAttempt(row.id, attempts); + if (!claimed) continue; + + try { + const result = await environmentRuntime.destroyRunLease({ + environment, + lease, + failureReason: "pending_cleanup_retry", + }); + if (result && result.status !== "pending_cleanup") { + destroyed += 1; + } + } catch { + // Log a constant errorKind only. The exception can carry a credential in + // its name, code, message, cause, or stack, so the sweep never reads it. + logger.warn( + { + errorKind: PENDING_CLEANUP_RETRY_ERROR_KIND, + leaseId: row.id, + environmentId: row.environmentId, + attempts: attempts + 1, + }, + "pending_cleanup lease retry failed", + ); + } + } + + return { swept: rows.length, destroyed, capped }; + } + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) { const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -13316,6 +13542,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (reaped.length > 0) { logger.warn({ reapedCount: reaped.length, runIds: reaped }, "reaped orphaned heartbeat runs"); } + + // Retry stranded pending_cleanup leases on the same tick. Isolate the sweep + // so its failure never hides the reaper result. The backoff equals the + // reaper staleness threshold. + try { + const sweep = await sweepPendingCleanupLeases({ backoffMs: staleThresholdMs }); + if (sweep.destroyed > 0 || sweep.capped > 0) { + logger.warn( + { destroyed: sweep.destroyed, capped: sweep.capped, swept: sweep.swept }, + "swept pending_cleanup environment leases", + ); + } + } catch { + // Log a constant errorKind only. The exception can carry a credential in + // its name, code, message, cause, or stack, so the sweep never reads it. + logger.error( + { errorKind: PENDING_CLEANUP_SWEEP_ERROR_KIND }, + "pending_cleanup lease sweep failed", + ); + } + return { reaped: reaped.length, runIds: reaped }; } @@ -18992,6 +19239,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) prepareHotRestartShutdown, reconcileHotRestartAdoption, reapOrphanedRuns, + sweepPendingCleanupLeases, // Override-aware scheduling-suppression check (honors the worktree // run-execution experimental setting). Callers outside the service that // gate on suppression should prefer this over the env-only resolver.