From e0e503e1bc11972560433aba40e47fc7641a5c08 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 19 Aug 2026 12:52:50 -0700 Subject: [PATCH] fix(server): make blockers-resolved wake dedup level-triggered (#11732) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Paperclip uses issue dependencies to pause work until blockers reach a ready state. > - A blocked issue with several blockers can miss its wake when the final blocker completes. > - The wake deduplication used a historical per-edge key, so an old completed wake hid the current ready state. > - This pull request adds a level-triggered key for the sorted set of blocker issue ids and uses one helper for all wake paths. > - The benefit is that the final blocker wake can repair a missed wake, while repeated reconciliation stays bounded. ## Linked Issues or Issue Description Refs #8009, #7853, and #6719. These public pull requests cover related dependency-wake and deduplication behavior. This pull request fixes a separate multi-blocker state-key gap. **What happened?** A blocked issue with multiple blockers received no `issue_blockers_resolved` wake when the final blocker completed. An earlier completed per-edge wake suppressed the wake for the current all-ready state. **Expected behavior** The final blocker completion must emit one wake for the current ready state. A later reconciliation pass must not emit a second wake for the same state. **Steps to reproduce** 1. Create a blocked issue with at least two blocker issues. 2. Complete one blocker and record its completed per-edge wake. 3. Complete the final blocker. 4. Run the route-time or reconciliation wake path. 5. Confirm that one level-triggered wake exists for the sorted blocker set. **Paperclip version or commit** `eed1e5cad91a37547e1b521232da04b9ddb316f0` **Deployment mode** Local dev from source. ## What Changed - Add a SHA-256 level-triggered idempotency key from the sorted blocker issue ids. - Share one deduplication helper across route-time, finalize-time, and periodic wake paths. - Treat state-key rows with idempotent statuses as duplicates. - Treat legacy per-edge rows as duplicates only while they remain in flight. - Record skipped route-time wakes without suppressing later finalize-time or periodic wakes. - Add regression coverage for a completed earlier-blocker wake and a second reconciliation pass. ## Verification - Run `npx vitest run server/src/__tests__/issue-dependency-wakeups-routes.test.ts`. - Run `npx vitest run server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts`. - Run `npx vitest run server/src/__tests__/heartbeat-dependency-scheduling.test.ts`. - Run `npx vitest run server/src/__tests__/issue-rewake-throttle.test.ts server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts`. - Run `tsc --noEmit` on the touched files. ## Risks The change alters wake deduplication for dependency reconciliation. The new key uses the full sorted blocker set, so a change in that set permits a new wake. The regression tests cover the missed-final-blocker case and repeated reconciliation. ## Model Used OpenAI Codex, GPT-5, tool-use model with code execution and repository review support. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- ...eartbeat-issue-liveness-escalation.test.ts | 106 +++++++++++++++- .../issue-dependency-wakeups-routes.test.ts | 7 +- server/src/routes/issues.ts | 22 ++-- .../src/services/issue-dependency-wakeups.ts | 119 ++++++++++++++---- server/src/services/recovery/service.ts | 24 ++-- 5 files changed, 222 insertions(+), 56 deletions(-) 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;