diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 32eb78055c1..4d51d6c33c1 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -4370,6 +4370,56 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(repairRun?.contextSnapshot).not.toHaveProperty("allowDeliverableWork"); }); + it("deduplicates concurrent immediate disposition-repair wakes inside the issue lock", async () => { + const { companyId, agentId, issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "cancelled", + retryReason: "issue_continuation_needed", + runErrorCode: "issue_continuation_waiting_on_review", + }); + const heartbeat = heartbeatService(db); + const idempotencyKey = `issue_disposition_repair:${issueId}:test-fingerprint:1`; + const wakeOptions = { + source: "automation" as const, + triggerDetail: "system" as const, + reason: "issue_disposition_repair", + idempotencyKey, + requestedByActorType: "system", + requestedByActorId: null, + payload: { + issueId, + dispositionRepairFingerprint: "test-fingerprint", + dispositionRepairAttempt: 1, + }, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_disposition_repair", + retryReason: "issue_disposition_repair", + dispositionRepairFingerprint: "test-fingerprint", + dispositionRepairAttempt: 1, + bypassContinuationSummaryPark: true, + }, + }; + + const [firstRun, secondRun] = await Promise.all([ + heartbeat.wakeup(agentId, wakeOptions), + heartbeat.wakeup(agentId, wakeOptions), + ]); + + expect(firstRun?.id).toBeTruthy(); + expect(secondRun?.id).toBe(firstRun?.id); + const requests = await db + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.idempotencyKey, idempotencyKey), + )); + expect(requests).toHaveLength(1); + expect(requests[0]?.runId).toBe(firstRun?.id); + }); + it("does not reset disposition repair for prose but does reset for durable source state", async () => { const { companyId, agentId, runId, issueId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5517197b02e..15a9643c16a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -17580,6 +17580,28 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { kind: "skipped" as const }; } + // Issue wakes are serialized by the source-row lock above. Recheck an + // idempotency key inside that lock so concurrent callers cannot both + // pass an earlier advisory lookup and create separate runs. Keep + // skipped requests retryable because they never dispatched work. + if (opts.idempotencyKey?.startsWith("issue_disposition_repair:")) { + const existingIdempotentWake = await tx + .select({ run: heartbeatRuns }) + .from(agentWakeupRequests) + .leftJoin(heartbeatRuns, eq(heartbeatRuns.wakeupRequestId, agentWakeupRequests.id)) + .where(and( + eq(agentWakeupRequests.companyId, agent.companyId), + eq(agentWakeupRequests.idempotencyKey, opts.idempotencyKey), + ne(agentWakeupRequests.status, "skipped"), + )) + .orderBy(desc(agentWakeupRequests.requestedAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (existingIdempotentWake) { + return { kind: "idempotent" as const, run: existingIdempotentWake.run }; + } + } + if (worktreeExecutionCutoff && issue.createdAt < worktreeExecutionCutoff) { await tx.insert(agentWakeupRequests).values({ companyId: agent.companyId, @@ -18308,6 +18330,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); if (outcome.kind === "deferred" || outcome.kind === "skipped") return null; + if (outcome.kind === "idempotent") return outcome.run; if (outcome.kind === "coalesced") { await startNextQueuedRunForAgent(agent.id); return outcome.run; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 852233f53d8..32394663ea4 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3492,6 +3492,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } } else { scheduledRun = await db.transaction(async (tx) => { + // Serialize every attempt for this source before the in-transaction + // idempotency recheck. This covers delayed attempts that bypass the + // normal heartbeat enqueue path and therefore do not acquire its lock. + await tx.execute( + sql`select id from issues where id = ${input.issue.id} and company_id = ${input.issue.companyId} for update`, + ); const existing = await tx .select({ run: heartbeatRuns }) .from(agentWakeupRequests)