From f20e7fa790fd7abe54a194ecc705473318469ff2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 27 Apr 2026 15:35:51 -0500 Subject: [PATCH] Add assigned todo dispatch backstop Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 142 ++++++++++++++++++ server/src/index.ts | 2 + server/src/services/recovery/service.ts | 36 ++++- 3 files changed, 178 insertions(+), 2 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 0723eb4e63a..a783499ef62 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -586,6 +586,48 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId, rootIssueId }; } + async function seedAssignedTodoNoRunFixture(input?: { + agentStatus?: "paused" | "idle" | "running"; + }) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = 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: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: input?.agentStatus ?? "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Assigned todo work that never received a heartbeat", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + assigneeUserId: null, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + + return { companyId, agentId, issueId }; + } + async function expectStrandedRecoveryArtifacts(input: { companyId: string; agentId: string; @@ -1173,6 +1215,105 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("dispatches assigned todo work with no prior run as a normal assignment wake", async () => { + const { companyId, agentId, issueId } = await seedAssignedTodoNoRunFixture(); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.assignmentDispatched).toBe(1); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]).toMatchObject({ + companyId, + agentId, + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: expect.objectContaining({ + issueId, + mutation: "assigned_todo_liveness_dispatch", + }), + }); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]?.retryOfRunId).toBeNull(); + expect(runs[0]?.contextSnapshot).toMatchObject({ + issueId, + taskId: issueId, + wakeReason: "issue_assigned", + source: "issue.assigned_todo_liveness_dispatch", + }); + expect((runs[0]?.contextSnapshot as Record)?.retryReason).toBeUndefined(); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("todo"); + + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + expect(recoveryIssues).toHaveLength(0); + await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([]); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(0); + + if (runs[0]?.id) { + await waitForRunToSettle(heartbeat, runs[0].id); + } + }); + + it("does not duplicate initial assigned todo dispatch when a queued wake already exists", async () => { + const { companyId, agentId, issueId } = await seedAssignedTodoNoRunFixture(); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId, mutation: "assigned_todo_liveness_dispatch" }, + status: "queued", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.assignmentDispatched).toBe(0); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.skipped).toBe(1); + expect(result.issueIds).toEqual([]); + + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups).toHaveLength(1); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(0); + }); + + it("does not dispatch assigned todo work with no prior run when the agent is paused", async () => { + const { agentId, issueId } = await seedAssignedTodoNoRunFixture({ agentStatus: "paused" }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.assignmentDispatched).toBe(0); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.skipped).toBe(1); + expect(result.issueIds).toEqual([]); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("todo"); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(0); + }); + it("re-enqueues assigned todo work when the last issue run died and no wake remains", async () => { const { agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "todo", @@ -1181,6 +1322,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { 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); diff --git a/server/src/index.ts b/server/src/index.ts index 35e2ca5f943..a3b7ddcefb3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -682,6 +682,7 @@ export async function startServer(): Promise { const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); if ( promotion.promoted > 0 || + reconciled.assignmentDispatched > 0 || reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.escalated > 0 @@ -740,6 +741,7 @@ export async function startServer(): Promise { const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); if ( promotion.promoted > 0 || + reconciled.assignmentDispatched > 0 || reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.escalated > 0 diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 0f435d3bca5..3aae3778919 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), - eq(agentWakeupRequests.status, "deferred_issue_execution"), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, ), ) @@ -386,6 +386,26 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return queued; } + async function enqueueInitialAssignedTodoDispatch(issue: typeof issues.$inferSelect, agentId: string) { + return deps.enqueueWakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: issue.id, + mutation: "assigned_todo_liveness_dispatch", + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: issue.id, + taskId: issue.id, + wakeReason: "issue_assigned", + source: "issue.assigned_todo_liveness_dispatch", + }, + }); + } + async function reconcileUnassignedBlockingIssues() { const candidates = await db .select({ @@ -1526,6 +1546,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) ); const result = { + assignmentDispatched: 0, dispatchRequeued: 0, continuationRequeued: 0, orphanBlockersAssigned: 0, @@ -1574,7 +1595,18 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } if (issue.status === "todo") { - if (!latestRun || latestRun.status === "succeeded") { + if (!latestRun) { + const queued = await enqueueInitialAssignedTodoDispatch(issue, agentId); + if (queued) { + result.assignmentDispatched += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + if (latestRun.status === "succeeded") { result.skipped += 1; continue; }