From deef1f479db4219a5aef596402c4e1d3a3739a03 Mon Sep 17 00:00:00 2001 From: Vladimir Balko <147043796+vbalko-claimate@users.noreply.github.com> Date: Fri, 12 Jun 2026 06:59:04 +0200 Subject: [PATCH] fix(heartbeat): release execution lock on cross-agent reassignment (#5110) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - Each issue can hold an execution lock via `issues.execution_run_id`, so concurrent wakes for the same task either coalesce into the active run or wait deferred > - When the issue is reassigned to a *different* agent (e.g. board operator changes `assigneeAgentId` from Coder → Reviewer + flips `status` to `in_review`), the new assignee's wake is correctly sent down the assignment-wakeup path > - But the lookup `activeExecutionRun` still finds the previous holder run as long as it is in `{queued, running, scheduled_retry}` — and `enqueueAssignmentWakeup` falls through to the deferred-wake branch when the holder agent does not match the new assignee > - The trouble is the **queued** holder for the old assignee will never start (the issue's status / target now belongs to someone else, the relevant assignment trigger was the original one), so the lock is never released, the deferred wake is never promoted, and the new assignee silently never wakes > - This pull request detects that situation right next to the existing `cancelStaleScheduledRetry` cleanup: if `activeExecutionRun.status !== 'running'` AND the holder agent differs from `issue.assigneeAgentId`, cancel the holder run, release the lock, and proceed with a normal queued wake instead of deferring > - The benefit is hand-offs across agents become reliable — no more silent stalls that operators have to unstick by manually cancelling a queued run ## Linked Issues or Issue Description - Closes #4058 ## What Changed - One new check in `reapOrphanedRuns()`'s peer function — the `enqueueAssignmentWakeup` defer-detection block in `server/src/services/heartbeat.ts` (around the lock-resolution code immediately following `cancelStaleScheduledRetry`): - If `activeExecutionRun` exists, its `status !== 'running'`, and `activeExecutionRun.agentId !== issue.assigneeAgentId`, mark the holder run as `cancelled` with errorCode `lock_released_on_reassignment`, cancel its corresponding wakeup request if any, and null `activeExecutionRun` so the lock-clear branch right below proceeds to release `executionRunId` / `executionAgentNameKey` / `executionLockedAt` and the wake gets enqueued normally. - `running` runs still defer (legitimate concurrency). - Same-agent queued/scheduled holders still defer (legitimate coalesce). - Total +37 lines, no API change, no schema change. ## Verification ```sh # Existing reaper tests still pass — exercises the lock-resolution path pnpm exec vitest run server/src/__tests__/heartbeat-process-recovery.test.ts --no-coverage # expected: Tests 39 passed (39) # New regression test for the cross-agent lock-release race pnpm exec vitest run server/src/__tests__/heartbeat-lock-release-on-reassignment.test.ts --no-coverage ``` Manual reproduction (matches an incident we hit running a small Coder + Reviewer company): 1. Coder pickup heartbeat schedule fires; paperclip queues a Coder run and pre-allocates the lock by recording `issues.execution_run_id = ` for the pickup issue. 2. The Coder run sits in `queued` because the agent's slot is busy elsewhere (`maxConcurrentRuns: 1`). 3. Operator (or CEO) PATCHes the issue: `assigneeAgentId: ` → `` together with `status: in_progress` → `in_review`. 4. Paperclip creates the Reviewer assignment wakeup, but stores it as `deferred_issue_execution` because `activeExecutionRun` is the queued Coder run. 5. **Before this PR**: Reviewer never wakes; the deferred wakeup waits for the queued Coder lock holder which never starts (the issue is no longer the Coder's). Operator has to `POST /api/heartbeat-runs//cancel` manually to unstick the chain. 6. **After this PR**: paperclip recognizes the holder is non-running and belongs to a now-foreign agent, cancels it inline, releases the lock, and queues the Reviewer wake normally — Reviewer wakes on the next heartbeat tick. ## Risks - **Low**. The new branch only fires when both conditions are true: - The holder run is **not** `running` — `running` runs still defer (we never interrupt active work). - `activeExecutionRun.agentId` is different from the issue's *current* `assigneeAgentId` — i.e. the assignee was just changed, the old holder is bound to the prior owner. - The cancel uses errorCode `lock_released_on_reassignment` so operators can grep for it; the corresponding wakeup is also cancelled in the same transaction so we do not leave an orphan wakeup request. - No DB schema change, no public API change, no UI change. - Sits next to the existing `cancelStaleScheduledRetry` cleanup pattern, so the behavior is locally consistent with how stale schedule retries are already cleared. ## Model Used - Claude Opus 4.7 (`claude-opus-4-7`), 1M-context build, extended thinking + tool use enabled. Used to trace the lock-acquire / defer / promote paths in `heartbeat.ts` from the live incident, design the minimal-blast-radius fix next to `cancelStaleScheduledRetry`, and produce this PR description. ## 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 run tests locally and they pass (39 in the directly affected suite, plus the new regression test) - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots — N/A (server-side wakeup routing) - [x] I have updated relevant documentation to reflect my changes — in-line code comment explains the new branch - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge ## Cross-references and status (maintainer) - `Closes #4058` ### Maintainer-added changes on top of the original commit A second commit was added on top of @vbalko-claimate's original to pin the cancel `UPDATE` for the queued/scheduled holder to the exact non-running status read just above it. Without that predicate, a worker that flipped the holder from `queued` → `running` between the `SELECT` and the `UPDATE` could have its freshly-claimed `running` row silently clobbered to `cancelled`. The new commit also gates the wakeup-request cancellation and the `activeExecutionRun = null` assignment on a non-empty `RETURNING`, so neither fires when the predicate misses. A dedicated regression test (`heartbeat-lock-release-on-reassignment.test.ts`) covers both paths: the legitimate-running-holder defer case and the queued→running race. --------- Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: Devin Foley --- ...tbeat-lock-release-on-reassignment.test.ts | 271 ++++++++++++++++++ server/src/services/heartbeat.ts | 53 ++++ 2 files changed, 324 insertions(+) create mode 100644 server/src/__tests__/heartbeat-lock-release-on-reassignment.test.ts diff --git a/server/src/__tests__/heartbeat-lock-release-on-reassignment.test.ts b/server/src/__tests__/heartbeat-lock-release-on-reassignment.test.ts new file mode 100644 index 00000000..167c4c18 --- /dev/null +++ b/server/src/__tests__/heartbeat-lock-release-on-reassignment.test.ts @@ -0,0 +1,271 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + agentWakeupRequests, + companies, + createDb, + heartbeatRunEvents, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres heartbeat lock-release-on-reassignment tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("heartbeat lock release on cross-agent reassignment", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-lock-release-on-reassignment-"); + db = createDb(tempDb.connectionString); + }, 60_000); + + afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedCrossAgentScenario(opts: { holderStatus: "queued" | "running" }) { + const companyId = randomUUID(); + const coderAgentId = randomUUID(); + const reviewerAgentId = randomUUID(); + const issueId = randomUUID(); + const holderRunId = randomUUID(); + const wakeupRequestId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: coderAgentId, + companyId, + name: "Coder", + role: "engineer", + status: "idle", + adapterType: "process", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: reviewerAgentId, + companyId, + name: "Reviewer", + role: "engineer", + status: "idle", + adapterType: "process", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId: coderAgentId, + source: "assignment", + status: "queued", + }); + + await db.insert(heartbeatRuns).values({ + id: holderRunId, + companyId, + agentId: coderAgentId, + invocationSource: "assignment", + triggerDetail: "system", + status: opts.holderStatus, + wakeupRequestId, + contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_assigned" }, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Cross-agent reassignment race", + status: "in_review", + priority: "medium", + assigneeAgentId: reviewerAgentId, + executionRunId: holderRunId, + executionAgentNameKey: "coder", + executionLockedAt: new Date(), + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + + return { + companyId, + coderAgentId, + reviewerAgentId, + issueId, + holderRunId, + wakeupRequestId, + }; + } + + it("defers a cross-agent wake while the holder is still running and leaves the holder alone", async () => { + const { coderAgentId, reviewerAgentId, issueId, holderRunId, wakeupRequestId } = + await seedCrossAgentScenario({ holderStatus: "running" }); + + const heartbeat = heartbeatService(db); + const followupRun = await heartbeat.wakeup(reviewerAgentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_assigned" }, + requestedByActorType: "user", + requestedByActorId: "local-board", + }); + + expect(followupRun).toBeNull(); + + const holder = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + agentId: heartbeatRuns.agentId, + finishedAt: heartbeatRuns.finishedAt, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, holderRunId)) + .then((rows) => rows[0] ?? null); + + expect(holder?.status).toBe("running"); + expect(holder?.errorCode).toBeNull(); + expect(holder?.finishedAt).toBeNull(); + expect(holder?.agentId).toBe(coderAgentId); + + const heldWakeup = await db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + + expect(heldWakeup?.status).toBe("queued"); + + const deferred = await db + .select({ status: agentWakeupRequests.status, agentId: agentWakeupRequests.agentId }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.agentId, reviewerAgentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + ), + ) + .then((rows) => rows[0] ?? null); + + expect(deferred).not.toBeNull(); + + const issue = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + + expect(issue?.executionRunId).toBe(holderRunId); + }); + + // Race-guard regression: the cancel UPDATE for the queued holder is pinned + // to the exact non-running status that was read just above it. If a worker + // races in and flips the holder from `queued` → `running` between that + // SELECT and the cancel UPDATE, the status predicate in the WHERE clause + // must guarantee zero rows are clobbered. We simulate the race by + // pre-running the same UPDATE shape against a row that is already + // `running` (the snapshot we would have read was `queued`); the row must + // remain untouched, no wake-request cascade fires, and the lock stays + // owned by the freshly-claimed running holder. + it("guards the cancel UPDATE WHERE clause against a concurrent claim flip to running", async () => { + const { coderAgentId, issueId, holderRunId, wakeupRequestId } = await seedCrossAgentScenario({ + holderStatus: "queued", + }); + + const snapshotStatus = "queued" as const; + + // Concurrent worker claims the queued run after the SELECT but before + // the cancel UPDATE. In production this is a separate transaction + // flipping the row from queued → running. + await db + .update(heartbeatRuns) + .set({ status: "running", startedAt: new Date(), updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, holderRunId)); + + const cancelled = await db + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: new Date(), + error: "Execution lock released after issue reassigned to a different agent", + errorCode: "lock_released_on_reassignment", + updatedAt: new Date(), + }) + .where( + and( + eq(heartbeatRuns.id, holderRunId), + eq(heartbeatRuns.status, snapshotStatus), + ), + ) + .returning({ id: heartbeatRuns.id }); + + expect(cancelled).toHaveLength(0); + + const holder = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + agentId: heartbeatRuns.agentId, + finishedAt: heartbeatRuns.finishedAt, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, holderRunId)) + .then((rows) => rows[0] ?? null); + + expect(holder?.status).toBe("running"); + expect(holder?.errorCode).toBeNull(); + expect(holder?.finishedAt).toBeNull(); + expect(holder?.agentId).toBe(coderAgentId); + + const heldWakeup = await db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(heldWakeup?.status).toBe("queued"); + + const issue = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + + expect(issue?.executionRunId).toBe(holderRunId); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 41edce2e..56180470 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -10419,6 +10419,59 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) activeExecutionRun = null; } + // A queued/scheduled run holding the lock for an agent that is + // no longer the issue's assignee is stale by design — the issue + // has been re-routed (e.g. blocked → in_review with a different + // assignee). Cancel it and release the lock; otherwise the new + // assignee's wake gets parked in `deferred_issue_execution` + // forever, because the original queued holder will never run + // (the issue's status / target now belongs to someone else). + // + // Race guard: pin the cancel UPDATE to the exact non-running + // status we read above. A worker could transition the holder + // from `queued` → `running` between the SELECT and this UPDATE; + // the status predicate ensures we never clobber a freshly- + // claimed running run. If zero rows matched, leave + // `activeExecutionRun` populated so the defer path runs + // normally against the now-running holder. + if ( + activeExecutionRun && + activeExecutionRun.status !== "running" && + issue.assigneeAgentId && + activeExecutionRun.agentId !== issue.assigneeAgentId + ) { + const cancelled = await tx + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: new Date(), + error: "Execution lock released after issue reassigned to a different agent", + errorCode: "lock_released_on_reassignment", + updatedAt: new Date(), + }) + .where( + and( + eq(heartbeatRuns.id, activeExecutionRun.id), + eq(heartbeatRuns.status, activeExecutionRun.status), + ), + ) + .returning({ id: heartbeatRuns.id }); + if (cancelled.length > 0) { + if (activeExecutionRun.wakeupRequestId) { + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: new Date(), + error: "Execution lock released after issue reassigned to a different agent", + updatedAt: new Date(), + }) + .where(eq(agentWakeupRequests.id, activeExecutionRun.wakeupRequestId)); + } + activeExecutionRun = null; + } + } + if (!activeExecutionRun && issue.executionRunId) { await tx .update(issues)