Merge repaired plugin scheduling into tool health migration

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

server/src/__tests__/approval-queue-seeding-work.test.ts

@@ -282,6 +282,68 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     draining = false; await seeded(approval.id); await quiet();
   });
 
+  it("backs off persistent projection failures without forgetting the owed membership", async () => {
+    const service = approvalQueueSeedingService(db);
+    await service.sweep();
+    const approval = await pending();
+    service.request(new Set([companyId]));
+    await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin raise exception 'projection unavailable'; end $$`);
+    await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail()`);
+    const transactions = vi.spyOn(db, "transaction");
+    try {
+      for (let second = 0; second < 120; second++) {
+        service.request(new Set([companyId]));
+        await service.sweep().catch(() => undefined);
+        await vi.advanceTimersByTimeAsync(1000);
+      }
+      expect(transactions.mock.calls.length).toBeLessThanOrEqual(8);
+      expect(transactions.mock.calls.length).toBeGreaterThan(1);
+      expect(await memberships()).toEqual([]);
+      expect(await service.next()).not.toBeNull();
+    } finally {
+      transactions.mockRestore();
+      await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`);
+    }
+    await vi.advanceTimersByTimeAsync(60_000);
+    await service.sweep();
+    expect(await memberships()).toEqual([expect.objectContaining({ sourceId: approval.id })]);
+    expect(await service.next()).toBeNull();
+  });
+
+  it("keeps backoff SQL-quiet while another company's committed work proceeds", async () => {
+    const service = approvalQueueSeedingService(db), sweep = service.sweep;
+    let running = false;
+    service.sweep = async signal => { running = true; try { await sweep(signal); } finally { running = false; } };
+    await start(service); await quiet();
+    await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin
+      if new.company_id::text = TG_ARGV[0] then raise exception 'projection unavailable'; end if; return new; end $$`);
+    await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + companyId + "'")})`);
+    const approval = await pending();
+    for (let count = 1; count <= 7; count++) {
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThanOrEqual(count); }, { timeout: 5000 });
+      await vi.waitFor(() => expect(running).toBe(false));
+    }
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+    reads.mockRestore(); writes.mockRestore();
+    const otherId = randomUUID();
+    await db.insert(companies).values({ id: otherId, name: "Healthy tenant", issuePrefix: `H${otherId.slice(0, 6)}` });
+    const [otherIssue] = await db.insert(issues).values({ companyId: otherId, title: "Other PR", status: "todo" }).returning();
+    await workProductService(db).createForIssue(otherIssue.id, otherId, { type: "pull_request", provider: "github", title: "Other PR", status: "open" });
+    const otherApproval = await approvalRecords(db).create(otherId, { type: "request_board_approval", payload: {} });
+    await issueApprovalService(db).link(otherIssue.id, otherApproval.id);
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, otherId)))
+        .toEqual([expect.objectContaining({ sourceId: otherApproval.id })]);
+    }, { timeout: 5000 });
+    expect(await memberships()).toEqual([]);
+    await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); errors.length = 0;
+    await vi.advanceTimersByTimeAsync(60_000);
+    await seeded(approval.id); await quiet();
+  });
+
   it("drains multiple pages without duplicating memberships or audit events", async () => {
     const target = await issue(); await product(target.id);
     const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning();
@@ -309,6 +371,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + last + "'")})`);
     const service = approvalQueueSeedingService(db);
     await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed");
+    await vi.advanceTimersByTimeAsync(1000);
     await service.sweep(); expect((await memberships()).map(row => row.sourceId)).toEqual([last]);
     await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`);
     await service.sweep(); expect(await memberships()).toHaveLength(101); expect(await service.next()).toBeNull();

server/src/__tests__/plugin-scheduled-work.test.ts

@@ -1,3 +1,4 @@
+import { finishServiceShutdown } from "../shutdown.js";
 import { EventEmitter } from "node:events";
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -61,7 +62,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     }, { timeout: 5000 });
   }
   async function quiet() {
-    await settled(); expect(coordinator!.nextWakeAt()).toBeNull();
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0);
+      expect(coordinator!.nextWakeAt()).toBeNull();
+    }, { timeout: 5000 });
     const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute");
     await vi.advanceTimersByTimeAsync(180_000);
     expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
@@ -80,6 +85,23 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   }
 
   it("does not query repeatedly after an empty startup reconciliation", async () => { await start(); await quiet(); });
+  it("bounds full refreshes while many job status writes arrive sequentially", async () => {
+    await pluginJobStore(db).syncJobDeclarations(pluginId, Array.from({ length: 40 }, (_, i) => ({ jobKey: `manual-${i}`, schedule: "" })));
+    const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId));
+    await start(); await quiet();
+    const reads = vi.spyOn(db, "select");
+    for (const job of jobs.slice(0, 15)) {
+      const run = await pluginJobStore(db).createRun({ jobId: job.id, pluginId, trigger: "manual" });
+      await pluginJobStore(db).markRunning(run.id);
+      await pluginJobStore(db).completeRun(run.id, { status: "succeeded" });
+      await vi.advanceTimersByTimeAsync(1100);
+    }
+    expect(reads.mock.calls.length).toBeLessThanOrEqual(1);
+    reads.mockRestore();
+    await quiet();
+    expect(await runs()).toHaveLength(15);
+    expect(rpc).not.toHaveBeenCalled();
+  });
   it("keeps configured manual-only and invalid schedules quiet", async () => {
     await job("manual", ""); await job("invalid", "not cron"); await start(); await quiet(); expect(rpc).not.toHaveBeenCalled();
   });
@@ -147,7 +169,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`);
     await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`);
     await due(); await start();
-    await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); expect(rpc).toHaveBeenCalledTimes(1);
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); expect(rpc).toHaveBeenCalledTimes(1);
     await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
     await completed(); expect(rpc).toHaveBeenCalledTimes(1);
   });
@@ -155,7 +177,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`);
     await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`);
     await due(); await start();
-    await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0));
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 });
     draining = true;
     await coordinator!.stop();
     await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
@@ -163,11 +185,28 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect((await runs())[0].status).toBe("succeeded");
     expect(rpc).toHaveBeenCalledTimes(1);
   });
+  it("finishes service cleanup and retries a failed settlement before closing the pool", async () => {
+    await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`);
+    await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`);
+    await due(); await start();
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 });
+    draining = true; await coordinator!.stop(); errors.length = 0;
+    const cleanup = vi.fn(async () => {
+      expect((await runs())[0].status).toBe("running");
+      await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`);
+    });
+    const stop = vi.fn(() => scheduler!.stop());
+    await finishServiceShutdown(stop, cleanup);
+    expect(cleanup).toHaveBeenCalledOnce();
+    expect(stop).toHaveBeenCalledTimes(2);
+    expect((await runs())[0].status).toBe("succeeded");
+    expect(rpc).toHaveBeenCalledTimes(1);
+  });
   it("reports shutdown persistence failure and retains the result for another stop attempt", async () => {
     await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`);
     await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`);
     await due(); await start();
-    await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0));
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 });
     draining = true; await coordinator!.stop(); errors.length = 0;
     try {
       await expect(scheduler!.stop()).rejects.toThrow("Plugin job completion persistence failed during shutdown");
@@ -183,7 +222,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure(${sql.raw("'" + blocked.id + "'")})`);
     try {
       await start();
-      await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0));
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 });
       const independent = await due("independent");
       await vi.waitFor(async () => {
         await vi.advanceTimersByTimeAsync(1000);
@@ -244,7 +283,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   it("does no SQL while waiting for a future configured deadline", async () => {
     const row = await job(); const at = new Date(Date.now() + 600_000);
     await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at);
-    await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    await start();
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    }, { timeout: 5000 });
     const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
     await vi.advanceTimersByTimeAsync(180_000);
     expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
@@ -254,7 +297,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     const row = await job(); const at = new Date(Date.now() + 600_000);
     await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at);
     await start(); await settled(); await scheduler!.stop(); await coordinator!.stop();
-    await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    await start();
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    }, { timeout: 5000 });
     await vi.advanceTimersByTimeAsync(at.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1);
   });
   it("rechecks a paused definition after waiting for its lock", async () => {

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -331,7 +331,7 @@ vi.mock("../services/join-request-retention.js", async importOriginal => ({
 
 vi.mock("../services/approval-queue-seeding-work.js", async importOriginal => ({
   ...await importOriginal<typeof import("../services/approval-queue-seeding-work.js")>(),
-  approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(async () => null) })),
+  approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(() => null), request: vi.fn() })),
 }));
 
 vi.mock("../services/approval-retention.js", async importOriginal => ({
@@ -545,7 +545,10 @@ describe("startServer feedback export wiring", () => {
     if (enabled) {
       const service = vi.mocked(approvalQueueSeedingService).mock.results.at(-1)!.value;
       await registration![1].run(new AbortController().signal);
-      expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled();
+      expect(service.sweep).toHaveBeenCalled();
+      expect(registration![1].nextRunAt()).toBeNull();
+      expect(await registration![1].hasPending()).toBe(false);
+      expect(service.next).toHaveBeenCalled();
     }
   });
 

server/src/app.ts

@@ -1,3 +1,4 @@
+import { finishServiceShutdown } from "./shutdown.js";
 import { promptInjectionPreventionRoutes } from "./routes/prompt-injection-prevention.js";
 import { voiceSessionRoutes, voiceWebhookRoutes } from "./routes/voice-sessions.js";
 import { eq } from "drizzle-orm";
@@ -1455,10 +1456,7 @@ export async function createApp(
   let appServicesShutdown: Promise<void> | null = null;
   const shutdownAppServices = (): Promise<void> => {
     if (appServicesShutdown) return appServicesShutdown;
-    appServicesShutdown = (async () => {
-      // The scheduler tick queries the database. Stop it here, inside the
-      // awaited teardown, so no tick runs after the caller ends the pool.
-      await scheduler.stop();
+    appServicesShutdown = finishServiceShutdown(() => scheduler.stop(), async () => {
       await agentLifecycle.stop();
       await publicMcpEvents?.stop();
       await dotMcpEvents?.stop();
@@ -1493,7 +1491,7 @@ export async function createApp(
       // caller stops the database and the provider. A lease release that
       // fails stays a durable record for the startup reaper.
       await setupTokenLoginService?.shutdown();
-    })();
+    });
     return appServicesShutdown;
   };
   app.locals.paperclipShutdown = shutdownAppServices;

server/src/services/approval-queue-seeding-work.ts

@@ -1,6 +1,7 @@
 import { and, asc, eq, gt, sql, type SQL } from "drizzle-orm";
-import { approvals, companies, issues, type Db } from "@paperclipai/db";
+import { approvals, companies, issues, issueWorkProducts, type Db } from "@paperclipai/db";
 import { executionIssueCondition } from "./issue-visibility.js";
+import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js";
 import { decisionQueueService } from "./decision-queues.js";
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -10,12 +11,12 @@ const linkedIssue = sql<string>`(select linked.issue_id from issue_approvals lin
   order by linked.issue_id limit 1)`;
 
 export function approvalQueueSeedingService(db: Db) {
-  const pending = new Map<string | undefined, { cursor?: string; generation: number }>([[undefined, { generation: 0 }]]);
+  const pending = new Map<string | undefined, { cursor?: string; generation: number; failures: number; retryAt: number }>([[undefined, { generation: 0, failures: 0, retryAt: 0 }]]);
   function request(companyIds: ReadonlySet<string> | undefined) {
     for (const companyId of companyIds ?? []) {
       const existing = pending.get(companyId);
       if (existing) existing.generation++;
-      else pending.set(companyId, { generation: 0 });
+      else pending.set(companyId, { generation: 0, failures: 0, retryAt: 0 });
     }
   }
   function candidates(database: Db, companyId: string | undefined, condition?: SQL) {
@@ -24,47 +25,56 @@ export function approvalQueueSeedingService(db: Db) {
       .where(and(eq(approvals.status, "pending"), eq(companies.status, "active"), companyId === undefined ? undefined : eq(approvals.companyId, companyId), condition,
         sql`exists (select 1 from ${issues} where ${issues.id} = ${linkedIssue}
           and ${issues.companyId} = ${approvals.companyId} and ${executionIssueCondition()}
-          and exists (select 1 from issue_work_products product where product.company_id = ${approvals.companyId}
-            and product.issue_id = ${issues.id} and product.type = 'pull_request'))`,
+          and exists (select 1 from ${issueWorkProducts} where ${pullRequestSeedProductCondition(approvals.companyId)}
+            and ${issueWorkProducts.issueId} = ${issues.id}))`,
         sql`not exists (select 1 from decision_queues queue where queue.company_id = ${approvals.companyId} and queue.key = 'prs'
           and (not queue.seed_rules_enabled or exists (select 1 from decision_queue_items item where item.queue_id = queue.id
             and item.company_id = ${approvals.companyId} and item.source_kind = 'approval' and item.source_id = ${approvals.id}::text)))`,
       )).orderBy(asc(approvals.id));
   }
   async function sweep(signal?: AbortSignal) {
-    const scope = pending.entries().next().value;
+    const scope = [...pending.entries()].find(([, state]) => state.retryAt <= Date.now());
     if (!scope) return;
     const [companyId, state] = scope;
-    const generation = state.generation;
-    const rows = await candidates(db, companyId, state.cursor ? gt(approvals.id, state.cursor) : undefined).limit(100);
-    const errors: unknown[] = [];
-    for (const row of rows) {
-      if (signal?.aborted) break;
-      try {
-        await db.transaction(async tx => {
-          const database = tx as unknown as Db;
-          const [current] = await candidates(database, row.companyId, eq(approvals.id, row.id)).for("update", { of: approvals });
-          if (!current) return;
-          await decisionQueueService(database).materializeSeedSources(current.companyId, [{
-            sourceKind: "approval", sourceId: current.id, issueId: current.issueId,
-          }]);
-        });
-      } catch (error) { errors.push(error); }
-      state.cursor = row.id;
+    try {
+      const generation = state.generation;
+      const rows = await candidates(db, companyId, state.cursor ? gt(approvals.id, state.cursor) : undefined).limit(100);
+      const errors: unknown[] = [];
+      for (const row of rows) {
+        if (signal?.aborted) break;
+        try {
+          await db.transaction(async tx => {
+            const database = tx as unknown as Db;
+            const [current] = await candidates(database, row.companyId, eq(approvals.id, row.id)).for("update", { of: approvals });
+            if (!current) return;
+            await decisionQueueService(database).materializeSeedSources(current.companyId, [{
+              sourceKind: "approval", sourceId: current.id, issueId: current.issueId,
+            }]);
+          });
+        } catch (error) { errors.push(error); }
+        state.cursor = row.id;
+      }
+      if (rows.length < 100) state.cursor = undefined;
+      const remaining = await candidates(db, companyId).limit(1);
+      if (!signal?.aborted && generation === state.generation && remaining.length === 0) pending.delete(companyId);
+      if (errors.length === 1) throw errors[0];
+      if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed");
+      state.retryAt = 0;
+    } catch (error) {
+      state.failures = Math.min(state.failures + 1, 7);
+      state.retryAt = Date.now() + Math.min(60_000, 1_000 * 2 ** (state.failures - 1));
+      throw error;
+    } finally {
+      if (pending.has(companyId)) {
+        pending.delete(companyId);
+        pending.set(companyId, state);
+      }
     }
-    if (rows.length < 100) state.cursor = undefined;
-    const remaining = await candidates(db, companyId).limit(1);
-    if (!signal?.aborted && generation === state.generation && remaining.length === 0) {
-      pending.delete(companyId);
-    } else {
-      pending.delete(companyId);
-      pending.set(companyId, state);
-    }
-    if (errors.length === 1) throw errors[0];
-    if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed");
   }
-  async function next() {
-    return pending.size ? Date.now() : null;
+  function next() {
+    let earliest: number | null = null;
+    for (const state of pending.values()) earliest = earliest === null ? state.retryAt : Math.min(earliest, state.retryAt);
+    return earliest;
   }
   return { sweep, next, request };
 }
@@ -75,19 +85,20 @@ export function registerApprovalQueueSeedingWork(
   canRun: () => boolean,
 ) {
   let earliest = 0;
-  let next: number | null = null;
   return coordinator.register(DELIVERY_QUEUES.approvalQueueSeeding, {
     retryMs: 1_000,
     pausePassiveMaintenance: true,
     onSettled: service.request,
     run: async signal => {
       if (!canRun()) return;
-      if (Date.now() < earliest) { next = earliest; return; }
+      if (Date.now() < earliest) return;
       earliest = Date.now() + 1_000;
       await service.sweep(signal);
-      next = await service.next();
     },
-    hasPending: async () => (await service.next()) !== null,
-    nextRunAt: () => next,
+    hasPending: async () => (service.next() ?? Infinity) <= Date.now(),
+    nextRunAt: () => {
+      const next = service.next();
+      return next === null ? null : Math.max(earliest, next);
+    },
   });
 }

server/src/services/decision-queue-seed-rules.ts

@@ -0,0 +1,6 @@
+import { and, eq, type SQLWrapper } from "drizzle-orm";
+import { issueWorkProducts } from "@paperclipai/db";
+
+export function pullRequestSeedProductCondition(companyId: string | SQLWrapper) {
+  return and(eq(issueWorkProducts.companyId, companyId), eq(issueWorkProducts.type, "pull_request"));
+}

server/src/services/decision-queues.ts

@@ -1,3 +1,4 @@
+import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js";
 import { notifyApprovalQueueWork } from "./approval-work-notifications.js";
 import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -402,8 +403,7 @@ export function decisionQueueService(db: Db) {
       .select({ issueId: issueWorkProducts.issueId })
       .from(issueWorkProducts)
       .where(and(
-        eq(issueWorkProducts.companyId, companyId),
-        eq(issueWorkProducts.type, "pull_request"),
+        pullRequestSeedProductCondition(companyId),
         inArray(issueWorkProducts.issueId, issueIds),
       )).then((rows) => rows.map((row) => row.issueId)));
 

server/src/services/plugin-job-scheduler.ts

@@ -45,7 +45,7 @@ function nextCron(schedule: string, now: Date): Date | null {
 }
 
 export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler {
-  const { db, jobStore, workerManager, tickIntervalMs = 1_000,
+  const { db, jobStore, workerManager, tickIntervalMs = 30_000,
     jobTimeoutMs = 5 * 60_000, maxConcurrentJobs = 10 } = options;
   const log = logger.child({ service: "plugin-job-scheduler" });
   const activeJobs = new Set<string>();

server/src/services/plugin-loader.ts

@@ -1,3 +1,4 @@
+import { finishServiceShutdown } from "../shutdown.js";
 /**
  * PluginLoader — discovery, installation, and runtime activation of plugins.
  *
@@ -2143,11 +2144,7 @@ export function pluginLoader(
 
       const { workerManager, jobScheduler } = runtimeServices;
 
-      // 1. Stop the job scheduler tick loop
-      await jobScheduler.stop();
-
-      // 2. Stop all worker processes
-      await workerManager.stopAll();
+      await finishServiceShutdown(() => jobScheduler.stop(), () => workerManager.stopAll());
 
       log.info("plugin-loader: all plugins shut down");
     },

server/src/shutdown.test.ts

@@ -5,6 +5,7 @@ import {
   coordinateHeartbeatSchedulerShutdown,
   drainRunExecutionFinalizersForShutdown,
   finalizeServerShutdown,
+  finishServiceShutdown,
   loadWithoutCoordinatedShutdownSignalHooks,
 } from "./shutdown.js";
 
@@ -523,3 +524,26 @@ describe("coordinateHeartbeatSchedulerShutdown", () => {
     });
   });
 });
+
+describe("service shutdown failure isolation", () => {
+  it("awaits remaining cleanup before reporting a scheduler failure", async () => {
+    const failure = new Error("plugin result could not be saved");
+    const gate = deferred();
+    const cleanup = vi.fn(() => gate.promise);
+    let finished = false;
+    const stopping = finishServiceShutdown(async () => { throw failure; }, cleanup)
+      .then(() => { finished = true; return undefined; }, error => { finished = true; return error; });
+    try {
+      await vi.waitFor(() => expect(cleanup).toHaveBeenCalledOnce());
+      expect(finished).toBe(false);
+    } finally { gate.resolve(); }
+    expect(await stopping).toBe(failure);
+  });
+  it("reports both failures after attempting cleanup", async () => {
+    const stopFailure = new Error("scheduler failure"), cleanupFailure = new Error("worker failure");
+    const cleanup = vi.fn(async () => { throw cleanupFailure; });
+    await expect(finishServiceShutdown(async () => { throw stopFailure; }, cleanup))
+      .rejects.toMatchObject({ errors: [stopFailure, cleanupFailure] });
+    expect(cleanup).toHaveBeenCalledOnce();
+  });
+});

server/src/shutdown.ts

@@ -272,3 +272,18 @@ export async function coordinateHeartbeatSchedulerShutdown<
     waitedForSchedulerIdle: true,
   };
 }
+
+export async function finishServiceShutdown(
+  stop: () => Promise<void>,
+  cleanup: () => Promise<void>,
+): Promise<void> {
+  const failures: unknown[] = [];
+  let retryStop = false;
+  try { await stop(); } catch { retryStop = true; }
+  try { await cleanup(); } catch (error) { failures.push(error); }
+  if (retryStop) {
+    try { await stop(); } catch (error) { failures.unshift(error); }
+  }
+  if (failures.length === 1) throw failures[0];
+  if (failures.length > 1) throw new AggregateError(failures, "Service shutdown failed");
+}