fix: ACP run lifecycle corrections — failure settlement, workspace sync-back, lease cleanup (#11454)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent adapters run ACP sessions and manage runtime, workspace, and lease resources. > - Several failure paths left runtime bridges, staged workspaces, or environment leases active after an error. > - These leaks reduce run reliability and can leave later runs without clean resources. > - This pull request closes the failure paths, applies one teardown policy, and adds regression tests. > - The benefit is consistent failure settlement and safer reuse of agent workspaces and leases. ## Linked Issues or Issue Description **What happened?** ACP runs could leave runtime bridges, staged workspaces, or environment leases active after failures. Claude and Gemini ACP runs did not restore the sandbox workspace on teardown. Lease release stopped when one lease returned an error. **Expected behavior** Each ACP failure must return an error result and settle its resources. Teardown must run each step, release leases independently, and restore the host workspace when the sandbox ends. Pending cleanup leases must receive bounded retry attempts. **Steps to reproduce** 1. Run an ACP session that fails after runtime creation or during turn preparation. 2. Run an ACP session that fails during a warm hit or staged runtime handoff. 3. Run lease cleanup with more than one lease when the first release returns an error. 4. Inspect the result phase, teardown calls, workspace state, and lease metadata. 5. Run the regression suites listed in the Verification section. ## What Changed - Settle every ACP failure after runtime creation with an error result and one sandbox.startup span closure. - Close the ACP runtime and remove warm entries after every pre-turn failure. - Run all teardown steps, record teardown errors, release staging leases in finally, and prevent duplicate teardown. - Dispose staged runtimes after seam failures and remove borrowed staged entries with identity guards. - Add fail-open workspace sync-back teardown for Claude and Gemini ACP adapters. - Isolate lease release errors and add bounded retry sweeps for stranded pending_cleanup leases. - Atomically claim pending_cleanup retries and clamp attempt readers to keep the five-attempt bound. - Default absent provider reusableLeases values to false and align the fake provider with its runtime declaration. - Add regression tests for engine, adapter, server, and shared environment behavior. ## Verification - [x] `npx vitest run packages/adapter-utils/src/acpx-engine/execute.test.ts` — 124 tests passed. - [x] `npx vitest run packages/adapters/codex-local/src/server/acp.test.ts packages/adapters/claude-local/src/server/acp.test.ts packages/adapters/gemini-local/src/server/acp.test.ts` — 61 tests passed. - [x] `npx vitest run server/src/__tests__/environment-runtime.test.ts server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts server/src/__tests__/reusable-leases-default.test.ts server/src/__tests__/environment-routes.test.ts packages/shared/src/environment-support.test.ts` — passed. - [x] All listed suites ran from the repository root. - [x] GitHub CI completed successfully for `cfc349c9f232711433897915112a1c52c0e462ca`. - [x] Greptile completed with a 5/5 confidence score and no blocking finding. ## Risks The engine changes affect failure settlement and teardown order across ACP runs. The server changes add retry state to existing lease metadata without a schema migration. The adapter changes restore workspaces after sandbox execution. Regression tests cover the changed paths. GitHub CI and Greptile passed for the current head. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. This change fixes runtime reliability and does not duplicate a roadmap feature. ## Model Used OpenAI GPT-5 Codex. The model used tool-based repository inspection, GitHub operations, and code review support. The runtime does not expose a context-window value. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (for example, `docs/...` or `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
38752c4e5e
commit
e52b8a343f
|
|
@ -3051,6 +3051,83 @@ describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / re
|
|||
expect(disposedRunIds).toContain("run-a");
|
||||
});
|
||||
|
||||
// F5 leak window 1: a seam that throws AFTER a successful stage() must not leave
|
||||
// the fresh in-sandbox runtime behind. The engine records the staged result in
|
||||
// the stage wrapper and disposes it when the seam throws.
|
||||
it("test_seam_throw_after_stage_disposes_fresh_staged_runtime", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const stagedRuntimes = new Map();
|
||||
let capturedRuntimeRoot: string | null = null;
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes,
|
||||
createRuntime: () => recordingRuntime({ ensureInputs: [] }) as never,
|
||||
prepareRemoteManagedHome: async (input) => {
|
||||
const staged = await input.stage([]);
|
||||
capturedRuntimeRoot = staged.runtimeRootDir;
|
||||
// The seam fails AFTER a successful stage (e.g. an in-sandbox config
|
||||
// materialize step throws). The fresh in-sandbox runtime is already shipped.
|
||||
throw new Error("seam failed after stage");
|
||||
},
|
||||
});
|
||||
const base = baseExecuteArgs({ stateDir, localCwd, executionTarget });
|
||||
|
||||
await expect(
|
||||
execute({ runId: "run-a", runtime: {}, ...base } as never),
|
||||
).rejects.toThrow("seam failed after stage");
|
||||
|
||||
// The stage shipped a real in-sandbox runtime root (local runner = host FS)...
|
||||
expect(typeof capturedRuntimeRoot).toBe("string");
|
||||
// ...and the seam throw disposed it, so the abandoned managed home is gone.
|
||||
expect(await pathExists(capturedRuntimeRoot!)).toBe(false);
|
||||
// The failed fresh stage was never cached for reuse.
|
||||
expect(stagedRuntimes.size).toBe(0);
|
||||
});
|
||||
|
||||
// F5 leak window 2: when a bridge fails during bring-up on a resume that BORROWED
|
||||
// the cached staged runtime, the rollback must remove the borrowed cache entry
|
||||
// through the same identity guard `discardStagedRuntime` uses — else a later
|
||||
// resume reuses an entry whose host staged-temp was already disposed.
|
||||
it("test_partial_bridge_rollback_removes_borrowed_staged_entry", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const ensureInputs: Array<Record<string, unknown>> = [];
|
||||
const stagedRuntimes = new Map();
|
||||
let disposeCalls = 0;
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes,
|
||||
createRuntime: () => recordingRuntime({ ensureInputs }) as never,
|
||||
prepareRemoteManagedHome: async (input) => ({
|
||||
stagedRuntime: await input.stage([]),
|
||||
disposeStaged: async () => {
|
||||
disposeCalls += 1;
|
||||
},
|
||||
}),
|
||||
});
|
||||
const base = baseExecuteArgs({ stateDir, localCwd, executionTarget });
|
||||
|
||||
const first = await execute({ runId: "run-a", runtime: {}, ...base } as never);
|
||||
expect(first.exitCode).toBe(0);
|
||||
// The clean turn cached the staged runtime for reuse; nothing disposed yet.
|
||||
expect(stagedRuntimes.size).toBe(1);
|
||||
expect(disposeCalls).toBe(0);
|
||||
|
||||
// Run B resumes the same session and borrows the cached staged runtime, but a
|
||||
// bridge fails during bring-up.
|
||||
vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(async () => {
|
||||
throw new Error("paperclip bridge boom");
|
||||
});
|
||||
await expect(
|
||||
execute({ runId: "run-b", runtime: { sessionParams: first.sessionParams }, ...base } as never),
|
||||
).rejects.toThrow("paperclip bridge boom");
|
||||
|
||||
// The rollback removed the borrowed cache entry through the identity guard, so a
|
||||
// later resume can never reuse an entry whose host staged-temp was disposed...
|
||||
expect(stagedRuntimes.size).toBe(0);
|
||||
// ...and it fired that entry's shared idempotent dispose exactly once.
|
||||
expect(disposeCalls).toBe(1);
|
||||
});
|
||||
|
||||
// Greptile P1 "Concurrent Runs Corrupt Cache Ownership": two overlapping runs
|
||||
// of the same session key must not ship into the same remote workspace at once.
|
||||
// The per-key staging lock serializes the stage-or-reuse section, so their
|
||||
|
|
@ -4377,3 +4454,785 @@ describe("ACPX engine per-step startup timing (run.startup.step events)", () =>
|
|||
expect(emitted.has("bridge.process-session")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("ACPX engine run lifecycle corrections (F1: settle every failure after buildRuntime)", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
// A remote sandbox setup that stages the host worktree through the real local
|
||||
// runner, so a run reaches the post-build window with live bridges and a held
|
||||
// staging lease.
|
||||
async function setupRemoteSandbox() {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
const remoteCwd = path.join(root, "remote-workspace");
|
||||
await fs.mkdir(localCwd, { recursive: true });
|
||||
await fs.mkdir(remoteCwd, { recursive: true });
|
||||
await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8");
|
||||
const executionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "fake-plugin",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
};
|
||||
return { root, stateDir, localCwd, remoteCwd, executionTarget };
|
||||
}
|
||||
|
||||
it("test_runtime_create_failure_returns_error_result_with_phase_create_runtime", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: () => {
|
||||
throw new Error("createRuntime boom");
|
||||
},
|
||||
});
|
||||
const result = await execute({
|
||||
runId: "run-create-fail",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") },
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
} as never);
|
||||
|
||||
// The post-build runtime-creation window failure returns a settled error
|
||||
// result instead of throwing.
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.resultJson?.phase).toBe("create_runtime");
|
||||
expect(result.summary).toContain("createRuntime boom");
|
||||
});
|
||||
|
||||
it("test_runtime_create_failure_stops_bridges_and_releases_staging_lease", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const paperclipStop = vi.fn(async () => {});
|
||||
const processStop = vi.fn(async () => {});
|
||||
vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce(
|
||||
async () => ({ env: {}, stop: paperclipStop }) as never,
|
||||
);
|
||||
vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementationOnce(
|
||||
async () => ({ agentCommand: null, stop: processStop }) as never,
|
||||
);
|
||||
const stagingLocks = new Map<string, Promise<unknown>>();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
stagingLocks,
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
createRuntime: () => {
|
||||
throw new Error("createRuntime boom");
|
||||
},
|
||||
});
|
||||
|
||||
const result = await execute({
|
||||
runId: "run-create-fail-remote",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd },
|
||||
context: {},
|
||||
authToken: "real-run-jwt",
|
||||
executionTarget,
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onEvent: async () => {},
|
||||
} as never);
|
||||
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.resultJson?.phase).toBe("create_runtime");
|
||||
// Both live bridges stop exactly once.
|
||||
expect(paperclipStop).toHaveBeenCalledTimes(1);
|
||||
expect(processStop).toHaveBeenCalledTimes(1);
|
||||
// The per-session staging lease released, so the lock map does not strand the
|
||||
// next same-session run.
|
||||
expect(stagingLocks.size).toBe(0);
|
||||
});
|
||||
|
||||
it("test_prompt_build_failure_returns_error_result_with_phase_prepare_turn", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const close = vi.fn(async () => {});
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close,
|
||||
}) as never,
|
||||
});
|
||||
|
||||
// A throwing accessor on a field only `buildPrompt` reads makes the prompt
|
||||
// build fail after the session handshake succeeds.
|
||||
const context: Record<string, unknown> = {};
|
||||
Object.defineProperty(context, "paperclipSessionHandoffMarkdown", {
|
||||
enumerable: false,
|
||||
get() {
|
||||
throw new Error("prompt build boom");
|
||||
},
|
||||
});
|
||||
|
||||
const result = await execute({
|
||||
runId: "run-prepare-fail",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") },
|
||||
context,
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
} as never);
|
||||
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.resultJson?.phase).toBe("prepare_turn");
|
||||
// The established runtime is closed on the pre-turn failure, so no session
|
||||
// leaks.
|
||||
expect(close).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("test_sandbox_startup_span_ends_exactly_once_on_every_exit_path", async () => {
|
||||
// Each scenario runs a fresh remote bring-up with a recording trace and asserts
|
||||
// the one sandbox.startup span ends exactly once, whatever the exit path.
|
||||
const failingEnsure = () =>
|
||||
({
|
||||
ensureSession: async () => {
|
||||
throw new Error("ensureSession boom");
|
||||
},
|
||||
startTurn: () => ({
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: async () => {},
|
||||
}) as never;
|
||||
const scenarios: Array<{ name: string; createRuntime: AcpxEngineExecutorOptions["createRuntime"] }> = [
|
||||
{ name: "success", createRuntime: () => buildRuntime() as never },
|
||||
{
|
||||
name: "create_runtime failure",
|
||||
createRuntime: () => {
|
||||
throw new Error("createRuntime boom");
|
||||
},
|
||||
},
|
||||
{ name: "ensure_session failure", createRuntime: () => failingEnsure() },
|
||||
];
|
||||
|
||||
for (const scenario of scenarios) {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const { traceContext, spans } = createRecordingStartupTrace();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
stagingLocks: new Map(),
|
||||
createRuntime: scenario.createRuntime,
|
||||
});
|
||||
|
||||
await execute({
|
||||
runId: `run-span-${scenario.name.replace(/\s+/g, "-")}`,
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd },
|
||||
context: {},
|
||||
authToken: "real-run-jwt",
|
||||
executionTarget,
|
||||
startupTraceContext: traceContext,
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onEvent: async () => {},
|
||||
} as never);
|
||||
|
||||
const startupSpans = spans.filter((span) => span.name === "sandbox.startup");
|
||||
expect(startupSpans, `scenario "${scenario.name}" must open one sandbox.startup span`).toHaveLength(1);
|
||||
expect(
|
||||
startupSpans[0]!.endCalls,
|
||||
`scenario "${scenario.name}" must end the sandbox.startup span exactly once`,
|
||||
).toBe(1);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("ACPX engine run lifecycle corrections (F2: close the runtime for every pre-turn failure)", () => {
|
||||
const startedAt = "2026-01-01T00:00:00.000Z";
|
||||
const handle = {
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
};
|
||||
|
||||
it("test_handshake_failure_closes_runtime_and_removes_warm_entry", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
const closeSpy = vi.fn(async () => {});
|
||||
let created = 0;
|
||||
const warmHandles = new Map();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles,
|
||||
createRuntime: (options) => {
|
||||
created += 1;
|
||||
const opts = options as AcpRuntimeOptions & {
|
||||
onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
|
||||
};
|
||||
return {
|
||||
ensureSession: async () => {
|
||||
await opts.onAgentSpawn?.({ pid: 4242, startedAt });
|
||||
return handle;
|
||||
},
|
||||
startTurn: () => ({
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: closeSpy,
|
||||
} as never;
|
||||
},
|
||||
});
|
||||
const config = {
|
||||
agent: "custom",
|
||||
agentCommand: "node ./fake-acp.js",
|
||||
stateDir,
|
||||
mode: "persistent",
|
||||
warmHandleIdleMs: 60_000,
|
||||
};
|
||||
|
||||
const first = await execute({
|
||||
runId: "warm-handshake-1",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config,
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onSpawn: async () => {},
|
||||
} as never);
|
||||
expect(first.exitCode).toBe(0);
|
||||
expect(warmHandles.size).toBe(1);
|
||||
expect(created).toBe(1);
|
||||
|
||||
// The warm-hit reuses runtime #1 and fails while persisting process identity.
|
||||
const second = await execute({
|
||||
runId: "warm-handshake-2",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: { sessionParams: first.sessionParams },
|
||||
config,
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onSpawn: async () => {
|
||||
throw new Error("onSpawn boom");
|
||||
},
|
||||
} as never);
|
||||
|
||||
expect(created).toBe(1);
|
||||
expect(second.exitCode).toBe(1);
|
||||
expect(second.resultJson?.phase).toBe("ensure_session");
|
||||
// The reused runtime is closed and the warm entry removed.
|
||||
expect(closeSpy).toHaveBeenCalledTimes(1);
|
||||
expect(warmHandles.size).toBe(0);
|
||||
});
|
||||
|
||||
it("test_missing_session_handle_closes_runtime", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const closeSpy = vi.fn(async () => {});
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: () =>
|
||||
({
|
||||
// A runtime that returns no session handle drives the missing-handle path.
|
||||
ensureSession: async () => undefined,
|
||||
startTurn: () => ({
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: closeSpy,
|
||||
}) as never,
|
||||
});
|
||||
|
||||
const result = await execute({
|
||||
runId: "missing-handle",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") },
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
} as never);
|
||||
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.resultJson?.phase).toBe("ensure_session");
|
||||
expect(result.errorCode).toBe("acpx_runtime_error");
|
||||
// The constructed runtime is closed so its child process cannot leak.
|
||||
expect(closeSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("test_warm_hit_failure_closes_runtime_and_does_not_rearm_idle_timer", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
let created = 0;
|
||||
const closes: number[] = [];
|
||||
const warmHandles = new Map();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles,
|
||||
createRuntime: (options) => {
|
||||
const id = (created += 1);
|
||||
const opts = options as AcpRuntimeOptions & {
|
||||
onAgentSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
|
||||
};
|
||||
return {
|
||||
ensureSession: async () => {
|
||||
await opts.onAgentSpawn?.({ pid: 1000 + id, startedAt });
|
||||
return handle;
|
||||
},
|
||||
startTurn: () => ({
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: async () => {
|
||||
closes.push(id);
|
||||
},
|
||||
} as never;
|
||||
},
|
||||
});
|
||||
const config = {
|
||||
agent: "custom",
|
||||
agentCommand: "node ./fake-acp.js",
|
||||
stateDir,
|
||||
mode: "persistent",
|
||||
warmHandleIdleMs: 60_000,
|
||||
};
|
||||
|
||||
const first = await execute({
|
||||
runId: "warm-timer-1",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config,
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onSpawn: async () => {},
|
||||
} as never);
|
||||
expect(first.exitCode).toBe(0);
|
||||
expect(warmHandles.size).toBe(1);
|
||||
|
||||
// A warm-hit whose identity-persist step fails reuses runtime #1 (no new create).
|
||||
const second = await execute({
|
||||
runId: "warm-timer-2",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: { sessionParams: first.sessionParams },
|
||||
config,
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onSpawn: async () => {
|
||||
throw new Error("onSpawn boom");
|
||||
},
|
||||
} as never);
|
||||
expect(created).toBe(1);
|
||||
expect(second.exitCode).toBe(1);
|
||||
// Runtime #1 is closed and its warm entry removed. The warm-hit already cleared
|
||||
// the idle timer, and the failure does not re-arm it, so the map is empty.
|
||||
expect(closes).toEqual([1]);
|
||||
expect(warmHandles.size).toBe(0);
|
||||
|
||||
// A later heartbeat finds no warm entry and constructs a fresh runtime, proving
|
||||
// the failed warm-hit did not leave a re-armed entry behind.
|
||||
const third = await execute({
|
||||
runId: "warm-timer-3",
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: { sessionParams: first.sessionParams },
|
||||
config,
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onSpawn: async () => {},
|
||||
} as never);
|
||||
expect(third.exitCode).toBe(0);
|
||||
expect(created).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("ACPX engine run lifecycle corrections (F3: one teardown error policy)", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
const okHandle = {
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
};
|
||||
|
||||
async function setupRemoteSandbox() {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
const remoteCwd = path.join(root, "remote-workspace");
|
||||
await fs.mkdir(localCwd, { recursive: true });
|
||||
await fs.mkdir(remoteCwd, { recursive: true });
|
||||
await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8");
|
||||
const executionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "fake-plugin",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
};
|
||||
return { root, stateDir, localCwd, remoteCwd, executionTarget };
|
||||
}
|
||||
|
||||
// Stub both sandbox bridges with stop spies collected per start, so a test can
|
||||
// assert the bridges stopped without running the real bridge transport.
|
||||
function stubBridges() {
|
||||
const paperclipStops: Array<ReturnType<typeof vi.fn>> = [];
|
||||
const processStops: Array<ReturnType<typeof vi.fn>> = [];
|
||||
vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementation(async () => {
|
||||
const stop = vi.fn(async () => {});
|
||||
paperclipStops.push(stop);
|
||||
return { env: {}, stop } as never;
|
||||
});
|
||||
vi.mocked(startAdapterExecutionTargetProcessSessionBridge).mockImplementation(async () => {
|
||||
const stop = vi.fn(async () => {});
|
||||
processStops.push(stop);
|
||||
return { agentCommand: null, stop } as never;
|
||||
});
|
||||
const anyStopped = (stops: Array<ReturnType<typeof vi.fn>>) =>
|
||||
stops.some((stop) => stop.mock.calls.length > 0);
|
||||
const stoppedCount = (stops: Array<ReturnType<typeof vi.fn>>) =>
|
||||
stops.filter((stop) => stop.mock.calls.length > 0).length;
|
||||
return { paperclipStops, processStops, anyStopped, stoppedCount };
|
||||
}
|
||||
|
||||
function throwingHandoffContext(): Record<string, unknown> {
|
||||
const context: Record<string, unknown> = {};
|
||||
Object.defineProperty(context, "paperclipSessionHandoffMarkdown", {
|
||||
enumerable: false,
|
||||
get() {
|
||||
throw new Error("prompt build boom");
|
||||
},
|
||||
});
|
||||
return context;
|
||||
}
|
||||
|
||||
function completedTurn() {
|
||||
return {
|
||||
events: (async function* () {})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
function throwingTurn() {
|
||||
return {
|
||||
events: (async function* () {
|
||||
throw new Error("turn upstream boom");
|
||||
})(),
|
||||
result: Promise.resolve({ status: "completed", stopReason: "end_turn" }),
|
||||
cancel: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
function remoteArgs(
|
||||
stateDir: string,
|
||||
localCwd: string,
|
||||
executionTarget: unknown,
|
||||
overrides: Record<string, unknown> = {},
|
||||
) {
|
||||
return {
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd },
|
||||
context: {},
|
||||
authToken: "real-run-jwt",
|
||||
executionTarget,
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onEvent: async () => {},
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
it("test_teardown_continues_after_one_teardown_step_fails", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const { paperclipStops, processStops, anyStopped } = stubBridges();
|
||||
const stagingLocks = new Map<string, Promise<unknown>>();
|
||||
const logs: Array<{ stream: string; text: string }> = [];
|
||||
const execute = createAcpxEngineExecutor({
|
||||
stagingLocks,
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => throwingTurn(),
|
||||
// A failing teardown step must not stop the later teardown steps.
|
||||
close: async () => {
|
||||
throw new Error("close boom");
|
||||
},
|
||||
}) as never,
|
||||
});
|
||||
|
||||
const result = await execute({
|
||||
runId: "td-continue",
|
||||
...remoteArgs(stateDir, localCwd, executionTarget, {
|
||||
onLog: async (stream: "stdout" | "stderr", text: string) => {
|
||||
logs.push({ stream, text });
|
||||
},
|
||||
}),
|
||||
} as never);
|
||||
|
||||
expect(result.exitCode).toBe(1);
|
||||
// The close failure did not stop the bridge stops or the lease release.
|
||||
expect(anyStopped(paperclipStops)).toBe(true);
|
||||
expect(anyStopped(processStops)).toBe(true);
|
||||
expect(stagingLocks.size).toBe(0);
|
||||
// The close failure was recorded, not silently dropped.
|
||||
expect(
|
||||
logs.some((entry) => entry.stream === "stderr" && entry.text.includes('teardown step "runtime-close" failed')),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("test_staging_lease_releases_in_finally_on_every_exit_path", async () => {
|
||||
const scenarios: Array<{
|
||||
name: string;
|
||||
createRuntime: AcpxEngineExecutorOptions["createRuntime"];
|
||||
context: Record<string, unknown>;
|
||||
}> = [
|
||||
{
|
||||
name: "create_runtime",
|
||||
createRuntime: () => {
|
||||
throw new Error("createRuntime boom");
|
||||
},
|
||||
context: {},
|
||||
},
|
||||
{
|
||||
name: "ensure_session",
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => {
|
||||
throw new Error("ensure boom");
|
||||
},
|
||||
startTurn: () => completedTurn(),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
context: {},
|
||||
},
|
||||
{
|
||||
name: "prepare_turn",
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => completedTurn(),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
context: throwingHandoffContext(),
|
||||
},
|
||||
{
|
||||
name: "turn",
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => throwingTurn(),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
context: {},
|
||||
},
|
||||
{
|
||||
name: "success",
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => completedTurn(),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
context: {},
|
||||
},
|
||||
];
|
||||
|
||||
for (const scenario of scenarios) {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
stubBridges();
|
||||
const stagingLocks = new Map<string, Promise<unknown>>();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
stagingLocks,
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
createRuntime: scenario.createRuntime,
|
||||
});
|
||||
|
||||
await execute({
|
||||
runId: `lease-${scenario.name}`,
|
||||
...remoteArgs(stateDir, localCwd, executionTarget, { context: scenario.context }),
|
||||
} as never).catch(() => {});
|
||||
|
||||
// The staging lease is released in a finally on every exit path, so the lock
|
||||
// map never strands the next same-session run.
|
||||
expect(stagingLocks.size, `lease must release on the ${scenario.name} path`).toBe(0);
|
||||
}
|
||||
});
|
||||
|
||||
it("test_result_emission_failure_does_not_skip_teardown", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const { paperclipStops, processStops, anyStopped } = stubBridges();
|
||||
const stagingLocks = new Map<string, Promise<unknown>>();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
stagingLocks,
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => throwingTurn(),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
});
|
||||
|
||||
await execute({
|
||||
runId: "emit-fail",
|
||||
...remoteArgs(stateDir, localCwd, executionTarget, {
|
||||
// Fail the acpx.error emission only, so teardown must still run.
|
||||
onLog: async (_stream: "stdout" | "stderr", text: string) => {
|
||||
if (text.includes('"type":"acpx.error"')) throw new Error("onLog boom");
|
||||
},
|
||||
}),
|
||||
} as never).catch(() => {});
|
||||
|
||||
expect(anyStopped(paperclipStops)).toBe(true);
|
||||
expect(anyStopped(processStops)).toBe(true);
|
||||
expect(stagingLocks.size).toBe(0);
|
||||
});
|
||||
|
||||
it("test_result_mapping_throw_after_close_does_not_rerun_teardown", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
const { paperclipStops, processStops, stoppedCount } = stubBridges();
|
||||
let closeCount = 0;
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
stagingLocks: new Map(),
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
yield { type: "done", stopReason: "end_turn" };
|
||||
})(),
|
||||
// A completed turn whose result mapping throws after the close is read.
|
||||
result: Promise.resolve({
|
||||
status: "completed",
|
||||
get stopReason(): string {
|
||||
throw new Error("mapping boom");
|
||||
},
|
||||
}),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: async () => {
|
||||
closeCount += 1;
|
||||
},
|
||||
}) as never,
|
||||
});
|
||||
|
||||
await execute({
|
||||
runId: "map-throw",
|
||||
...remoteArgs(stateDir, localCwd, executionTarget),
|
||||
} as never).catch(() => {});
|
||||
|
||||
// The completed turn closed the runtime once; the mapping throw did not re-run
|
||||
// the teardown through the turn catch.
|
||||
expect(closeCount).toBe(1);
|
||||
expect(stoppedCount(paperclipStops)).toBe(1);
|
||||
expect(stoppedCount(processStops)).toBe(1);
|
||||
});
|
||||
|
||||
it("test_teardown_failure_does_not_change_external_result", async () => {
|
||||
const { stateDir, localCwd, executionTarget } = await setupRemoteSandbox();
|
||||
stubBridges();
|
||||
const execute = createAcpxEngineExecutor({
|
||||
warmHandles: new Map(),
|
||||
stagedRuntimes: new Map(),
|
||||
stagingLocks: new Map(),
|
||||
createRuntime: () =>
|
||||
({
|
||||
ensureSession: async () => okHandle,
|
||||
startTurn: () => throwingTurn(),
|
||||
close: async () => {
|
||||
throw new Error("close boom");
|
||||
},
|
||||
}) as never,
|
||||
});
|
||||
|
||||
const result = await execute({
|
||||
runId: "td-result",
|
||||
...remoteArgs(stateDir, localCwd, executionTarget),
|
||||
} as never);
|
||||
|
||||
// The exit cause (the turn failure) determines the external result; the failing
|
||||
// teardown step never leaks into it.
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.resultJson?.phase).toBe("turn");
|
||||
expect(result.errorCode).toBe("acpx_turn_failed");
|
||||
expect(result.errorMessage).toContain("turn upstream boom");
|
||||
expect(result.errorMessage).not.toContain("close boom");
|
||||
});
|
||||
|
||||
it("test_flush_child_stderr_runs_on_every_exit_path", async () => {
|
||||
const writes: string[] = [];
|
||||
const spy = vi.spyOn(process.stderr, "write").mockImplementation(((chunk: unknown) => {
|
||||
writes.push(String(chunk));
|
||||
return true;
|
||||
}) as never);
|
||||
try {
|
||||
const scenarios: Array<{ name: string; mode: string; context: Record<string, unknown> }> = [
|
||||
{ name: "ensure_session", mode: "ensure_fail", context: {} },
|
||||
{ name: "prepare_turn", mode: "success", context: throwingHandoffContext() },
|
||||
{ name: "turn", mode: "turn_fail", context: {} },
|
||||
{ name: "success", mode: "success", context: {} },
|
||||
];
|
||||
|
||||
for (const scenario of scenarios) {
|
||||
const root = await makeTempRoot();
|
||||
const marker = `paperclip-flush-probe-${scenario.name}`;
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: (options) => {
|
||||
const opts = options as {
|
||||
onAgentStderr?: (chunk: string) => void;
|
||||
};
|
||||
// Emit a partial (newline-free) stderr chunk so it stays pending until a
|
||||
// teardown flush writes it to process.stderr.
|
||||
const emit = () => opts.onAgentStderr?.(marker);
|
||||
return {
|
||||
ensureSession: async () => {
|
||||
emit();
|
||||
if (scenario.mode === "ensure_fail") throw new Error("ensure boom");
|
||||
return okHandle;
|
||||
},
|
||||
startTurn: () =>
|
||||
scenario.mode === "turn_fail" ? throwingTurn() : completedTurn(),
|
||||
close: async () => {},
|
||||
} as never;
|
||||
},
|
||||
});
|
||||
|
||||
await execute({
|
||||
runId: `flush-${scenario.name}`,
|
||||
agent: { id: "agent-1", companyId: "company-1" },
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") },
|
||||
context: scenario.context,
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
} as never).catch(() => {});
|
||||
|
||||
expect(
|
||||
writes.some((write) => write.includes(marker)),
|
||||
`flushChildStderr must run on the ${scenario.name} path`,
|
||||
).toBe(true);
|
||||
}
|
||||
} finally {
|
||||
spy.mockRestore();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -618,6 +618,79 @@ describe("claude_local ACP lane", () => {
|
|||
expect(Object.keys(meta[0]?.env ?? {}).filter((key) => key.startsWith("XDG_"))).toEqual([]);
|
||||
});
|
||||
|
||||
it("test_claude_acp_seam_registers_workspace_sync_back", async () => {
|
||||
const root = await makeTempRoot("paperclip-claude-acp-syncback-");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
const remoteCwd = path.join(root, "remote-workspace");
|
||||
const sharedClaudeConfig = path.join(root, "shared-claude-config");
|
||||
await fs.mkdir(localCwd, { recursive: true });
|
||||
await fs.mkdir(remoteCwd, { recursive: true });
|
||||
await fs.mkdir(sharedClaudeConfig, { recursive: true });
|
||||
await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8");
|
||||
await fs.writeFile(
|
||||
path.join(sharedClaudeConfig, "settings.json"),
|
||||
JSON.stringify({ permissions: { defaultMode: "acceptEdits" } }),
|
||||
"utf8",
|
||||
);
|
||||
await fs.writeFile(path.join(sharedClaudeConfig, "CLAUDE.md"), "# shared guidance\n", "utf8");
|
||||
process.env.PAPERCLIP_HOME = path.join(root, "paperclip-home");
|
||||
process.env.PAPERCLIP_INSTANCE_ID = "test";
|
||||
process.env.CLAUDE_CONFIG_DIR = sharedClaudeConfig;
|
||||
|
||||
// The runtime writes a NEW file into the in-sandbox workspace during the turn.
|
||||
// The seam must register a workspace sync-back teardown, so the file lands in
|
||||
// the host worktree after the run.
|
||||
const runtime = new FakeRuntime({});
|
||||
const startTurn = runtime.startTurn.bind(runtime);
|
||||
runtime.startTurn = (input) => {
|
||||
const turn = startTurn(input);
|
||||
const remoteWorkspaceCwd = input.handle.cwd ?? remoteCwd;
|
||||
return {
|
||||
...turn,
|
||||
result: (async () => {
|
||||
await fs.writeFile(path.join(remoteWorkspaceCwd, "from-sandbox.txt"), "synced", "utf8");
|
||||
return await turn.result;
|
||||
})(),
|
||||
};
|
||||
};
|
||||
|
||||
const execute = createClaudeAcpExecutor({
|
||||
createRuntime: (options: FakeRuntimeOptions) => {
|
||||
Object.assign(runtime.options, options);
|
||||
return runtime as never;
|
||||
},
|
||||
});
|
||||
|
||||
const result = await execute(
|
||||
buildContext(localCwd, {
|
||||
config: {
|
||||
engine: "acp",
|
||||
cwd: localCwd,
|
||||
agentCommand: "node ./fake-acp.js",
|
||||
stateDir: path.join(root, "state"),
|
||||
promptTemplate: "Do the assigned work.",
|
||||
},
|
||||
context: {
|
||||
issueId: "issue-1",
|
||||
paperclipWorkspace: { cwd: localCwd, source: "project_workspace", workspaceId: "workspace-1" },
|
||||
},
|
||||
executionTarget: {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "fake-plugin",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
} as never,
|
||||
authToken: "real-run-jwt",
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result.exitCode).toBe(0);
|
||||
// The teardown fired `restoreWorkspace`, so the sandbox-authored file is now
|
||||
// in the host worktree.
|
||||
await expect(fs.readFile(path.join(localCwd, "from-sandbox.txt"), "utf8")).resolves.toBe("synced");
|
||||
});
|
||||
|
||||
it("remaps a workspace-relative explicit CLAUDE_CONFIG_DIR onto the in-sandbox workspace path", async () => {
|
||||
const root = await makeTempRoot("paperclip-claude-acp-explicit-inworkspace-");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
|
|
|
|||
|
|
@ -181,7 +181,8 @@ export function resolveClaudeAcpBillingIdentity(
|
|||
* asset, materialize it into an in-sandbox config dir (copying the sandbox's own
|
||||
* `$HOME/.claude` credentials in), then repoint `CLAUDE_CONFIG_DIR` onto that
|
||||
* in-sandbox config dir. Claude has no credential copy-back (its CLI lane has
|
||||
* none — mirroring the CLI is the contract), so no teardown hook.
|
||||
* none — mirroring the CLI is the contract). The teardown hook therefore only
|
||||
* syncs the sandbox workspace back to the host; it does not touch credentials.
|
||||
*
|
||||
* An explicit `CLAUDE_CONFIG_DIR` (user-managed) is honored only if it can reach
|
||||
* the remote sandbox; a host-only path cannot, so we do NOT forward it verbatim
|
||||
|
|
@ -193,6 +194,25 @@ async function prepareClaudeRemoteManagedHome(
|
|||
input: AcpxRemoteManagedHomeContext,
|
||||
): Promise<AcpxRemoteManagedHomeResult> {
|
||||
const { env, runId, onLog, executionTarget } = input;
|
||||
// Fail-open workspace sync-back for every exit path (mirrors the Claude CLI
|
||||
// lane's restore-hook finally and the Codex ACP seam's teardown). Claude has no
|
||||
// credential copy-back, so the teardown only syncs the sandbox workspace back to
|
||||
// the host. A restore miss is logged and never fails the run.
|
||||
const registerWorkspaceSyncBack = (
|
||||
stagedRuntime: AcpxRemoteManagedHomeResult["stagedRuntime"],
|
||||
): AcpxRemoteManagedHomeResult["teardown"] => async () => {
|
||||
try {
|
||||
await onLog("stdout", "[paperclip] Restoring workspace changes from the sandbox.\n");
|
||||
await stagedRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Claude ACP teardown workspace restore failed: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}\n`,
|
||||
);
|
||||
}
|
||||
};
|
||||
const envConfig = parseObject(input.config.env);
|
||||
const explicitClaudeConfigDir =
|
||||
typeof envConfig.CLAUDE_CONFIG_DIR === "string" && envConfig.CLAUDE_CONFIG_DIR.trim().length > 0
|
||||
|
|
@ -228,7 +248,7 @@ async function prepareClaudeRemoteManagedHome(
|
|||
"stdout",
|
||||
`[paperclip] Remapped operator CLAUDE_CONFIG_DIR from host path ${explicitClaudeConfigDir} onto the in-sandbox workspace path ${remappedConfigDir} for the remote ACP run.\n`,
|
||||
);
|
||||
return { stagedRuntime };
|
||||
return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) };
|
||||
}
|
||||
await onLog(
|
||||
"stderr",
|
||||
|
|
@ -266,7 +286,7 @@ async function prepareClaudeRemoteManagedHome(
|
|||
});
|
||||
// Repoint CLAUDE_CONFIG_DIR onto the in-sandbox config dir.
|
||||
env.CLAUDE_CONFIG_DIR = remoteClaudeConfigDir;
|
||||
return { stagedRuntime };
|
||||
return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) };
|
||||
}
|
||||
|
||||
function withClaudeAcpDefaults(options: ClaudeAcpExecutorOptions): AcpxEngineExecutorOptions {
|
||||
|
|
|
|||
|
|
@ -551,6 +551,68 @@ describe("gemini_local ACP lane", () => {
|
|||
expect(Object.keys(meta[0]?.env ?? {}).filter((key) => key.startsWith("XDG_"))).toEqual([]);
|
||||
});
|
||||
|
||||
it("test_gemini_acp_seam_registers_workspace_sync_back", async () => {
|
||||
const root = await makeTempRoot("paperclip-gemini-acp-syncback-");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
const remoteCwd = path.join(root, "remote-workspace");
|
||||
await fs.mkdir(localCwd, { recursive: true });
|
||||
await fs.mkdir(remoteCwd, { recursive: true });
|
||||
await fs.writeFile(path.join(localCwd, "hello.txt"), "hi", "utf8");
|
||||
|
||||
// The runtime writes a NEW file into the in-sandbox workspace during the turn.
|
||||
// The seam must register a workspace sync-back teardown, so the file lands in
|
||||
// the host worktree after the run.
|
||||
const runtime = new FakeRuntime({});
|
||||
const startTurn = runtime.startTurn.bind(runtime);
|
||||
runtime.startTurn = (input) => {
|
||||
const turn = startTurn(input);
|
||||
const remoteWorkspaceCwd = input.handle.cwd ?? remoteCwd;
|
||||
return {
|
||||
...turn,
|
||||
result: (async () => {
|
||||
await fs.writeFile(path.join(remoteWorkspaceCwd, "from-sandbox.txt"), "synced", "utf8");
|
||||
return await turn.result;
|
||||
})(),
|
||||
};
|
||||
};
|
||||
|
||||
const execute = createGeminiAcpExecutor({
|
||||
createRuntime: (options) => {
|
||||
Object.assign(runtime.options, options);
|
||||
return runtime as never;
|
||||
},
|
||||
});
|
||||
|
||||
const result = await execute(
|
||||
buildContext(localCwd, {
|
||||
config: {
|
||||
engine: "acp",
|
||||
cwd: localCwd,
|
||||
agentCommand: "node ./fake-acp.js",
|
||||
stateDir: path.join(root, "state"),
|
||||
promptTemplate: "Do the assigned work.",
|
||||
},
|
||||
context: {
|
||||
issueId: "issue-1",
|
||||
paperclipWorkspace: { cwd: localCwd, source: "project_workspace", workspaceId: "workspace-1" },
|
||||
},
|
||||
executionTarget: {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "fake-plugin",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
} as never,
|
||||
authToken: "real-run-jwt",
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result.exitCode).toBe(0);
|
||||
// The teardown fired `restoreWorkspace`, so the sandbox-authored file is now
|
||||
// in the host worktree.
|
||||
await expect(fs.readFile(path.join(localCwd, "from-sandbox.txt"), "utf8")).resolves.toBe("synced");
|
||||
});
|
||||
|
||||
it("does not persist an api-key auth selector from a host-only credential", async () => {
|
||||
const root = await makeTempRoot("paperclip-gemini-acp-hostkey-");
|
||||
const localCwd = path.join(root, "worktree");
|
||||
|
|
|
|||
|
|
@ -149,12 +149,32 @@ function resolveGeminiSkillsHome(config: Record<string, unknown>): string {
|
|||
*
|
||||
* The seed never writes key bytes: the key is only read as a boolean signal to
|
||||
* decide whether to persist the auth-method selector. Gemini has no credential
|
||||
* copy-back, so no teardown hook.
|
||||
* copy-back. The teardown hook therefore only syncs the sandbox workspace back to
|
||||
* the host; it does not touch credentials.
|
||||
*/
|
||||
async function prepareGeminiRemoteManagedHome(
|
||||
input: AcpxRemoteManagedHomeContext,
|
||||
): Promise<AcpxRemoteManagedHomeResult> {
|
||||
const { env, runId, onLog, executionTarget } = input;
|
||||
// Fail-open workspace sync-back for every exit path (mirrors the Gemini CLI
|
||||
// lane's restore-hook finally and the Codex ACP seam's teardown). Gemini has no
|
||||
// credential copy-back, so the teardown only syncs the sandbox workspace back to
|
||||
// the host. A restore miss is logged and never fails the run.
|
||||
const registerWorkspaceSyncBack = (
|
||||
stagedRuntime: AcpxRemoteManagedHomeResult["stagedRuntime"],
|
||||
): AcpxRemoteManagedHomeResult["teardown"] => async () => {
|
||||
try {
|
||||
await onLog("stdout", "[paperclip] Restoring workspace changes from the sandbox.\n");
|
||||
await stagedRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Gemini ACP teardown workspace restore failed: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}\n`,
|
||||
);
|
||||
}
|
||||
};
|
||||
const geminiSkillsHome = resolveGeminiSkillsHome(input.config);
|
||||
const stagedRuntime = await input.stage(
|
||||
geminiSkillsHome
|
||||
|
|
@ -168,8 +188,9 @@ async function prepareGeminiRemoteManagedHome(
|
|||
const managedRemoteHomeDir = stagedRuntime.runtimeRootDir;
|
||||
if (!managedRemoteHomeDir) {
|
||||
// No runtime root resolved — leave HOME as-is (host fallback) and skip the
|
||||
// in-sandbox seed; nothing to remap onto.
|
||||
return { stagedRuntime };
|
||||
// in-sandbox seed; nothing to remap onto. The workspace still staged, so the
|
||||
// sync-back teardown still applies.
|
||||
return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) };
|
||||
}
|
||||
env.HOME = managedRemoteHomeDir;
|
||||
|
||||
|
|
@ -222,7 +243,7 @@ async function prepareGeminiRemoteManagedHome(
|
|||
);
|
||||
}
|
||||
|
||||
return { stagedRuntime };
|
||||
return { stagedRuntime, teardown: registerWorkspaceSyncBack(stagedRuntime) };
|
||||
}
|
||||
|
||||
function withGeminiAcpDefaults(options: GeminiAcpExecutorOptions): AcpxEngineExecutorOptions {
|
||||
|
|
|
|||
|
|
@ -43,3 +43,26 @@ describe("isSandboxProviderSupportedForAdapter", () => {
|
|||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getEnvironmentCapabilities reusable leases default", () => {
|
||||
it("test_absent_reusable_leases_presents_as_false", () => {
|
||||
const capabilities = getEnvironmentCapabilities(["codex_local"], {
|
||||
sandboxProviders: {
|
||||
// The manifest does not declare supportsReusableLeases.
|
||||
"fake-plugin": { displayName: "Fake Plugin" },
|
||||
},
|
||||
});
|
||||
expect(capabilities.sandboxProviders["fake-plugin"].supportsReusableLeases).toBe(false);
|
||||
|
||||
// A manifest that declares true still presents as true.
|
||||
const declared = getEnvironmentCapabilities(["codex_local"], {
|
||||
sandboxProviders: {
|
||||
"reuse-plugin": { displayName: "Reuse Plugin", supportsReusableLeases: true },
|
||||
},
|
||||
});
|
||||
expect(declared.sandboxProviders["reuse-plugin"].supportsReusableLeases).toBe(true);
|
||||
|
||||
// The built-in fake provider presents as false.
|
||||
expect(capabilities.sandboxProviders.fake.supportsReusableLeases).toBe(false);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -122,7 +122,9 @@ export function getEnvironmentCapabilities(
|
|||
supportsSavedProbe: true,
|
||||
supportsUnsavedProbe: true,
|
||||
supportsRunExecution: false,
|
||||
supportsReusableLeases: true,
|
||||
// The fake provider runtime declares false. The presentation must match
|
||||
// it, so execution and presentation agree.
|
||||
supportsReusableLeases: false,
|
||||
supportsInteractiveSetup: false,
|
||||
interactiveSetupConnectionTypes: [],
|
||||
supportsTemplateCapture: false,
|
||||
|
|
@ -137,7 +139,9 @@ export function getEnvironmentCapabilities(
|
|||
supportsSavedProbe: capability.supportsSavedProbe ?? true,
|
||||
supportsUnsavedProbe: capability.supportsUnsavedProbe ?? true,
|
||||
supportsRunExecution: capability.supportsRunExecution ?? true,
|
||||
supportsReusableLeases: capability.supportsReusableLeases ?? true,
|
||||
// Default absent to false. A manifest that does not declare reusable
|
||||
// leases must present as false, so execution (=== true) agrees.
|
||||
supportsReusableLeases: capability.supportsReusableLeases ?? false,
|
||||
supportsInteractiveSetup: capability.supportsInteractiveSetup ?? false,
|
||||
interactiveSetupConnectionTypes: capability.interactiveSetupConnectionTypes ?? [],
|
||||
supportsTemplateCapture: capability.supportsTemplateCapture ?? false,
|
||||
|
|
|
|||
|
|
@ -1164,7 +1164,9 @@ describe("environment routes", () => {
|
|||
expect(res.body.sandboxProviders["secure-plugin"]).toMatchObject({
|
||||
status: "supported",
|
||||
supportsRunExecution: true,
|
||||
supportsReusableLeases: true,
|
||||
// The mock manifest omits reusable leases, so the capability defaults to
|
||||
// false and agrees with the execution guard.
|
||||
supportsReusableLeases: false,
|
||||
supportsInteractiveSetup: true,
|
||||
interactiveSetupConnectionTypes: ["ssh"],
|
||||
supportsTemplateCapture: true,
|
||||
|
|
|
|||
|
|
@ -3207,4 +3207,68 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
|||
expect(sshRelease).not.toHaveBeenCalled();
|
||||
expect(acquired.lease.metadata?.driver).toBe("local");
|
||||
});
|
||||
|
||||
it("test_release_run_leases_continues_after_first_release_fails", async () => {
|
||||
const { companyId, environment, runId } = await seedEnvironment();
|
||||
const environmentsSvc = environmentService(db);
|
||||
|
||||
// Seed two active leases for one run. The first driver release throws; the
|
||||
// second must still release.
|
||||
const failingLease = await environmentsSvc.acquireLease({
|
||||
companyId,
|
||||
environmentId: environment.id,
|
||||
heartbeatRunId: runId,
|
||||
provider: "local",
|
||||
providerLeaseId: "fail-release",
|
||||
metadata: { driver: "local" },
|
||||
});
|
||||
const healthyLease = await environmentsSvc.acquireLease({
|
||||
companyId,
|
||||
environmentId: environment.id,
|
||||
heartbeatRunId: runId,
|
||||
provider: "local",
|
||||
providerLeaseId: "healthy-release",
|
||||
metadata: { driver: "local" },
|
||||
});
|
||||
|
||||
const runtimeWithFailingDriver = environmentRuntimeService(db, {
|
||||
drivers: [
|
||||
{
|
||||
driver: "local",
|
||||
acquireRunLease: async () => {
|
||||
throw new Error("acquire should not be called");
|
||||
},
|
||||
releaseRunLease: async ({ lease, status }) => {
|
||||
if (lease.providerLeaseId === "fail-release") {
|
||||
throw new Error("driver release failed");
|
||||
}
|
||||
return await environmentsSvc.releaseLease(lease.id, status);
|
||||
},
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
const errors: Array<{ leaseId: string; error: unknown }> = [];
|
||||
const released = await runtimeWithFailingDriver.releaseRunLeases(
|
||||
runId,
|
||||
"released",
|
||||
(leaseId, error) => errors.push({ leaseId, error }),
|
||||
);
|
||||
|
||||
// The healthy lease released even though the first release failed.
|
||||
expect(released).toHaveLength(1);
|
||||
expect(released[0]?.lease.id).toBe(healthyLease.id);
|
||||
expect(released[0]?.lease.status).toBe("released");
|
||||
|
||||
// The failed release reported one lease-specific error.
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(errors[0]?.leaseId).toBe(failingLease.id);
|
||||
expect((errors[0]?.error as Error).message).toBe("driver release failed");
|
||||
|
||||
// The database confirms the isolation. The healthy lease released; the
|
||||
// failing lease stayed active.
|
||||
const rows = await db.select().from(environmentLeases);
|
||||
expect(rows.find((row) => row.id === healthyLease.id)?.status).toBe("released");
|
||||
expect(rows.find((row) => row.id === failingLease.id)?.status).toBe("active");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -0,0 +1,941 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
environments,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
|
||||
const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() }));
|
||||
vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient }));
|
||||
|
||||
vi.mock("../middleware/logger.js", () => ({
|
||||
logger: {
|
||||
child: vi.fn(function child() {
|
||||
return this;
|
||||
}),
|
||||
trace: vi.fn(),
|
||||
debug: vi.fn(),
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
fatal: vi.fn(),
|
||||
},
|
||||
httpLogger: vi.fn(),
|
||||
}));
|
||||
|
||||
import { logger } from "../middleware/logger.ts";
|
||||
import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.ts";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
console.warn(
|
||||
`Skipping embedded Postgres pending_cleanup sweep tests on this host: ${
|
||||
embeddedPostgresSupport.reason ?? "unsupported environment"
|
||||
}`,
|
||||
);
|
||||
}
|
||||
|
||||
// The sweep reads and writes its retry state under these lease metadata keys.
|
||||
const ATTEMPTS_KEY = "pendingCleanupRetryAttempts";
|
||||
const CAP_WARNED_KEY = "pendingCleanupRetryCapWarned";
|
||||
const ATTEMPT_CAP = 5;
|
||||
|
||||
describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pending-cleanup-sweep-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
beforeEach(() => {
|
||||
vi.mocked(logger.warn).mockClear();
|
||||
vi.mocked(logger.error).mockClear();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(environments);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seedCompanyAndEnvironment() {
|
||||
const companyId = randomUUID();
|
||||
const environmentId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(environments).values({
|
||||
id: environmentId,
|
||||
name: "Fake Sandbox",
|
||||
driver: "sandbox",
|
||||
status: "active",
|
||||
config: { provider: "fake", image: "ubuntu:24.04" },
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
return { companyId, environmentId };
|
||||
}
|
||||
|
||||
async function insertPendingCleanupLease(input: {
|
||||
companyId: string;
|
||||
environmentId: string;
|
||||
updatedAt: Date;
|
||||
metadata?: Record<string, unknown>;
|
||||
}): Promise<string> {
|
||||
const id = randomUUID();
|
||||
await db.insert(environmentLeases).values({
|
||||
id,
|
||||
companyId: input.companyId,
|
||||
environmentId: input.environmentId,
|
||||
status: "pending_cleanup",
|
||||
leasePolicy: "reuse_by_environment",
|
||||
provider: "fake",
|
||||
providerLeaseId: `sandbox://fake/${id}`,
|
||||
cleanupStatus: "failed",
|
||||
metadata: { driver: "sandbox", ...(input.metadata ?? {}) },
|
||||
acquiredAt: input.updatedAt,
|
||||
lastUsedAt: input.updatedAt,
|
||||
releasedAt: input.updatedAt,
|
||||
createdAt: input.updatedAt,
|
||||
updatedAt: input.updatedAt,
|
||||
});
|
||||
return id;
|
||||
}
|
||||
|
||||
function fakeRuntime(
|
||||
destroyRunLease: HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
): HeartbeatEnvironmentRuntime {
|
||||
return { destroyRunLease } as unknown as HeartbeatEnvironmentRuntime;
|
||||
}
|
||||
|
||||
it("test_pending_cleanup_sweep_retries_and_destroys_lease", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy succeeds on retry and moves the lease out of pending_cleanup.
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
const now = new Date();
|
||||
const row = await db
|
||||
.update(environmentLeases)
|
||||
.set({ status: "expired", cleanupStatus: "success", updatedAt: now })
|
||||
.where(eq(environmentLeases.id, lease.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row ? { ...row, status: "expired" as const } : null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]),
|
||||
});
|
||||
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
|
||||
expect(result).toEqual({ swept: 1, destroyed: 1, capped: 0 });
|
||||
expect(destroyRunLease).toHaveBeenCalledTimes(1);
|
||||
expect(destroyRunLease.mock.calls[0]?.[0]).toMatchObject({
|
||||
failureReason: "pending_cleanup_retry",
|
||||
lease: expect.objectContaining({ id: leaseId }),
|
||||
});
|
||||
|
||||
const status = await db
|
||||
.select({ status: environmentLeases.status })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.then((rows) => rows[0]?.status);
|
||||
expect(status).toBe("expired");
|
||||
});
|
||||
|
||||
it("test_pending_cleanup_sweep_respects_backoff_and_page_size", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// A lease touched inside the backoff window is not eligible yet.
|
||||
await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]),
|
||||
});
|
||||
|
||||
const backoffResult = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60 * 1000 });
|
||||
expect(backoffResult.swept).toBe(0);
|
||||
expect(destroyRunLease).not.toHaveBeenCalled();
|
||||
|
||||
// Seed more eligible leases than one page holds. The sweep processes one
|
||||
// page only.
|
||||
const eligibleCount = 22;
|
||||
for (let index = 0; index < eligibleCount; index += 1) {
|
||||
await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
}
|
||||
|
||||
const pageResult = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60 * 1000 });
|
||||
expect(pageResult.swept).toBe(20);
|
||||
expect(destroyRunLease).toHaveBeenCalledTimes(20);
|
||||
});
|
||||
|
||||
it("test_pending_cleanup_sweep_stops_at_attempt_cap_and_warns_once", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP },
|
||||
});
|
||||
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"]),
|
||||
});
|
||||
|
||||
const first = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(first).toEqual({ swept: 1, destroyed: 0, capped: 1 });
|
||||
// A capped lease is not retried.
|
||||
expect(destroyRunLease).not.toHaveBeenCalled();
|
||||
|
||||
const metadataAfterFirst = await db
|
||||
.select({ metadata: environmentLeases.metadata })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.then((rows) => rows[0]?.metadata as Record<string, unknown> | null);
|
||||
expect(metadataAfterFirst?.[CAP_WARNED_KEY]).toBe(true);
|
||||
expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1);
|
||||
|
||||
// A second sweep warns no more; the lease keeps its warned flag.
|
||||
const second = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(second).toEqual({ swept: 1, destroyed: 0, capped: 1 });
|
||||
expect(destroyRunLease).not.toHaveBeenCalled();
|
||||
expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// Two sweep ticks can overlap. Without an atomic claim, both read the same
|
||||
// attempt count, both destroy the same lease, and the retry cap counts one
|
||||
// attempt for two destroys. The atomic claim must let only one sweep destroy
|
||||
// the lease per attempt.
|
||||
it("test_pending_cleanup_overlapping_sweeps_destroy_lease_once", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy fails, so the lease stays in pending_cleanup. The counter
|
||||
// records how many sweeps reached the destroy for this lease.
|
||||
let destroyCount = 0;
|
||||
const destroyRunLease = vi.fn(async () => {
|
||||
destroyCount += 1;
|
||||
return null;
|
||||
});
|
||||
|
||||
// Two sweeps run on two separate database clients, so they truly overlap.
|
||||
// A single client serializes the queries on one connection and hides the
|
||||
// race. The second client connects to the same embedded database.
|
||||
const dbB = createDb(tempDb!.connectionString);
|
||||
try {
|
||||
// Warm up the second connection first. A cold connection would add setup
|
||||
// latency and let the first sweep finish before the second sweep reads.
|
||||
await dbB.select({ id: environmentLeases.id }).from(environmentLeases).limit(1);
|
||||
|
||||
const heartbeatA = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
const heartbeatB = heartbeatService(dbB, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// Both sweeps use the production backoff and run at the same time.
|
||||
const backoffMs = 5 * 60 * 1000;
|
||||
const [first, second] = await Promise.all([
|
||||
heartbeatA.sweepPendingCleanupLeases({ backoffMs }),
|
||||
heartbeatB.sweepPendingCleanupLeases({ backoffMs }),
|
||||
]);
|
||||
|
||||
// Only one sweep claimed the attempt and destroyed the lease.
|
||||
expect(destroyCount).toBe(1);
|
||||
expect(first.destroyed + second.destroyed).toBe(0);
|
||||
} finally {
|
||||
await (dbB as unknown as { $client: { end: () => Promise<void> } }).$client.end();
|
||||
}
|
||||
|
||||
// The attempt count advanced by exactly one; the cap is not exceeded.
|
||||
const metadata = await db
|
||||
.select({ metadata: environmentLeases.metadata, status: environmentLeases.status })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.then((rows) => rows[0]);
|
||||
expect((metadata?.metadata as Record<string, unknown> | null)?.[ATTEMPTS_KEY]).toBe(1);
|
||||
expect(metadata?.status).toBe("pending_cleanup");
|
||||
});
|
||||
|
||||
// Read the current retry attempt count from a lease's metadata.
|
||||
async function readAttempts(leaseId: string): Promise<number> {
|
||||
const metadata = await db
|
||||
.select({ metadata: environmentLeases.metadata })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.then((rows) => rows[0]?.metadata as Record<string, unknown> | null);
|
||||
const value = metadata?.[ATTEMPTS_KEY];
|
||||
return typeof value === "number" ? value : 0;
|
||||
}
|
||||
|
||||
// Run two sweeps at the same time on two separate database clients, so they
|
||||
// truly overlap. A single client serializes the queries on one connection and
|
||||
// hides the race. The second client connects to the same embedded database.
|
||||
// The caller reuses the returned destroy spy to count the real destroys.
|
||||
async function runOverlappingSweeps(
|
||||
destroyRunLease: HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
backoffMs: number,
|
||||
): Promise<void> {
|
||||
const dbB = createDb(tempDb!.connectionString);
|
||||
try {
|
||||
// Warm up the second connection first. A cold connection would add setup
|
||||
// latency and let the first sweep finish before the second sweep reads.
|
||||
await dbB.select({ id: environmentLeases.id }).from(environmentLeases).limit(1);
|
||||
|
||||
const heartbeatA = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(destroyRunLease),
|
||||
});
|
||||
const heartbeatB = heartbeatService(dbB, {
|
||||
environmentRuntime: fakeRuntime(destroyRunLease),
|
||||
});
|
||||
|
||||
await Promise.all([
|
||||
heartbeatA.sweepPendingCleanupLeases({ backoffMs }),
|
||||
heartbeatB.sweepPendingCleanupLeases({ backoffMs }),
|
||||
]);
|
||||
} finally {
|
||||
await (dbB as unknown as { $client: { end: () => Promise<void> } }).$client.end();
|
||||
}
|
||||
}
|
||||
|
||||
// Two overlapping sweeps read the same pending_cleanup lease. The atomic claim
|
||||
// must let only one sweep destroy the sandbox. This test runs the two sweeps
|
||||
// on two clients, then asserts a single destroy and a single attempt
|
||||
// increment.
|
||||
it("test_concurrent_sweeps_destroy_a_lease_at_most_once", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy fails, so the lease stays in pending_cleanup and the test can
|
||||
// read the final attempt count.
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
|
||||
await runOverlappingSweeps(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
5 * 60 * 1000,
|
||||
);
|
||||
|
||||
// Only one sweep wins the claim, so the destroy runs once and the attempt
|
||||
// count advances by one.
|
||||
expect(destroyRunLease).toHaveBeenCalledTimes(1);
|
||||
expect(await readAttempts(leaseId)).toBe(1);
|
||||
});
|
||||
|
||||
// A lost increment lets the real attempt count pass the cap: two destroys run
|
||||
// but the counter advances by one. The atomic claim prevents the lost
|
||||
// increment. This test starts a lease at cap - 1, overlaps two sweeps, then
|
||||
// asserts one destroy, one increment to the cap, and at most one cap warning.
|
||||
it("test_concurrent_sweeps_never_exceed_attempt_cap", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 },
|
||||
});
|
||||
|
||||
// The destroy fails, so the lease stays in pending_cleanup and the test can
|
||||
// read the final attempt count.
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
|
||||
await runOverlappingSweeps(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
5 * 60 * 1000,
|
||||
);
|
||||
|
||||
// One real retry ran, so the count advances by one to the cap and never
|
||||
// passes it.
|
||||
expect(destroyRunLease).toHaveBeenCalledTimes(1);
|
||||
const attempts = await readAttempts(leaseId);
|
||||
expect(attempts).toBe(ATTEMPT_CAP);
|
||||
expect(attempts).toBeLessThanOrEqual(ATTEMPT_CAP);
|
||||
|
||||
// The cap warning logs at most once across the overlap.
|
||||
const capWarnings = vi
|
||||
.mocked(logger.warn)
|
||||
.mock.calls.filter(
|
||||
(call) =>
|
||||
call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup",
|
||||
);
|
||||
expect(capWarnings.length).toBeLessThanOrEqual(1);
|
||||
});
|
||||
|
||||
// Run `inject` exactly once, right before the sweep's first UPDATE on the
|
||||
// environmentLeases table runs. The sweep's first update is always its claim
|
||||
// (the retry claim or the cap-warning claim). This lets a test place a
|
||||
// concurrent writer between the sweep's read and its claim.
|
||||
function injectBeforeFirstLeaseClaim(inject: () => Promise<void>): { restore: () => void } {
|
||||
const realUpdate = db.update.bind(db);
|
||||
let injected = false;
|
||||
const spy = vi.spyOn(db, "update").mockImplementation(((table: unknown) => {
|
||||
const builder = (realUpdate as (t: unknown) => unknown)(table) as {
|
||||
set: (values: unknown) => unknown;
|
||||
};
|
||||
if (table === environmentLeases && !injected) {
|
||||
const realSet = builder.set.bind(builder);
|
||||
builder.set = (values: unknown) => {
|
||||
const afterSet = realSet(values) as {
|
||||
returning: (...args: unknown[]) => unknown;
|
||||
};
|
||||
if (!injected) {
|
||||
injected = true;
|
||||
const realReturning = afterSet.returning.bind(afterSet);
|
||||
afterSet.returning = (...args: unknown[]) =>
|
||||
(async () => {
|
||||
await inject();
|
||||
return realReturning(...args);
|
||||
})();
|
||||
}
|
||||
return afterSet;
|
||||
};
|
||||
}
|
||||
return builder;
|
||||
}) as unknown as typeof db.update);
|
||||
return { restore: () => spy.mockRestore() };
|
||||
}
|
||||
|
||||
// Read the full metadata object of a lease.
|
||||
async function readMetadata(leaseId: string): Promise<Record<string, unknown> | null> {
|
||||
return db
|
||||
.select({ metadata: environmentLeases.metadata })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, leaseId))
|
||||
.then((rows) => (rows[0]?.metadata as Record<string, unknown> | null) ?? null);
|
||||
}
|
||||
|
||||
// A provider or plugin destroy rejection can carry a credential in its `name`
|
||||
// or `code`, not only its message. The per-lease retry log must never
|
||||
// serialize any error-derived string.
|
||||
it("test_pending_cleanup_retry_log_omits_sentinel_in_error_name_and_code", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy rejects with a credential-shaped sentinel in both the error
|
||||
// name and the error code.
|
||||
const sentinel = "Bearer sk-SENTINEL-a1b2c3";
|
||||
const rejection = new Error("provider destroy failed");
|
||||
rejection.name = `ProviderDestroyError ${sentinel}`;
|
||||
(rejection as { code?: string }).code = `EPROVIDER ${sentinel}`;
|
||||
const destroyRunLease = vi.fn(async () => {
|
||||
throw rejection;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result).toEqual({ swept: 1, destroyed: 0, capped: 0 });
|
||||
expect(destroyRunLease).toHaveBeenCalledTimes(1);
|
||||
|
||||
const retryCall = vi
|
||||
.mocked(logger.warn)
|
||||
.mock.calls.find((call) => call[1] === "pending_cleanup lease retry failed");
|
||||
expect(retryCall).toBeDefined();
|
||||
const record = retryCall?.[0] as Record<string, unknown>;
|
||||
|
||||
// No error-derived string reaches the log. The sentinel never appears, and
|
||||
// no error field is present.
|
||||
expect(JSON.stringify(record)).not.toContain(sentinel);
|
||||
expect(record).not.toHaveProperty("err");
|
||||
expect(record).not.toHaveProperty("errorName");
|
||||
expect(record).not.toHaveProperty("errorCode");
|
||||
expect(record).not.toHaveProperty("message");
|
||||
expect(record).not.toHaveProperty("stack");
|
||||
expect(record).not.toHaveProperty("cause");
|
||||
|
||||
// The log carries only fixed, locally generated fields.
|
||||
expect(record).toMatchObject({
|
||||
leaseId,
|
||||
environmentId,
|
||||
attempts: 1,
|
||||
errorKind: "destroy_failed",
|
||||
});
|
||||
});
|
||||
|
||||
// The outer sweep catch logs a failure of the sweep itself. It must log a
|
||||
// constant errorKind only, never an error-derived string in the name or code.
|
||||
it("test_pending_cleanup_sweep_failure_log_omits_sentinel_in_error_name_and_code", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// Make the sweep itself throw. The lease query rejects with an error that
|
||||
// carries a credential-shaped sentinel in both the name and the message. The
|
||||
// reaper's own queries pass through, so only the sweep fails.
|
||||
const sentinel = "Bearer sk-SENTINEL-d4e5f6";
|
||||
const realSelect = db.select.bind(db);
|
||||
const selectSpy = vi
|
||||
.spyOn(db, "select")
|
||||
.mockImplementation((...args: Parameters<typeof db.select>) => {
|
||||
const builder = realSelect(...args);
|
||||
const realFrom = builder.from.bind(builder);
|
||||
(builder as { from: unknown }).from = (table: unknown) => {
|
||||
if (table === environmentLeases) {
|
||||
const failure = new Error(`sweep query failed: ${sentinel}`);
|
||||
failure.name = `SweepQueryError ${sentinel}`;
|
||||
(failure as { code?: string }).code = `ESWEEP ${sentinel}`;
|
||||
throw failure;
|
||||
}
|
||||
return realFrom(table as Parameters<typeof realFrom>[0]);
|
||||
};
|
||||
return builder;
|
||||
});
|
||||
|
||||
try {
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The reaper isolates the sweep, so the reaper itself resolves.
|
||||
await expect(heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 })).resolves.toBeDefined();
|
||||
|
||||
const sweepCall = vi
|
||||
.mocked(logger.error)
|
||||
.mock.calls.find((call) => call[1] === "pending_cleanup lease sweep failed");
|
||||
expect(sweepCall).toBeDefined();
|
||||
const record = sweepCall?.[0] as Record<string, unknown>;
|
||||
|
||||
expect(JSON.stringify(record)).not.toContain(sentinel);
|
||||
expect(record).not.toHaveProperty("err");
|
||||
expect(record).not.toHaveProperty("errorName");
|
||||
expect(record).not.toHaveProperty("errorCode");
|
||||
expect(record).not.toHaveProperty("message");
|
||||
expect(record).not.toHaveProperty("stack");
|
||||
expect(record).toMatchObject({ errorKind: "sweep_failed" });
|
||||
} finally {
|
||||
selectSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
// The claim writes only its own attempts key with `jsonb_set`. A concurrent
|
||||
// writer that sets an unrelated metadata key between the sweep's read and its
|
||||
// claim must keep that key. A full copied-metadata write would lose it.
|
||||
it("test_pending_cleanup_claim_preserves_unrelated_concurrent_metadata_key", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
const leaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy fails, so the lease stays pending_cleanup and the test can
|
||||
// read the final metadata.
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// A concurrent writer sets an unrelated key right before the claim runs.
|
||||
const hook = injectBeforeFirstLeaseClaim(async () => {
|
||||
await db
|
||||
.update(environmentLeases)
|
||||
.set({
|
||||
metadata: sql`jsonb_set(coalesce(${environmentLeases.metadata}, '{}'::jsonb), array['concurrentKey'], to_jsonb('keep-me'::text), true)`,
|
||||
})
|
||||
.where(eq(environmentLeases.id, leaseId));
|
||||
});
|
||||
|
||||
try {
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result).toEqual({ swept: 1, destroyed: 0, capped: 0 });
|
||||
} finally {
|
||||
hook.restore();
|
||||
}
|
||||
|
||||
// The claim advanced its own attempts key and kept the concurrent key.
|
||||
const metadata = await readMetadata(leaseId);
|
||||
expect(metadata?.[ATTEMPTS_KEY]).toBe(1);
|
||||
expect(metadata?.concurrentKey).toBe("keep-me");
|
||||
});
|
||||
|
||||
// A provider can write a malformed value under the attempts key. The guarded
|
||||
// cast must read it as zero, not throw. One malformed lease must never abort
|
||||
// the page sweep, so the sweep still processes the other leases on the page.
|
||||
it("test_pending_cleanup_malformed_retry_metadata_does_not_abort_sweep", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// The malformed lease sorts first, so a throw would abort before the sweep
|
||||
// reaches the healthy lease.
|
||||
const malformedLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 120 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: "corrupt" },
|
||||
});
|
||||
const healthyLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy fails, so both leases stay pending_cleanup and the test can
|
||||
// count the retries.
|
||||
const destroyedLeaseIds: string[] = [];
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
destroyedLeaseIds.push(lease.id);
|
||||
return null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The sweep resolves; the malformed cast does not throw.
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result.swept).toBe(2);
|
||||
|
||||
// The sweep still processed the healthy lease.
|
||||
expect(destroyedLeaseIds).toContain(healthyLeaseId);
|
||||
|
||||
// The malformed attempts value normalized to a real number on the claim.
|
||||
const malformedMetadata = await readMetadata(malformedLeaseId);
|
||||
expect(malformedMetadata?.[ATTEMPTS_KEY]).toBe(1);
|
||||
});
|
||||
|
||||
// The cap warning fires only for a pending_cleanup lease at or above the cap.
|
||||
// A lease that leaves pending_cleanup between the read and the claim must get
|
||||
// no warn-flag write, because the claim carries a status predicate.
|
||||
it("test_pending_cleanup_cap_warning_requires_pending_cleanup_status_and_cap", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// A lease at the cap. It leaves pending_cleanup right before the warn claim.
|
||||
const cappedLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP },
|
||||
});
|
||||
|
||||
const destroyRunLease = vi.fn(async () => null);
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// A concurrent path moves the lease out of pending_cleanup right before the
|
||||
// warn claim runs.
|
||||
const hook = injectBeforeFirstLeaseClaim(async () => {
|
||||
await db
|
||||
.update(environmentLeases)
|
||||
.set({ status: "expired" })
|
||||
.where(eq(environmentLeases.id, cappedLeaseId));
|
||||
});
|
||||
|
||||
try {
|
||||
await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
} finally {
|
||||
hook.restore();
|
||||
}
|
||||
|
||||
// The lease left pending_cleanup, so the warn claim wrote no flag.
|
||||
const metadata = await readMetadata(cappedLeaseId);
|
||||
expect(metadata?.[CAP_WARNED_KEY]).not.toBe(true);
|
||||
|
||||
// The cap warning did not log.
|
||||
const capWarnings = vi
|
||||
.mocked(logger.warn)
|
||||
.mock.calls.filter(
|
||||
(call) =>
|
||||
call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup",
|
||||
);
|
||||
expect(capWarnings.length).toBe(0);
|
||||
|
||||
// A lease below the cap gets no warn flag either. The retry path never
|
||||
// reaches the cap branch.
|
||||
const belowCapLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 },
|
||||
});
|
||||
await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
const belowCapMetadata = await readMetadata(belowCapLeaseId);
|
||||
expect(belowCapMetadata?.[CAP_WARNED_KEY]).not.toBe(true);
|
||||
});
|
||||
|
||||
// Override a lease metadata root with a raw JSON value. The insert helper
|
||||
// always writes an object root, so a scalar or array root test needs a raw
|
||||
// update. The value binds as text and casts to jsonb, so `"5"` writes a number
|
||||
// scalar root and `"[1,2,3]"` writes an array root.
|
||||
async function setRawMetadataRoot(leaseId: string, rawJson: string): Promise<void> {
|
||||
await db
|
||||
.update(environmentLeases)
|
||||
.set({ metadata: sql`${rawJson}::jsonb` })
|
||||
.where(eq(environmentLeases.id, leaseId));
|
||||
}
|
||||
|
||||
// A provider can write a negative attempt count. The SQL reader and the
|
||||
// TypeScript reader both clamp a negative value to zero, so the claim predicate
|
||||
// matches. The lease still claims one attempt and the destroy still runs.
|
||||
it("test_pending_cleanup_negative_attempts_metadata_normalizes_and_lease_still_destroys", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// The negative-attempts lease sorts first. A thrown or mismatched claim would
|
||||
// strand it before the sweep reaches the healthy lease.
|
||||
const negativeLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 120 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: -3 },
|
||||
});
|
||||
const healthyLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy succeeds and moves each lease out of pending_cleanup.
|
||||
const destroyedLeaseIds: string[] = [];
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
destroyedLeaseIds.push(lease.id);
|
||||
const now = new Date();
|
||||
const row = await db
|
||||
.update(environmentLeases)
|
||||
.set({ status: "expired", cleanupStatus: "success", updatedAt: now })
|
||||
.where(eq(environmentLeases.id, lease.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row ? { ...row, status: "expired" as const } : null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The sweep resolves; the negative value does not strand the lease.
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result.swept).toBe(2);
|
||||
|
||||
// The negative-attempts lease claimed one attempt and destroyed.
|
||||
expect(destroyedLeaseIds).toContain(negativeLeaseId);
|
||||
expect(destroyedLeaseIds).toContain(healthyLeaseId);
|
||||
expect(result.destroyed).toBe(2);
|
||||
|
||||
// The negative value normalized to zero, so the claim advanced it to one.
|
||||
const negativeMetadata = await readMetadata(negativeLeaseId);
|
||||
expect(negativeMetadata?.[ATTEMPTS_KEY]).toBe(1);
|
||||
|
||||
const negativeStatus = await db
|
||||
.select({ status: environmentLeases.status })
|
||||
.from(environmentLeases)
|
||||
.where(eq(environmentLeases.id, negativeLeaseId))
|
||||
.then((rows) => rows[0]?.status);
|
||||
expect(negativeStatus).toBe("expired");
|
||||
});
|
||||
|
||||
// A provider can write a finite number outside the 32-bit integer range. The
|
||||
// SQL reader computes as numeric and clamps to the cap, so the `::int` cast
|
||||
// never runs on the stored value. The sweep does not throw, and the malformed
|
||||
// lease normalizes into the cap behavior.
|
||||
it("test_pending_cleanup_out_of_range_attempts_metadata_does_not_abort_sweep", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// The out-of-range lease sorts first. A thrown cast would abort before the
|
||||
// sweep reaches the healthy lease.
|
||||
const outOfRangeLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 120 * 60 * 1000),
|
||||
metadata: { [ATTEMPTS_KEY]: 1e300 },
|
||||
});
|
||||
const healthyLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy fails, so the healthy lease stays pending_cleanup and the test
|
||||
// can confirm the sweep reached it.
|
||||
const destroyedLeaseIds: string[] = [];
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
destroyedLeaseIds.push(lease.id);
|
||||
return null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The sweep resolves; the out-of-range value does not abort the page.
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result.swept).toBe(2);
|
||||
|
||||
// The value clamped to the cap, so the lease took the cap branch, not a
|
||||
// destroy. The warn-once flag is set exactly once.
|
||||
expect(result.capped).toBe(1);
|
||||
expect(destroyedLeaseIds).not.toContain(outOfRangeLeaseId);
|
||||
const outOfRangeMetadata = await readMetadata(outOfRangeLeaseId);
|
||||
expect(outOfRangeMetadata?.[CAP_WARNED_KEY]).toBe(true);
|
||||
|
||||
// The sweep still processed the healthy lease.
|
||||
expect(destroyedLeaseIds).toContain(healthyLeaseId);
|
||||
});
|
||||
|
||||
// A provider can write a scalar metadata root. `jsonb_set` fails on a
|
||||
// non-object root, so the reader falls back to an empty object. The sweep does
|
||||
// not throw, and the lease still claims one attempt and destroys.
|
||||
it("test_pending_cleanup_scalar_metadata_root_does_not_abort_sweep", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// The scalar-root lease sorts first. A failed `jsonb_set` would abort before
|
||||
// the sweep reaches the healthy lease.
|
||||
const scalarLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 120 * 60 * 1000),
|
||||
});
|
||||
await setRawMetadataRoot(scalarLeaseId, "5");
|
||||
const healthyLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy succeeds and moves each lease out of pending_cleanup.
|
||||
const destroyedLeaseIds: string[] = [];
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
destroyedLeaseIds.push(lease.id);
|
||||
const now = new Date();
|
||||
const row = await db
|
||||
.update(environmentLeases)
|
||||
.set({ status: "expired", cleanupStatus: "success", updatedAt: now })
|
||||
.where(eq(environmentLeases.id, lease.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row ? { ...row, status: "expired" as const } : null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The sweep resolves; the scalar root does not break the claim write.
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result.swept).toBe(2);
|
||||
expect(result.destroyed).toBe(2);
|
||||
expect(destroyedLeaseIds).toContain(scalarLeaseId);
|
||||
expect(destroyedLeaseIds).toContain(healthyLeaseId);
|
||||
|
||||
// The claim replaced the scalar root with an object that holds one attempt.
|
||||
const scalarMetadata = await readMetadata(scalarLeaseId);
|
||||
expect(scalarMetadata?.[ATTEMPTS_KEY]).toBe(1);
|
||||
});
|
||||
|
||||
// A provider can write an array metadata root. `jsonb_set` fails on a
|
||||
// non-object root, so the reader falls back to an empty object. The sweep does
|
||||
// not throw, and the lease still claims one attempt and destroys.
|
||||
it("test_pending_cleanup_array_metadata_root_does_not_abort_sweep", async () => {
|
||||
const { companyId, environmentId } = await seedCompanyAndEnvironment();
|
||||
|
||||
// The array-root lease sorts first. A failed `jsonb_set` would abort before
|
||||
// the sweep reaches the healthy lease.
|
||||
const arrayLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 120 * 60 * 1000),
|
||||
});
|
||||
await setRawMetadataRoot(arrayLeaseId, "[1, 2, 3]");
|
||||
const healthyLeaseId = await insertPendingCleanupLease({
|
||||
companyId,
|
||||
environmentId,
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||
});
|
||||
|
||||
// The destroy succeeds and moves each lease out of pending_cleanup.
|
||||
const destroyedLeaseIds: string[] = [];
|
||||
const destroyRunLease = vi.fn(async ({ lease }: { lease: { id: string } }) => {
|
||||
destroyedLeaseIds.push(lease.id);
|
||||
const now = new Date();
|
||||
const row = await db
|
||||
.update(environmentLeases)
|
||||
.set({ status: "expired", cleanupStatus: "success", updatedAt: now })
|
||||
.where(eq(environmentLeases.id, lease.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return row ? { ...row, status: "expired" as const } : null;
|
||||
});
|
||||
const heartbeat = heartbeatService(db, {
|
||||
environmentRuntime: fakeRuntime(
|
||||
destroyRunLease as unknown as HeartbeatEnvironmentRuntime["destroyRunLease"],
|
||||
),
|
||||
});
|
||||
|
||||
// The sweep resolves; the array root does not break the claim write.
|
||||
const result = await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(result.swept).toBe(2);
|
||||
expect(result.destroyed).toBe(2);
|
||||
expect(destroyedLeaseIds).toContain(arrayLeaseId);
|
||||
expect(destroyedLeaseIds).toContain(healthyLeaseId);
|
||||
|
||||
// The claim replaced the array root with an object that holds one attempt.
|
||||
const arrayMetadata = await readMetadata(arrayLeaseId);
|
||||
expect(arrayMetadata?.[ATTEMPTS_KEY]).toBe(1);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,52 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { getEnvironmentCapabilities } from "@paperclipai/shared";
|
||||
import { getSandboxProvider } from "../services/sandbox-provider-runtime.ts";
|
||||
|
||||
// The execution guard in environment-runtime.ts admits a reusable lease only
|
||||
// when the provider declares supportsReusableLeases === true. This helper
|
||||
// mirrors that guard, so the tests can compare the presentation default with
|
||||
// the execution decision.
|
||||
function executionSupportsReusableLeases(declared: boolean | undefined): boolean {
|
||||
return declared === true;
|
||||
}
|
||||
|
||||
describe("reusable leases default", () => {
|
||||
it("test_fake_provider_reusable_leases_matches_runtime_declaration", () => {
|
||||
const fakeRuntime = getSandboxProvider("fake");
|
||||
expect(fakeRuntime).not.toBeNull();
|
||||
expect(fakeRuntime?.supportsReusableLeases).toBe(false);
|
||||
|
||||
const capabilities = getEnvironmentCapabilities(["codex_local"]);
|
||||
expect(capabilities.sandboxProviders.fake.supportsReusableLeases).toBe(
|
||||
fakeRuntime?.supportsReusableLeases ?? false,
|
||||
);
|
||||
});
|
||||
|
||||
it("test_api_presentation_agrees_with_runtime_execution_for_reusable_leases", () => {
|
||||
// Built-in fake provider: the runtime declares false, so execution and
|
||||
// presentation both resolve to false.
|
||||
const fakeRuntime = getSandboxProvider("fake");
|
||||
const builtin = getEnvironmentCapabilities(["codex_local"]);
|
||||
expect(builtin.sandboxProviders.fake.supportsReusableLeases).toBe(
|
||||
executionSupportsReusableLeases(fakeRuntime?.supportsReusableLeases),
|
||||
);
|
||||
|
||||
// Plug-in provider whose manifest does not declare reusable leases.
|
||||
const absent = getEnvironmentCapabilities(["codex_local"], {
|
||||
sandboxProviders: { "no-reuse-plugin": { displayName: "No Reuse" } },
|
||||
});
|
||||
expect(absent.sandboxProviders["no-reuse-plugin"].supportsReusableLeases).toBe(
|
||||
executionSupportsReusableLeases(undefined),
|
||||
);
|
||||
|
||||
// Plug-in provider whose manifest declares reusable leases.
|
||||
const declared = getEnvironmentCapabilities(["codex_local"], {
|
||||
sandboxProviders: {
|
||||
"reuse-plugin": { displayName: "Reuse", supportsReusableLeases: true },
|
||||
},
|
||||
});
|
||||
expect(declared.sandboxProviders["reuse-plugin"].supportsReusableLeases).toBe(
|
||||
executionSupportsReusableLeases(true),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
@ -715,7 +715,9 @@ export function environmentRoutes(
|
|||
supportsSavedProbe: true,
|
||||
supportsUnsavedProbe: true,
|
||||
supportsRunExecution: true,
|
||||
supportsReusableLeases: driver.supportsReusableLeases ?? true,
|
||||
// Default absent to false, so the presentation agrees with the
|
||||
// execution guard (=== true).
|
||||
supportsReusableLeases: driver.supportsReusableLeases ?? false,
|
||||
supportsInteractiveSetup: driver.supportsInteractiveSetup,
|
||||
interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes,
|
||||
supportsTemplateCapture: driver.supportsTemplateCapture,
|
||||
|
|
|
|||
|
|
@ -561,7 +561,11 @@ export function environmentRunOrchestrator(
|
|||
|
||||
let releasedLeases: EnvironmentRuntimeLeaseRecord[];
|
||||
try {
|
||||
releasedLeases = await environmentRuntime.releaseRunLeases(input.heartbeatRunId, status);
|
||||
releasedLeases = await environmentRuntime.releaseRunLeases(
|
||||
input.heartbeatRunId,
|
||||
status,
|
||||
(leaseId, error) => result.errors.push({ leaseId, error }),
|
||||
);
|
||||
} catch (err) {
|
||||
result.errors.push({ leaseId: "*", error: err });
|
||||
return result;
|
||||
|
|
|
|||
|
|
@ -1968,6 +1968,7 @@ export function environmentRuntimeService(
|
|||
async releaseRunLeases(
|
||||
heartbeatRunId: string,
|
||||
status: Extract<EnvironmentLeaseStatus, "released" | "expired" | "failed"> = "released",
|
||||
onLeaseReleaseError?: (leaseId: string, error: unknown) => void,
|
||||
): Promise<EnvironmentRuntimeLeaseRecord[]> {
|
||||
const leaseRows = await db
|
||||
.select()
|
||||
|
|
@ -1982,31 +1983,39 @@ export function environmentRuntimeService(
|
|||
return [];
|
||||
}
|
||||
|
||||
// Release each lease in its own try/catch. One driver error must not stop
|
||||
// the release of the later leases. The caller records each lease-specific
|
||||
// error through `onLeaseReleaseError` for its log path. Keep the order
|
||||
// serial.
|
||||
const released: EnvironmentRuntimeLeaseRecord[] = [];
|
||||
for (const leaseRow of leaseRows) {
|
||||
const environment = await environmentsSvc.getById(leaseRow.environmentId);
|
||||
if (!environment) continue;
|
||||
try {
|
||||
const environment = await environmentsSvc.getById(leaseRow.environmentId);
|
||||
if (!environment) continue;
|
||||
|
||||
const leaseSnapshot = toEnvironmentLeaseSnapshot(leaseRow);
|
||||
const driver = getDriver(getLeaseDriverKey(leaseSnapshot, environment));
|
||||
const lease = driver
|
||||
? await driver.releaseRunLease({
|
||||
environment,
|
||||
lease: leaseSnapshot,
|
||||
status,
|
||||
})
|
||||
: await environmentsSvc.releaseLease(leaseRow.id, status);
|
||||
if (!lease) continue;
|
||||
const leaseSnapshot = toEnvironmentLeaseSnapshot(leaseRow);
|
||||
const driver = getDriver(getLeaseDriverKey(leaseSnapshot, environment));
|
||||
const lease = driver
|
||||
? await driver.releaseRunLease({
|
||||
environment,
|
||||
lease: leaseSnapshot,
|
||||
status,
|
||||
})
|
||||
: await environmentsSvc.releaseLease(leaseRow.id, status);
|
||||
if (!lease) continue;
|
||||
|
||||
released.push({
|
||||
environment,
|
||||
lease,
|
||||
leaseContext: {
|
||||
executionWorkspaceId: lease.executionWorkspaceId,
|
||||
executionWorkspaceMode:
|
||||
(lease.metadata?.executionWorkspaceMode as ExecutionWorkspace["mode"] | null | undefined) ?? null,
|
||||
},
|
||||
});
|
||||
released.push({
|
||||
environment,
|
||||
lease,
|
||||
leaseContext: {
|
||||
executionWorkspaceId: lease.executionWorkspaceId,
|
||||
executionWorkspaceMode:
|
||||
(lease.metadata?.executionWorkspaceMode as ExecutionWorkspace["mode"] | null | undefined) ?? null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
onLeaseReleaseError?.(leaseRow.id, error);
|
||||
}
|
||||
}
|
||||
|
||||
return released;
|
||||
|
|
|
|||
|
|
@ -44,6 +44,7 @@ import {
|
|||
documentAnnotationComments,
|
||||
documentAnnotationThreads,
|
||||
documentRevisions,
|
||||
environmentLeases,
|
||||
issueDocuments,
|
||||
executionWorkspaces,
|
||||
heartbeatRunEvents,
|
||||
|
|
@ -356,6 +357,67 @@ const PAPERCLIP_WAKE_PAYLOAD_KEY = "paperclipWake";
|
|||
const PAPERCLIP_AGENT_MESSAGE_KEY = "paperclipAgentMessage";
|
||||
const PAPERCLIP_HARNESS_CHECKOUT_KEY = "paperclipHarnessCheckedOut";
|
||||
const DETACHED_PROCESS_ERROR_CODE = "process_detached";
|
||||
// The reaper sweeps at most this many pending_cleanup leases per tick.
|
||||
const PENDING_CLEANUP_SWEEP_PAGE_SIZE = 20;
|
||||
// The reaper stops retrying a pending_cleanup lease after this many attempts.
|
||||
const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5;
|
||||
// The reaper stores its retry state under these keys in the lease metadata.
|
||||
const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts";
|
||||
const PENDING_CLEANUP_CAP_WARNED_METADATA_KEY = "pendingCleanupRetryCapWarned";
|
||||
|
||||
// A provider or plugin destroy rejection can carry a bearer credential, a
|
||||
// signed URL, or provider response detail in its name, code, message, cause, or
|
||||
// stack. The exception fields cross the server boundary, so they are not a
|
||||
// trusted enum. The pending_cleanup sweep logs never read the exception. Each
|
||||
// catch site logs a constant, locally generated `errorKind` instead.
|
||||
const PENDING_CLEANUP_RETRY_ERROR_KIND = "destroy_failed";
|
||||
const PENDING_CLEANUP_SWEEP_ERROR_KIND = "sweep_failed";
|
||||
|
||||
// Read the stored retry attempt count as a safe value, directly in SQL. A
|
||||
// provider can write a malformed value under the attempts key. The type guard
|
||||
// makes any non-number value read as zero. The reader computes as numeric and
|
||||
// never casts to int, so a finite number outside the 32-bit range (for example
|
||||
// 1e300) never throws. The reader clamps a negative value to zero and a positive
|
||||
// value to the attempt cap. One malformed lease therefore never aborts the page
|
||||
// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`,
|
||||
// which clamps to the same range. The claim predicate compares the two readers,
|
||||
// so both must yield the same value for every input.
|
||||
function pendingCleanupAttemptsSql() {
|
||||
return sql`
|
||||
case
|
||||
when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number'
|
||||
then least(
|
||||
greatest(
|
||||
floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric),
|
||||
0
|
||||
),
|
||||
${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}
|
||||
)
|
||||
else 0
|
||||
end`;
|
||||
}
|
||||
|
||||
// Choose the `jsonb_set` target root. A provider can write a scalar or array
|
||||
// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the
|
||||
// stored metadata only when its root is an object. A NULL, scalar, or array root
|
||||
// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also
|
||||
// covers a NULL root.
|
||||
function pendingCleanupMetadataObjectSql() {
|
||||
return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`;
|
||||
}
|
||||
|
||||
// Read the stored cap-warned flag as a safe boolean, directly in SQL. A
|
||||
// malformed value reads as false, so the boolean cast never throws.
|
||||
function pendingCleanupCapWarnedSql() {
|
||||
return sql`coalesce(
|
||||
case
|
||||
when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY}) = 'boolean'
|
||||
then (${environmentLeases.metadata} ->> ${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY})::boolean
|
||||
else false
|
||||
end,
|
||||
false
|
||||
)`;
|
||||
}
|
||||
const REPO_ONLY_CWD_SENTINEL = "/__paperclip_repo_only__";
|
||||
const MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS = 10 * 60 * 1000;
|
||||
const MAX_INLINE_WAKE_COMMENTS = 8;
|
||||
|
|
@ -13123,6 +13185,170 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
// Clamp the stored attempt count to the range [0, cap]. The SQL reader
|
||||
// `pendingCleanupAttemptsSql` clamps to the same range, so both readers yield
|
||||
// the same value for every input. The claim predicate compares the two values,
|
||||
// so this alignment lets the claim match for a malformed lease.
|
||||
function readPendingCleanupRetryAttempts(metadata: Record<string, unknown>): number {
|
||||
const value = metadata[PENDING_CLEANUP_ATTEMPTS_METADATA_KEY];
|
||||
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) return 0;
|
||||
return Math.min(Math.floor(value), PENDING_CLEANUP_SWEEP_ATTEMPT_CAP);
|
||||
}
|
||||
|
||||
// Atomically claim one retry attempt on a pending_cleanup lease. The update
|
||||
// only matches when the lease is still pending_cleanup and its stored attempt
|
||||
// count still equals `expectedAttempts`. Two concurrent sweeps read the same
|
||||
// count, but Postgres serializes the two updates on the row and only the first
|
||||
// matches the guard. The loser gets zero rows and skips the lease. This bounds
|
||||
// the retries to the cap and stops a second destroy of the same lease.
|
||||
// Returns true only for the sweep that won the claim.
|
||||
//
|
||||
// The update writes only the attempts key with `jsonb_set`. It never writes a
|
||||
// copied metadata object, so a concurrent write to an unrelated metadata key
|
||||
// survives. The guard reads the stored count through the safe SQL reader, so a
|
||||
// malformed value never throws.
|
||||
async function claimPendingCleanupRetryAttempt(
|
||||
leaseId: string,
|
||||
expectedAttempts: number,
|
||||
): Promise<boolean> {
|
||||
const now = new Date();
|
||||
const claimed = await db
|
||||
.update(environmentLeases)
|
||||
.set({
|
||||
metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], to_jsonb(${expectedAttempts + 1}::int), true)`,
|
||||
lastUsedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(environmentLeases.id, leaseId),
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${pendingCleanupAttemptsSql()} = ${expectedAttempts}`,
|
||||
),
|
||||
)
|
||||
.returning({ id: environmentLeases.id });
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
// Atomically claim the one-time cap warning for a lease. The update only
|
||||
// matches when the lease is still pending_cleanup, its stored attempt count is
|
||||
// at or above the cap, and it has not yet carried the warned flag. Two
|
||||
// concurrent sweeps that both reach the cap race here, but only one update
|
||||
// sets the flag and returns a row. The loser skips the warning. This keeps the
|
||||
// warning to one log line per lease.
|
||||
//
|
||||
// The status and cap predicates are the last line of defense. They stop a warn
|
||||
// flag write to a lease that left pending_cleanup or dropped below the cap
|
||||
// between the read and this claim. The update writes only the warned key with
|
||||
// `jsonb_set`, so a concurrent write to an unrelated metadata key survives.
|
||||
async function claimPendingCleanupCapWarning(leaseId: string): Promise<boolean> {
|
||||
const now = new Date();
|
||||
const claimed = await db
|
||||
.update(environmentLeases)
|
||||
.set({
|
||||
metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_CAP_WARNED_METADATA_KEY}], to_jsonb(true), true)`,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(environmentLeases.id, leaseId),
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${pendingCleanupAttemptsSql()} >= ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}`,
|
||||
sql`${pendingCleanupCapWarnedSql()} = false`,
|
||||
),
|
||||
)
|
||||
.returning({ id: environmentLeases.id });
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
// Retry the leases stranded in "pending_cleanup". A failed destroy leaves a
|
||||
// lease in that state forever without this sweep. The reaper tick runs the
|
||||
// sweep. The backoff equals the reaper staleness threshold, so a lease waits
|
||||
// for that period between attempts. The sweep reads and writes the attempt
|
||||
// count in the lease metadata. It warns once when a lease reaches the attempt
|
||||
// cap and then stops the retries for that lease.
|
||||
async function sweepPendingCleanupLeases(opts?: { backoffMs?: number }): Promise<{
|
||||
swept: number;
|
||||
destroyed: number;
|
||||
capped: number;
|
||||
}> {
|
||||
const backoffMs = opts?.backoffMs ?? 0;
|
||||
const now = new Date();
|
||||
const cutoff = new Date(now.getTime() - backoffMs);
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(environmentLeases)
|
||||
.where(
|
||||
and(
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
backoffMs > 0 ? lte(environmentLeases.updatedAt, cutoff) : undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(environmentLeases.updatedAt))
|
||||
.limit(PENDING_CLEANUP_SWEEP_PAGE_SIZE);
|
||||
|
||||
let destroyed = 0;
|
||||
let capped = 0;
|
||||
for (const row of rows) {
|
||||
const metadata = { ...(row.metadata ?? {}) } as Record<string, unknown>;
|
||||
const attempts = readPendingCleanupRetryAttempts(metadata);
|
||||
|
||||
if (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) {
|
||||
capped += 1;
|
||||
// Warn once, then leave the lease for manual cleanup. The atomic claim
|
||||
// keeps the warning to one log line even when two sweeps overlap.
|
||||
if (metadata[PENDING_CLEANUP_CAP_WARNED_METADATA_KEY] !== true) {
|
||||
const warned = await claimPendingCleanupCapWarning(row.id);
|
||||
if (warned) {
|
||||
logger.warn(
|
||||
{ leaseId: row.id, environmentId: row.environmentId, attempts },
|
||||
"environment lease reached the pending_cleanup retry cap; left for manual cleanup",
|
||||
);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const environment = await environmentsSvc.getById(row.environmentId);
|
||||
const lease = await environmentsSvc.getLeaseById(row.id);
|
||||
if (!environment || !lease) continue;
|
||||
|
||||
// Atomically claim the attempt before the retry. Only the winning sweep
|
||||
// increments the count and destroys the lease, so an overlapping sweep
|
||||
// never destroys the same lease twice or exceeds the attempt cap. The
|
||||
// claim records the attempt before the retry, so a thrown driver error
|
||||
// still counts against the cap.
|
||||
const claimed = await claimPendingCleanupRetryAttempt(row.id, attempts);
|
||||
if (!claimed) continue;
|
||||
|
||||
try {
|
||||
const result = await environmentRuntime.destroyRunLease({
|
||||
environment,
|
||||
lease,
|
||||
failureReason: "pending_cleanup_retry",
|
||||
});
|
||||
if (result && result.status !== "pending_cleanup") {
|
||||
destroyed += 1;
|
||||
}
|
||||
} catch {
|
||||
// Log a constant errorKind only. The exception can carry a credential in
|
||||
// its name, code, message, cause, or stack, so the sweep never reads it.
|
||||
logger.warn(
|
||||
{
|
||||
errorKind: PENDING_CLEANUP_RETRY_ERROR_KIND,
|
||||
leaseId: row.id,
|
||||
environmentId: row.environmentId,
|
||||
attempts: attempts + 1,
|
||||
},
|
||||
"pending_cleanup lease retry failed",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { swept: rows.length, destroyed, capped };
|
||||
}
|
||||
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) {
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
|
|
@ -13316,6 +13542,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
if (reaped.length > 0) {
|
||||
logger.warn({ reapedCount: reaped.length, runIds: reaped }, "reaped orphaned heartbeat runs");
|
||||
}
|
||||
|
||||
// Retry stranded pending_cleanup leases on the same tick. Isolate the sweep
|
||||
// so its failure never hides the reaper result. The backoff equals the
|
||||
// reaper staleness threshold.
|
||||
try {
|
||||
const sweep = await sweepPendingCleanupLeases({ backoffMs: staleThresholdMs });
|
||||
if (sweep.destroyed > 0 || sweep.capped > 0) {
|
||||
logger.warn(
|
||||
{ destroyed: sweep.destroyed, capped: sweep.capped, swept: sweep.swept },
|
||||
"swept pending_cleanup environment leases",
|
||||
);
|
||||
}
|
||||
} catch {
|
||||
// Log a constant errorKind only. The exception can carry a credential in
|
||||
// its name, code, message, cause, or stack, so the sweep never reads it.
|
||||
logger.error(
|
||||
{ errorKind: PENDING_CLEANUP_SWEEP_ERROR_KIND },
|
||||
"pending_cleanup lease sweep failed",
|
||||
);
|
||||
}
|
||||
|
||||
return { reaped: reaped.length, runIds: reaped };
|
||||
}
|
||||
|
||||
|
|
@ -18992,6 +19239,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
prepareHotRestartShutdown,
|
||||
reconcileHotRestartAdoption,
|
||||
reapOrphanedRuns,
|
||||
sweepPendingCleanupLeases,
|
||||
// Override-aware scheduling-suppression check (honors the worktree
|
||||
// run-execution experimental setting). Callers outside the service that
|
||||
// gate on suppression should prefer this over the env-only resolver.
|
||||
|
|
|
|||
Loading…
Reference in New Issue