From 3a21bb3b486c5eaf0f031c1cb198661d89b9f359 Mon Sep 17 00:00:00 2001 From: Dev Agent Date: Thu, 27 Aug 2026 14:01:37 +0000 Subject: [PATCH] fix(heartbeat): stop blocked wake churn and prune terminal requests Co-Authored-By: Paperclip --- .../heartbeat-dependency-scheduling.test.ts | 50 +++--- .../heartbeat-process-recovery.test.ts | 40 +++++ .../server-startup-feedback-export.test.ts | 10 ++ .../wakeup-request-retention.test.ts | 155 ++++++++++++++++++ server/src/index.ts | 34 ++++ server/src/services/heartbeat.ts | 21 +-- server/src/services/index.ts | 5 + .../src/services/wakeup-request-retention.ts | 58 +++++++ 8 files changed, 331 insertions(+), 42 deletions(-) create mode 100644 server/src/__tests__/wakeup-request-retention.test.ts create mode 100644 server/src/services/wakeup-request-retention.ts diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index bb875a3f21b..05ce1f22246 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -169,7 +169,7 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = await tempDb?.cleanup(); }); - it("keeps blocked descendants idle until their blockers resolve", async () => { + it("keeps blocked descendants idle without persisting per-tick assignment skips until their blockers resolve", async () => { const companyId = randomUUID(); const agentId = randomUUID(); const blockerId = randomUUID(); @@ -243,28 +243,28 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = }); expect(blockedWake).toBeNull(); - const blockedWakeRequest = await waitForCondition(async () => { - const wakeup = await db - .select({ - status: agentWakeupRequests.status, - reason: agentWakeupRequests.reason, - }) - .from(agentWakeupRequests) - .where( - and( - eq(agentWakeupRequests.agentId, agentId), - sql`${agentWakeupRequests.payload} ->> 'issueId' = ${blockedIssueId}`, - ), - ) - .orderBy(agentWakeupRequests.requestedAt) - .then((rows) => rows[0] ?? null); - return Boolean( - wakeup && - wakeup.status === "skipped" && - wakeup.reason === "issue_dependencies_blocked", - ); - }); - expect(blockedWakeRequest).toBe(true); + for (let tick = 0; tick < 3; tick += 1) { + await expect(heartbeat.wakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId: blockedIssueId }, + contextSnapshot: { issueId: blockedIssueId, wakeReason: "issue_assigned" }, + })).resolves.toBeNull(); + } + + const blockedWakeRequestCount = await db + .select({ count: sql`count(*)::int` }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.reason, "issue_dependencies_blocked"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${blockedIssueId}`, + ), + ) + .then((rows) => rows[0]?.count ?? 0); + expect(blockedWakeRequestCount).toBe(0); const blockedRunsBeforeResolution = await db .select({ count: sql`count(*)::int` }) @@ -396,7 +396,7 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = .from(heartbeatRuns) .where(eq(heartbeatRuns.id, promotedWake!.id)) .then((rows) => rows[0] ?? null); - const blockedWakeRequestCount = await db + const allBlockedIssueWakeRequestCount = await db .select({ count: sql`count(*)::int` }) .from(agentWakeupRequests) .where( @@ -408,7 +408,7 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = .then((rows) => rows[0]?.count ?? 0); expect(promotedBlockedRun?.status).toBe("succeeded"); - expect(blockedWakeRequestCount).toBeGreaterThanOrEqual(2); + expect(allBlockedIssueWakeRequestCount).toBeGreaterThanOrEqual(2); const noActiveRuns = await waitForCondition(async () => { const rows = await db diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 174f086b56c..abda6b1a881 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -5061,6 +5061,46 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } }); + it("does not persist dependency-blocked assignment skips across recovery scheduler ticks", async () => { + const { companyId, agentId, issueId } = await seedAssignedTodoNoRunFixture(); + const blockerIssueId = randomUUID(); + await db.insert(issues).values({ + id: blockerIssueId, + companyId, + title: "Open prerequisite", + status: "todo", + priority: "high", + responsibleUserId: "responsible-user", + }); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerIssueId, + relatedIssueId: issueId, + type: "blocks", + }); + const heartbeat = heartbeatService(db); + + for (let tick = 0; tick < 3; tick += 1) { + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.assignmentDispatched).toBe(0); + expect(result.skipped).toBeGreaterThanOrEqual(1); + } + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.reason, "issue_dependencies_blocked"), + )); + expect(wakeups).toHaveLength(0); + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(0); + }); + it("does not duplicate initial assigned todo dispatch when a queued wake already exists", async () => { const { companyId, agentId, issueId } = await seedAssignedTodoNoRunFixture(); await db.insert(agentWakeupRequests).values({ diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 9477122b7f8..f80bb9e1c5d 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -32,6 +32,7 @@ const { resolveHeartbeatSchedulingSuppressionMock, routineServiceFactoryMock, routineServiceMock, + wakeupRequestRetentionServiceFactoryMock, } = vi.hoisted(() => { const createAppMock = vi.fn(async () => ((_: unknown, __: unknown) => {}) as never); const createBetterAuthInstanceMock = vi.fn(() => ({})); @@ -109,6 +110,13 @@ const { tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })), }; const routineServiceFactoryMock = vi.fn(() => routineServiceMock); + const wakeupRequestRetentionServiceFactoryMock = vi.fn(() => ({ + pruneTerminalRequests: vi.fn(async () => ({ + deleted: 0, + hasMore: false, + cutoff: new Date(0), + })), + })); const feedbackExportServiceMock = { flushPendingFeedbackTraces: vi.fn(async () => ({ attempted: 0, sent: 0, failed: 0 })), }; @@ -148,6 +156,7 @@ const { resolveHeartbeatSchedulingSuppressionMock, routineServiceFactoryMock, routineServiceMock, + wakeupRequestRetentionServiceFactoryMock, }; }); @@ -312,6 +321,7 @@ vi.mock("../services/index.js", () => ({ failed: 0, })), })), + wakeupRequestRetentionService: wakeupRequestRetentionServiceFactoryMock, })); vi.mock("../services/secret-proposals.js", () => ({ diff --git a/server/src/__tests__/wakeup-request-retention.test.ts b/server/src/__tests__/wakeup-request-retention.test.ts new file mode 100644 index 00000000000..9b59e490976 --- /dev/null +++ b/server/src/__tests__/wakeup-request-retention.test.ts @@ -0,0 +1,155 @@ +import { randomUUID } from "node:crypto"; +import { inArray } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + agentWakeupRequests, + companies, + createDb, + heartbeatRuns, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { wakeupRequestRetentionService } from "../services/wakeup-request-retention.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +describeEmbeddedPostgres("agent wakeup request retention", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-wakeup-request-retention-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + it("prunes old skipped and coalesced rows in bounded batches while preserving audit and referenced rows", async () => { + const now = new Date("2026-08-27T12:00:00.000Z"); + const old = new Date("2026-08-19T12:00:00.000Z"); + const recent = new Date("2026-08-26T12:00:00.000Z"); + const companyId = randomUUID(); + const agentId = randomUUID(); + const ids = { + skipped: randomUUID(), + coalesced: randomUUID(), + completed: randomUUID(), + failed: randomUUID(), + recentSkipped: randomUUID(), + referencedSkipped: randomUUID(), + }; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: "RET", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Retention Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(agentWakeupRequests).values([ + { + id: ids.skipped, + companyId, + agentId, + source: "assignment", + status: "skipped", + requestedAt: old, + finishedAt: old, + }, + { + id: ids.coalesced, + companyId, + agentId, + source: "assignment", + status: "coalesced", + requestedAt: old, + finishedAt: old, + runId: randomUUID(), + }, + { + id: ids.completed, + companyId, + agentId, + source: "assignment", + status: "completed", + requestedAt: old, + finishedAt: old, + }, + { + id: ids.failed, + companyId, + agentId, + source: "assignment", + status: "failed", + requestedAt: old, + finishedAt: old, + }, + { + id: ids.recentSkipped, + companyId, + agentId, + source: "assignment", + status: "skipped", + requestedAt: recent, + finishedAt: recent, + }, + { + id: ids.referencedSkipped, + companyId, + agentId, + source: "assignment", + status: "skipped", + requestedAt: old, + finishedAt: old, + runId: randomUUID(), + }, + ]); + await db.insert(heartbeatRuns).values({ + companyId, + agentId, + invocationSource: "assignment", + status: "failed", + wakeupRequestId: ids.referencedSkipped, + finishedAt: old, + }); + + const retention = wakeupRequestRetentionService(db); + await expect(retention.pruneTerminalRequests({ now, batchSize: 1 })).resolves.toMatchObject({ + deleted: 1, + hasMore: true, + }); + await expect(retention.pruneTerminalRequests({ now, batchSize: 100 })).resolves.toMatchObject({ + deleted: 1, + hasMore: false, + }); + + const remaining = await db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where(inArray(agentWakeupRequests.id, Object.values(ids))) + .then((rows) => rows.map((row) => row.id).sort()); + + expect(remaining).toEqual([ + ids.completed, + ids.failed, + ids.recentSkipped, + ids.referencedSkipped, + ].sort()); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 3890076b041..dbe45df520b 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -65,6 +65,7 @@ import { routineService, statusCardService, toolAccessService, + wakeupRequestRetentionService, workspaceOperationService, } from "./services/index.js"; import { queueIssueAssignmentWakeup } from "./services/issue-assignment-wakeup.js"; @@ -1132,6 +1133,37 @@ export async function startServer(): Promise { trackHeartbeatSchedulerWork(runEnvironmentLeaseCleanupSweep(ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS)); }; + const wakeupRequestRetention = wakeupRequestRetentionService(db as any); + let wakeupRequestRetentionRunning = false; + const runWakeupRequestRetentionSweep = async (drainBacklog: boolean) => { + if (wakeupRequestRetentionRunning) return; + wakeupRequestRetentionRunning = true; + let deleted = 0; + try { + do { + const result = await wakeupRequestRetention.pruneTerminalRequests(); + deleted += result.deleted; + if (!drainBacklog || !result.hasMore || heartbeatSchedulerStopped) break; + await new Promise((resolveYield) => setImmediate(resolveYield)); + } while (true); + if (deleted > 0) { + logger.info({ deleted }, "agent wakeup request retention sweep pruned terminal rows"); + } + } finally { + wakeupRequestRetentionRunning = false; + } + }; + const scheduleWakeupRequestRetentionSweep = (drainBacklog = false) => { + if (heartbeatSchedulerStopped) return; + trackHeartbeatSchedulerWork(runWakeupRequestRetentionSweep(drainBacklog).catch((err) => { + logger.error({ err }, "agent wakeup request retention sweep failed"); + })); + }; + + // Drain historical skipped/coalesced rows without delaying server startup. + // Later scheduler ticks keep the table inside the same retention bound. + scheduleWakeupRequestRetentionSweep(true); + if (heartbeat) { const secretProposals = createSecretProposalsService(db as any); const decisionExecutor = decisionService(db as any, decisionServiceOptions); @@ -1438,6 +1470,7 @@ export async function startServer(): Promise { trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => { logger.error({ err }, "decision retention sweep failed"); })); + scheduleWakeupRequestRetentionSweep(); const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses(); if (sweptRuntimeStatuses > 0) { logger.info( @@ -1613,6 +1646,7 @@ export async function startServer(): Promise { startHeartbeatSchedulerInterval(() => { scheduleExternalObjectRefreshSweep(new Date()); scheduleEnvironmentLeaseCleanupSweep(); + scheduleWakeupRequestRetentionSweep(); }); } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 69419dad9de..6f63a648551 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -18486,23 +18486,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } if (!activeExecutionRun && dependencyReadiness && !dependencyReadiness.isDependencyReady && !blockedInteractionWake) { - await tx.insert(agentWakeupRequests).values({ - companyId: agent.companyId, - agentId, - source, - triggerDetail, - reason: "issue_dependencies_blocked", - payload: { - ...(payload ?? {}), - issueId, - unresolvedBlockerIssueIds: dependencyReadiness.unresolvedBlockerIssueIds, - }, - status: "skipped", - requestedByActorType: opts.requestedByActorType ?? null, - requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, - finishedAt: new Date(), - }); + // Dependency resolution owns the durable wake path. Persisting a + // skipped assignment request here makes the scheduler insert one row + // per blocked issue on every reconciliation tick without changing any + // runnable state. return { kind: "skipped" as const }; } diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 54e56fc936a..85831209001 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -177,6 +177,11 @@ export { type WorkspaceGitSchedulerSnapshot, } from "./workspace-git-operation-scheduler.js"; export { workProductService } from "./work-products.js"; +export { + wakeupRequestRetentionService, + WAKEUP_REQUEST_RETENTION_BATCH_SIZE, + WAKEUP_REQUEST_TERMINAL_RETENTION_MS, +} from "./wakeup-request-retention.js"; export { logActivity, persistActivity, diff --git a/server/src/services/wakeup-request-retention.ts b/server/src/services/wakeup-request-retention.ts new file mode 100644 index 00000000000..5c21bb0a47e --- /dev/null +++ b/server/src/services/wakeup-request-retention.ts @@ -0,0 +1,58 @@ +import { sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; + +export const WAKEUP_REQUEST_TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60 * 1000; +export const WAKEUP_REQUEST_RETENTION_BATCH_SIZE = 10_000; + +export function wakeupRequestRetentionService(db: Db) { + return { + async pruneTerminalRequests(opts?: { now?: Date; batchSize?: number }) { + const now = opts?.now ?? new Date(); + const batchSize = Math.max(1, Math.floor(opts?.batchSize ?? WAKEUP_REQUEST_RETENTION_BATCH_SIZE)); + const cutoff = new Date(now.getTime() - WAKEUP_REQUEST_TERMINAL_RETENTION_MS); + const cutoffIso = cutoff.toISOString(); + const companyRows = await db.execute(sql<{ id: string }>` + select id + from companies + order by id + `); + + let deleted = 0; + for (const company of Array.from(companyRows)) { + const remaining = batchSize - deleted; + if (remaining <= 0) break; + + // The company/requested_at index bounds candidate discovery. A skipped + // request with a run_id became part of the heartbeat audit trail and is + // retained. Coalesced requests point at another request's run and never + // own a heartbeat_runs.wakeup_request_id reference. + const deletedRows = await db.execute(sql<{ id: string }>` + with candidates as materialized ( + select request.id + from agent_wakeup_requests request + where request.company_id = ${company.id} + and request.requested_at < ${cutoffIso}::timestamptz + and ( + (request.status = 'skipped' and request.run_id is null) + or request.status = 'coalesced' + ) + order by request.requested_at, request.id + limit ${remaining} + for update skip locked + ) + delete from agent_wakeup_requests request + using candidates + where request.id = candidates.id + returning request.id + `); + deleted += Array.from(deletedRows).length; + } + + return { + deleted, + hasMore: deleted === batchSize, + cutoff, + }; + }, + }; +}