diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 1f6d5b72a8..7619972cd2 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -72,7 +72,10 @@ 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"; +import { + buildIssueBlockersResolvedWakeStateKey, + buildIssueBlockersResolvedWakeStateKeyWithoutCycle, +} from "../services/issue-dependency-wakeups.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -812,6 +815,101 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { expect(stateKeyWakes).toHaveLength(1); }); + it("heals a blocked dependent after a terminal reset when a previous-cycle old-key wake exists", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const previousCycleWakeAt = new Date("2026-07-01T12:00:00.000Z"); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db + .update(issues) + .set({ blockedTransitionAt, updatedAt: blockedTransitionAt }) + .where(eq(issues.id, blockedIssueId)); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { + issueId: blockedIssueId, + resolvedBlockerIssueId: blockerIssueId, + blockerIssueIds: [blockerIssueId], + }, + status: "completed", + finishedAt: previousCycleWakeAt, + requestedAt: previousCycleWakeAt, + idempotencyKey: buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), + }); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.dependencyWakesHealed).toBe(1); + expect(result.dependencyWakeIssueIds).toEqual([blockedIssueId]); + expect(result.dependencyWakeExistingSkipped).toBe(0); + + const cycleKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt, + }); + const healedWake = await db + .select({ status: agentWakeupRequests.status, idempotencyKey: agentWakeupRequests.idempotencyKey }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.idempotencyKey, cycleKey))) + .then((rows) => rows[0] ?? null); + expect(healedWake).not.toBeNull(); + expect(["queued", "claimed", "completed"]).toContain(healedWake?.status); + + const secondPass = await heartbeatService(db).reconcileIssueGraphLiveness(); + expect(secondPass.dependencyWakesHealed).toBe(0); + + const cycleKeyWakes = await db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.idempotencyKey, cycleKey))); + expect(cycleKeyWakes).toHaveLength(1); + }); + + it("does not re-heal when a completed old-key wake is from the current blocked cycle", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + const sameCycleWakeAt = new Date("2026-08-01T12:00:01.000Z"); + await db + .update(issues) + .set({ blockedTransitionAt, updatedAt: blockedTransitionAt }) + .where(eq(issues.id, blockedIssueId)); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { + issueId: blockedIssueId, + resolvedBlockerIssueId: blockerIssueId, + blockerIssueIds: [blockerIssueId], + }, + status: "completed", + finishedAt: sameCycleWakeAt, + requestedAt: sameCycleWakeAt, + idempotencyKey: buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId: blockedIssueId, + blockerIssueIds: [blockerIssueId], + }), + }); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.dependencyWakesHealed).toBe(0); + expect(result.dependencyWakeExistingSkipped).toBe(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 a840a81a4a..16cf34a20a 100644 --- a/server/src/__tests__/issue-dependency-wakeups-routes.test.ts +++ b/server/src/__tests__/issue-dependency-wakeups-routes.test.ts @@ -1,6 +1,9 @@ import express from "express"; import request from "supertest"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import { + buildIssueBlockersResolvedWakeStateKey, +} from "../services/issue-dependency-wakeups.ts"; // The first test in this suite imports the large `routes/issues.ts` module // through `vi.importActual` inside `createApp`. `vi.resetModules()` in @@ -11,7 +14,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; // default 5000ms test timeout and produce a flaky "Test timed out in 5000ms" // failure. Give the suite generous headroom, far above the observed cold-start // yet still below the 30s hook timeout. -vi.setConfig({ testTimeout: 15000 }); +vi.setConfig({ testTimeout: 30000 }); const mockWakeup = vi.hoisted(() => vi.fn(async () => undefined)); const mockFindExistingIssueBlockersResolvedWakeForReadyState = vi.hoisted(() => vi.fn(async () => null)); @@ -402,4 +405,280 @@ describe("issue dependency wakeups in issue routes", () => { ); }); }); + + function issueRecord(overrides: Record = {}) { + return { + id: "issue-1", + companyId: "company-1", + identifier: "PAP-100", + title: "Issue", + description: null, + status: "todo", + priority: "medium", + parentId: null, + assigneeAgentId: "agent-1", + assigneeUserId: null, + createdByAgentId: null, + createdByUserId: null, + executionWorkspaceId: null, + blockedTransitionAt: null, + labels: [], + labelIds: [], + ...overrides, + }; + } + + it("wakes a Release-like dependent after a terminal reset using the current blocked cycle", async () => { + const reviewIssueId = "11111111-1111-4111-8111-111111111111"; + const releaseIssueId = "22222222-2222-4222-8222-222222222222"; + const releaseBlockedAt = new Date("2026-08-01T15:00:00.000Z"); + mockIssueService.getById.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "in_progress", + })); + mockIssueService.update.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "done", + })); + mockIssueService.listWakeableBlockedDependents.mockResolvedValue([ + { + id: releaseIssueId, + assigneeAgentId: "agent-release", + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: releaseBlockedAt, + }, + ]); + + const res = await request(await createApp()).patch(`/api/issues/${reviewIssueId}`).send({ status: "done" }); + expect(res.status).toBe(200); + await vi.waitFor(() => { + expect(mockFindExistingIssueBlockersResolvedWakeForReadyState).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + companyId: "company-1", + dependentIssueId: releaseIssueId, + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: releaseBlockedAt, + }), + ); + expect(mockWakeup).toHaveBeenCalledTimes(1); + expect(mockWakeup).toHaveBeenCalledWith( + "agent-release", + expect.objectContaining({ + reason: "issue_blockers_resolved", + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: releaseIssueId, + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: releaseBlockedAt, + }), + payload: expect.objectContaining({ + issueId: releaseIssueId, + resolvedBlockerIssueId: reviewIssueId, + mutation: "blocker_done", + }), + }), + ); + }); + }); + + it("does not enqueue a second wake when the blocker is already done", async () => { + const reviewIssueId = "11111111-1111-4111-8111-111111111111"; + mockIssueService.getById.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "done", + })); + mockIssueService.update.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "done", + })); + mockIssueService.listWakeableBlockedDependents.mockResolvedValue([ + { + id: "22222222-2222-4222-8222-222222222222", + assigneeAgentId: "agent-release", + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: new Date("2026-08-01T15:00:00.000Z"), + }, + ]); + + const res = await request(await createApp()).patch(`/api/issues/${reviewIssueId}`).send({ status: "done" }); + expect(res.status).toBe(200); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(mockWakeup).not.toHaveBeenCalled(); + expect(mockIssueService.listWakeableBlockedDependents).not.toHaveBeenCalled(); + }); + + it("wakes a QA-like chain one dependent at a time after each blocker completes", async () => { + const reviewIssueId = "11111111-1111-4111-8111-111111111111"; + const releaseIssueId = "22222222-2222-4222-8222-222222222222"; + const qaIssueId = "33333333-3333-4333-8333-333333333333"; + const releaseBlockedAt = new Date("2026-08-02T10:00:00.000Z"); + const qaBlockedAt = new Date("2026-08-02T10:05:00.000Z"); + const app = await createApp(); + + mockIssueService.getById.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "in_progress", + })); + mockIssueService.update.mockResolvedValue(issueRecord({ + id: reviewIssueId, + identifier: "PAP-REVIEW", + title: "Review", + status: "done", + })); + mockIssueService.listWakeableBlockedDependents.mockResolvedValue([ + { + id: releaseIssueId, + assigneeAgentId: "agent-release", + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: releaseBlockedAt, + }, + ]); + + expect((await request(app).patch(`/api/issues/${reviewIssueId}`).send({ status: "done" })).status).toBe(200); + await vi.waitFor(() => { + expect(mockWakeup).toHaveBeenCalledTimes(1); + }); + expect(mockWakeup).toHaveBeenCalledWith( + "agent-release", + expect.objectContaining({ + payload: expect.objectContaining({ issueId: releaseIssueId }), + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: releaseIssueId, + blockerIssueIds: [reviewIssueId], + blockedTransitionAt: releaseBlockedAt, + }), + }), + ); + + mockWakeup.mockClear(); + mockFindExistingIssueBlockersResolvedWakeForReadyState.mockClear(); + mockIssueService.getById.mockResolvedValue(issueRecord({ + id: releaseIssueId, + identifier: "PAP-RELEASE", + title: "Release", + status: "blocked", + assigneeAgentId: "agent-release", + blockedTransitionAt: releaseBlockedAt, + })); + mockIssueService.update.mockResolvedValue(issueRecord({ + id: releaseIssueId, + identifier: "PAP-RELEASE", + title: "Release", + status: "done", + assigneeAgentId: "agent-release", + })); + mockIssueService.listWakeableBlockedDependents.mockResolvedValue([ + { + id: qaIssueId, + assigneeAgentId: "agent-qa", + blockerIssueIds: [releaseIssueId], + blockedTransitionAt: qaBlockedAt, + }, + ]); + + expect((await request(app).patch(`/api/issues/${releaseIssueId}`).send({ status: "done" })).status).toBe(200); + await vi.waitFor(() => { + expect(mockWakeup).toHaveBeenCalledTimes(1); + }); + expect(mockWakeup).toHaveBeenCalledWith( + "agent-qa", + expect.objectContaining({ + payload: expect.objectContaining({ + issueId: qaIssueId, + resolvedBlockerIssueId: releaseIssueId, + }), + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: qaIssueId, + blockerIssueIds: [releaseIssueId], + blockedTransitionAt: qaBlockedAt, + }), + }), + ); + expect(mockWakeup).not.toHaveBeenCalledWith("agent-release", expect.anything()); + }); + + it("restores a blocked-and-ready dependent under the new blocked cycle key", async () => { + const parentIssueId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + const childIssueId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; + const blockedTransitionAt = new Date("2026-08-03T18:00:00.000Z"); + mockIssueService.getById.mockResolvedValue(issueRecord({ + id: parentIssueId, + identifier: "PAP-200", + title: "Blocked after completion", + status: "done", + assigneeAgentId: "agent-2", + })); + mockIssueService.update.mockResolvedValue(issueRecord({ + id: parentIssueId, + identifier: "PAP-200", + title: "Blocked after completion", + status: "blocked", + assigneeAgentId: "agent-2", + blockedTransitionAt, + })); + mockIssueService.getDependencyReadiness.mockResolvedValue({ + issueId: parentIssueId, + blockerIssueIds: [childIssueId], + unresolvedBlockerIssueIds: [], + unresolvedBlockerCount: 0, + pendingFinalizeBlockerIssueIds: [], + allBlockersDone: true, + isDependencyReady: true, + }); + + const res = await request(await createApp()) + .patch(`/api/issues/${parentIssueId}`) + .send({ + status: "blocked", + blockedByIssueIds: [childIssueId], + unblockDescriptor: { owner: "board", action: "Review the restored dependency" }, + }); + + expect(res.status).toBe(200); + await vi.waitFor(() => { + expect(mockFindExistingIssueBlockersResolvedWakeForReadyState).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + dependentIssueId: parentIssueId, + blockerIssueIds: [childIssueId], + blockedTransitionAt, + }), + ); + expect(mockWakeup).toHaveBeenCalledWith( + "agent-2", + expect.objectContaining({ + reason: "issue_blockers_resolved", + idempotencyKey: buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId: parentIssueId, + blockerIssueIds: [childIssueId], + blockedTransitionAt, + }), + payload: expect.objectContaining({ + mutation: "blocked_dependency_restored", + }), + }), + ); + }); + }); + + it("does not emit a dependency wake when an unresolved or cancelled blocker remains", async () => { + mockIssueService.getById.mockResolvedValue(issueRecord({ status: "in_progress" })); + mockIssueService.update.mockResolvedValue(issueRecord({ status: "done" })); + mockIssueService.listWakeableBlockedDependents.mockResolvedValue([]); + + const res = await request(await createApp()).patch("/api/issues/issue-1").send({ status: "done" }); + expect(res.status).toBe(200); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(mockWakeup).not.toHaveBeenCalled(); + }); }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index ec1604b950..b5e6a3adb2 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -10557,18 +10557,21 @@ export function issueRoutes( dependentIssueId: string; resolvedBlockerIssueId: string; blockerIssueIds: string[]; + blockedTransitionAt?: Date | string | null; source: string; mutation: string; }) => { const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: input.dependentIssueId, blockerIssueIds: input.blockerIssueIds, + blockedTransitionAt: input.blockedTransitionAt, }); try { const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId: issue.companyId, dependentIssueId: input.dependentIssueId, blockerIssueIds: input.blockerIssueIds, + blockedTransitionAt: input.blockedTransitionAt, }); if (existingWake) return; } catch (err) { @@ -10753,6 +10756,7 @@ export function issueRoutes( dependentIssueId: dependent.id, resolvedBlockerIssueId: issue.id, blockerIssueIds: dependent.blockerIssueIds, + blockedTransitionAt: dependent.blockedTransitionAt, source: "issue.blockers_resolved", mutation: "blocker_done", }); @@ -10780,6 +10784,7 @@ export function issueRoutes( dependentIssueId: issue.id, resolvedBlockerIssueId, blockerIssueIds: readiness.blockerIssueIds, + blockedTransitionAt: issue.blockedTransitionAt, source: "issue.blockers_restored", mutation: "blocked_dependency_restored", }); @@ -12612,16 +12617,19 @@ export function issueRoutes( dependentIssueId: string; resolvedBlockerIssueId: string; blockerIssueIds: string[]; + blockedTransitionAt?: Date | string | null; }) => { const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: input.dependentIssueId, blockerIssueIds: input.blockerIssueIds, + blockedTransitionAt: input.blockedTransitionAt, }); try { const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId: currentIssue.companyId, dependentIssueId: input.dependentIssueId, blockerIssueIds: input.blockerIssueIds, + blockedTransitionAt: input.blockedTransitionAt, }); if (existingWake) return; } catch (err) { @@ -12785,6 +12793,7 @@ export function issueRoutes( dependentIssueId: dependent.id, resolvedBlockerIssueId: currentIssue.id, blockerIssueIds: dependent.blockerIssueIds, + blockedTransitionAt: dependent.blockedTransitionAt, }); } } diff --git a/server/src/services/issue-dependency-wakeups.test.ts b/server/src/services/issue-dependency-wakeups.test.ts new file mode 100644 index 0000000000..3ed0555417 --- /dev/null +++ b/server/src/services/issue-dependency-wakeups.test.ts @@ -0,0 +1,231 @@ +import { describe, expect, it } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { + buildIssueBlockersResolvedWakeIdempotencyKey, + buildIssueBlockersResolvedWakeStateKey, + buildIssueBlockersResolvedWakeStateKeyWithoutCycle, + findExistingIssueBlockersResolvedWakeForReadyState, +} from "./issue-dependency-wakeups.js"; + +const dependentIssueId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; +const blockerIssueId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; +const companyId = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"; +const firstCycle = new Date("2026-04-01T12:00:00.000Z"); +const secondCycle = new Date("2026-08-01T09:30:00.000Z"); + +type WakeRow = { + id: string; + status: string; + idempotencyKey: string | null; + requestedAt: Date; +}; + +function dbWithWakes(rows: WakeRow[]): Db { + return { + select() { + return { + from() { + return { + where() { + return Promise.resolve(rows); + }, + }; + }, + }; + }, + } as unknown as Db; +} + +describe("buildIssueBlockersResolvedWakeStateKey", () => { + it("is identical for the same dependent, blockers, and blockedTransitionAt", () => { + const first = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: firstCycle, + }); + const second = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: firstCycle.toISOString(), + }); + expect(first).toBe(second); + expect(first).toContain(dependentIssueId); + }); + + it("changes when blockedTransitionAt changes", () => { + const first = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: firstCycle, + }); + const second = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: secondCycle, + }); + expect(first).not.toBe(second); + }); + + it("hashes a null cycle as none and differs from any timestamp", () => { + const noneKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: null, + }); + const omittedKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }); + const datedKey = buildIssueBlockersResolvedWakeStateKey({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: firstCycle, + }); + expect(noneKey).toBe(omittedKey); + expect(noneKey).not.toBe(datedKey); + expect(noneKey).not.toBe( + buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }), + ); + }); +}); + +describe("findExistingIssueBlockersResolvedWakeForReadyState", () => { + const readyState = { + companyId, + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: secondCycle, + }; + + it("suppresses a completed wake on the cycle-aware state key", async () => { + const cycleKey = buildIssueBlockersResolvedWakeStateKey(readyState); + const existing = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-cycle", + status: "completed", + idempotencyKey: cycleKey, + requestedAt: secondCycle, + }, + ]), + readyState, + ); + expect(existing?.id).toBe("wake-cycle"); + }); + + it("does not let a completed old-key wake from a previous blocked cycle suppress", async () => { + const oldKey = buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }); + const existing = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-old-previous-cycle", + status: "completed", + idempotencyKey: oldKey, + requestedAt: firstCycle, + }, + ]), + readyState, + ); + expect(existing).toBeNull(); + }); + + it("suppresses a completed old-key wake requested at or after blockedTransitionAt", async () => { + const oldKey = buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }); + const existing = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-old-same-cycle", + status: "completed", + idempotencyKey: oldKey, + requestedAt: secondCycle, + }, + ]), + readyState, + ); + expect(existing?.id).toBe("wake-old-same-cycle"); + }); + + it("suppresses a completed old-key wake when blockedTransitionAt is null", async () => { + const oldKey = buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }); + const existing = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-old-no-cycle", + status: "completed", + idempotencyKey: oldKey, + requestedAt: firstCycle, + }, + ]), + { + companyId, + dependentIssueId, + blockerIssueIds: [blockerIssueId], + blockedTransitionAt: null, + }, + ); + expect(existing?.id).toBe("wake-old-no-cycle"); + }); + + it("suppresses an in-flight old-key wake across a later blocked cycle", async () => { + const oldKey = buildIssueBlockersResolvedWakeStateKeyWithoutCycle({ + dependentIssueId, + blockerIssueIds: [blockerIssueId], + }); + const existing = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-old-queued", + status: "queued", + idempotencyKey: oldKey, + requestedAt: firstCycle, + }, + ]), + readyState, + ); + expect(existing?.id).toBe("wake-old-queued"); + }); + + it("keeps legacy per-edge matching in-flight only", async () => { + const legacyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId, + resolvedBlockerIssueId: blockerIssueId, + }); + const inFlight = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-legacy-claimed", + status: "claimed", + idempotencyKey: legacyKey, + requestedAt: firstCycle, + }, + ]), + readyState, + ); + expect(inFlight?.id).toBe("wake-legacy-claimed"); + + const completed = await findExistingIssueBlockersResolvedWakeForReadyState( + dbWithWakes([ + { + id: "wake-legacy-completed", + status: "completed", + idempotencyKey: legacyKey, + requestedAt: firstCycle, + }, + ]), + readyState, + ); + expect(completed).toBeNull(); + }); +}); diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts index bcd565a8b4..c40d7ff0fd 100644 --- a/server/src/services/issue-dependency-wakeups.ts +++ b/server/src/services/issue-dependency-wakeups.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, or } from "drizzle-orm"; +import { and, eq, inArray } from "drizzle-orm"; import { createHash } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests } from "@paperclipai/db"; @@ -19,14 +19,62 @@ const IDEMPOTENT_DEPENDENCY_WAKE_STATUSES = [ // 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. +// current ready state. The dedup uses this set only for the legacy per-edge key +// and for old no-cycle state keys that are still queued after a deploy. const IN_FLIGHT_DEPENDENCY_WAKE_STATUSES = [ "queued", "deferred_issue_execution", "claimed", ] as const; +const IDEMPOTENT_DEPENDENCY_WAKE_STATUS_SET = new Set(IDEMPOTENT_DEPENDENCY_WAKE_STATUSES); +const IN_FLIGHT_DEPENDENCY_WAKE_STATUS_SET = new Set(IN_FLIGHT_DEPENDENCY_WAKE_STATUSES); + +export type IssueBlockersResolvedWakeCycleInput = Date | string | null | undefined; + +export type IssueBlockersResolvedReadyStateInput = { + dependentIssueId: string; + blockerIssueIds: string[]; + blockedTransitionAt?: IssueBlockersResolvedWakeCycleInput; +}; + +/** + * Canonical blocked-cycle stamp for the dependency-ready state key. + * `blockedTransitionAt` is UTC ISO-8601, or `none` when the dependent has no + * recorded transition into `blocked`. + */ +export function formatIssueBlockersResolvedWakeCycle( + blockedTransitionAt: IssueBlockersResolvedWakeCycleInput, +): string { + if (blockedTransitionAt == null || blockedTransitionAt === "") return "none"; + const parsed = blockedTransitionAt instanceof Date + ? blockedTransitionAt + : new Date(blockedTransitionAt); + if (Number.isNaN(parsed.getTime())) return "none"; + return parsed.toISOString(); +} + +function uniqueSortedBlockerIssueIds(blockerIssueIds: string[]): string[] { + return [...new Set(blockerIssueIds.filter(Boolean))].sort(); +} + +function hashBlockerReadyStateDigest(sortedBlockerIssueIds: string[], cycle: string | null): string { + const payload = cycle == null + ? sortedBlockerIssueIds.join(",") + : `${sortedBlockerIssueIds.join(",")}\n${cycle}`; + return createHash("sha256").update(payload).digest("hex").slice(0, 32); +} + +function buildStateKey(dependentIssueId: string, digest: string, blockerCount: number): string { + return [ + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + "state", + dependentIssueId, + String(blockerCount), + digest, + ].join(":"); +} + /** * Legacy per-edge idempotency key. One key encodes a single resolved blocker * edge `issue_blockers_resolved:{dependentIssueId}:{resolvedBlockerIssueId}`. @@ -45,42 +93,96 @@ export function buildIssueBlockersResolvedWakeIdempotencyKey(input: { } /** - * 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. + * Pre-cycle level-triggered key. Rows written before the ready state included + * `blockedTransitionAt` hashed only the sorted blocker ids. Lookup still reads + * this format so an in-flight deploy-overlap wake can suppress a duplicate. */ -export function buildIssueBlockersResolvedWakeStateKey(input: { +export function buildIssueBlockersResolvedWakeStateKeyWithoutCycle(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", + const sortedBlockerIssueIds = uniqueSortedBlockerIssueIds(input.blockerIssueIds); + return buildStateKey( input.dependentIssueId, - String(sortedBlockerIssueIds.length), - digest, - ].join(":"); + hashBlockerReadyStateDigest(sortedBlockerIssueIds, null), + sortedBlockerIssueIds.length, + ); +} + +/** + * Level-triggered idempotency key. One key encodes the full set of blockers that + * defines the current dependency-ready state plus the dependent's current + * blocked cycle (`blockedTransitionAt`, or `none`). Two wakes for the same ready + * state share the key. A wake from an earlier blocked cycle has a different + * cycle stamp, 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: IssueBlockersResolvedReadyStateInput) { + const sortedBlockerIssueIds = uniqueSortedBlockerIssueIds(input.blockerIssueIds); + const cycle = formatIssueBlockersResolvedWakeCycle(input.blockedTransitionAt); + return buildStateKey( + input.dependentIssueId, + hashBlockerReadyStateDigest(sortedBlockerIssueIds, cycle), + sortedBlockerIssueIds.length, + ); +} + +function parseWakeCycleDate(blockedTransitionAt: IssueBlockersResolvedWakeCycleInput): Date | null { + if (blockedTransitionAt == null || blockedTransitionAt === "") return null; + const parsed = blockedTransitionAt instanceof Date + ? blockedTransitionAt + : new Date(blockedTransitionAt); + if (Number.isNaN(parsed.getTime())) return null; + return parsed; +} + +function wakeCoversIssueBlockersResolvedReadyState( + wake: { + status: string; + idempotencyKey: string | null; + requestedAt: Date; + }, + keys: { + cycleKey: string; + oldStateKey: string; + legacyKeys: Set; + blockedTransitionAt: Date | null; + }, +): boolean { + const idempotencyKey = wake.idempotencyKey; + if (!idempotencyKey) return false; + + if (idempotencyKey === keys.cycleKey) { + return IDEMPOTENT_DEPENDENCY_WAKE_STATUS_SET.has(wake.status); + } + + if (idempotencyKey === keys.oldStateKey) { + if (IN_FLIGHT_DEPENDENCY_WAKE_STATUS_SET.has(wake.status)) return true; + if (wake.status !== "completed") return false; + if (!keys.blockedTransitionAt) return true; + return wake.requestedAt.getTime() >= keys.blockedTransitionAt.getTime(); + } + + if (keys.legacyKeys.has(idempotencyKey)) { + return IN_FLIGHT_DEPENDENCY_WAKE_STATUS_SET.has(wake.status); + } + + return false; } /** * Find a wake that already covers the current dependency-ready state of the - * dependent issue. The check is level-triggered: + * dependent issue. The check is level-triggered and cycle-aware: * - * - 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. + * - The cycle-aware state key matches a wake in any idempotent status + * (including `completed`). This suppresses a duplicate for the SAME ready + * state, including the current blocked cycle. + * - The old no-cycle state key matches in-flight statuses (deploy overlap), + * or a `completed` wake whose `requestedAt` is at or after the current + * `blockedTransitionAt` (same cycle). A completed old-key wake from a + * previous cycle does not suppress. + * - Each legacy per-edge key matches only a wake that is still in flight. * * Returns the first matching wake or `null`. */ @@ -90,13 +192,12 @@ export async function findExistingIssueBlockersResolvedWakeForReadyState( companyId: string; dependentIssueId: string; blockerIssueIds: string[]; + blockedTransitionAt?: IssueBlockersResolvedWakeCycleInput; }, ) { - const stateKey = buildIssueBlockersResolvedWakeStateKey({ - dependentIssueId: input.dependentIssueId, - blockerIssueIds: input.blockerIssueIds, - }); - const legacyKeys = [ + const cycleKey = buildIssueBlockersResolvedWakeStateKey(input); + const oldStateKey = buildIssueBlockersResolvedWakeStateKeyWithoutCycle(input); + const legacyKeyList = [ ...new Set( input.blockerIssueIds .filter(Boolean) @@ -108,32 +209,31 @@ export async function findExistingIssueBlockersResolvedWakeForReadyState( ), ), ]; + const lookupKeys = [...new Set([cycleKey, oldStateKey, ...legacyKeyList])]; + const blockedTransitionAt = parseWakeCycleDate(input.blockedTransitionAt); - 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 + const rows = await db .select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status, idempotencyKey: agentWakeupRequests.idempotencyKey, + requestedAt: agentWakeupRequests.requestedAt, }) .from(agentWakeupRequests) .where( and( eq(agentWakeupRequests.companyId, input.companyId), - legacyMatch ? or(stateMatch, legacyMatch) : stateMatch, + inArray(agentWakeupRequests.idempotencyKey, lookupKeys), ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); + ); + + const covering = rows.find((row) => + wakeCoversIssueBlockersResolvedReadyState(row, { + cycleKey, + oldStateKey, + legacyKeys: new Set(legacyKeyList), + blockedTransitionAt, + }), + ); + return covering ?? null; } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 39d2f2545b..655f6c5af4 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -6571,6 +6571,7 @@ export function issueService(db: Db) { id: issues.id, assigneeAgentId: issues.assigneeAgentId, status: issues.status, + blockedTransitionAt: issues.blockedTransitionAt, }) .from(issueRelations) .innerJoin(issues, eq(issueRelations.relatedIssueId, issues.id)) @@ -6610,6 +6611,7 @@ export function issueService(db: Db) { id: candidate.id, assigneeAgentId: candidate.assigneeAgentId!, blockerIssueIds: readiness.blockerIssueIds, + blockedTransitionAt: candidate.blockedTransitionAt, })); }, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f715479178..4d3f08d9ab 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5712,6 +5712,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) companyId: issues.companyId, identifier: issues.identifier, assigneeAgentId: issues.assigneeAgentId, + blockedTransitionAt: issues.blockedTransitionAt, totalCount: sql`count(*) over()::int`, }) .from(issueRelations) @@ -5727,6 +5728,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) companyId: issues.companyId, identifier: issues.identifier, assigneeAgentId: issues.assigneeAgentId, + blockedTransitionAt: issues.blockedTransitionAt, totalCount: sql`count(*) over()::int`, }) .from(issues) @@ -5800,11 +5802,13 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) const idempotencyKey = buildIssueBlockersResolvedWakeStateKey({ dependentIssueId: candidate.id, blockerIssueIds: readiness.blockerIssueIds, + blockedTransitionAt: candidate.blockedTransitionAt, }); const existingWake = await findExistingIssueBlockersResolvedWakeForReadyState(db, { companyId, dependentIssueId: candidate.id, blockerIssueIds: readiness.blockerIssueIds, + blockedTransitionAt: candidate.blockedTransitionAt, }); if (existingWake) { result.existingWakeSkipped += 1;