From e52b8a343f0ea4d7bae56c696c6c60c6f329d82a Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 15 Aug 2026 18:52:08 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20ACP=20run=20lifecycle=20corrections=20?= =?UTF-8?q?=E2=80=94=20failure=20settlement,=20workspace=20sync-back,=20le?= =?UTF-8?q?ase=20cleanup=20(#11454)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent adapters run ACP sessions and manage runtime, workspace, and lease resources. > - Several failure paths left runtime bridges, staged workspaces, or environment leases active after an error. > - These leaks reduce run reliability and can leave later runs without clean resources. > - This pull request closes the failure paths, applies one teardown policy, and adds regression tests. > - The benefit is consistent failure settlement and safer reuse of agent workspaces and leases. ## Linked Issues or Issue Description **What happened?** ACP runs could leave runtime bridges, staged workspaces, or environment leases active after failures. Claude and Gemini ACP runs did not restore the sandbox workspace on teardown. Lease release stopped when one lease returned an error. **Expected behavior** Each ACP failure must return an error result and settle its resources. Teardown must run each step, release leases independently, and restore the host workspace when the sandbox ends. Pending cleanup leases must receive bounded retry attempts. **Steps to reproduce** 1. Run an ACP session that fails after runtime creation or during turn preparation. 2. Run an ACP session that fails during a warm hit or staged runtime handoff. 3. Run lease cleanup with more than one lease when the first release returns an error. 4. Inspect the result phase, teardown calls, workspace state, and lease metadata. 5. Run the regression suites listed in the Verification section. ## What Changed - Settle every ACP failure after runtime creation with an error result and one sandbox.startup span closure. - Close the ACP runtime and remove warm entries after every pre-turn failure. - Run all teardown steps, record teardown errors, release staging leases in finally, and prevent duplicate teardown. - Dispose staged runtimes after seam failures and remove borrowed staged entries with identity guards. - Add fail-open workspace sync-back teardown for Claude and Gemini ACP adapters. - Isolate lease release errors and add bounded retry sweeps for stranded pending_cleanup leases. - Atomically claim pending_cleanup retries and clamp attempt readers to keep the five-attempt bound. - Default absent provider reusableLeases values to false and align the fake provider with its runtime declaration. - Add regression tests for engine, adapter, server, and shared environment behavior. ## Verification - [x] `npx vitest run packages/adapter-utils/src/acpx-engine/execute.test.ts` — 124 tests passed. - [x] `npx vitest run packages/adapters/codex-local/src/server/acp.test.ts packages/adapters/claude-local/src/server/acp.test.ts packages/adapters/gemini-local/src/server/acp.test.ts` — 61 tests passed. - [x] `npx vitest run server/src/__tests__/environment-runtime.test.ts server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts server/src/__tests__/reusable-leases-default.test.ts server/src/__tests__/environment-routes.test.ts packages/shared/src/environment-support.test.ts` — passed. - [x] All listed suites ran from the repository root. - [x] GitHub CI completed successfully for `cfc349c9f232711433897915112a1c52c0e462ca`. - [x] Greptile completed with a 5/5 confidence score and no blocking finding. ## Risks The engine changes affect failure settlement and teardown order across ACP runs. The server changes add retry state to existing lease metadata without a schema migration. The adapter changes restore workspaces after sandbox execution. Regression tests cover the changed paths. GitHub CI and Greptile passed for the current head. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. This change fixes runtime reliability and does not duplicate a roadmap feature. ## Model Used OpenAI GPT-5 Codex. The model used tool-based repository inspection, GitHub operations, and code review support. The runtime does not expose a context-window value. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (for example, `docs/...` or `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../src/acpx-engine/execute.test.ts | 859 +++++++++++++++ .../adapter-utils/src/acpx-engine/execute.ts | 990 +++++++++++------- .../claude-local/src/server/acp.test.ts | 73 ++ .../adapters/claude-local/src/server/acp.ts | 26 +- .../gemini-local/src/server/acp.test.ts | 62 ++ .../adapters/gemini-local/src/server/acp.ts | 29 +- .../shared/src/environment-support.test.ts | 23 + packages/shared/src/environment-support.ts | 8 +- .../src/__tests__/environment-routes.test.ts | 4 +- .../src/__tests__/environment-runtime.test.ts | 64 ++ .../heartbeat-pending-cleanup-sweep.test.ts | 941 +++++++++++++++++ .../__tests__/reusable-leases-default.test.ts | 52 + server/src/routes/environments.ts | 4 +- .../services/environment-run-orchestrator.ts | 6 +- server/src/services/environment-runtime.ts | 51 +- server/src/services/heartbeat.ts | 248 +++++ 16 files changed, 3046 insertions(+), 394 deletions(-) create mode 100644 server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts create mode 100644 server/src/__tests__/reusable-leases-default.test.ts 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.