diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 0412cfaf85..1f6d5b72a8 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -72,6 +72,7 @@ import { instanceSettingsService } from "../services/instance-settings.ts"; import { issueService } from "../services/issues.ts"; import { runningProcesses } from "../adapters/index.ts"; import { DEFAULT_LIVENESS_REESCALATION_COOLDOWN_MS } from "../services/recovery/service.ts"; +import { buildIssueBlockersResolvedWakeStateKey } from "../services/issue-dependency-wakeups.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -484,7 +485,12 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { .then((rows) => rows[0] ?? null); expect(wake?.reason).toBe("issue_blockers_resolved"); - expect(wake?.idempotencyKey).toBe(`issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`); + expect(wake?.idempotencyKey).toBe( + buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), + ); expect(["queued", "claimed", "completed"]).toContain(wake?.status); const events = await db @@ -522,7 +528,12 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { .then((rows) => rows[0] ?? null); expect(wake?.reason).toBe("issue_blockers_resolved"); - expect(wake?.idempotencyKey).toBe(`issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`); + expect(wake?.idempotencyKey).toBe( + buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), + ); expect(["queued", "claimed", "completed"]).toContain(wake?.status); const events = await db @@ -564,14 +575,23 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { .then((rows) => rows[0] ?? null); expect(wake).toMatchObject({ reason: "issue_blockers_resolved", - idempotencyKey: `issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`, + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), }); }); it("retries a resolved dependency wake when the prior wake was skipped as stale", async () => { const { companyId, agentId, blockedIssueId, blockerIssueId } = await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); - const idempotencyKey = `issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`; + // The route-time wake writes the level-triggered state key. A skip records a + // `skipped` row with that key. `skipped` is not an in-flight status, so the + // backstop must still re-emit for the same ready state. + const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }); await db.insert(agentWakeupRequests).values({ companyId, agentId, @@ -653,7 +673,10 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { .then((rows) => rows[0] ?? null); expect(wake).toMatchObject({ reason: "issue_blockers_resolved", - idempotencyKey: `issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`, + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), }); }); @@ -716,6 +739,79 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { ); }); + it("heals a multi-blocker dependent when only a completed wake for an earlier blocker exists", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const secondBlockerIssueId = randomUUID(); + await db.insert(issues).values({ + id: secondBlockerIssueId, + companyId, + title: "Earlier completed blocker", + status: "done", + priority: "medium", + issueNumber: 3, + identifier: "R-MULTI-3", + }); + await db.insert(issueRelations).values({ + companyId, + issueId: secondBlockerIssueId, + relatedIssueId: blockedIssueId, + type: "blocks", + }); + + // An earlier partial resolution left a `completed` per-edge wake. The bug was + // that this stale wake suppressed the wake for the current ready state. The + // level-triggered dedup keys on the full blocker set, so this completed wake + // no longer strands the dependent. + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { + issueId: blockedIssueId, + resolvedBlockerIssueId: secondBlockerIssueId, + }, + status: "completed", + finishedAt: new Date(), + idempotencyKey: `issue_blockers_resolved:${blockedIssueId}:${secondBlockerIssueId}`, + }); + + const readiness = await issueService(db).getDependencyReadiness(blockedIssueId); + expect(readiness.isDependencyReady).toBe(true); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.dependencyWakesHealed).toBe(1); + expect(result.dependencyWakeIssueIds).toEqual([blockedIssueId]); + expect(result.dependencyWakeExistingSkipped).toBe(0); + + const stateKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: readiness.blockerIssueIds, + }); + const healedWake = await db + .select({ status: agentWakeupRequests.status, idempotencyKey: agentWakeupRequests.idempotencyKey }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.idempotencyKey, stateKey))) + .then((rows) => rows[0] ?? null); + expect(healedWake).not.toBeNull(); + expect(["queued", "claimed", "completed"]).toContain(healedWake?.status); + + // A second reconciliation pass finds the state-key wake and stays bounded: + // it heals nothing more and never enqueues a second wake for the same state. + const secondPass = await heartbeatService(db).reconcileIssueGraphLiveness(); + expect(secondPass.dependencyWakesHealed).toBe(0); + + const stateKeyWakes = await db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.idempotencyKey, stateKey))); + expect(stateKeyWakes).toHaveLength(1); + }); + it("counts null dependency wake returns as deferred instead of enqueue failures", async () => { await enableAutoRecovery(); const { companyId, agentId } = diff --git a/server/src/__tests__/issue-dependency-wakeups-routes.test.ts b/server/src/__tests__/issue-dependency-wakeups-routes.test.ts index c8b76d429f..a5d22dc1da 100644 --- a/server/src/__tests__/issue-dependency-wakeups-routes.test.ts +++ b/server/src/__tests__/issue-dependency-wakeups-routes.test.ts @@ -3,7 +3,7 @@ import request from "supertest"; import { beforeEach, describe, expect, it, vi } from "vitest"; const mockWakeup = vi.hoisted(() => vi.fn(async () => undefined)); -const mockFindExistingIssueBlockersResolvedWake = vi.hoisted(() => vi.fn(async () => null)); +const mockFindExistingIssueBlockersResolvedWakeForReadyState = vi.hoisted(() => vi.fn(async () => null)); const mockIssueService = vi.hoisted(() => ({ getAncestors: vi.fn(), getById: vi.fn(), @@ -98,7 +98,8 @@ vi.mock("../services/issue-dependency-wakeups.js", async () => { ); return { ...actual, - findExistingIssueBlockersResolvedWake: mockFindExistingIssueBlockersResolvedWake, + findExistingIssueBlockersResolvedWakeForReadyState: + mockFindExistingIssueBlockersResolvedWakeForReadyState, }; }); @@ -145,7 +146,7 @@ describe("issue dependency wakeups in issue routes", () => { vi.doUnmock("../routes/authz.js"); vi.doUnmock("../middleware/index.js"); vi.clearAllMocks(); - mockFindExistingIssueBlockersResolvedWake.mockResolvedValue(null); + mockFindExistingIssueBlockersResolvedWakeForReadyState.mockResolvedValue(null); mockIssueService.getAncestors.mockResolvedValue([]); mockIssueService.getByIdForUpdate.mockImplementation(async () => mockIssueService.getById()); mockIssueService.getComment.mockResolvedValue(null); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 4207e2d2c0..566a973a7d 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -179,8 +179,8 @@ import { } from "../services/onboarding-greeting.js"; import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, - buildIssueBlockersResolvedWakeIdempotencyKey, - findExistingIssueBlockersResolvedWake, + buildIssueBlockersResolvedWakeStateKey, + findExistingIssueBlockersResolvedWakeForReadyState, } from "../services/issue-dependency-wakeups.js"; import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; import { @@ -10204,14 +10204,15 @@ export function issueRoutes( source: string; mutation: string; }) => { - const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: input.dependentIssueId, - resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, }); try { - const existingWake = await findExistingIssueBlockersResolvedWake(db, { + const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId: issue.companyId, - idempotencyKey, + dependentIssueId: input.dependentIssueId, + blockerIssueIds: input.blockerIssueIds, }); if (existingWake) return; } catch (err) { @@ -12256,14 +12257,15 @@ export function issueRoutes( resolvedBlockerIssueId: string; blockerIssueIds: string[]; }) => { - const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: input.dependentIssueId, - resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, }); try { - const existingWake = await findExistingIssueBlockersResolvedWake(db, { + const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId: currentIssue.companyId, - idempotencyKey, + dependentIssueId: input.dependentIssueId, + blockerIssueIds: input.blockerIssueIds, }); if (existingWake) return; } catch (err) { diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts index 019a9a9cd5..bcd565a8b4 100644 --- a/server/src/services/issue-dependency-wakeups.ts +++ b/server/src/services/issue-dependency-wakeups.ts @@ -1,9 +1,14 @@ -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq, inArray, or } from "drizzle-orm"; +import { createHash } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests } from "@paperclipai/db"; export const ISSUE_BLOCKERS_RESOLVED_WAKE_REASON = "issue_blockers_resolved"; +// A wake counts as "already delivered or in flight for the current ready state" +// for these statuses. The level-triggered state key uses this full set so that +// one wake for a ready state suppresses further wakes for the SAME state. This +// bounds reconciliation: after one wake, later passes find the completed row. const IDEMPOTENT_DEPENDENCY_WAKE_STATUSES = [ "queued", "deferred_issue_execution", @@ -11,6 +16,23 @@ const IDEMPOTENT_DEPENDENCY_WAKE_STATUSES = [ "completed", ] as const; +// A wake counts as "still in flight" for these statuses. The `completed` status +// is not in this set on purpose. Dependency readiness is level-triggered, so a +// historical completed per-edge wake must never suppress a new wake for the +// current ready state. The dedup uses this set only for the legacy per-edge key, +// to avoid a duplicate while an old-format wake is still queued or claimed. +const IN_FLIGHT_DEPENDENCY_WAKE_STATUSES = [ + "queued", + "deferred_issue_execution", + "claimed", +] as const; + +/** + * Legacy per-edge idempotency key. One key encodes a single resolved blocker + * edge `issue_blockers_resolved:{dependentIssueId}:{resolvedBlockerIssueId}`. + * The dedup keeps this format only to read wake rows written before the + * level-triggered state key existed. + */ export function buildIssueBlockersResolvedWakeIdempotencyKey(input: { dependentIssueId: string; resolvedBlockerIssueId: string; @@ -22,36 +44,82 @@ export function buildIssueBlockersResolvedWakeIdempotencyKey(input: { ].join(":"); } -export async function findExistingIssueBlockersResolvedWake( - db: Db, - input: { - companyId: string; - idempotencyKey: string; - }, -) { - return db - .select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status }) - .from(agentWakeupRequests) - .where( - and( - eq(agentWakeupRequests.companyId, input.companyId), - eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey), - inArray(agentWakeupRequests.status, [...IDEMPOTENT_DEPENDENCY_WAKE_STATUSES]), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); +/** + * Level-triggered idempotency key. One key encodes the full set of blockers that + * defines the current dependency-ready state. Two wakes for the same ready state + * share the key. A wake for an earlier partial state has a different blocker set, + * so it produces a different key and never suppresses the current wake. All three + * emit paths (route-time, finalize-time, periodic backstop) use this key so they + * share one idempotency rule. + */ +export function buildIssueBlockersResolvedWakeStateKey(input: { + dependentIssueId: string; + blockerIssueIds: string[]; +}) { + const sortedBlockerIssueIds = [...new Set(input.blockerIssueIds.filter(Boolean))].sort(); + const digest = createHash("sha256") + .update(sortedBlockerIssueIds.join(",")) + .digest("hex") + .slice(0, 32); + return [ + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + "state", + input.dependentIssueId, + String(sortedBlockerIssueIds.length), + digest, + ].join(":"); } -export async function findExistingIssueBlockersResolvedWakeForAnyKey( +/** + * Find a wake that already covers the current dependency-ready state of the + * dependent issue. The check is level-triggered: + * + * - The state key matches a wake in any idempotent status (including + * `completed`). This suppresses a duplicate wake for the SAME ready state and + * bounds reconciliation. + * - Each legacy per-edge key matches only a wake that is still in flight + * (`queued`, `deferred_issue_execution`, `claimed`). This prevents a duplicate + * wake while an old-format wake is still pending after a deploy, but it never + * lets a historical completed per-edge wake strand the issue. + * + * Returns the first matching wake or `null`. + */ +export async function findExistingIssueBlockersResolvedWakeForReadyState( db: Db, input: { companyId: string; - idempotencyKeys: string[]; + dependentIssueId: string; + blockerIssueIds: string[]; }, ) { - const idempotencyKeys = [...new Set(input.idempotencyKeys.filter(Boolean))]; - if (idempotencyKeys.length === 0) return null; + const stateKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: input.dependentIssueId, + blockerIssueIds: input.blockerIssueIds, + }); + const legacyKeys = [ + ...new Set( + input.blockerIssueIds + .filter(Boolean) + .map((resolvedBlockerIssueId) => + buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: input.dependentIssueId, + resolvedBlockerIssueId, + }), + ), + ), + ]; + + const stateMatch = and( + eq(agentWakeupRequests.idempotencyKey, stateKey), + inArray(agentWakeupRequests.status, [...IDEMPOTENT_DEPENDENCY_WAKE_STATUSES]), + ); + const legacyMatch = + legacyKeys.length > 0 + ? and( + inArray(agentWakeupRequests.idempotencyKey, legacyKeys), + inArray(agentWakeupRequests.status, [...IN_FLIGHT_DEPENDENCY_WAKE_STATUSES]), + ) + : null; return db .select({ @@ -63,8 +131,7 @@ export async function findExistingIssueBlockersResolvedWakeForAnyKey( .where( and( eq(agentWakeupRequests.companyId, input.companyId), - inArray(agentWakeupRequests.idempotencyKey, idempotencyKeys), - inArray(agentWakeupRequests.status, [...IDEMPOTENT_DEPENDENCY_WAKE_STATUSES]), + legacyMatch ? or(stateMatch, legacyMatch) : stateMatch, ), ) .limit(1) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 193782770d..ec31c026c8 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -48,8 +48,8 @@ import { } from "../issue-execution-policy.js"; import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, - buildIssueBlockersResolvedWakeIdempotencyKey, - findExistingIssueBlockersResolvedWakeForAnyKey, + buildIssueBlockersResolvedWakeStateKey, + findExistingIssueBlockersResolvedWakeForReadyState, } from "../issue-dependency-wakeups.js"; import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; @@ -5248,19 +5248,19 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - const idempotencyKeys = readiness.blockerIssueIds.map((blockerIssueId) => - buildIssueBlockersResolvedWakeIdempotencyKey({ - dependentIssueId: candidate.id, - resolvedBlockerIssueId: blockerIssueId, - }) - ); - const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + // Level-triggered dedup: key on the full blocker set (the current ready + // state), not on any single resolved edge. An older completed per-edge + // wake for an earlier partial resolution has a different key, so it does + // not suppress this wake. The shared helper still suppresses a duplicate + // wake for the SAME ready state, which bounds reconciliation. + const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: candidate.id, - resolvedBlockerIssueId, + blockerIssueIds: readiness.blockerIssueIds, }); - const existingWake = await findExistingIssueBlockersResolvedWakeForAnyKey(db, { + const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId, - idempotencyKeys, + dependentIssueId: candidate.id, + blockerIssueIds: readiness.blockerIssueIds, }); if (existingWake) { result.existingWakeSkipped += 1;