diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 76cd8f3304..c0c0a8eb0f 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -1981,3 +1981,619 @@ describe("ACPX engine remote managed-home seam (PR 2: per-adapter home seed)", ( expect(sessionInputs[0]?.cwd).toBe(remoteCwd); }); }); + +describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / reuse on compatible resume)", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + 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 }; + } + + // A runtime double that records ensureSession inputs and can be told to make + // the turn fail (to exercise the teardown/eviction path). + function recordingRuntime(input: { + ensureInputs: Array>; + terminalStatus?: "completed" | "failed"; + }) { + return { + ensureSession: async (session: Record) => { + input.ensureInputs.push(session); + return { + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }; + }, + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: + input.terminalStatus === "failed" + ? Promise.resolve({ status: "failed", error: new Error("boom") }) + : Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async () => {}, + close: async () => {}, + }; + } + + function baseExecuteArgs(input: { + stateDir: string; + localCwd: string; + executionTarget: Record; + env?: Record; + }) { + return { + agent: { id: "agent-1", companyId: "company-1" }, + config: { + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: input.stateDir, + cwd: input.localCwd, + mode: "persistent", + warmHandleIdleMs: 60_000, + ...(input.env ? { env: input.env } : {}), + }, + context: {}, + authToken: "real-run-jwt", + executionTarget: input.executionTarget, + onLog: async () => {}, + onMeta: async () => {}, + }; + } + + it("test_acp_resume_compatible_session_does_not_restage", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + const second = await execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // Staging (workspace ship + home seed) ran exactly ONCE across both runs: + // the compatible resume reused the already-staged in-sandbox runtime. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(1); + // Both runs bind session/new (and resume) to the in-sandbox workspace cwd... + expect(ensureInputs[0]?.cwd).toBe(remoteCwd); + expect(ensureInputs[1]?.cwd).toBe(remoteCwd); + // ...and the second run RESUMES the first session rather than starting fresh. + expect(ensureInputs[1]?.resumeSessionId).toBe(first.sessionId); + }); + + it("test_acp_resume_incompatible_fingerprint_stages_fresh", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + }); + + const first = await execute({ + runId: "run-a", + runtime: {}, + ...baseExecuteArgs({ stateDir, localCwd, executionTarget, env: { FOO: "a" } }), + } as never); + // A changed adapter env value shifts the session fingerprint → a different + // sessionKey → the cache slot does not match, so staging runs fresh. + const second = await execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...baseExecuteArgs({ stateDir, localCwd, executionTarget, env: { FOO: "b" } }), + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // Incompatible fingerprint → staged fresh, no stale reuse. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(2); + expect(ensureInputs[0]?.cwd).toBe(remoteCwd); + expect(ensureInputs[1]?.cwd).toBe(remoteCwd); + // The second run does NOT resume the first session (fingerprint differs). + expect(ensureInputs[1]?.resumeSessionId).toBeUndefined(); + }); + + it("test_warm_handle_scoped_per_fingerprint_no_cross_session_credential_reuse", async () => { + const { root, stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + // Two managed homes, one per session, each carrying a distinct credential + // marker. The seam seeds whichever home belongs to the current run. + const homeA = path.join(root, "home-a"); + const homeB = path.join(root, "home-b"); + await fs.mkdir(homeA, { recursive: true }); + await fs.mkdir(homeB, { recursive: true }); + await fs.writeFile(path.join(homeA, "auth.json"), JSON.stringify({ token: "SECRET-A" }), "utf8"); + await fs.writeFile(path.join(homeB, "auth.json"), JSON.stringify({ token: "SECRET-B" }), "utf8"); + + const seededHomeEnv: string[] = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => { + const localHome = input.env.SESSION_MARKER === "b" ? homeB : homeA; + const stagedRuntime = await input.stage([ + { key: "home", localDir: localHome, followSymlinks: true }, + ]); + input.env.MANAGED_HOME = stagedRuntime.assetDirs.home ?? ""; + seededHomeEnv.push(input.env.MANAGED_HOME); + return { stagedRuntime }; + }, + }); + + const first = await execute({ + runId: "run-a", + runtime: {}, + ...baseExecuteArgs({ stateDir, localCwd, executionTarget, env: { SESSION_MARKER: "a" } }), + } as never); + // Different fingerprint (SESSION_MARKER changed) → different sessionKey. If the + // cache were NOT fingerprint-scoped, this run could silently inherit session A's + // staged auth.json without re-seeding. It must instead seed its own home. + const second = await execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...baseExecuteArgs({ stateDir, localCwd, executionTarget, env: { SESSION_MARKER: "b" } }), + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // Each session staged its OWN managed home — no cross-session reuse. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(2); + expect(seededHomeEnv).toHaveLength(2); + // Session B's staged home holds session B's credential, never session A's. + const bHome = seededHomeEnv[1]!; + await expect(fs.readFile(path.join(bHome, "auth.json"), "utf8")).resolves.toContain("SECRET-B"); + }); + + it("test_acp_failed_turn_evicts_staged_runtime_so_resume_restages", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + // The first turn fails; the second (compatible) run then completes. + createRuntime: (() => { + let call = 0; + return () => { + call += 1; + return recordingRuntime({ + ensureInputs, + terminalStatus: call === 1 ? "failed" : "completed", + }) as never; + }; + })(), + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + const second = await execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + + expect(first.exitCode).toBe(1); + expect(second.exitCode).toBe(0); + // A failed turn discards the staged runtime, so the next run stages fresh + // instead of reusing a torn-down session's staged credentials. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(2); + }); + + // Greptile P1 "Cache Reuse Bypasses Session Compatibility": a fresh invocation + // that shares company/agent/task/fingerprint (hence sessionKey) with a prior + // run but carries NO sessionParams starts a new ACP session — it must NOT + // inherit the prior session's staged workspace + managed home. + it("test_acp_reuse_requires_compatible_resume_not_just_session_key", async () => { + const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + let seamCalls = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => { + seamCalls += 1; + return { stagedRuntime: await input.stage([]) }; + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + // Same config (identical sessionKey) but sessionParams cleared → this is a + // NEW session, not a resume of A. The old code reused A's staged runtime on a + // bare sessionKey hit; the compatibility gate now forces a fresh stage. + const second = await execute({ runId: "run-b", runtime: {}, ...base } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // Staged (and re-seeded the managed home) fresh for the new session — no + // silent inheritance of the prior session's staged credentials. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(2); + expect(seamCalls).toBe(2); + // B binds a fresh session/new (no resumeSessionId), it does not resume A. + expect(ensureInputs[1]?.cwd).toBe(remoteCwd); + expect(ensureInputs[1]?.resumeSessionId).toBeUndefined(); + }); + + // Greptile P1 "Teardown Invalidates Cached Runtime": the per-run copy-back must + // fire on every run (incl. a reused resume) while the one-time host staged-temp + // cleanup must NOT fire between clean runs — otherwise the reused staged runtime + // would be invalidated before the next resume. + it("test_reused_resume_copies_back_per_run_without_disposing_staged_temp", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + let teardownCalls = 0; + let disposeCalls = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => { + const stagedRuntime = await input.stage([]); + return { + stagedRuntime, + teardown: async () => { + teardownCalls += 1; + }, + disposeStaged: async () => { + disposeCalls += 1; + }, + }; + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + const second = await execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + + expect(first.exitCode).toBe(0); + expect(second.exitCode).toBe(0); + // Staged once, reused on the compatible resume. + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(1); + // Per-run copy-back fired on BOTH runs — cadence unchanged. + expect(teardownCalls).toBe(2); + // The staged temp was never disposed while the entry stayed warm for reuse, + // so the resume found its staged home intact. + expect(disposeCalls).toBe(0); + expect(ensureInputs[1]?.resumeSessionId).toBe(first.sessionId); + }); + + // The one-time dispose DOES fire when the staged runtime is actually dropped + // (here: a failed turn), releasing the host staged-temp — the copy-back also + // still fires on the failure path. + it("test_dropped_staged_runtime_disposes_host_temp", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + let teardownCalls = 0; + let disposeCalls = 0; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs, terminalStatus: "failed" }) as never, + prepareRemoteManagedHome: async (input) => ({ + stagedRuntime: await input.stage([]), + teardown: async () => { + teardownCalls += 1; + }, + disposeStaged: async () => { + disposeCalls += 1; + }, + }), + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const result = await execute({ runId: "run-a", runtime: {}, ...base } as never); + + expect(result.exitCode).toBe(1); + // Failed turn → staged runtime dropped → host staged-temp disposed once, and + // the per-run copy-back still fired. + expect(teardownCalls).toBe(1); + expect(disposeCalls).toBe(1); + }); + + it("test_idle_staged_runtime_cleanup_waits_for_active_turn_release", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const events: string[] = []; + let currentNow = 0; + let releaseTurn!: () => void; + let signalTurnStarted!: () => void; + const turnStarted = new Promise((resolve) => { + signalTurnStarted = resolve; + }); + const turnCompleted = new Promise((resolve) => { + releaseTurn = resolve; + }); + const execute = createAcpxEngineExecutor({ + now: () => currentNow, + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: (() => { + let call = 0; + return () => { + call += 1; + return { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => { + if (call === 2) signalTurnStarted(); + return { + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: + call === 2 + ? turnCompleted.then(() => ({ status: "completed", stopReason: "end_turn" })) + : Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; + }, + setConfigOption: async () => {}, + close: async () => {}, + } as never; + }; + })(), + prepareRemoteManagedHome: async (input) => { + events.push(`stage:${input.runId}`); + return { + stagedRuntime: await input.stage([]), + disposeStaged: async () => { + events.push(`dispose:${input.runId}`); + }, + }; + }, + }); + const base = baseExecuteArgs({ + stateDir, + localCwd, + executionTarget, + env: { SESSION_MARKER: "idle-eviction" }, + }); + + const first = await execute({ runId: "run-a", runtime: {}, ...base } as never); + expect(first.exitCode).toBe(0); + + const second = execute({ + runId: "run-b", + runtime: { sessionParams: first.sessionParams }, + ...base, + } as never); + await turnStarted; + currentNow = 10_000; + const third = execute({ runId: "run-c", runtime: {}, ...base } as never); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(events).toEqual(["stage:run-a"]); + + releaseTurn(); + const [resultB, resultC] = await Promise.all([second, third]); + + expect(resultB.exitCode).toBe(0); + expect(resultC.exitCode).toBe(0); + expect(events).toEqual(["stage:run-a", "dispose:run-a", "stage:run-c"]); + }); + + // Superseding an incompatible session that collides on sessionKey re-stages + // fresh AND releases the superseded entry's host staged-temp (no leak, no + // reuse of the old session's staged credentials). + it("test_incompatible_restage_disposes_superseded_staged_temp", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const disposedRunIds: string[] = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => ({ + stagedRuntime: await input.stage([]), + disposeStaged: async () => { + disposedRunIds.push(input.runId); + }, + }), + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + // Run A completes cleanly and caches its staged runtime. + await execute({ runId: "run-a", runtime: {}, ...base } as never); + // Run B: same sessionKey, no sessionParams → not a compatible resume. It must + // drop + dispose A's superseded staged entry, then stage fresh. + await execute({ runId: "run-b", runtime: {}, ...base } as never); + + expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(2); + // A's staged temp was disposed when B superseded it. + expect(disposedRunIds).toContain("run-a"); + }); + + // 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 + // staging windows never overlap. + it("test_concurrent_same_session_staging_is_serialized", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const ensureInputs: Array> = []; + const events: string[] = []; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => recordingRuntime({ ensureInputs }) as never, + prepareRemoteManagedHome: async (input) => { + events.push(`enter:${input.runId}`); + // Yield to the event loop so an unserialized second run would interleave + // its own enter here before we finish staging. + await new Promise((resolve) => setTimeout(resolve, 5)); + const stagedRuntime = await input.stage([]); + events.push(`exit:${input.runId}`); + return { stagedRuntime }; + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + // Both runs share the sessionKey (identical config) and start concurrently. + const [a, b] = await Promise.all([ + execute({ runId: "run-a", runtime: {}, ...base } as never), + execute({ runId: "run-b", runtime: {}, ...base } as never), + ]); + + expect(a.exitCode).toBe(0); + expect(b.exitCode).toBe(0); + // Each staging window is a matched enter/exit pair with no interleaving — the + // lock serialized them (never enter,enter,...,exit,exit). + expect(events).toHaveLength(4); + expect(events[0]).toMatch(/^enter:/); + expect(events[1]).toBe(`exit:${events[0]!.slice("enter:".length)}`); + expect(events[2]).toMatch(/^enter:/); + expect(events[3]).toBe(`exit:${events[2]!.slice("enter:".length)}`); + }); + + // Greptile P1 "Lock Ends Before Use": a same-session re-stage must wait for + // the prior run's active turn and cleanup to finish before it can touch the + // staged remote workspace again. + it("test_concurrent_same_session_staging_waits_for_active_turn_cleanup", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const events: string[] = []; + let releaseTurn!: () => void; + let signalTurnStarted!: () => void; + const turnStarted = new Promise((resolve) => { + signalTurnStarted = resolve; + }); + const turnCompleted = new Promise((resolve) => { + releaseTurn = resolve; + }); + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => ({ + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => { + signalTurnStarted(); + return { + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: turnCompleted.then(() => ({ status: "completed", stopReason: "end_turn" })), + cancel: async () => {}, + }; + }, + setConfigOption: async () => {}, + close: async () => {}, + }) as never, + prepareRemoteManagedHome: async (input) => { + events.push(`enter:${input.runId}`); + await new Promise((resolve) => setTimeout(resolve, 5)); + const stagedRuntime = await input.stage([]); + events.push(`exit:${input.runId}`); + return { stagedRuntime }; + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + const runA = execute({ runId: "run-a", runtime: {}, ...base } as never); + await turnStarted; + const runB = execute({ runId: "run-b", runtime: {}, ...base } as never); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(events).not.toContain("enter:run-b"); + + releaseTurn(); + await runA; + events.push("run-a-finished"); + await runB; + + expect(events).toContain("enter:run-b"); + expect(events.indexOf("enter:run-b")).toBeGreaterThan(events.indexOf("run-a-finished")); + }); + + // The per-session lease must be released when a run is abandoned before it + // reaches the executor's cleanup (e.g. staging or a bridge fails to start), + // otherwise the next run of the same session waits on the lease forever. Here + // the first run's staging throws; the second run of the same session must + // still acquire the lease and stage instead of deadlocking. + it("test_failed_staging_releases_lease_so_next_same_session_run_proceeds", async () => { + const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox(); + const events: string[] = []; + let failNextStaging = true; + const execute = createAcpxEngineExecutor({ + warmHandles: new Map(), + stagedRuntimes: new Map(), + stagingLocks: new Map(), + createRuntime: () => ({ + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + setConfigOption: async () => {}, + close: async () => {}, + }) as never, + prepareRemoteManagedHome: async (input) => { + events.push(`enter:${input.runId}`); + if (failNextStaging) { + failNextStaging = false; + throw new Error("staging boom"); + } + const stagedRuntime = await input.stage([]); + events.push(`exit:${input.runId}`); + return { stagedRuntime }; + }, + }); + const base = baseExecuteArgs({ stateDir, localCwd, executionTarget }); + + await expect(execute({ runId: "run-a", runtime: {}, ...base } as never)).rejects.toThrow( + "staging boom", + ); + // If the failed run had stranded its lease, this second same-session run + // would hang on it and the test would time out. + const resultB = await execute({ runId: "run-b", runtime: {}, ...base } as never); + + expect(resultB.exitCode).toBe(0); + expect(events).toContain("enter:run-b"); + expect(events).toContain("exit:run-b"); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index d244558897..4403901506 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -129,6 +129,52 @@ export interface RuntimeCacheEntry { cleanupTimer?: NodeJS.Timeout; } +/** + * A remote runner-backed session's staged runtime, kept warm across runs so a + * compatible resume reuses it instead of re-shipping the workspace / re-seeding + * the managed home (PR 3: "stage once per session"). Keyed by the session's + * `sessionKey` (`paperclip:companyId:agentId:taskKey:fingerprint`) — the SAME + * fingerprint scoping the warm handle uses — so one session can never read + * another session's staged credentials: a different agent/task/config hashes to + * a different key, misses this cache, and stages its own home. + * + * Remote sessions are never held in the warm-handle cache (their agent process + * lives behind a per-run process-session bridge, torn down each run and resumed + * via `session/load`); the only thing that survives between their runs is the + * in-sandbox staged workspace + home, which this cache reuses. + */ +export interface StagedRuntimeCacheEntry { + stagedRuntime: PreparedAdapterExecutionTargetRuntime; + /** + * The env keys the per-adapter managed-home seam mutated when it staged (e.g. + * `CODEX_HOME` repointed onto the in-sandbox home). Re-applied verbatim on a + * reused run so the spawned agent still receives the in-sandbox home paths + * without re-invoking the seam. These values are deterministic (derived from + * the staged asset dirs), so they are identical across the session's runs. + */ + envDelta: Record; + /** + * The seam's per-run copy-back (codex auth copy-back via `restoreWorkspace()`), + * or null for adapters/customs with no seam. Reused on every run's teardown so + * the copy-back cadence stays exactly per-run — unchanged from PR 2. + * `restoreWorkspace()` reads the sandbox live through the stable (stateless) + * runner, so reusing the closure across resumes copies back the current + * in-sandbox credential, not a stale snapshot. It never removes the staged + * in-sandbox home, so re-running it on each reuse can't invalidate this entry. + */ + teardown: (() => Promise) | null; + /** + * The seam's one-time host-side staged-resource cleanup (e.g. remove the + * staged home temp dir), or null. Fired ONLY when this entry is dropped — + * failed/cancelled/timed-out turn, incompatible re-stage, or idle eviction — + * never while the entry stays warm for reuse. Kept separate from `teardown` + * so a clean turn's per-run copy-back can't delete resources the next + * compatible resume still relies on. + */ + dispose: (() => Promise) | null; + lastUsedAt: number; +} + interface AcpxEngineSettings { adapterType: string; moduleDir: string; @@ -198,20 +244,55 @@ export interface AcpxRemoteManagedHomeContext { export interface AcpxRemoteManagedHomeResult { stagedRuntime: PreparedAdapterExecutionTargetRuntime; /** - * Invoked once on every teardown/exit path (mirrors the CLI restore-hook + - * staged-temp cleanup finally). For codex this runs `restoreWorkspace()` — the - * seam that fires the auth copy-back — and removes the staged home temp dir. - * Failures are logged by the seam, never fatal to the run result (an - * unclean-teardown copy-back miss is the accepted `refresh_token_reused` - * residual, loud on the next host Codex use, never silent). + * Per-run copy-back, invoked once on every teardown/exit path (mirrors the CLI + * restore-hook finally). For codex this runs `restoreWorkspace()` — the seam + * that fires the auth copy-back. It reads the sandbox live and does NOT remove + * the staged in-sandbox home/workspace, so it is safe to re-run on every + * compatible resume that reuses the staged runtime — the copy-back cadence + * stays exactly per-run. Failures are logged by the seam, never fatal to the + * run result (an unclean-teardown copy-back miss is the accepted + * `refresh_token_reused` residual, loud on the next host Codex use, never + * silent). + * + * Host-side staged-resource cleanup (e.g. removing the staged home temp dir) + * is NOT done here — it moved to {@link disposeStaged} so that reusing the + * cached staged runtime across resumes never destroys resources a later run + * still needs. */ teardown?: () => Promise; + /** + * One-time cleanup of host-side staged resources (e.g. the curated staged + * home temp dir). Split out from {@link teardown} so it fires ONLY when the + * staged runtime is actually dropped — a failed/cancelled/timed-out turn, an + * incompatible re-stage, or idle eviction — never on a clean turn that keeps + * the staged runtime warm for the next compatible resume. Idempotent (safe to + * call more than once — it force-removes and swallows already-gone paths). + * Null for adapters that seed from a managed cache and hold no disposable + * temp. + */ + disposeStaged?: () => Promise; } export interface AcpxEngineExecutorOptions { createRuntime?: AcpxRuntimeFactory; now?: () => number; warmHandles?: Map; + /** + * Per-session staged-runtime cache for the remote runner-backed lane (PR 3). + * Keyed by `sessionKey`. Reused across runs so a compatible resume does not + * re-ship the workspace / re-seed the managed home. Defaults to a shared + * module-level map; tests pass an isolated map. + */ + stagedRuntimes?: Map; + /** + * Per-`sessionKey` staging mutex for the remote runner-backed lane (PR 3). + * Serializes the stage-or-reuse decision so two overlapping runs of the same + * session can never ship into the same remote workspace concurrently (one + * stages while the other waits, then re-checks the cache). Defaults to a + * shared module-level map; tests pass an isolated map. Entries are ephemeral — + * cleared as soon as the last waiter for a key finishes staging. + */ + stagingLocks?: Map>; adapterType?: string; moduleDir?: string; packageRootDir?: string; @@ -262,11 +343,28 @@ interface AcpxPreparedRuntime { // are what PR 2 (managed-home seeding + codex copy-back) and PR 3 (session // lifecycle re-staging) build on. stagedRuntime: PreparedAdapterExecutionTargetRuntime | null; - // Teardown hook from the per-adapter remote managed-home seam: runs the - // codex auth copy-back (via `restoreWorkspace()`) and removes staged temp - // dirs. Invoked once on every exit path by `cleanupRemoteBridges`. Null for - // local runs, the runner-less fallback, and adapters with no seam. + // Per-run copy-back hook from the per-adapter remote managed-home seam: runs + // the codex auth copy-back (via `restoreWorkspace()`). Invoked once on every + // exit path by `cleanupRemoteBridges`; it never removes staged temp, so it is + // safe on every compatible resume. Null for local runs, the runner-less + // fallback, and adapters with no seam. remoteManagedHomeTeardown: (() => Promise) | null; + // One-time host-side staged-resource cleanup from the seam (remove staged temp + // dirs). Fired ONLY when the staged runtime is dropped (failed/cancelled/timed + // -out turn, incompatible re-stage, idle eviction), not on a clean turn that + // keeps the runtime warm. Null for local runs, the runner-less fallback, and + // adapters with no disposable temp. + remoteStagingDispose: (() => Promise) | null; + // PR 3: for the remote runner-backed lane, the env keys the managed-home seam + // mutated on this run (or the reused delta on a compatible resume), so the + // executor can cache/refresh the staged-runtime entry after a clean turn. + // Null for local runs, the runner-less fallback, and non-remote lanes. + remoteStagingEnvDelta: Record | null; + // Per-session staging lease held from the initial stage-or-reuse decision + // through the active turn and released only after bridge cleanup completes. + // This keeps later overlapping runs from re-staging into the same remote + // workspace while a prior turn is still using it. + sessionStagingLeaseRelease: (() => void) | null; remoteExecutionIdentity: Record | null; skillPromptInstructions: string; skillsIdentity: Record; @@ -277,6 +375,8 @@ interface AcpxPreparedRuntime { } const defaultWarmHandles = new Map(); +const defaultStagedRuntimes = new Map(); +const defaultStagingLocks = new Map>(); function resolveEngineSettings(options: AcpxEngineExecutorOptions): AcpxEngineSettings { const moduleDir = path.resolve(options.moduleDir ?? defaultModuleDir); @@ -1070,6 +1170,9 @@ async function stageAcpRemoteRuntime(input: { target: AdapterExecutionTarget; adapterKey: string; workspaceLocalDir: string; + // Pin the in-sandbox workspace dir so it provably equals the deterministic + // `sessionCwd` the engine folded into the session fingerprint (PR 3). + workspaceRemoteDir?: string; timeoutSec: number; assets?: AdapterManagedRuntimeAsset[]; onLog: AdapterExecutionContext["onLog"]; @@ -1085,6 +1188,7 @@ async function stageAcpRemoteRuntime(input: { adapterKey: input.adapterKey, timeoutSec: input.timeoutSec, workspaceLocalDir: input.workspaceLocalDir, + ...(input.workspaceRemoteDir ? { workspaceRemoteDir: input.workspaceRemoteDir } : {}), ...(input.assets && input.assets.length > 0 ? { assets: input.assets } : {}), onProgress: (line) => input.onLog("stdout", line), onRuntimeProgress: input.onRuntimeProgress, @@ -1334,108 +1438,25 @@ async function buildRuntime(input: { executionTarget.transport === "sandbox" && Boolean(executionTarget.runner) && Boolean(agentCommandShell); - // Ship the workspace into the sandbox and capture `{ workspaceRemoteDir, - // runtimeRootDir, assetDirs, restoreWorkspace }`. Done once here, before the - // bridges, so both bridges receive the real (non-null) `runtimeRootDir`. - // - // PR 2: on the remote lane, delegate staging to the per-adapter - // `prepareRemoteManagedHome` seam when the adapter supplies one. The seam - // ships the adapter's managed home as an `assets` entry (through the `stage` - // callback = `stageAcpRemoteRuntime`), repoints the home env var (`env`) onto - // the in-sandbox `assetDirs.*` path, and returns a `teardown` that fires the - // codex auth copy-back (`restoreWorkspace()`) and removes staged temp dirs. - // Without a seam (custom agents / shared-engine tests) the engine stages the - // workspace with no home asset — identical to the PR-1 behavior. - let stagedRuntime: PreparedAdapterExecutionTargetRuntime | null = null; - let remoteManagedHomeTeardown: (() => Promise) | null = null; - if (useRemoteProcessSession) { - const stage = (assets: AdapterManagedRuntimeAsset[]) => - stageAcpRemoteRuntime({ - runId, - target: executionTarget, - adapterKey: input.engine.adapterType, - workspaceLocalDir: cwd, - timeoutSec, - assets, - onLog: input.ctx.onLog, - onRuntimeProgress: input.ctx.onRuntimeProgress, - }); - if (input.deps.prepareRemoteManagedHome) { - const seeded = await input.deps.prepareRemoteManagedHome({ - acpxAgent, - companyId: agent.companyId, - runId, - config, - executionTarget, - workspaceLocalDir: cwd, - timeoutSec, - env, - onLog: input.ctx.onLog, - onRuntimeProgress: input.ctx.onRuntimeProgress, - stage, - }); - stagedRuntime = seeded.stagedRuntime; - remoteManagedHomeTeardown = seeded.teardown ?? null; - } else { - stagedRuntime = await stage([]); - } - } // The ACP `session/new` cwd and every cwd-keyed session-state site // (fingerprint, compat, persist, ensureSession, error) bind to THIS single // value so a warm/resumable session created with the in-sandbox cwd is reused - // — not invalidated — on the next run. Remote runner-backed → the staged - // in-sandbox workspace dir; local and the runner-less fallback → the HOST cwd, + // — not invalidated — on the next run. Remote runner-backed → the in-sandbox + // workspace dir; local and the runner-less fallback → the HOST cwd, // byte-identical to today. - const sessionCwd = stagedRuntime?.workspaceRemoteDir ?? cwd; - let paperclipBridge: AdapterExecutionTargetPaperclipBridgeHandle | null = null; - if (useRemoteProcessSession) { - paperclipBridge = await startAdapterExecutionTargetPaperclipBridge({ - runId, - target: { ...executionTarget, streamRunLogs: false }, - runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, - adapterKey: input.engine.adapterType, - timeoutSec, - hostApiToken: env.PAPERCLIP_API_KEY, - onLog: input.ctx.onLog, - }); - if (paperclipBridge) { - Object.assign(env, paperclipBridge.env); - await input.ctx.onLog("stdout", "[paperclip] Sandbox ACP API callback bridge enabled for this run.\n"); - } - } - const runtimeEnv = Object.fromEntries( - Object.entries(ensurePathInEnv({ ...process.env, ...env })).filter( - (entry): entry is [string, string] => typeof entry[1] === "string", - ), - ); - let processSessionBridge: AdapterExecutionTargetProcessSessionBridgeHandle | null = null; - try { - processSessionBridge = useRemoteProcessSession - ? await startAdapterExecutionTargetProcessSessionBridge({ - runId, - target: executionTarget, - runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, - adapterKey: input.engine.adapterType, - command: "sh", - args: ["-lc", `exec ${agentCommandShell}`], - cwd: sessionCwd, - env: runtimeEnv, - timeoutSec, - onLog: input.ctx.onLog, - }) - : null; - } catch (err) { - await paperclipBridge?.stop().catch(() => {}); - // The staged home / copy-back teardown must run even if a bridge fails to - // start after the workspace + managed home were already staged into the - // sandbox, so a refreshed credential is copied back and staged temp dirs - // are removed on this error path too. - await remoteManagedHomeTeardown?.().catch(() => {}); - throw err; - } - const overrideCommand = processSessionBridge?.agentCommand ?? agentCommand; - const overrides = overrideCommand ? { [acpxAgent]: overrideCommand } : undefined; - const agentRegistry = createAgentRegistry({ overrides }); + // + // PR 3: the staging transport derives the in-sandbox workspace dir + // deterministically from the target's `remoteCwd` (it is exactly `remoteCwd` + // for the sandbox transport), so we resolve `sessionCwd` — and therefore the + // session fingerprint / cache key — BEFORE staging. That lets a compatible + // resume decide to reuse an already-staged runtime instead of re-shipping the + // workspace / re-seeding the managed home. The stage call below pins its + // `workspaceRemoteDir` to this same value, so the staged cwd can never + // diverge from the cwd that fed the fingerprint. + const sessionCwd = + useRemoteProcessSession && executionTarget?.kind === "remote" + ? executionTarget.remoteCwd + : cwd; const fingerprint = shortHash({ acpxAgent, agentCommand: agentCommand ?? acpxAgent, @@ -1469,6 +1490,223 @@ async function buildRuntime(input: { }); const taskKey = asString(input.ctx.runtime.taskKey, "") || wakeTaskId || workspaceId || "default"; const sessionKey = `paperclip:${agent.companyId}:${agent.id}:${taskKey}:${fingerprint}`; + + // Ship the workspace into the sandbox and capture `{ workspaceRemoteDir, + // runtimeRootDir, assetDirs, restoreWorkspace }`. Done once here, before the + // bridges, so both bridges receive the real (non-null) `runtimeRootDir`. + // + // PR 2: on the remote lane, delegate staging to the per-adapter + // `prepareRemoteManagedHome` seam when the adapter supplies one. The seam + // ships the adapter's managed home as an `assets` entry (through the `stage` + // callback = `stageAcpRemoteRuntime`), repoints the home env var (`env`) onto + // the in-sandbox `assetDirs.*` path, and returns a `teardown` (per-run codex + // auth copy-back via `restoreWorkspace()`) plus a `disposeStaged` (one-time + // staged-temp cleanup). Without a seam (custom agents / shared-engine tests) + // the engine stages the workspace with no home asset — identical to PR-1. + // + // PR 3 (stage once per session): a COMPATIBLE resume whose fingerprint matches + // this exact `sessionKey` reuses the already-staged in-sandbox runtime — no + // workspace re-ship, no home re-seed — while an incompatible fingerprint (a + // different key) misses the cache and stages fresh. The `sessionKey` + // (`companyId:agentId:taskKey:fingerprint`) is the single scoping key, so one + // session can never read another session's staged credentials. The cache is + // populated by the executor only after a clean turn and dropped on + // failure/cancel/timeout, so it always holds a known-good staged runtime. + // + // Two guards close the concurrency / cross-session windows Greptile flagged: + // * Compatibility gate: reuse only when the supplied session params actually + // resume THIS staged session (the same `isCompatibleSession` predicate the + // warm-handle path uses). A fresh invocation with missing/cleared + // `sessionParams` starts a new ACP session via `session/new`, so it must + // NOT inherit the prior session's staged home/credentials — it stages + // fresh even when company/agent/task/fingerprint (and hence sessionKey) + // collide. + // * Per-key staging lock: the stage-or-reuse decision runs under a + // `sessionKey` mutex so two overlapping runs of the same session can never + // ship into the same remote workspace at once (the loser waits, then + // re-checks the cache before deciding). + const stagedRuntimes = input.deps.stagedRuntimes ?? defaultStagedRuntimes; + const stagingLocks = input.deps.stagingLocks ?? defaultStagingLocks; + const nowMs = input.deps.now ?? (() => Date.now()); + const previousParams = parseObject(input.ctx.runtime.sessionParams); + const isCompatibleResume = isCompatibleSession(previousParams, { + fingerprint, + sessionKey, + cwd: sessionCwd, + mode, + acpxAgent, + remoteExecutionIdentity, + }); + let stagedRuntime: PreparedAdapterExecutionTargetRuntime | null = null; + let remoteManagedHomeTeardown: (() => Promise) | null = null; + let remoteStagingDispose: (() => Promise) | null = null; + let remoteStagingEnvDelta: Record | null = null; + let sessionStagingLeaseRelease: (() => void) | null = null; + if (useRemoteProcessSession && executionTarget?.kind === "remote") { + const remoteTarget = executionTarget; + const staged = await withSessionStagingLease(stagingLocks, sessionKey, async (): Promise<{ + stagedRuntime: PreparedAdapterExecutionTargetRuntime; + teardown: (() => Promise) | null; + dispose: (() => Promise) | null; + envDelta: Record; + }> => { + const cachedStaged = isCompatibleResume ? stagedRuntimes.get(sessionKey) : undefined; + if (cachedStaged) { + // Reuse the already-staged in-sandbox workspace + managed home. Re-apply + // the env keys the seam repointed onto the in-sandbox home (deterministic, + // identical across the session's runs) and reuse the seam's per-run + // copy-back so the codex auth copy-back still fires on THIS run's teardown + // — the copy-back cadence stays exactly per-run, unchanged from PR 2. The + // copy-back reads the sandbox auth.json live at teardown, so the reused + // closure copies back the current credential, never a stale snapshot, and + // it never removes the staged in-sandbox home (host staged-temp cleanup + // moved to `dispose`, fired only when the entry is dropped), so reusing it + // can't leave this run without its staged home. + // (The workspace restore in that same closure diffs against the ORIGINAL + // staging run's host baseline — an accepted consequence of "reuse, don't + // re-ship": the in-sandbox workspace is the source of truth mid-session + // and the host stays synced from it each run.) + Object.assign(env, cachedStaged.envDelta); + cachedStaged.lastUsedAt = nowMs(); + await input.ctx.onLog( + "stdout", + "[paperclip] Reusing the staged in-sandbox runtime for this resumed session (no workspace re-ship / managed-home re-seed).\n", + ); + return { + stagedRuntime: cachedStaged.stagedRuntime, + teardown: cachedStaged.teardown, + dispose: cachedStaged.dispose, + envDelta: cachedStaged.envDelta, + }; + } + // Not a compatible resume (or no cache entry): stage fresh. If a stale + // entry sits at this key (e.g. an incompatible new session colliding on + // company/agent/task/fingerprint), drop it and release its host staged + // resources first so we neither reuse nor leak it. + const stale = stagedRuntimes.get(sessionKey); + if (stale) { + stagedRuntimes.delete(sessionKey); + if (stale.dispose) await stale.dispose().catch(() => {}); + } + const stage = (assets: AdapterManagedRuntimeAsset[]) => + stageAcpRemoteRuntime({ + runId, + target: remoteTarget, + adapterKey: input.engine.adapterType, + workspaceLocalDir: cwd, + workspaceRemoteDir: sessionCwd, + timeoutSec, + assets, + onLog: input.ctx.onLog, + onRuntimeProgress: input.ctx.onRuntimeProgress, + }); + // 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 + // (never deletes) its home env var, so a set-based delta is complete. + const envBeforeStage = { ...env }; + let freshStagedRuntime: PreparedAdapterExecutionTargetRuntime; + let freshTeardown: (() => Promise) | null = null; + let freshDispose: (() => Promise) | null = 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, + }); + freshStagedRuntime = seeded.stagedRuntime; + freshTeardown = seeded.teardown ?? null; + freshDispose = seeded.disposeStaged ?? null; + } else { + freshStagedRuntime = await stage([]); + } + const delta: Record = {}; + for (const [key, value] of Object.entries(env)) { + if (envBeforeStage[key] !== value) delta[key] = value; + } + return { + stagedRuntime: freshStagedRuntime, + teardown: freshTeardown, + dispose: freshDispose, + envDelta: delta, + }; + }); + sessionStagingLeaseRelease = staged.release; + stagedRuntime = staged.value.stagedRuntime; + remoteManagedHomeTeardown = staged.value.teardown; + remoteStagingDispose = staged.value.dispose; + remoteStagingEnvDelta = staged.value.envDelta; + } + // Both bridge starts run under one try so a failure at EITHER — including the + // paperclip callback bridge — fires the same abandon-path cleanup. The + // paperclip bridge starts after the workspace + managed home were already + // staged and the per-session staging lease is already held, so leaving it + // outside the catch would strand the lease (and the staged temp) on a + // start failure and deadlock the next run of this session. + let paperclipBridge: AdapterExecutionTargetPaperclipBridgeHandle | null = null; + let processSessionBridge: AdapterExecutionTargetProcessSessionBridgeHandle | null = null; + let runtimeEnv: Record = {}; + try { + if (useRemoteProcessSession) { + paperclipBridge = await startAdapterExecutionTargetPaperclipBridge({ + runId, + target: { ...executionTarget, streamRunLogs: false }, + runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, + adapterKey: input.engine.adapterType, + timeoutSec, + hostApiToken: env.PAPERCLIP_API_KEY, + onLog: input.ctx.onLog, + }); + if (paperclipBridge) { + Object.assign(env, paperclipBridge.env); + await input.ctx.onLog("stdout", "[paperclip] Sandbox ACP API callback bridge enabled for this run.\n"); + } + } + runtimeEnv = Object.fromEntries( + Object.entries(ensurePathInEnv({ ...process.env, ...env })).filter( + (entry): entry is [string, string] => typeof entry[1] === "string", + ), + ); + processSessionBridge = useRemoteProcessSession + ? await startAdapterExecutionTargetProcessSessionBridge({ + runId, + target: executionTarget, + runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, + adapterKey: input.engine.adapterType, + command: "sh", + args: ["-lc", `exec ${agentCommandShell}`], + cwd: sessionCwd, + env: runtimeEnv, + timeoutSec, + onLog: input.ctx.onLog, + }) + : null; + } catch (err) { + await paperclipBridge?.stop().catch(() => {}); + // The staged home / copy-back teardown must run even if a bridge fails to + // start after the workspace + managed home were already staged into the + // sandbox, so a refreshed credential is copied back on this error path too. + // This run never reaches the executor, so also fire the one-time staged-temp + // dispose here (it no longer rides the per-run copy-back) — the run is being + // 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). + await remoteManagedHomeTeardown?.().catch(() => {}); + await remoteStagingDispose?.().catch(() => {}); + sessionStagingLeaseRelease?.(); + throw err; + } + const overrideCommand = processSessionBridge?.agentCommand ?? agentCommand; + const overrides = overrideCommand ? { [acpxAgent]: overrideCommand } : undefined; + const agentRegistry = createAgentRegistry({ overrides }); const loggedEnv = buildInvocationEnvForLogs(env, { runtimeEnv, includeRuntimeKeys: ["HOME"], @@ -1503,6 +1741,9 @@ async function buildRuntime(input: { paperclipBridge, stagedRuntime, remoteManagedHomeTeardown, + remoteStagingDispose, + remoteStagingEnvDelta, + sessionStagingLeaseRelease, remoteExecutionIdentity, skillPromptInstructions, skillsIdentity: { @@ -1584,6 +1825,7 @@ async function cleanupRemoteBridges(prepared: AcpxPreparedRuntime): Promise {}); } + prepared.sessionStagingLeaseRelease?.(); } function renderPaperclipEnvNote(env: Record): string { @@ -2044,6 +2286,116 @@ async function cleanupIdleHandles(input: { } } +// Drop staged-runtime entries the session has not touched within the warm-idle +// window, so the cache does not accumulate abandoned sessions (e.g. every time +// a config change shifts the fingerprint to a new key). The per-run copy-back +// already ran on the entry's last run's `cleanupRemoteBridges`; eviction fires +// the entry's one-time `dispose` (host staged-temp cleanup) — the only place +// the staged temp is removed now that it no longer rides the per-run teardown. +// A later run of the same session simply re-stages fresh (re-shipping into the +// still-persistent sandbox, which the inbound monotonic auth-merge keeps safe). +async function cleanupIdleStagedRuntimes(input: { + handles: Map; + locks: Map>; + now: () => number; + idleMs: number; +}) { + if (input.idleMs <= 0) return; + const stale: Array<[string, StagedRuntimeCacheEntry]> = []; + for (const entry of input.handles.entries()) { + if (input.now() - entry[1].lastUsedAt >= input.idleMs) stale.push(entry); + } + for (const [key, entry] of stale) { + const lease = await withSessionStagingLease(input.locks, key, async () => { + const current = input.handles.get(key); + if (current !== entry) return; + if (input.now() - current.lastUsedAt < input.idleMs) return; + input.handles.delete(key); + if (entry.dispose) await entry.dispose().catch(() => {}); + }); + lease.release(); + } +} + +// Persist a remote runner-backed session's staged runtime for reuse on the next +// compatible resume. Called ONLY after a clean turn, so the cache never offers a +// half-staged or failed session for reuse. Non-remote lanes carry a null +// stagedRuntime / null envDelta and are skipped. +function saveStagedRuntimeAfterCleanTurn(input: { + handles: Map; + prepared: AcpxPreparedRuntime; + now: number; +}) { + const { prepared } = input; + if (!prepared.stagedRuntime || prepared.remoteStagingEnvDelta === null) return; + input.handles.set(prepared.sessionKey, { + stagedRuntime: prepared.stagedRuntime, + envDelta: prepared.remoteStagingEnvDelta, + teardown: prepared.remoteManagedHomeTeardown, + dispose: prepared.remoteStagingDispose, + lastUsedAt: input.now, + }); +} + +// Drop the staged-runtime entry a finished run owns and release its host-side +// staged resources. Two guards make this safe under overlapping runs of the same +// session key (PR 3 fix — "Concurrent Runs Corrupt Cache Ownership"): +// 1. Ownership guard: only delete the map entry when it is still the exact +// staged runtime THIS run installed/reused (object identity). A concurrent +// run that installed a different clean entry keeps it — a failed run can no +// longer evict another run's good cache entry. +// 2. `dispose` is fired for THIS run's own staged resources regardless, so a +// failed/cancelled run always frees its own staged temp. `dispose` is +// idempotent, so a shared closure re-fired across a reuse chain is safe. +async function discardStagedRuntime(input: { + handles: Map; + 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); + } + if (prepared.remoteStagingDispose) await prepared.remoteStagingDispose().catch(() => {}); +} + +// Per-`sessionKey` async lease: chains each caller after the previous one so +// the stage-or-reuse decision for a session runs serially, then keeps the +// lease held until the active turn finishes and bridge cleanup runs. That means +// overlapping runs of the same session can never stage fresh into the same +// remote workspace while a prior turn is still using it: the loser waits, then +// re-checks the cache before deciding to reuse or re-stage. +async function withSessionStagingLease( + locks: Map>, + key: string, + fn: () => Promise, +): Promise<{ value: T; release: () => void }> { + const prev = locks.get(key) ?? Promise.resolve(); + let releaseGate!: () => void; + const gate = new Promise((resolve) => { + releaseGate = resolve; + }); + // The next waiter's `prev` is this promise; it settles only once we release + // the gate below, so callers run one at a time. + const mine: Promise = prev.then(() => gate); + locks.set(key, mine); + await prev.catch(() => {}); + let released = false; + const release = () => { + if (released) return; + released = true; + releaseGate(); + // GC the lock if no later caller has chained after us. + if (locks.get(key) === mine) locks.delete(key); + }; + try { + return { value: await fn(), release }; + } catch (error) { + if (!released) release(); + throw error; + } +} + function clearWarmHandleTimer(entry: RuntimeCacheEntry) { if (!entry.cleanupTimer) return; clearTimeout(entry.cleanupTimer); @@ -2112,6 +2464,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const createRuntime = deps.createRuntime ?? createAcpRuntime; const now = deps.now ?? (() => Date.now()); const warmHandles = deps.warmHandles ?? defaultWarmHandles; + const stagedRuntimes = deps.stagedRuntimes ?? defaultStagedRuntimes; + const stagingLocks = deps.stagingLocks ?? defaultStagingLocks; const engine = resolveEngineSettings(deps); return async function executeAcpxEngine(ctx: AdapterExecutionContext): Promise { @@ -2126,6 +2480,16 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { ...(billingIdentity?.biller ? { biller: billingIdentity.biller } : {}), billingType: billingIdentity?.billingType ?? ("unknown" as const), }; + const warmIdleMs = asNumber(ctx.config.warmHandleIdleMs, DEFAULT_ACP_ENGINE_WARM_HANDLE_IDLE_MS); + // Evict idle staged runtimes BEFORE building the runtime, since buildRuntime + // consults the staged cache to decide whether a compatible resume may reuse + // an already-staged runtime — an expired entry must not be reused. + await cleanupIdleStagedRuntimes({ + handles: stagedRuntimes, + locks: stagingLocks, + now, + idleMs: warmIdleMs, + }); const prepared = await buildRuntime({ ctx, engine, deps }); // 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: @@ -2135,7 +2499,6 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { "stderr", `[paperclip] ${formatAdapterExecutionTimeoutStartLogLine(prepared.timeoutResolution)}\n`, ); - const warmIdleMs = asNumber(ctx.config.warmHandleIdleMs, DEFAULT_ACP_ENGINE_WARM_HANDLE_IDLE_MS); await cleanupIdleHandles({ handles: warmHandles, now: now(), idleMs: warmIdleMs }); const previousParams = parseObject(ctx.runtime.sessionParams); @@ -2209,6 +2572,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { err, phase: "ensure_session", }); + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); await cleanupRemoteBridges(prepared); return { exitCode: 1, @@ -2225,6 +2589,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { } if (!handle) { + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); await cleanupRemoteBridges(prepared); return { exitCode: 1, @@ -2263,6 +2628,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { clearWarmHandleTimer(existing); warmHandles.delete(prepared.sessionKey); } + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); await cleanupRemoteBridges(prepared); return { exitCode: 1, @@ -2438,6 +2804,17 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { } } + // PR 3: keep the staged runtime warm for the next compatible resume only + // after a clean turn; a failed/cancelled/timed-out turn discards it so the + // next run stages fresh instead of reusing a torn-down session's staged + // credentials. Copy-back still fires for every outcome via + // `cleanupRemoteBridges` below (unchanged from PR 2). + if (terminal.status === "completed" && !timedOut) { + saveStagedRuntimeAfterCleanTurn({ handles: stagedRuntimes, prepared, now: now() }); + } else { + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); + } + const errorMessage = timedOut ? formatAdapterExecutionTimeoutErrorMessage(prepared.timeoutResolution) : resultErrorMessage(terminal); @@ -2498,6 +2875,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { clearWarmHandleTimer(existing); warmHandles.delete(prepared.sessionKey); } + await discardStagedRuntime({ handles: stagedRuntimes, prepared }); const { classified, message } = await emitAcpxFailure({ ctx, prepared, diff --git a/packages/adapters/codex-local/src/server/acp.test.ts b/packages/adapters/codex-local/src/server/acp.test.ts index bc686bcaba..9af33f26c3 100644 --- a/packages/adapters/codex-local/src/server/acp.test.ts +++ b/packages/adapters/codex-local/src/server/acp.test.ts @@ -92,6 +92,16 @@ function subscriptionAuthJson(accountId: string, lastRefresh: string, marker: st ); } +// Enumerate the host staged-home temp dirs `stageCodexHomeForSync` created for a +// given runId (`paperclip-codex-home-sync--` under os.tmpdir()). +// A unique per-test runId scopes the match to this run's staging dirs only, so +// the assertion is not disturbed by other tests/processes sharing the tmp dir. +async function listCodexHomeSyncDirs(runId: string): Promise { + const prefix = `paperclip-codex-home-sync-${runId}-`; + const entries = await fs.readdir(os.tmpdir()); + return entries.filter((name) => name.startsWith(prefix)).map((name) => path.join(os.tmpdir(), name)); +} + function setNodeVersion(version: string): void { Object.defineProperty(process, "version", { configurable: true, @@ -788,6 +798,158 @@ describe("codex_local ACP lane", () => { expect(hostAuth.tokens.refresh_token).toBe("ref-host-newer"); }); + it("keeps the host staged Codex home after a clean teardown so a compatible resume can reuse it", async () => { + // Session-re-staging guardrail: the per-run copy-back (`teardown`) must NOT + // remove the host staged-home temp dir — that removal moved to the one-time + // `disposeStaged`, fired only when the runtime is dropped. So after a CLEAN + // turn the engine caches the staged runtime warm and its host staged home is + // still on disk for the next compatible resume to reuse. + const runId = "run-keep-staged-home"; + const root = await makeTempRoot("paperclip-codex-acp-keep-staged-"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + const sourceHome = path.join(root, "codex-home"); + const sharedHostHome = path.join(root, "shared-codex-home"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.mkdir(sourceHome, { recursive: true }); + await fs.mkdir(sharedHostHome, { recursive: true }); + // Strictly-newer sandbox credential so the per-run copy-back has real work. + await fs.writeFile( + path.join(sourceHome, "auth.json"), + subscriptionAuthJson("acct-same", NEWER_REFRESH, "sandbox-newer"), + { mode: 0o600 }, + ); + await fs.writeFile( + path.join(sharedHostHome, "auth.json"), + subscriptionAuthJson("acct-same", OLDER_REFRESH, "host-older"), + { mode: 0o600 }, + ); + process.env.CODEX_HOME = sharedHostHome; + + // Isolated staged-runtime cache so this test observes only its own entry. + const stagedRuntimes = new Map(); + const execute = createCodexAcpExecutor({ + createRuntime: (options: FakeRuntimeOptions) => new FakeRuntime(options) as never, + stagedRuntimes, + stagingLocks: new Map(), + }); + const result = await execute( + buildContext(localCwd, { + runId, + config: { + engine: "acp", + cwd: localCwd, + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + env: { CODEX_HOME: sourceHome }, + 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); + // Guardrail: `teardown` ran the copy-back but left the host staged home in + // place, and the clean turn cached the staged runtime warm for reuse. + const stagedDirs = await listCodexHomeSyncDirs(runId); + expect(stagedDirs).toHaveLength(1); + await expect(fs.stat(stagedDirs[0]!)).resolves.toBeDefined(); + expect(stagedRuntimes.size).toBe(1); + // The per-run copy-back still fired: the strictly-newer sandbox credential + // landed on the shared host under the monotonic guard. + const hostAuth = JSON.parse(await fs.readFile(path.join(sharedHostHome, "auth.json"), "utf8")); + expect(hostAuth.last_refresh).toBe(NEWER_REFRESH); + expect(hostAuth.tokens.refresh_token).toBe("ref-sandbox-newer"); + // No `disposeStaged` fires while the entry stays warm, so remove the + // intentionally-persisted staged temp ourselves to avoid leaking it. + await Promise.all(stagedDirs.map((dir) => fs.rm(dir, { recursive: true, force: true }))); + }); + + it("removes the host staged Codex home when a failed turn drops the staged runtime", async () => { + // The complementary guardrail: when the staged runtime IS dropped (here, a + // failed turn), the one-time `disposeStaged` fires and removes the host + // staged-home temp dir — while the per-run copy-back (`teardown`) STILL fires + // on the unclean exit path, so a rotated sandbox credential is never lost. + const runId = "run-drop-staged-home"; + const root = await makeTempRoot("paperclip-codex-acp-drop-staged-"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + const sourceHome = path.join(root, "codex-home"); + const sharedHostHome = path.join(root, "shared-codex-home"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.mkdir(sourceHome, { recursive: true }); + await fs.mkdir(sharedHostHome, { recursive: true }); + await fs.writeFile( + path.join(sourceHome, "auth.json"), + subscriptionAuthJson("acct-same", NEWER_REFRESH, "sandbox-newer"), + { mode: 0o600 }, + ); + await fs.writeFile( + path.join(sharedHostHome, "auth.json"), + subscriptionAuthJson("acct-same", OLDER_REFRESH, "host-older"), + { mode: 0o600 }, + ); + process.env.CODEX_HOME = sharedHostHome; + + const stagedRuntimes = new Map(); + const execute = createCodexAcpExecutor({ + // A failed turn drives the drop path (discard staged runtime + dispose). + createRuntime: (options: FakeRuntimeOptions) => + new FakeRuntime(options, [], { status: "failed", stopReason: "error" }) as never, + stagedRuntimes, + stagingLocks: new Map(), + }); + const result = await execute( + buildContext(localCwd, { + runId, + config: { + engine: "acp", + cwd: localCwd, + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + env: { CODEX_HOME: sourceHome }, + 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(1); + // Guardrail: the dropped staged runtime disposed its host staged home and + // left nothing cached for reuse. + await expect(listCodexHomeSyncDirs(runId)).resolves.toEqual([]); + expect(stagedRuntimes.size).toBe(0); + // ...yet the per-run copy-back still ran on the failure teardown path, so the + // strictly-newer sandbox credential was not lost. + const hostAuth = JSON.parse(await fs.readFile(path.join(sharedHostHome, "auth.json"), "utf8")); + expect(hostAuth.last_refresh).toBe(NEWER_REFRESH); + expect(hostAuth.tokens.refresh_token).toBe("ref-sandbox-newer"); + }); + it("falls back to the CLI lane for a runner-less sandbox even when the ACP command is set", async () => { setNodeVersion("v22.13.0"); // Isolate the missing bidirectional runner as the sole fallback cause: diff --git a/packages/adapters/codex-local/src/server/acp.ts b/packages/adapters/codex-local/src/server/acp.ts index e9bd0bdb5a..cf543a2a4e 100644 --- a/packages/adapters/codex-local/src/server/acp.ts +++ b/packages/adapters/codex-local/src/server/acp.ts @@ -204,6 +204,13 @@ async function prepareCodexRemoteManagedHome( return { stagedRuntime, + // Per-run copy-back: fires on EVERY run's teardown (including a compatible + // resume that reuses this staged runtime). It reads the sandbox auth.json / + // workspace live and copies back to the host; it does NOT remove the staged + // in-sandbox home, so re-running it across resumes can't leave a later run + // without its staged home. Host staged-temp removal is deliberately NOT here + // — see `disposeStaged` — so caching this runtime for reuse never destroys + // resources the next resume needs. teardown: async () => { try { await onLog( @@ -222,17 +229,23 @@ async function prepareCodexRemoteManagedHome( err instanceof Error ? err.message : String(err) }\n`, ); - } finally { - await fs.rm(stagedCodexHomeDir, { recursive: true, force: true }).catch(async (error) => { - await onLog( - "stderr", - `[paperclip] Failed to remove staged Codex home "${stagedCodexHomeDir}": ${ - error instanceof Error ? error.message : String(error) - }\n`, - ); - }); } }, + // One-time cleanup of the HOST staged home temp dir. Fired ONLY when the + // staged runtime is dropped (failed/cancelled/timed-out turn, incompatible + // re-stage, idle eviction) — never on a clean turn that keeps the runtime + // warm — so it can't remove the staged home while a reuse still depends on + // it. Idempotent: `force: true` no-ops if it was already removed. + disposeStaged: async () => { + await fs.rm(stagedCodexHomeDir, { recursive: true, force: true }).catch(async (error) => { + await onLog( + "stderr", + `[paperclip] Failed to remove staged Codex home "${stagedCodexHomeDir}": ${ + error instanceof Error ? error.message : String(error) + }\n`, + ); + }); + }, }; }