test(heartbeat): unblock merge queue after lock invariant change

server/src/__tests__/heartbeat-finalize-cancelled-skip-dispatch.test.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { eq, sql } from "drizzle-orm";
+import { eq } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import {
   agents,
@@ -11,6 +11,7 @@ import {
   getEmbeddedPostgresTestSupport,
   startEmbeddedPostgresTestDatabase,
 } from "./helpers/embedded-postgres.js";
+import { cleanupHeartbeatTestState } from "./helpers/cleanup-heartbeat-test-state.js";
 import { heartbeatService } from "../services/heartbeat.js";
 
 const mockAdapterExecute = vi.hoisted(() =>
@@ -83,10 +84,10 @@ describeEmbeddedPostgres("executeRun finalize: cancelled status skips next-queue
       provider: "test",
       model: "test-model",
     }));
-    // TRUNCATE CASCADE handles the activity_log FK to heartbeat_runs that the
-    // claim + finalize paths populate. Plain row-delete hits a 23503 ordering
-    // problem because activity_log has no ON DELETE CASCADE on run_id.
-    await db.execute(sql.raw(`TRUNCATE TABLE "companies" CASCADE`));
+    await cleanupHeartbeatTestState(db, heartbeat, {
+      errorLabel: "finalize-cancelled cleanup",
+      drainTimeoutMs: 30_000,
+    });
   }, 120_000);
 
   afterAll(async () => {

server/src/__tests__/heartbeat-retry-scheduling.test.ts

@@ -2613,7 +2613,7 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
   );
 
   it.each(["session_unavailable", "zero_token_session_reset"] as const)(
-    "schedules %s retries for an assigned todo issue while retaining its execution lock",
+    "schedules %s retries for an assigned todo issue while leaving its execution lock free until claim",
     async (retryReason) => {
       const fixture = await seedMaxTurnFixture({ issueStatus: "todo" });
       const scheduled = await heartbeat.scheduleBoundedRetry(fixture.runId, {
@@ -2636,7 +2636,7 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
         .from(issues)
         .where(eq(issues.id, fixture.issueId))
         .then((rows) => rows[0] ?? null);
-      expect(issue).toEqual({ executionRunId: scheduled.run.id, status: "todo" });
+      expect(issue).toEqual({ executionRunId: null, status: "todo" });
     },
   );
 

server/src/services/heartbeat.ts

@@ -24305,8 +24305,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
    * escalation then moves the issue to `blocked` while its replacement is
    * already pending, which is exactly the strand this path exists to prevent.
    *
-   * Mirrors `hasQueuedIssueWake` in recovery/service.ts. Callers MUST evaluate
-   * this last in their condition chain so the query stays off the hot path.
+   * A materialized run is an executable replacement when it matches the issue,
+   * participant, and active stage. A bare wake is only sufficient when it is the
+   * dedicated participant-recovery wake with those same exact coordinates.
+   * Callers MUST evaluate this last in their condition chain so the queries stay
+   * off the hot path.
    */
   async function hasQueuedReplacementIssueWake(
     dbOrTx: typeof db | Parameters<Parameters<typeof db.transaction>[0]>[0],
@@ -24315,6 +24318,25 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     participantAgentId: string,
     stageId: string,
   ) {
+    const replacementRun = await dbOrTx
+      .select({ id: heartbeatRuns.id })
+      .from(heartbeatRuns)
+      .where(
+        and(
+          eq(heartbeatRuns.companyId, companyId),
+          eq(heartbeatRuns.agentId, participantAgentId),
+          inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
+          sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
+          sql`coalesce(
+            ${heartbeatRuns.contextSnapshot} -> 'executionStage' ->> 'stageId',
+            ${heartbeatRuns.contextSnapshot} ->> 'currentStageId'
+          ) = ${stageId}`,
+        ),
+      )
+      .limit(1)
+      .then((rows) => rows[0] ?? null);
+    if (replacementRun) return true;
+
     return dbOrTx
       .select({ id: agentWakeupRequests.id })
       .from(agentWakeupRequests)

server/src/services/recovery/service.ts

@@ -1757,6 +1757,32 @@ export function recoveryService(
       .then((rows) => Boolean(rows[0]));
   }
 
+  async function hasQueuedExecutionReviewParticipantRecoveryWake(
+    companyId: string,
+    issueId: string,
+    participantAgentId: string,
+    stageId: string,
+  ) {
+    return db
+      .select({ id: agentWakeupRequests.id })
+      .from(agentWakeupRequests)
+      .where(
+        and(
+          eq(agentWakeupRequests.companyId, companyId),
+          eq(agentWakeupRequests.agentId, participantAgentId),
+          eq(agentWakeupRequests.status, "queued"),
+          eq(agentWakeupRequests.reason, EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON),
+          sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
+          sql`coalesce(
+            ${agentWakeupRequests.payload} ->> 'currentStageId',
+            ${agentWakeupRequests.payload} -> 'executionStage' ->> 'stageId'
+          ) = ${stageId}`,
+        ),
+      )
+      .limit(1)
+      .then((rows) => Boolean(rows[0]));
+  }
+
   async function latestDependencyReadinessTransitionAt(companyId: string, blockerIssueIds: string[]) {
     const uniqueBlockerIssueIds = [...new Set(blockerIssueIds.filter(Boolean))];
     if (uniqueBlockerIssueIds.length === 0) return null;
@@ -6432,10 +6458,22 @@ export function recoveryService(
         }
 
         const participantContinuationClassification = classifyContinuationFailure(participantLatestRun);
+        const queuedParticipantRecovery = agentInvokable
+          ? await hasQueuedExecutionReviewParticipantRecoveryWake(
+              issue.companyId,
+              issue.id,
+              participantAgentId,
+              pendingExecutionState.currentStageId,
+            )
+          : false;
         if (
           isUnsuccessfulTerminalIssueRun(participantLatestRun) &&
           participantContinuationClassification.kind === "non_retryable"
         ) {
+          if (queuedParticipantRecovery) {
+            result.skipped += 1;
+            continue;
+          }
           if (await latestRunPredatesLatestUnblock(issue.companyId, issue.id, participantLatestRun)) {
             result.skipped += 1;
             continue;
@@ -6535,6 +6573,11 @@ export function recoveryService(
           continue;
         }
 
+        if (queuedParticipantRecovery) {
+          result.skipped += 1;
+          continue;
+        }
+
         if (didAutomaticRecoveryFail(participantLatestRun, EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON)) {
           const updated = await escalateStrandedAssignedIssue({
             issue,