From 17b860bdfabafaf1de1b15fbb16f8023100540e1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 24 Apr 2026 15:23:27 -0500 Subject: [PATCH] Fix reassigned issue scheduled retries Cancel stale scheduled retries when an issue is reassigned so the new assignee can queue immediately instead of waiting behind the old assignee's retry. Co-Authored-By: Paperclip --- .../heartbeat-retry-scheduling.test.ts | 277 ++++++++++++++++++ server/src/services/heartbeat.ts | 168 ++++++++++- 2 files changed, 430 insertions(+), 15 deletions(-) diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 162ea539ed4..e5b452f4067 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -3,11 +3,14 @@ import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { agents, + agentRuntimeState, agentWakeupRequests, companies, createDb, + environmentLeases, heartbeatRunEvents, heartbeatRuns, + issues, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -40,8 +43,11 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { afterEach(async () => { await db.delete(heartbeatRunEvents); + await db.delete(environmentLeases); + await db.delete(issues); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); + await db.delete(agentRuntimeState); await db.delete(agents); await db.delete(companies); }); @@ -212,6 +218,277 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { expect(promotedRun?.status).toBe("queued"); }); + it("does not defer a new assignee behind the previous assignee's scheduled retry", async () => { + const companyId = randomUUID(); + const oldAgentId = randomUUID(); + const newAgentId = randomUUID(); + const issueId = randomUUID(); + const sourceRunId = randomUUID(); + const now = new Date("2026-04-20T13:00:00.000Z"); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: oldAgentId, + companyId, + name: "ClaudeCoder", + role: "engineer", + status: "active", + adapterType: "claude_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }, + { + id: newAgentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }, + ]); + + await db.insert(heartbeatRuns).values({ + id: sourceRunId, + companyId, + agentId: oldAgentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "failed", + error: "upstream overload", + errorCode: "adapter_failed", + finishedAt: now, + contextSnapshot: { + issueId, + wakeReason: "issue_assigned", + }, + updatedAt: now, + createdAt: now, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Retry reassignment", + status: "todo", + priority: "medium", + assigneeAgentId: oldAgentId, + executionRunId: sourceRunId, + executionAgentNameKey: "claudecoder", + executionLockedAt: now, + issueNumber: 1, + identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-1`, + }); + + const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, { + now, + random: () => 0.5, + }); + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + await db.update(issues).set({ + assigneeAgentId: newAgentId, + updatedAt: now, + }).where(eq(issues.id, issueId)); + + // Keep the new agent's queue from auto-claiming/executing during this unit test. + await db.insert(heartbeatRuns).values( + Array.from({ length: 5 }, () => ({ + id: randomUUID(), + companyId, + agentId: newAgentId, + invocationSource: "automation", + triggerDetail: "system", + status: "running", + contextSnapshot: { + wakeReason: "test_busy_slot", + }, + startedAt: now, + updatedAt: now, + createdAt: now, + })), + ); + + const newAssigneeRun = await heartbeat.wakeup(newAgentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId, + mutation: "update", + }, + contextSnapshot: { + issueId, + source: "issue.update", + }, + requestedByActorType: "user", + requestedByActorId: "local-board", + }); + + expect(newAssigneeRun).not.toBeNull(); + expect(newAssigneeRun?.agentId).toBe(newAgentId); + expect(newAssigneeRun?.status).toBe("queued"); + + const oldRetry = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(oldRetry).toEqual({ + status: "cancelled", + errorCode: "issue_reassigned", + }); + + const deferredWakeups = await db + .select({ count: sql`count(*)::int` }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.status, "deferred_issue_execution")) + .then((rows) => rows[0]?.count ?? 0); + expect(deferredWakeups).toBe(0); + }); + + it("does not promote a scheduled retry after issue ownership changes", async () => { + const companyId = randomUUID(); + const oldAgentId = randomUUID(); + const newAgentId = randomUUID(); + const issueId = randomUUID(); + const sourceRunId = randomUUID(); + const now = new Date("2026-04-20T14:00:00.000Z"); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: oldAgentId, + companyId, + name: "ClaudeCoder", + role: "engineer", + status: "active", + adapterType: "claude_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }, + { + id: newAgentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }, + ]); + + await db.insert(heartbeatRuns).values({ + id: sourceRunId, + companyId, + agentId: oldAgentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "failed", + error: "upstream overload", + errorCode: "adapter_failed", + finishedAt: now, + contextSnapshot: { + issueId, + wakeReason: "issue_assigned", + }, + updatedAt: now, + createdAt: now, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Retry promotion reassignment", + status: "todo", + priority: "medium", + assigneeAgentId: oldAgentId, + executionRunId: sourceRunId, + executionAgentNameKey: "claudecoder", + executionLockedAt: now, + issueNumber: 1, + identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-2`, + }); + + const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, { + now, + random: () => 0.5, + }); + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + await db.update(issues).set({ + assigneeAgentId: newAgentId, + updatedAt: now, + }).where(eq(issues.id, issueId)); + + const promotion = await heartbeat.promoteDueScheduledRetries(scheduled.dueAt); + expect(promotion).toEqual({ promoted: 0, runIds: [] }); + + const oldRetry = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(oldRetry).toEqual({ + status: "cancelled", + errorCode: "issue_reassigned", + }); + + const issue = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.executionRunId).toBeNull(); + }); + it("exhausts bounded retries after the hard cap", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1f510878f9e..a1a77df64de 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -3594,6 +3594,83 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const promotedRunIds: string[] = []; for (const dueRun of dueRuns) { + const dueRunIssueId = readNonEmptyString(parseObject(dueRun.contextSnapshot).issueId); + if (dueRunIssueId) { + const issue = await db + .select({ + id: issues.id, + assigneeAgentId: issues.assigneeAgentId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(and(eq(issues.id, dueRunIssueId), eq(issues.companyId, dueRun.companyId))) + .then((rows) => rows[0] ?? null); + + if (issue && issue.assigneeAgentId !== dueRun.agentId) { + const reason = "Cancelled because the issue was reassigned before the scheduled retry became due"; + const cancelled = await db + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + errorCode: "issue_reassigned", + updatedAt: now, + }) + .where( + and( + eq(heartbeatRuns.id, dueRun.id), + eq(heartbeatRuns.status, "scheduled_retry"), + lte(heartbeatRuns.scheduledRetryAt, now), + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + + if (!cancelled) continue; + + if (cancelled.wakeupRequestId) { + await db + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, cancelled.wakeupRequestId)); + } + + if (issue.executionRunId === cancelled.id) { + await db + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: now, + }) + .where(and(eq(issues.id, issue.id), eq(issues.executionRunId, cancelled.id))); + } + + await appendRunEvent(cancelled, await nextRunEventSeq(cancelled.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Scheduled retry cancelled because issue ownership changed before it became due", + payload: { + issueId: issue.id, + scheduledRetryAttempt: cancelled.scheduledRetryAttempt, + scheduledRetryAt: cancelled.scheduledRetryAt ? new Date(cancelled.scheduledRetryAt).toISOString() : null, + scheduledRetryReason: cancelled.scheduledRetryReason, + previousRetryAgentId: cancelled.agentId, + currentAssigneeAgentId: issue.assigneeAgentId, + }, + }); + continue; + } + } + const promoted = await db .update(heartbeatRuns) .set({ @@ -6277,6 +6354,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .select({ id: issues.id, companyId: issues.companyId, + assigneeAgentId: issues.assigneeAgentId, executionRunId: issues.executionRunId, executionAgentNameKey: issues.executionAgentNameKey, }) @@ -6301,6 +6379,58 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { kind: "skipped" as const }; } + const cancelStaleScheduledRetry = async (scheduledRun: typeof heartbeatRuns.$inferSelect) => { + if ( + scheduledRun.status !== "scheduled_retry" || + scheduledRun.agentId === issue.assigneeAgentId + ) { + return false; + } + + const now = new Date(); + const reason = "Cancelled because the issue was reassigned before the scheduled retry became due"; + const cancelled = await tx + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + errorCode: "issue_reassigned", + updatedAt: now, + }) + .where(and(eq(heartbeatRuns.id, scheduledRun.id), eq(heartbeatRuns.status, "scheduled_retry"))) + .returning() + .then((rows) => rows[0] ?? null); + + if (!cancelled) return false; + + if (scheduledRun.wakeupRequestId) { + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, scheduledRun.wakeupRequestId)); + } + + if (issue.executionRunId === scheduledRun.id) { + await tx + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: now, + }) + .where(and(eq(issues.id, issue.id), eq(issues.executionRunId, scheduledRun.id))); + } + + return true; + }; + let activeExecutionRun = issue.executionRunId ? await tx .select() @@ -6318,6 +6448,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) activeExecutionRun = null; } + if (activeExecutionRun && await cancelStaleScheduledRetry(activeExecutionRun)) { + activeExecutionRun = null; + } + if (!activeExecutionRun && issue.executionRunId) { await tx .update(issues) @@ -6349,21 +6483,25 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); if (legacyRun) { - activeExecutionRun = legacyRun; - const legacyAgent = await tx - .select({ name: agents.name }) - .from(agents) - .where(eq(agents.id, legacyRun.agentId)) - .then((rows) => rows[0] ?? null); - await tx - .update(issues) - .set({ - executionRunId: legacyRun.id, - executionAgentNameKey: normalizeAgentNameKey(legacyAgent?.name), - executionLockedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(issues.id, issue.id)); + if (await cancelStaleScheduledRetry(legacyRun)) { + activeExecutionRun = null; + } else { + activeExecutionRun = legacyRun; + const legacyAgent = await tx + .select({ name: agents.name }) + .from(agents) + .where(eq(agents.id, legacyRun.agentId)) + .then((rows) => rows[0] ?? null); + await tx + .update(issues) + .set({ + executionRunId: legacyRun.id, + executionAgentNameKey: normalizeAgentNameKey(legacyAgent?.name), + executionLockedAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(issues.id, issue.id)); + } } }