From a7990f1a1f23790ca494376ce672c9fc13481274 Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Tue, 11 Aug 2026 00:01:32 -0700 Subject: [PATCH 1/2] test(heartbeat): align retry lock and cleanup invariants --- ...heartbeat-finalize-cancelled-skip-dispatch.test.ts | 11 ++++++----- .../src/__tests__/heartbeat-retry-scheduling.test.ts | 4 ++-- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/server/src/__tests__/heartbeat-finalize-cancelled-skip-dispatch.test.ts b/server/src/__tests__/heartbeat-finalize-cancelled-skip-dispatch.test.ts index b90dc12eb0c0..358d93323056 100644 --- a/server/src/__tests__/heartbeat-finalize-cancelled-skip-dispatch.test.ts +++ b/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 () => { diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 169cbcb69116..9cfccd0f9a83 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/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" }); }, ); From b6461bb23e0e9bf0ad1b66eed5b02d341684d7d4 Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Tue, 11 Aug 2026 01:23:50 -0700 Subject: [PATCH 2/2] fix(recovery): preserve queued review-stage replacements --- server/src/services/heartbeat.ts | 26 +++++++++++++-- server/src/services/recovery/service.ts | 43 +++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 970a20844c77..338b76990026 100644 --- a/server/src/services/heartbeat.ts +++ b/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[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) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index c7d5bb797c48..83806b8c277c 100644 --- a/server/src/services/recovery/service.ts +++ b/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,