feat(acpx): stage once per remote ACP session - compatible resume reuses staged runtime, no cross-session credential reuse (#10089)
## Thinking Path > - Paperclip coordinates autonomous agent work across isolated company-scoped sessions > - The ACP remote lane stages workspaces and managed home state inside the sandbox so sessions can resume safely > - If a compatible resume re-staged everything every time, it would waste work and risk inconsistent session reuse behavior > - If an incompatible resume reused the wrong staged runtime, it could cross session boundaries or leak credentials > - This pull request keeps the session fingerprint as the scoping key and adds a staged-runtime cache keyed to that fingerprint > - Compatible resumes now reuse the already staged runtime while incompatible fingerprints stage fresh > - The benefit is faster safe resumes without weakening session isolation or credential separation ## Linked Issues or Issue Description ### Problem or motivation The ACP remote lane currently needs to preserve safe resume behavior without repeatedly re-staging work that is already valid for the same session. The failure mode to avoid is letting one session reuse another session's staged workspace or credentials. ### Proposed solution Keep the session fingerprint as the scoping key and add a staged-runtime cache keyed to that fingerprint. When the fingerprint matches, reuse the already staged workspace and managed home. When the fingerprint changes, stage fresh. ### Alternatives considered - Always restage on resume: safest mechanically, but wastes work and breaks the compatible-resume optimization. - Reuse without fingerprint scoping: too risky because it could cross session boundaries. ### Roadmap alignment This is a narrow implementation change for the ACP remote lane and does not duplicate any broader roadmap item I could find in `ROADMAP.md`. ### Additional context The change preserves the existing session fingerprint contents and codex auth copy-back cadence while adding tests for compatible reuse, incompatible fresh staging, no cross-session credential reuse, and failed-turn eviction. ## What Changed - Added a staged-runtime cache in the ACP remote execution path keyed by the session fingerprint. - Reused the existing staged workspace and managed home for compatible resumes. - Kept incompatible fingerprints on the fresh staging path. - Preserved the existing session fingerprint contents and codex auth copy-back cadence. - Added tests for compatible reuse, incompatible fresh staging, no cross-session credential reuse, and failed-turn eviction. ## Verification - `pnpm exec vitest run packages/adapter-utils/src/acpx-engine/execute.test.ts` (74/74 pass, including the active-turn lease regression) - `pnpm exec tsc -p packages/adapter-utils/tsconfig.json --noEmit` - Verified the PR touches only `packages/adapter-utils/src/acpx-engine/execute.ts` and `packages/adapter-utils/src/acpx-engine/execute.test.ts` ## Risks - A cache eviction bug could cause an unavailable or partially staged runtime to be reused. - If the fingerprint scoping regressed, a session could incorrectly reuse another session's state. - The change is isolated to the ACP remote lane, but it still affects resume behavior for that path. ## Model Used OpenAI Codex, GPT-5, reasoning-capable coding agent, tool-enabled session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
a7186dce4b
commit
d36ea13e08
|
|
@ -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<Record<string, unknown>>;
|
||||
terminalStatus?: "completed" | "failed";
|
||||
}) {
|
||||
return {
|
||||
ensureSession: async (session: Record<string, unknown>) => {
|
||||
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<string, unknown>;
|
||||
env?: Record<string, string>;
|
||||
}) {
|
||||
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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
// 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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
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<void>((resolve) => {
|
||||
signalTurnStarted = resolve;
|
||||
});
|
||||
const turnCompleted = new Promise<void>((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<Record<string, unknown>> = [];
|
||||
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<Record<string, unknown>> = [];
|
||||
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<void>((resolve) => {
|
||||
signalTurnStarted = resolve;
|
||||
});
|
||||
const turnCompleted = new Promise<void>((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");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, string>;
|
||||
/**
|
||||
* 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<void>) | 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<void>) | 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<void>;
|
||||
/**
|
||||
* 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<void>;
|
||||
}
|
||||
|
||||
export interface AcpxEngineExecutorOptions {
|
||||
createRuntime?: AcpxRuntimeFactory;
|
||||
now?: () => number;
|
||||
warmHandles?: Map<string, RuntimeCacheEntry>;
|
||||
/**
|
||||
* 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<string, StagedRuntimeCacheEntry>;
|
||||
/**
|
||||
* 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<string, Promise<unknown>>;
|
||||
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<void>) | 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<void>) | 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<string, string> | 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<string, unknown> | null;
|
||||
skillPromptInstructions: string;
|
||||
skillsIdentity: Record<string, unknown>;
|
||||
|
|
@ -277,6 +375,8 @@ interface AcpxPreparedRuntime {
|
|||
}
|
||||
|
||||
const defaultWarmHandles = new Map<string, RuntimeCacheEntry>();
|
||||
const defaultStagedRuntimes = new Map<string, StagedRuntimeCacheEntry>();
|
||||
const defaultStagingLocks = new Map<string, Promise<unknown>>();
|
||||
|
||||
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<void>) | 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<void>) | null = null;
|
||||
let remoteStagingDispose: (() => Promise<void>) | null = null;
|
||||
let remoteStagingEnvDelta: Record<string, string> | 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<void>) | null;
|
||||
dispose: (() => Promise<void>) | null;
|
||||
envDelta: Record<string, string>;
|
||||
}> => {
|
||||
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<void>) | null = null;
|
||||
let freshDispose: (() => Promise<void>) | 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<string, string> = {};
|
||||
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<string, string> = {};
|
||||
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<void
|
|||
if (prepared.remoteManagedHomeTeardown) {
|
||||
await prepared.remoteManagedHomeTeardown().catch(() => {});
|
||||
}
|
||||
prepared.sessionStagingLeaseRelease?.();
|
||||
}
|
||||
|
||||
function renderPaperclipEnvNote(env: Record<string, string>): 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<string, StagedRuntimeCacheEntry>;
|
||||
locks: Map<string, Promise<unknown>>;
|
||||
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<string, StagedRuntimeCacheEntry>;
|
||||
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<string, StagedRuntimeCacheEntry>;
|
||||
prepared: AcpxPreparedRuntime;
|
||||
}): Promise<void> {
|
||||
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<T>(
|
||||
locks: Map<string, Promise<unknown>>,
|
||||
key: string,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<{ value: T; release: () => void }> {
|
||||
const prev = locks.get(key) ?? Promise.resolve();
|
||||
let releaseGate!: () => void;
|
||||
const gate = new Promise<void>((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<unknown> = 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<AdapterExecutionResult> {
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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-<runId>-<random>` 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<string[]> {
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -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`,
|
||||
);
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue