From 940d681974eaa6e5f83c3ccf377bbd19c8b69fee Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:53:54 -0600 Subject: [PATCH] fix(heartbeat): hold a queued native run while its task's previous session finishes cleanup (#21031) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runner runs keep a provider session per task. When a run stops, its session close continues in the server process > - The runner refuses every new session of the task until that close settles. A close that cannot prove the old runner suspended needs a stopped-process proof before it is released > - A reassignment stops the old assignee's run mid-turn. The new assignee's run was dispatched one second later into that gate, and it failed with `native_session_cleanup_quarantined` > - Only a Retry continued the task. Recovery must work without a person, and the needed-retry rate must be 0% > - This pull request keeps such a run queued until the close settles or its old owner is proven stopped > - The benefit is that the new run starts by itself after cleanup, without a failed run or a Retry ## Linked Issues or Issue Description **What happened?** In the paid chat eval `tasks.edit-properties` (and `tasks.assignment-conflict` in an earlier cohort), the CEO reassigned a task while the old assignee's native run was mid-turn, with a Paperclip tool call in flight. The reassignment cancelled that run. Its session close then failed with `native_session_close_unrecoverable: runner did not durably suspend before checkpoint`, and the runner kept an operator-required cleanup gate for the task. The new assignee's run had started 1 second after the cancel. The runner's admission waited for the in-flight close. When the close escalated, admission refused the run with `native_session_cleanup_quarantined`. No run was queued again. **Steps to reproduce** 1. Run a native task and stop its run while its session close cannot settle at once. 2. Queue a run of another agent for the same task. 3. The run is dispatched into the cleanup gate and fails before it launches. **Expected behavior** The run waits for the cleanup, and starts once the close settles or the old owner is proven stopped. It never fails, and no Retry is needed. ## What Changed - `server/src/services/native-runtime/native-cleanup-admission-hold.ts`: `nativeCleanupHoldsAdmission` reads the task's cleanup gate. For an operator-required gate it first tries `retireStoppedSessionCleanupGates`, which retires the gate when the old owner's runner and provider processes are proven stopped. This is the same proof that Retry uses. - `server/src/services/heartbeat/queue.ts`: `claimQueuedRun` keeps an unstarted `paperclip_runner` run queued while its task has a settling or unproven close. This matches the existing `predecessor_settling_in_process` hold, which covers only the same agent. A reassignment changes the agent. - `server/src/services/native-runtime/native-cleanup-release-wake.ts`: a released close now also wakes queued-run dispatch. A close that escalates to operator-required also wakes it, so admission tries the stopped-process proof at once. The queued-run liveness redrive covers any later proof. - `server/src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts`: an assembled heartbeat test. It is red on master (the run is claimed into a pending close and into an operator-required close). It is green here (the run stays queued, then is admitted after the close settles). A control case shows that a run for another task is not held. ## Verification - `cd server && npx vitest run src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts` (3 passed; 2 red on master) - `cd server && npx vitest run src/services/quarantined-task-retry.test.ts src/services/heartbeat/queue.test.ts src/__tests__/heartbeat-workspace-busy.test.ts src/services/queued-run-dispatch-liveness.test.ts` (180 passed) - `cd server && npx tsc --noEmit` passes. - Paid E2E run on a combined branch: https://github.com/paperclipai/paperclip/actions/runs/43161593494 (results in a comment below). ## Risks - A run whose task has an operator-required gate that is never proven stopped now stays queued instead of failing. Before, that run failed, and Retry was blocked by the same proof. The liveness redrive re-checks it with back-off. - The hold applies only to unstarted `paperclip_runner` runs of the gated task. Other tasks and agents are not affected. - Previous-release saved state: no persisted shape changes. The gate lives in process memory, and runs saved by the previous release are claimed by the same rule. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- ...beat-native-cleanup-admission-hold.test.ts | 178 ++++++++++++++++++ server/src/services/heartbeat/queue.ts | 10 + .../native-cleanup-admission-hold.ts | 15 ++ .../native-cleanup-release-wake.ts | 15 +- 4 files changed, 215 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts create mode 100644 server/src/services/native-runtime/native-cleanup-admission-hold.ts diff --git a/server/src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts b/server/src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts new file mode 100644 index 00000000000..3bcdfa8b9cb --- /dev/null +++ b/server/src/__tests__/heartbeat-native-cleanup-admission-hold.test.ts @@ -0,0 +1,178 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { hostname } from "node:os"; +import { agents, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { + executeNativeSession, + NativeSessionCloseUnrecoverableError, + nativeSessionCleanupBlock, + parseNativeExecutionInput, + type NativeSession, +} from "../vendor/paperclip-runner/index.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; +import { registerNativeCleanupReleaseWake } from "../services/native-runtime/native-cleanup-release-wake.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +describeEmbeddedPostgres("queued native run behind the task's previous session cleanup", () => { + let tempDb: Awaited>; + let db: ReturnType; + let base: string; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-native-cleanup-hold-"); + db = createDb(tempDb.connectionString); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true }); + base = await mkdtemp(join(tmpdir(), "native-cleanup-hold-")); + }, 90_000); + + afterAll(async () => { + await db?.$client.end(); + await tempDb?.cleanup(); + if (base) await rm(base, { recursive: true, force: true }); + }); + + async function seed() { + const companyId = randomUUID(), previousId = randomUUID(), nextId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Cleanup Hold", defaultResponsibleUserId: "board", + issuePrefix: `H${companyId.replaceAll("-", "").slice(0, 8)}` }); + await db.insert(agents).values([previousId, nextId].map(id => ({ id, companyId, name: `Worker ${id.slice(0, 4)}`, + adapterType: "paperclip_runner", status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }))); + await db.insert(issues).values({ id: issueId, companyId, title: "Edit the task", status: "todo", assigneeAgentId: nextId }); + return { companyId, previousId, nextId, issueId }; + } + + async function retainPreviousSessionClose(f: Awaited>, close: () => Promise, + owner: { runId: string; runnerInstanceId: string } = { runId: randomUUID(), runnerInstanceId: randomUUID() }) { + const runId = owner.runId; + const identity = { runId, sessionId: `closing-${runId}`, companyId: f.companyId, issueId: f.issueId, agentId: f.previousId }; + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true }; + const session: NativeSession = { + identity: () => identity, capabilities: async () => capabilities, + async *events() { throw new Error("stopped for reassignment"); }, + startTurn: async () => ({ turnId: "previous-turn" }), result: async () => null, + async snapshot() { return { backendKind: "mock", sessionId: "previous-driver", identity, + providerSessionId: "previous-provider", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; }, + close, + }; + await expect(executeNativeSession({ + input: parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId: f.companyId, agentId: f.previousId, issueId: f.issueId, runId, executionWorkspaceId: runId }, + task: { identifier: "TEST", title: "Edit the task", description: null, prompt: "Edit the task", workMode: "standard" }, + workspace: { cwd: base, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: "contract", sha256: "sha", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Edit the task", criteria: [{ id: "done", requirement: "Save the edit" }] } }, + interactionResponses: [], credentialBindings: [] }), + backend: { descriptor: async () => ({ kind: "runner", name: "reassigned-backend", version: "1", capabilities }), + openSession: async () => session }, + controlPlane: { openRun: async () => {}, checkpointSession: async () => {}, + appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }), + replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }), completeRun: async () => {} }, + runnerInstanceId: owner.runnerInstanceId, controlPlaneInstanceId: "control", + })).rejects.toThrow("stopped for reassignment"); + } + + async function queueNextAssigneeRun(f: Awaited>) { + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.nextId, + invocationSource: "assignment", triggerDetail: "system", status: "queued", + contextSnapshot: { issueId: f.issueId, wakeReason: "issue_assigned" } }).returning(); + return run!; + } + + const readRun = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + + function dispatchWorker() { + const heartbeat = heartbeatService(db); + const wakeFailure = vi.fn(); + const unregisterWake = registerNativeCleanupReleaseWake(db, wakeFailure); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.queuedRunDispatch, () => { + void heartbeat.resumeQueuedRuns(); + }); + return { wakeFailure, stop: () => { unsubscribe(); unregisterWake(); } }; + } + + async function recordStoppedOwner(f: Awaited>, owner: { runId: string; runnerInstanceId: string }) { + await db.insert(heartbeatRuns).values({ id: owner.runId, companyId: f.companyId, agentId: f.previousId, + invocationSource: "assignment", status: "cancelled", errorCode: "issue_reassigned", runtimeMode: "native", + nativeIssueId: f.issueId, runnerInstanceId: owner.runnerInstanceId, startedAt: new Date(Date.now() - 60_000), + finishedAt: new Date(), contextSnapshot: { issueId: f.issueId } }); + const processStartedAt = new Date(Date.now() - 60_000).toISOString(); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: owner.runId, agentId: f.previousId, + eventType: "native.local_process_stopped", + payload: { processPid: 99_999_993, processGroupId: 99_999_993, processStartedAt, hostname: hostname() } }); + } + + it("starts the new assignee's run by itself once the previous close settles", async () => { + const worker = dispatchWorker(); + try { + const f = await seed(); + let settle = () => {}; + await retainPreviousSessionClose(f, () => new Promise(resolve => { settle = resolve; })); + expect(nativeSessionCleanupBlock({ companyId: f.companyId, issueId: f.issueId })).toBe("pending"); + + const run = await queueNextAssigneeRun(f); + await heartbeatService(db).resumeQueuedRuns(); + expect(await readRun(run.id)).toMatchObject({ status: "queued", startedAt: null }); + + settle(); + await expect.poll(async () => (await readRun(run.id)).status, { timeout: 15_000 }).not.toBe("queued"); + expect((await readRun(run.id)).errorCode).not.toBe("native_session_cleanup_quarantined"); + expect(worker.wakeFailure).not.toHaveBeenCalled(); + } finally { + worker.stop(); + } + }); + + it("starts the new assignee's run by itself when an unrecoverable close's owner is proven stopped", async () => { + const worker = dispatchWorker(); + try { + const f = await seed(); + const owner = { runId: randomUUID(), runnerInstanceId: randomUUID() }; + await recordStoppedOwner(f, owner); + const run = await queueNextAssigneeRun(f); + await retainPreviousSessionClose(f, async () => { + throw new NativeSessionCloseUnrecoverableError({ runnerSuspended: true, providerDrained: false }); + }, owner); + await expect.poll(async () => (await readRun(run.id)).status, { timeout: 15_000 }).not.toBe("queued"); + expect((await readRun(run.id)).errorCode).not.toBe("native_session_cleanup_quarantined"); + expect(nativeSessionCleanupBlock({ companyId: f.companyId, issueId: f.issueId })).toBeNull(); + } finally { + worker.stop(); + } + }); + + it("keeps the run queued while an unrecoverable close's owner is not proven stopped", async () => { + const f = await seed(); + await retainPreviousSessionClose(f, async () => { + throw new NativeSessionCloseUnrecoverableError({ runnerSuspended: true, providerDrained: false }); + }); + expect(nativeSessionCleanupBlock({ companyId: f.companyId, issueId: f.issueId })).toBe("operator_required"); + + const run = await queueNextAssigneeRun(f); + await heartbeatService(db).resumeQueuedRuns(); + expect(await readRun(run.id)).toMatchObject({ status: "queued", startedAt: null, errorCode: null }); + }); + + it("does not hold a run for a different task", async () => { + const f = await seed(); + await retainPreviousSessionClose(f, () => new Promise(() => {})); + const other = await seed(); + const run = await queueNextAssigneeRun(other); + await heartbeatService(db).resumeQueuedRuns(); + await expect.poll(async () => (await readRun(run.id)).status).not.toBe("queued"); + }); +}); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 307edfc74d9..8378eea4404 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -69,6 +69,7 @@ import { readNativeReviewAssignmentContext, } from "../native-runtime/native-review-participant.js"; import { claimQueuedNativeReviewRun } from "../native-runtime/native-review-dispatch.js"; +import { nativeCleanupHoldsAdmission } from "../native-runtime/native-cleanup-admission-hold.js"; import { adapterExecutionControls } from "../adapter-execution-control.js"; import { preserveTaskRemoteWorkspaceWriter } from "../native-runtime/native-failed-workspace-recovery.js"; import { executionFailureRetryCount } from "../execution-recovery-attempt.js"; @@ -1066,6 +1067,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return null; } } + if (issueId && !run.startedAt && agent.adapterType === "paperclip_runner") { + const cleanupBlock = await nativeCleanupHoldsAdmission(db, { companyId: run.companyId, issueId }); + if (cleanupBlock) { + logger.info({ runId: run.id, agentId: run.agentId, companyId: run.companyId, issueId, + reason: "native_session_cleanup_settling", cleanupBlock }, + "claimQueuedRun: queued run kept queued while the task's previous session finishes cleanup"); + return null; + } + } if (issueId) { const activePauseHold = await treeControlSvc.getActivePauseHoldGate( run.companyId, diff --git a/server/src/services/native-runtime/native-cleanup-admission-hold.ts b/server/src/services/native-runtime/native-cleanup-admission-hold.ts new file mode 100644 index 00000000000..56581df8453 --- /dev/null +++ b/server/src/services/native-runtime/native-cleanup-admission-hold.ts @@ -0,0 +1,15 @@ +import type { Db } from "@paperclipai/db"; +import { nativeSessionCleanupBlock } from "../../vendor/paperclip-runner/index.js"; + +export async function nativeCleanupHoldsAdmission( + db: Db, + input: { companyId: string; issueId: string }, +): Promise<"pending" | "operator_required" | null> { + let block = nativeSessionCleanupBlock(input); + if (block === "operator_required") { + const { retireStoppedSessionCleanupGates } = await import("./native-session-executor.js"); + await retireStoppedSessionCleanupGates(db, input).catch(() => false); + block = nativeSessionCleanupBlock(input); + } + return block; +} diff --git a/server/src/services/native-runtime/native-cleanup-release-wake.ts b/server/src/services/native-runtime/native-cleanup-release-wake.ts index 8d23f85b240..fe258495ed9 100644 --- a/server/src/services/native-runtime/native-cleanup-release-wake.ts +++ b/server/src/services/native-runtime/native-cleanup-release-wake.ts @@ -1,6 +1,7 @@ import type { Db } from "@paperclipai/db"; import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js"; -import { onNativeSessionCleanupSettled } from "../../vendor/paperclip-runner/index.js"; +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; +import { onNativeSessionCleanupQuarantined, onNativeSessionCleanupSettled } from "../../vendor/paperclip-runner/index.js"; /** * A retained native session close holds task input in the runner's process @@ -9,9 +10,17 @@ import { onNativeSessionCleanupSettled } from "../../vendor/paperclip-runner/ind * dispatch repeats every admission check under its own locks. */ export function registerNativeCleanupReleaseWake(db: Db, onError: (error: unknown, companyId: string) => void): () => void { - return onNativeSessionCleanupSettled(({ companyId }) => { + const unsubscribeSettled = onNativeSessionCleanupSettled(({ companyId }) => { if (!companyId) return; - void db.transaction(async tx => notifyDeferredExecutionWork(tx, { companyId })) + void db.transaction(async tx => { + await notifyDeferredExecutionWork(tx, { companyId }); + await notifyQueuedRunDispatch(tx, { companyId }); + }).catch((error: unknown) => onError(error, companyId)); + }); + const unsubscribeQuarantined = onNativeSessionCleanupQuarantined(({ companyId, operatorRecoveryRequired }) => { + if (!companyId || !operatorRecoveryRequired) return; + void db.transaction(async tx => notifyQueuedRunDispatch(tx, { companyId })) .catch((error: unknown) => onError(error, companyId)); }); + return () => { unsubscribeSettled(); unsubscribeQuarantined(); }; }