diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index a2faad9fd81..211f119a110 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1417,6 +1417,42 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } }); + it("still re-enqueues stranded assigned todo recovery when an old queued wake exists", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "todo", + runStatus: "failed", + }); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + status: "queued", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.assignmentDispatched).toBe(0); + expect(result.dispatchRequeued).toBe(1); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + + const retryRun = runs.find((row) => row.id !== runId); + expect((retryRun?.contextSnapshot as Record)?.retryReason).toBe("assignment_recovery"); + if (retryRun) { + await waitForRunToSettle(heartbeat, retryRun.id); + } + }); + it("blocks assigned todo work after the one automatic dispatch recovery was already used", async () => { const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "todo", diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f239195058e..40d12555a9c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -332,7 +332,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .where( and( eq(agentWakeupRequests.companyId, companyId), - inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, ), ) @@ -343,6 +343,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return Boolean(run || deferredWake); } + async function hasQueuedIssueWake(companyId: string, issueId: string) { + return db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "queued"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + async function enqueueStrandedIssueRecovery(input: { issueId: string; agentId: string; @@ -1604,6 +1619,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (issue.status === "todo") { if (!latestRun) { + if (await hasQueuedIssueWake(issue.companyId, issue.id)) { + result.skipped += 1; + continue; + } + if (await isInvocationBudgetBlocked(issue, agentId)) { result.skipped += 1; continue;