diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 25f031c9457..c7793de1ac5 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -120,6 +120,7 @@ import { noticeMetadataReferencesRecoveryAction, } from "../services/recovery/index.ts"; import { collectDispositionRepairSourceState } from "../services/recovery/disposition-repair.ts"; +import { recoveryService } from "../services/recovery/service.ts"; import { UNMANAGED_BACKGROUND_TASK_LIVENESS_REASON, UNMANAGED_BACKGROUND_TASK_STOP_REASON, @@ -7241,6 +7242,120 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } }); + it("ignores a newer non-retryable run from another agent when recovering assigned work", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "process_lost", + }); + const foreignAgentId = randomUUID(); + const foreignRunId = randomUUID(); + + await db.insert(agents).values({ + id: foreignAgentId, + companyId, + name: "RecommendationAgent", + role: "advisor", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: foreignRunId, + companyId, + agentId: foreignAgentId, + invocationSource: "automation", + triggerDetail: "system", + status: "cancelled", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + }, + errorCode: "agent_not_invokable", + error: "Recommendation agent is paused.", + startedAt: new Date("2026-03-19T00:10:00.000Z"), + finishedAt: new Date("2026-03-19T00:11:00.000Z"), + createdAt: new Date(Date.now() + 1_000), + updatedAt: new Date("2026-03-19T00:11:00.000Z"), + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + + expect(result.continuationRequeued).toBe(1); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const issue = await db + .select() + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("in_progress"); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(0); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)) + .then((rows) => rows.find((row) => row.id !== runId) ?? null); + expect(retryRun).toMatchObject({ retryOfRunId: runId }); + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId, + retryReason: "issue_continuation_needed", + source: "issue.continuation_recovery", + }); + if (retryRun) { + await waitForRunToSettle(heartbeat, retryRun.id); + } + }); + + it("does not block an issue that changed status after recovery selected it", async () => { + const { issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "budget_blocked", + }); + const staleIssue = await db + .select() + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]!); + const latestRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0]!); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); + + const recovery = recoveryService(db, { enqueueWakeup: vi.fn() }); + const result = await recovery.escalateStrandedAssignedIssue({ + issue: staleIssue, + previousStatus: "in_progress", + latestRun, + notice: { + body: "Stale recovery candidate.", + title: "Continuation failed", + tone: "danger", + }, + }); + + expect(result).toBeNull(); + const issue = await db + .select() + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("in_review"); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(0); + }); + it("leaves the productive-but-stranded continuation path unchanged under the new classifier", async () => { const { agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 8b00c7d07bb..092570a5f97 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -811,32 +811,6 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return (await evaluateAgentInvokabilityFromDb(db, agent)).invokable; } - async function getLatestIssueRun(companyId: string, issueId: string): Promise { - return db - .select({ - id: heartbeatRuns.id, - agentId: heartbeatRuns.agentId, - status: heartbeatRuns.status, - error: heartbeatRuns.error, - errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, - livenessState: heartbeatRuns.livenessState, - resultJson: heartbeatRuns.resultJson, - startedAt: heartbeatRuns.startedAt, - createdAt: heartbeatRuns.createdAt, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, companyId), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, - ), - ) - .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - async function getLatestIssueRunForAgent( companyId: string, issueId: string, @@ -3198,7 +3172,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) previousStatus: StrandedPreviousStatus; latestRun: LatestIssueRun; }) { - const updated = await issuesSvc.update(input.issue.id, { status: "blocked" }); + const updated = await updateIssueToBlockedIfStatusUnchanged({ + issueId: input.issue.id, + expectedStatus: input.previousStatus, + }); if (!updated) return null; const prefix = await getCompanyIssuePrefix(input.issue.companyId); @@ -3261,6 +3238,33 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return updated; } + async function updateIssueToBlockedIfStatusUnchanged(input: { + issueId: string; + expectedStatus: StrandedPreviousStatus; + blockedByIssueIds?: string[]; + assigneeAgentId?: string | null; + }) { + return db.transaction(async (tx) => { + const current = await tx + .select({ status: issues.status }) + .from(issues) + .where(eq(issues.id, input.issueId)) + .for("update") + .then((rows) => rows[0] ?? null); + if (!current || current.status !== input.expectedStatus) return null; + + return issuesSvc.update(input.issueId, { + status: "blocked", + ...(input.blockedByIssueIds !== undefined + ? { blockedByIssueIds: input.blockedByIssueIds } + : {}), + ...(input.assigneeAgentId !== undefined + ? { assigneeAgentId: input.assigneeAgentId } + : {}), + }, tx); + }); + } + async function existingBlockerIssueIds(companyId: string, issueId: string) { return db .select({ blockerIssueId: issueRelations.issueId }) @@ -4393,8 +4397,9 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) }); } const blockerIds = await existingUnresolvedBlockerIssueIds(input.issue.companyId, input.issue.id); - const updated = await issuesSvc.update(input.issue.id, { - status: "blocked", + const updated = await updateIssueToBlockedIfStatusUnchanged({ + issueId: input.issue.id, + expectedStatus: input.previousStatus, blockedByIssueIds: blockerIds, // Recovery-action ownership is intentionally separate from deliverable // ownership. Automatic escalation must never transfer the source task. @@ -4552,8 +4557,9 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) (currentIssue.status !== "blocked" || currentIssue.assigneeAgentId !== input.issue.assigneeAgentId) ) { - const reblocked = await issuesSvc.update(input.issue.id, { - status: "blocked", + const reblocked = await updateIssueToBlockedIfStatusUnchanged({ + issueId: input.issue.id, + expectedStatus: input.previousStatus, blockedByIssueIds: blockerIds, assigneeAgentId: input.issue.assigneeAgentId, }); @@ -4749,7 +4755,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - let latestRun = await getLatestIssueRun(issue.companyId, issue.id); + let latestRun = await getLatestIssueRunForAgent(issue.companyId, issue.id, agentId); const agent = await getAgent(agentId); const agentInvokable = agent && agent.companyId === issue.companyId @@ -4777,7 +4783,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (await hasActiveExecutionPath( issue.companyId, issue.id, - issue.status === "in_review" ? agentId : null, + agentId, )) { result.skipped += 1; continue; @@ -4803,7 +4809,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } const recoveryNow = new Date(); const participantLatestRunForRecovery = issue.status === "in_review" && participantAgentId - ? await getLatestIssueRunForAgent(issue.companyId, issue.id, participantAgentId) + ? latestRun : null; const providerQuotaMonitorRun = issue.status === "in_review" ? participantLatestRunForRecovery @@ -5138,7 +5144,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (issue.status === "todo") { if (!latestRun) { - if (await hasQueuedIssueWake(issue.companyId, issue.id)) { + if (await hasQueuedIssueWake(issue.companyId, issue.id, agentId)) { result.skipped += 1; continue; }