mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
fix(heartbeat): hold a queued native run while its task's previous session finishes cleanup (#21031)
## 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
ebaffddb01
commit
940d681974
4 files changed
+215
-3
No files matched your search
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
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<ReturnType<typeof seed>>, close: () => Promise<void>,
|
||||
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<ReturnType<typeof seed>>) {
|
||||
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<ReturnType<typeof seed>>, 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<void>(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<void>(() => {}));
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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(); };
|
||||
}
|
||||
Reference in new issue
Block a user