From 568f1d351744f8345c156c933e9f6552f24fca6d Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 8 Jun 2026 21:18:24 +0000 Subject: [PATCH] Harden blocker release wake idempotency --- .../heartbeat-dependency-scheduling.test.ts | 182 +++++++++++++ server/src/services/heartbeat.ts | 244 +++++++++++++++++- 2 files changed, 416 insertions(+), 10 deletions(-) diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index 12b70a45269..19fd6388505 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -399,6 +399,188 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = expect(noActiveRuns).toBe(true); }); + it("does not let skipped blocker-resolved wakes suppress the released barrier and dedupes repeats", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const firstBlockerId = randomUUID(); + const secondBlockerId = randomUUID(); + const blockedIssueId = randomUUID(); + + 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: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + await db.insert(issues).values([ + { + id: firstBlockerId, + companyId, + title: "First blocker", + status: "done", + priority: "high", + }, + { + id: secondBlockerId, + companyId, + title: "Second blocker", + status: "todo", + priority: "high", + }, + { + id: blockedIssueId, + companyId, + title: "Dependent work", + status: "blocked", + priority: "medium", + assigneeAgentId: agentId, + }, + ]); + await db.insert(issueRelations).values([ + { + companyId, + issueId: firstBlockerId, + relatedIssueId: blockedIssueId, + type: "blocks", + }, + { + companyId, + issueId: secondBlockerId, + relatedIssueId: blockedIssueId, + type: "blocks", + }, + ]); + + const barrierIds = [firstBlockerId, secondBlockerId]; + const sortedBarrierIds = [...barrierIds].sort(); + const releaseKey = `issue_blockers_resolved:${blockedIssueId}:${sortedBarrierIds.join(",")}`; + const prematureWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { issueId: blockedIssueId, resolvedBlockerIssueId: firstBlockerId, blockerIssueIds: barrierIds }, + contextSnapshot: { + issueId: blockedIssueId, + wakeReason: "issue_blockers_resolved", + resolvedBlockerIssueId: firstBlockerId, + blockerIssueIds: barrierIds, + }, + }); + expect(prematureWake).toBeNull(); + + const skippedWake = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(sql`${agentWakeupRequests.payload} ->> 'issueId' = ${blockedIssueId}`) + .then((rows) => rows[0] ?? null); + expect(skippedWake).toMatchObject({ + status: "skipped", + reason: "issue_dependencies_blocked", + idempotencyKey: releaseKey, + }); + + await db + .update(issues) + .set({ status: "done", updatedAt: new Date() }) + .where(eq(issues.id, secondBlockerId)); + + let finishReleaseRun!: () => void; + const releaseRunFinished = new Promise((resolve) => { + finishReleaseRun = resolve; + }); + mockAdapterExecute.mockImplementationOnce(async () => { + await releaseRunFinished; + return { + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Released blocker barrier run complete.", + provider: "test", + model: "test-model", + }; + }); + + const releasedWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { issueId: blockedIssueId, resolvedBlockerIssueId: secondBlockerId, blockerIssueIds: barrierIds }, + contextSnapshot: { + issueId: blockedIssueId, + wakeReason: "issue_blockers_resolved", + resolvedBlockerIssueId: secondBlockerId, + blockerIssueIds: barrierIds, + }, + }); + expect(releasedWake).not.toBeNull(); + + const duplicateWake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { issueId: blockedIssueId, resolvedBlockerIssueId: secondBlockerId, blockerIssueIds: barrierIds }, + contextSnapshot: { + issueId: blockedIssueId, + wakeReason: "issue_blockers_resolved", + resolvedBlockerIssueId: secondBlockerId, + blockerIssueIds: barrierIds, + }, + }); + expect(duplicateWake?.id ?? null).toBe(releasedWake!.id); + + finishReleaseRun(); + await waitForCondition(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, releasedWake!.id)) + .then((rows) => rows[0] ?? null); + return run?.status === "succeeded"; + }); + + const blockerReleaseRunCount = await db + .select({ count: sql`count(*)::int` }) + .from(heartbeatRuns) + .where( + and( + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${blockedIssueId}`, + sql`${heartbeatRuns.contextSnapshot} ->> 'wakeReason' = 'issue_blockers_resolved'`, + ), + ) + .then((rows) => rows[0]?.count ?? 0); + const releaseWakeRows = await db + .select({ + status: agentWakeupRequests.status, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.idempotencyKey, releaseKey)); + + expect(blockerReleaseRunCount).toBe(1); + expect(releaseWakeRows.map((row) => row.status).sort()).toEqual(["coalesced", "completed", "skipped"]); + }, 30_000); + it("honors maxConcurrentRuns 1 by leaving a second assignment wake queued", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4b0a0c00bd5..b97c3dbca6b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1376,6 +1376,92 @@ interface WakeupOptions { contextSnapshot?: Record; } +const BLOCKER_RELEASE_WAKE_REASON = "issue_blockers_resolved"; +const BLOCKER_RELEASE_IDEMPOTENT_WAKE_STATUSES = [ + "queued", + "deferred_issue_execution", + "claimed", + "coalesced", + "completed", +]; + +function normalizeBlockerReleaseIds(value: unknown) { + return Array.isArray(value) + ? [...new Set(value.filter((item): item is string => typeof item === "string" && item.trim().length > 0))] + .sort() + : []; +} + +function buildBlockerReleaseIdempotencyKey(input: { + issueId: string; + blockerIssueIds: string[]; +}) { + const blockerFingerprint = input.blockerIssueIds.length > 0 ? input.blockerIssueIds.join(",") : "no-blockers"; + return `${BLOCKER_RELEASE_WAKE_REASON}:${input.issueId}:${blockerFingerprint}`; +} + +async function listIssueBlockerIdsForReleaseFingerprint( + dbOrTx: Pick, + companyId: string, + issueId: string, +) { + const rows = await dbOrTx + .select({ id: issueRelations.issueId }) + .from(issueRelations) + .where( + and( + eq(issueRelations.companyId, companyId), + eq(issueRelations.relatedIssueId, issueId), + eq(issueRelations.type, "blocks"), + ), + ); + return normalizeBlockerReleaseIds(rows.map((row) => row.id)); +} + +async function resolveBlockerReleaseIdempotencyKey(input: { + dbOrTx: Pick; + companyId: string; + issueId: string; + payload: Record | null; + contextSnapshot: Record; +}) { + const payloadBlockerIds = normalizeBlockerReleaseIds(input.payload?.blockerIssueIds); + const contextBlockerIds = normalizeBlockerReleaseIds(input.contextSnapshot.blockerIssueIds); + const blockerIssueIds = contextBlockerIds.length > 0 + ? contextBlockerIds + : payloadBlockerIds.length > 0 + ? payloadBlockerIds + : await listIssueBlockerIdsForReleaseFingerprint(input.dbOrTx, input.companyId, input.issueId); + return buildBlockerReleaseIdempotencyKey({ issueId: input.issueId, blockerIssueIds }); +} + +async function findExistingBlockerReleaseWake( + dbOrTx: Pick, + input: { + companyId: string; + idempotencyKey: string; + }, +) { + return dbOrTx + .select({ + id: agentWakeupRequests.id, + runId: agentWakeupRequests.runId, + status: agentWakeupRequests.status, + coalescedCount: agentWakeupRequests.coalescedCount, + }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey), + inArray(agentWakeupRequests.status, BLOCKER_RELEASE_IDEMPOTENT_WAKE_STATUSES), + ), + ) + .orderBy(asc(agentWakeupRequests.requestedAt), asc(agentWakeupRequests.id)) + .limit(1) + .then((rows) => rows[0] ?? null); +} + type UsageTotals = { inputTokens: number; cachedInputTokens: number; @@ -6833,9 +6919,91 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }, }); + if (readNonEmptyString(parseObject(cancelled.contextSnapshot).wakeReason) === BLOCKER_RELEASE_WAKE_REASON) { + await requeueBlockerReleaseWakeIfDependencyGateCleared({ + companyId: cancelled.companyId, + agentId: cancelled.agentId, + issueId, + contextSnapshot: parseObject(cancelled.contextSnapshot), + payload: { + issueId, + unresolvedBlockerIssueIds, + cancelledDependencyGateRunId: cancelled.id, + }, + sourceRun: cancelled, + }); + } + return cancelled; } + async function requeueBlockerReleaseWakeIfDependencyGateCleared(input: { + companyId: string; + agentId: string; + issueId: string; + contextSnapshot: Record; + payload: Record | null; + sourceRun?: typeof heartbeatRuns.$inferSelect; + }) { + const readiness = await issuesSvc + .listDependencyReadiness(input.companyId, [input.issueId]) + .then((rows) => rows.get(input.issueId) ?? null); + if (!readiness?.isDependencyReady || readiness.blockerIssueIds.length === 0) return null; + + const contextSnapshot = { + ...input.contextSnapshot, + issueId: input.issueId, + taskId: input.issueId, + wakeReason: BLOCKER_RELEASE_WAKE_REASON, + source: "issue.dependencies_gate_requeue", + blockerIssueIds: readiness.blockerIssueIds, + ...(input.sourceRun ? { requeuedAfterDependencyGateRunId: input.sourceRun.id } : {}), + }; + const payload = { + ...(input.payload ?? {}), + issueId: input.issueId, + blockerIssueIds: readiness.blockerIssueIds, + ...(input.sourceRun ? { requeuedAfterDependencyGateRunId: input.sourceRun.id } : {}), + }; + const idempotencyKey = await resolveBlockerReleaseIdempotencyKey({ + dbOrTx: db, + companyId: input.companyId, + issueId: input.issueId, + payload, + contextSnapshot, + }); + const existingWake = await findExistingBlockerReleaseWake(db, { + companyId: input.companyId, + idempotencyKey, + }); + if (existingWake) return null; + + if (input.sourceRun) { + await appendRunEvent(input.sourceRun, await nextRunEventSeq(input.sourceRun.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Requeued blocker-resolved wake because the dependency gate cleared after cancellation", + payload: { + issueId: input.issueId, + blockerIssueIds: readiness.blockerIssueIds, + idempotencyKey, + }, + }); + } + + return enqueueWakeup(input.agentId, { + source: "automation", + triggerDetail: "system", + reason: BLOCKER_RELEASE_WAKE_REASON, + payload, + contextSnapshot, + idempotencyKey, + requestedByActorType: "system", + requestedByActorId: "heartbeat", + }); + } + type QueuedRunStaleness = | { stale: false } | { @@ -9800,6 +9968,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) payload, }); let issueId = readNonEmptyString(enrichedContextSnapshot.issueId) ?? issueIdFromPayload; + let effectiveIdempotencyKey = opts.idempotencyKey ?? null; const agent = await getAgent(agentId); if (!agent) throw notFound("Agent not found"); @@ -9818,7 +9987,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "skipped", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, finishedAt: new Date(), ...patch, }); @@ -9898,6 +10067,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) enrichedContextSnapshot.projectId = projectId; } + if (reason === BLOCKER_RELEASE_WAKE_REASON && issueId && !effectiveIdempotencyKey) { + effectiveIdempotencyKey = await resolveBlockerReleaseIdempotencyKey({ + dbOrTx: db, + companyId: agent.companyId, + issueId, + payload, + contextSnapshot: enrichedContextSnapshot, + }); + enrichedContextSnapshot.blockerReleaseFingerprint = effectiveIdempotencyKey; + } + const budgetBlock = await budgets.getInvocationBlock(agent.companyId, agentId, { issueId, projectId, @@ -10018,7 +10198,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "skipped", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, finishedAt: new Date(), }); return { kind: "skipped" as const }; @@ -10221,10 +10401,42 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "skipped", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, finishedAt: new Date(), }); - return { kind: "skipped" as const }; + return { kind: "skipped" as const, dependencyGate: true as const }; + } + + if (!activeExecutionRun && reason === BLOCKER_RELEASE_WAKE_REASON && effectiveIdempotencyKey) { + const existingReleaseWake = await findExistingBlockerReleaseWake(tx, { + companyId: issue.companyId, + idempotencyKey: effectiveIdempotencyKey, + }); + if (existingReleaseWake) { + await tx.insert(agentWakeupRequests).values({ + companyId: agent.companyId, + agentId, + source, + triggerDetail, + reason, + payload, + status: "coalesced", + coalescedCount: 1, + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: effectiveIdempotencyKey, + runId: existingReleaseWake.runId, + finishedAt: new Date(), + }); + await tx + .update(agentWakeupRequests) + .set({ + coalescedCount: (existingReleaseWake.coalescedCount ?? 0) + 1, + updatedAt: new Date(), + }) + .where(eq(agentWakeupRequests.id, existingReleaseWake.id)); + return { kind: "duplicate" as const }; + } } if (activeExecutionRun) { @@ -10269,7 +10481,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) coalescedCount: 1, requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, runId: mergedRun.id, finishedAt: new Date(), }); @@ -10334,7 +10546,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "deferred_issue_execution", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, }); return { kind: "deferred" as const }; @@ -10352,7 +10564,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "queued", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, }) .returning() .then((rows) => rows[0]); @@ -10388,7 +10600,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { kind: "queued" as const, run: newRun }; }); - if (outcome.kind === "deferred" || outcome.kind === "skipped") return null; + if (outcome.kind === "deferred" || outcome.kind === "duplicate") return null; + if (outcome.kind === "skipped") { + if ("dependencyGate" in outcome && outcome.dependencyGate && reason === BLOCKER_RELEASE_WAKE_REASON) { + await requeueBlockerReleaseWakeIfDependencyGateCleared({ + companyId: agent.companyId, + agentId, + issueId, + contextSnapshot: enrichedContextSnapshot, + payload, + }); + } + return null; + } if (outcome.kind === "coalesced") { await startNextQueuedRunForAgent(agent.id); return outcome.run; @@ -10462,7 +10686,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) coalescedCount: 1, requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, runId: mergedRun.id, finishedAt: new Date(), }); @@ -10481,7 +10705,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "queued", requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, + idempotencyKey: effectiveIdempotencyKey, }) .returning() .then((rows) => rows[0]);