Harden blocker release wake idempotency

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<void>((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<number>`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();

server/src/services/heartbeat.ts

@@ -1376,6 +1376,92 @@ interface WakeupOptions {
   contextSnapshot?: Record<string, unknown>;
 }
 
+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<Db, "select">,
+  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<Db, "select">;
+  companyId: string;
+  issueId: string;
+  payload: Record<string, unknown> | null;
+  contextSnapshot: Record<string, unknown>;
+}) {
+  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<Db, "select">,
+  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<string, unknown>;
+    payload: Record<string, unknown> | 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]);