fix(heartbeat): stop blocked wake churn and prune terminal requests

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<number>`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<number>`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<number>`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

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({

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", () => ({

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<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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());
+  });
+});

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<StartedServer> {
     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<void>((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<StartedServer> {
         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<StartedServer> {
     startHeartbeatSchedulerInterval(() => {
       scheduleExternalObjectRefreshSweep(new Date());
       scheduleEnvironmentLeaseCleanupSweep();
+      scheduleWakeupRequestRetentionSweep();
     });
   }
   

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 };
         }
 

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,

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,
+      };
+    },
+  };
+}