perf(server): wake five delivery queues from committed work

Replace empty feedback, completion, connection, question, and tool receipt scans with commit-aware notifications and conditional recovery workers. Preserve startup recovery, claims, shutdown, and conservative handling of ambiguous commits.

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

doc/DATABASE.md

@@ -681,3 +681,18 @@ and are atomically deleted before code exchange. The `code_verifier` column hold
 this non-secret binding for this namespace; Slack bot installation does not use
 PKCE. Removal and manual recovery invalidate outstanding attempts under the same
 credential-mutation lease used by configuration.
+
+
+## Durable work notifications
+
+`createDb` supports process-local `signalDatabaseWork` and
+`subscribeDatabaseWork` hints for delivery queues. Signal inside the transaction
+before its durable write; nested signals publish only after the outermost
+transaction settles. For an autocommit write, signal after success, and signal
+with `uncertainCommit: true` (the third argument) if its acknowledgement fails.
+Consumers subscribe before an initial durable reconciliation. They must retain
+retries while work or an uncertain commit is outstanding and release their
+subscription on shutdown. A signal is a hint, not a persisted queue or a
+cross-process notification. Every supported producer must use the same
+`createDb` instance (or one of its transactions) as its consumer. See the
+[delivery migration scope](plans/2026-10-08-instance-sleep-and-scheduler-inventory.md#first-implementation-five-delivery-queues).

doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md

@@ -2,7 +2,7 @@
 
 Date: 2026-10-08
 
-Status: Implementation checklist; application behavior has not been changed
+Status: First five delivery queues migrated; remaining inventory is open
 
 Code baseline: `0ac194450`
 
@@ -159,6 +159,64 @@ This order supersedes starting by routing the feedback polling loop through a
 generic scheduler. Event-driven improvements can ship before that scheduler is
 complete. The resulting service still needs an owned way to retry failures.
 
+## First implementation: five delivery queues
+
+The first change removes empty normal-path polling from these five lanes:
+
+- [x] **JOB-04 export dispatch:** feedback export enqueue wakes the worker.
+  The request handler and worker serialize uploads through the same service.
+- [x] **JOB-22 completion dispatch:** the task status transaction wakes the
+  completion worker without relying on a later activity publication.
+- [x] **JOB-23 continuation dispatch:** resolving a connection interaction
+  wakes its durable continuation worker.
+- [x] **JOB-25 answer dispatch:** the answer transaction wakes its delivery worker.
+- [x] **JOB-24 receipt dispatch:** creation of a tool-action receipt activates
+  recovery until its delivery completes. Review execution and expiry sweeps
+  remain separate, unchanged work.
+
+These checks cover dispatch, not the full completion criteria of the original
+JOB items below. An empty queue now has no retry timer and issues no periodic
+query. Outstanding work retains the former recovery cadence: five seconds for
+exports and the configured heartbeat scheduler interval for the other queues.
+Blocked deliveries and unanswered tool reviews still own recovery timers.
+Refining those into specific dependency events/deadlines remains future work.
+
+[`delivery-queue-worker.ts`](../../server/src/services/delivery-queue-worker.ts)
+centralizes admission, single-flight processing, retry timers, and shutdown for
+these five lanes. Its retry scheduling is a small boundary to adapt to SCH-01;
+it does not implement the instance-wide scheduler or external waker yet. The
+main heartbeat scheduler continues to own other jobs, so this change alone
+neither makes an instance sleep nor proves provider cost savings.
+
+[`work-signals.ts`](../../packages/db/src/work-signals.ts) adds process-local
+notifications to `createDb` transactions. The producers register a topic in
+scope of the write, and publication waits for the outer transaction to settle,
+including caller-owned transactions and nested savepoints. Rollback may cause
+one harmless empty reconciliation. A successful commit does not depend on
+subsequent telemetry, activity publication, or HTTP response code succeeding.
+
+Workers subscribe before startup recovery, remember writes that arrive during
+a sweep, and retry errors. If a transaction callback succeeds but the commit
+acknowledgement fails, the worker conservatively keeps recovery checks active
+until the process restarts. Neither an empty read nor delivery of an unrelated
+record can prove that an ambiguous commit will not appear later. An autocommit receipt
+write that fails uses the same conservative recovery mode. This exceptional
+path can keep an otherwise empty queue awake; it deliberately favors delivery
+recovery. Normal successful transactions have no empty-queue backstop.
+
+The supported writer boundary is one server process using its `createDb`
+instance and the instrumented service mutations. Startup recovers durable rows
+left by a crashed process. Direct SQL writers, a second independent database
+client, and other server replicas do not emit these local signals; supporting
+those writers requires an external notification/recovery design before relying
+on this consumer model. No persistent `LISTEN` connection is introduced.
+
+Verification covers all five producer signals against temporary PostgreSQL,
+completion delivery without route callbacks, consumer restart, nested commits,
+rollback and ambiguous commit, concurrent signals, retry, drain/standby gating,
+shutdown, and an hour of simulated empty-queue inactivity. Full queue claim and
+idempotency behavior remains covered by the existing delivery suites.
+
 ## Provider behavior and cost model
 
 | Layer | Documented behavior | Consequence |

packages/db/src/client.ts

@@ -3,6 +3,7 @@ import { drizzle as drizzlePg } from "drizzle-orm/postgres-js";
 import { readFile, readdir } from "node:fs/promises";
 import { fileURLToPath } from "node:url";
 import postgres from "postgres";
+import { installDatabaseWorkSignals } from "./work-signals.js";
 import * as schema from "./schema/index.js";
 
 const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url));
@@ -275,7 +276,7 @@ export function createDb(url: string, options?: DatabaseClientOptions) {
   // postgres.js calls that error "write CONNECTION_CLOSED" too, so the
   // message cannot establish that replay is safe. Leave retries to callers
   // that know the complete operation is idempotent.
-  const db = drizzlePg(sql, { schema });
+  const db = installDatabaseWorkSignals(drizzlePg(sql, { schema }));
   dedicatedDbFactories.set(db, () => createDb(url, {
     ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock",
   }));

packages/db/src/index.ts

@@ -42,3 +42,5 @@ export { loadWithoutEmbeddedPostgresExitHooks } from "./embedded-postgres-lifecy
 export { issueRelations } from "./schema/issue_relations.js";
 export { issueReferenceMentions } from "./schema/issue_reference_mentions.js";
 export * from "./schema/index.js";
+
+export { signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js";

packages/db/src/work-signals.test.ts

@@ -0,0 +1,70 @@
+import { describe, expect, it, vi } from "vitest";
+import { installDatabaseWorkSignals, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js";
+
+type Connection = { transaction<T>(callback: (tx: Connection) => Promise<T>): Promise<T> };
+function connection(commit?: () => void): Connection {
+  return { async transaction<T>(callback: (tx: ReturnType<typeof connection>) => Promise<T>): Promise<T> {
+    const result = await callback(connection());
+    commit?.();
+    return result;
+  } };
+}
+
+describe("database work signals", () => {
+  it("waits for the outer commit, including nested savepoints, and coalesces hints", async () => {
+    const listener = vi.fn();
+    const db = installDatabaseWorkSignals(connection(() => expect(listener).not.toHaveBeenCalled()));
+    subscribeDatabaseWork(db, "queue", listener);
+    await db.transaction(async tx => {
+      signalDatabaseWork(tx, "queue");
+      await tx.transaction(async nested => { signalDatabaseWork(nested, "queue"); });
+      expect(listener).not.toHaveBeenCalled();
+    });
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+
+  it("reconciles uncertain commits and rollbacks without changing their errors", async () => {
+    const error = new Error("lost COMMIT response");
+    const db = installDatabaseWorkSignals(connection(() => { throw error; }));
+    const listener = vi.fn();
+    subscribeDatabaseWork(db, "queue", listener);
+    await expect(db.transaction(async tx => { signalDatabaseWork(tx, "queue"); })).rejects.toBe(error);
+    expect(listener).toHaveBeenCalledTimes(1);
+    expect(listener).toHaveBeenLastCalledWith(true);
+    await expect(db.transaction(async tx => {
+      await tx.transaction(async nested => { signalDatabaseWork(nested, "queue"); throw error; });
+    })).rejects.toBe(error);
+    expect(listener).toHaveBeenCalledTimes(2);
+    expect(listener).toHaveBeenLastCalledWith(false);
+  });
+
+  it("keeps concurrent transaction hints isolated until their own commits", async () => {
+    const db = installDatabaseWorkSignals(connection());
+    const first = vi.fn(), second = vi.fn();
+    subscribeDatabaseWork(db, "first", first);
+    subscribeDatabaseWork(db, "second", second);
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const pending = db.transaction(async tx => { signalDatabaseWork(tx, "first"); await gate; });
+    await db.transaction(async tx => { signalDatabaseWork(tx, "second"); });
+    expect(first).not.toHaveBeenCalled();
+    expect(second).toHaveBeenCalledTimes(1);
+    release(); await pending;
+    expect(first).toHaveBeenCalledTimes(1);
+  });
+
+  it("isolates databases and topics and releases subscriptions", () => {
+    const db = installDatabaseWorkSignals(connection());
+    const other = installDatabaseWorkSignals(connection());
+    const listener = vi.fn();
+    const unsubscribe = subscribeDatabaseWork(db, "queue", listener);
+    signalDatabaseWork(other, "queue");
+    signalDatabaseWork(db, "other");
+    expect(listener).not.toHaveBeenCalled();
+    signalDatabaseWork(db, "queue");
+    expect(listener).toHaveBeenCalledTimes(1);
+    unsubscribe();
+    signalDatabaseWork(db, "queue");
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+});

packages/db/src/work-signals.ts

@@ -0,0 +1,67 @@
+/** Process-local hints for durable work. The database remains the source of truth. */
+type Listener = (uncertainCommit: boolean) => void;
+type Scope = { listeners: Map<string, Set<Listener>> };
+type Context = { scope: Scope; pending: Set<string> | null };
+const contexts = new WeakMap<object, Context>();
+
+// Drizzle transactions can contain savepoints and can be supplied by callers.
+// Instrument each transaction at creation so a hint cannot escape an outer
+// transaction before commit. No SQL, connection, or timer is added here.
+type Transactional = { transaction: (...args: any[]) => Promise<any> };
+export function installDatabaseWorkSignals<T extends Transactional>(db: T): T {
+  install(db, { scope: { listeners: new Map() }, pending: null });
+  return db;
+}
+
+function publish(scope: Scope, topic: string, uncertainCommit = false) {
+  for (const listener of scope.listeners.get(topic) ?? []) {
+    try { listener(uncertainCommit); }
+    catch (error) { process.emitWarning(`Database work listener failed: ${String(error)}`); }
+  }
+}
+
+function install(connection: Transactional, context: Context) {
+  contexts.set(connection, context);
+  const transaction = connection.transaction;
+  connection.transaction = async function (callback, ...options) {
+    const pending = new Set<string>();
+    let callbackCompleted = false;
+    let committed = false;
+    try {
+      const result = await transaction.call(this, async (tx: Transactional) => {
+        install(tx, { scope: context.scope, pending });
+        const result = await callback(tx);
+        callbackCompleted = true;
+        return result;
+      }, ...options);
+      committed = true;
+      return result;
+    } finally {
+      // Also reconcile after failure: a lost COMMIT response is ambiguous.
+      // A rollback merely causes one harmless empty reconciliation. Nested
+      // hints are deferred until the outermost transaction settles.
+      for (const topic of pending) {
+        if (context.pending) context.pending.add(topic);
+        else publish(context.scope, topic, callbackCompleted && !committed);
+      }
+    }
+  };
+}
+
+/** Signal before a transactional write; signal autocommit success/failure after settlement. */
+export function signalDatabaseWork(connection: object, topic: string, uncertainCommit = false): void {
+  const context = contexts.get(connection);
+  if (!context) return; // Non-runtime database doubles do not have consumers.
+  if (context.pending) context.pending.add(topic);
+  else publish(context.scope, topic, uncertainCommit);
+}
+
+/** Subscribe before the initial durable reconciliation to close the startup gap. */
+export function subscribeDatabaseWork(connection: object, topic: string, listener: Listener): () => void {
+  const context = contexts.get(connection);
+  if (!context) throw new Error("Database work subscriptions require createDb");
+  let listeners = context.scope.listeners.get(topic);
+  if (!listeners) context.scope.listeners.set(topic, listeners = new Set());
+  listeners.add(listener);
+  return () => { listeners.delete(listener); };
+}

server/src/__tests__/chat-completion-delivery.test.ts

@@ -1,3 +1,4 @@
+import { createDeliveryQueueWorker, DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
 import { randomUUID } from "node:crypto";
 import { and, asc, eq } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -51,6 +52,34 @@ const support = await getEmbeddedPostgresTestSupport();
     };
     return { companyId, agentId, sourceId, runId, task, create, finish, rows, due, wakeup, service, run };
   }
+  it("dispatches a committed completion without activity publication and recovers after restart", async () => {
+    const f = await seed();
+    const run = vi.fn(() => f.service.sweepPending());
+    const makeWorker = () => createDeliveryQueueWorker({
+      db, topic: DELIVERY_QUEUES.chatCompletion, retryMs: 60_000,
+      run, hasPending: f.service.hasPending, canRun: () => true,
+      onError: error => { throw error; },
+    });
+    const worker = makeWorker();
+    try {
+      await worker.ready;
+      expect(await f.service.hasPending()).toBe(false);
+      expect(run).toHaveBeenCalledTimes(1);
+      await f.finish(); // Service mutation alone: no route activity callback.
+      await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledTimes(1));
+      expect(await f.service.hasPending()).toBe(true); // queued receipt still owed
+    } finally { await worker.stop(); }
+    // Admit the first reporting turn; later completions need a new turn.
+    await f.run();
+    // A second task commits while no consumer is registered.
+    const next = await f.create();
+    await f.finish(next.id);
+    const restarted = makeWorker();
+    try {
+      await restarted.ready;
+      expect(f.wakeup).toHaveBeenCalledTimes(2);
+    } finally { await restarted.stop(); }
+  });
   it("records authenticated origins and atomically creates one event per Done transition", async () => {
     const f = await seed();
     expect(await db.select().from(handoffs).where(eq(handoffs.taskId, f.task.id))).toMatchObject([{ conversationId: f.sourceId, sessionGeneration: 0 }]);

server/src/__tests__/connection-intents-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
@@ -773,7 +775,11 @@ describeEmbeddedPostgres("connectionIntentService", () => {
   it("recovers a resolution after a failed dispatch and acknowledges an already queued wake exactly once", async () => {
     const service = connectionIntentService(db);
     const pending = await service.request(claims, "airtable");
+    const notified = vi.fn();
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.connection, notified);
     await service.decline(pending.interactionId!, claims.responsible_user_id);
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
     const wakeup = vi.fn().mockRejectedValueOnce(new Error("simulated crash before durable enqueue"));
     const deliveries = connectionIntentDeliveryService(db, { wakeup } as never);
     await expect(deliveries.deliver(pending.interactionId!)).rejects.toThrow("simulated crash");

server/src/__tests__/delivery-queue-worker.test.ts

@@ -0,0 +1,100 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { installDatabaseWorkSignals, signalDatabaseWork } from "../../../packages/db/src/work-signals.js";
+// Use the same module instance for the subscription and the test connection.
+vi.mock("@paperclipai/db", async () => import("../../../packages/db/src/work-signals.js"));
+import { createDeliveryQueueWorker, DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const workers: ReturnType<typeof createDeliveryQueueWorker>[] = [];
+afterEach(async () => { for (const worker of workers.splice(0)) await worker.stop(); vi.useRealTimers(); });
+function setup(topic: string = DELIVERY_QUEUES.feedback) {
+  vi.useFakeTimers();
+  const db = installDatabaseWorkSignals({ transaction: vi.fn() });
+  const input = { db, topic, retryMs: 1000, run: vi.fn(async () => {}),
+    hasPending: vi.fn(async () => false), canRun: vi.fn(() => true), onError: vi.fn() };
+  const worker = createDeliveryQueueWorker(input);
+  workers.push(worker);
+  return { ...input, worker, signal: () => signalDatabaseWork(db, topic) };
+}
+
+describe("delivery queue workers", () => {
+  it.each(Object.values(DELIVERY_QUEUES))("%s stops all scans after startup and after a committed enqueue drains", async topic => {
+    const s = setup(topic);
+    await vi.advanceTimersByTimeAsync(0);
+    expect(s.run).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+    await vi.advanceTimersByTimeAsync(60 * 60 * 1000);
+    expect(s.run).toHaveBeenCalledTimes(1);
+    s.signal(); s.signal();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(s.run).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("retries outstanding work and failures, then becomes quiet", async () => {
+    const s = setup();
+    s.run.mockRejectedValueOnce(new Error("DB unavailable"));
+    s.hasPending.mockResolvedValueOnce(true);
+    await vi.advanceTimersByTimeAsync(0);
+    expect(s.onError).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(s.run).toHaveBeenCalledTimes(3);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("keeps a recovery backstop after an uncertain commit even if unrelated work drains", async () => {
+    const s = setup();
+    await s.worker.ready;
+    signalDatabaseWork(s.db, s.topic, true);
+    await vi.advanceTimersByTimeAsync(3000);
+    expect(s.run.mock.calls.length).toBeGreaterThan(3);
+    expect(vi.getTimerCount()).toBe(1);
+    s.hasPending.mockResolvedValueOnce(true);
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(vi.getTimerCount()).toBe(1);
+  });
+
+  it("does not lose a commit racing the final empty check or overlap deliveries", async () => {
+    const s = setup();
+    let release!: (value: boolean) => void;
+    s.hasPending.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    await vi.advanceTimersByTimeAsync(0);
+    expect(idleWorkSnapshot().active).toBe(1);
+    s.signal(); s.signal();
+    await vi.advanceTimersByTimeAsync(5000);
+    expect(s.run).toHaveBeenCalledTimes(1);
+    release(false);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(s.run).toHaveBeenCalledTimes(2);
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("defers without querying during idle drain/standby and resumes pending recovery", async () => {
+    const s = setup();
+    s.canRun.mockReturnValue(false);
+    s.signal();
+    await vi.advanceTimersByTimeAsync(4000);
+    expect(s.run).not.toHaveBeenCalled();
+    expect(s.hasPending).not.toHaveBeenCalled();
+    s.canRun.mockReturnValue(true);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(s.run).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it("waits for accepted work on shutdown and ignores subsequent writes", async () => {
+    const s = setup();
+    let release!: () => void;
+    s.run.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    await vi.advanceTimersByTimeAsync(0);
+    const stopped = vi.fn();
+    const stopping = s.worker.stop().then(stopped);
+    s.signal();
+    expect(stopped).not.toHaveBeenCalled();
+    release(); await stopping;
+    await vi.advanceTimersByTimeAsync(10000);
+    expect(s.run).toHaveBeenCalledTimes(1);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+});

server/src/__tests__/feedback-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
 import { randomUUID } from "node:crypto";
 import fs from "node:fs";
 import os from "node:os";
@@ -800,6 +802,8 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => {
       sessionId,
     });
 
+    const notified = vi.fn();
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.feedback, notified);
     await flushingSvc.saveIssueVote({
       issueId,
       targetType: "issue_comment",
@@ -808,7 +812,12 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => {
       authorUserId: "user-1",
       allowSharing: true,
     });
-    await flushingSvc.flushPendingFeedbackTraces();
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(true);
+    // The route and worker may request a flush together after the commit.
+    await Promise.all([flushingSvc.flushPendingFeedbackTraces(), flushingSvc.flushPendingFeedbackTraces()]);
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(false);
 
     expect(uploadTraceBundle).toHaveBeenCalledTimes(1);
     const bundle = uploadTraceBundle.mock.calls[0]?.[0] as Record<string, unknown> | undefined;

server/src/__tests__/issue-thread-interactions-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
 import { randomUUID } from "node:crypto";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -231,11 +233,15 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => {
     expect(created.payload.questions.map((question) => question.id)).toEqual(["repo", "scope", "hosting"]);
     expect(await interactionsSvc.create(issue, input, { userId: "local-board" })).toEqual(created);
     await expect(interactionsSvc.answerQuestions(issue, created.id, { answers: [{ questionId: "repo", optionIds: [], otherText: "https://example.com/repo" }] }, { userId: "local-board" })).rejects.toThrow("requires an answer");
+    const notified = vi.fn();
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.question, notified);
     const answered = await interactionsSvc.answerQuestions(issue, created.id, { answers: [
       { questionId: "repo", optionIds: [], otherText: "https://example.com/repo" },
       { questionId: "scope", optionIds: ["all"] },
       { questionId: "hosting", optionIds: ["existing", "new"] },
     ] }, { userId: "local-board" });
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
     expect(answered.status).toBe("answered");
   });
 

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

@@ -45,7 +45,7 @@ const {
         sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
         cleanupExpiredSessions: vi.fn(async () => ({ deletedCount: 0 })),
       },
-      toolActionDeliveries: { sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) },
+      toolActionDeliveries: { hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) },
     },
   }) as never);
   const createBetterAuthInstanceMock = vi.fn(() => ({}));
@@ -130,6 +130,7 @@ const {
   };
   const routineServiceFactoryMock = vi.fn(() => routineServiceMock);
   const feedbackExportServiceMock = {
+    hasPendingFeedbackTraces: vi.fn(async () => false),
     flushPendingFeedbackTraces: vi.fn(async () => ({ attempted: 0, sent: 0, failed: 0 })),
   };
   const feedbackServiceFactoryMock = vi.fn(() => feedbackExportServiceMock);
@@ -223,6 +224,7 @@ vi.mock("detect-port", () => ({
 
 vi.mock("@paperclipai/db", () => ({
   createDb: createDbMock,
+  subscribeDatabaseWork: vi.fn(() => () => {}),
   ensurePostgresDatabase: vi.fn(),
   getPostgresDataDirectory: vi.fn(),
   inspectMigrations: vi.fn(async () => ({ status: "upToDate" })),
@@ -360,16 +362,18 @@ vi.mock("../services/index.js", () => ({
   })),
 }));
 
-vi.mock("../services/chat-completion-delivery.js", () => ({ chatCompletionDeliveryService: () => ({ sweepPending: completionSweepMock }) }));
+vi.mock("../services/chat-completion-delivery.js", () => ({ chatCompletionDeliveryService: () => ({ hasPending: vi.fn(async () => false), sweepPending: completionSweepMock }) }));
 
 vi.mock("../services/connection-intent-delivery.js", () => ({
   connectionIntentDeliveryService: vi.fn(() => ({
+    hasPending: vi.fn(async () => false),
     sweepPending: vi.fn(async () => ({ scanned: 0, failed: 0 })),
   })),
 }));
 
 vi.mock("../services/question-response-delivery.js", () => ({
   questionResponseDeliveryService: vi.fn(() => ({
+    hasPending: vi.fn(async () => false),
     sweepPending: vi.fn(async () => ({
       scanned: 0,
       steered: 0,

server/src/__tests__/tool-gateway-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "../services/delivery-queue-worker.js";
 import { randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -219,6 +221,8 @@ describeEmbeddedPostgres("tool gateway service", () => {
       runId: run.id,
     });
 
+    const notified = vi.fn();
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.toolAction, notified);
     await expect(gateway.executeTool({
       sessionToken: session.token,
       tool: "mcp-remote-fixture:update_note",
@@ -228,6 +232,9 @@ describeEmbeddedPostgres("tool gateway service", () => {
       details: { instructions: expect.stringContaining("A human approval card was posted on task") },
     });
 
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
+
     await expect(gateway.executeTool({
       sessionToken: session.token,
       tool: "mcp-remote-fixture:update_note",

server/src/app.ts

@@ -1,3 +1,4 @@
+import { createDeliveryQueueWorker, DELIVERY_QUEUES } from "./services/delivery-queue-worker.js";
 import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "./middleware/idle-admission.js";
 import { isIdleTaskDrainActive, trackIdleWork } from "./services/task-admission.js";
 import { customerSuccessRoutes } from "./routes/customer-success.js";
@@ -482,6 +483,7 @@ export async function createApp(
     serverPort: number;
     storageService: StorageService;
     feedbackExportService?: {
+      hasPendingFeedbackTraces(): Promise<boolean>;
       flushPendingFeedbackTraces(input?: {
         companyId?: string;
         traceId?: string;
@@ -1187,41 +1189,17 @@ export async function createApp(
 
   jobCoordinator.start();
   scheduler.start();
-  let feedbackExportShuttingDown = false;
-  let feedbackExportTimer: ReturnType<typeof setInterval> | null = null;
-  const disableFeedbackExportFlushes = () => {
-    feedbackExportShuttingDown = true;
-    if (feedbackExportTimer) {
-      clearInterval(feedbackExportTimer);
-      feedbackExportTimer = null;
-    }
-  };
-  const flushPendingFeedbackExports = async () => {
-    if (feedbackExportShuttingDown || isWarmStandby() || isIdleTaskDrainActive()) return;
-    try {
-      await opts.feedbackExportService?.flushPendingFeedbackTraces();
-    } catch (err) {
-      if (isDatabaseConnectionUnavailableError(err)) {
-        disableFeedbackExportFlushes();
-        logger.warn(
-          { err },
-          "Disabling pending feedback export flushes because the database is unavailable",
-        );
-        return;
-      }
-      logger.error({ err }, "Failed to flush pending feedback exports");
-    }
-  };
-
-  feedbackExportTimer = opts.feedbackExportService
-    ? setInterval(() => {
-        void flushPendingFeedbackExports();
-      }, FEEDBACK_EXPORT_FLUSH_INTERVAL_MS)
+  const feedbackExportWorker = opts.feedbackExportService
+    ? createDeliveryQueueWorker({
+        db,
+        topic: DELIVERY_QUEUES.feedback,
+        retryMs: FEEDBACK_EXPORT_FLUSH_INTERVAL_MS,
+        run: () => opts.feedbackExportService!.flushPendingFeedbackTraces(),
+        hasPending: () => opts.feedbackExportService!.hasPendingFeedbackTraces(),
+        canRun: () => !isWarmStandby() && !isIdleTaskDrainActive(),
+        onError: (err) => logger.error({ err }, "Failed to flush pending feedback exports"),
+      })
     : null;
-  feedbackExportTimer?.unref?.();
-  if (opts.feedbackExportService) {
-    void flushPendingFeedbackExports();
-  }
   emailChannels.start();
   const flushChatPublications = async () => {
     await chatChannels.schedulePendingPublications();
@@ -1386,7 +1364,7 @@ export async function createApp(
       await publicMcpEvents?.stop();
       await dotMcpEvents?.stop();
       jobCoordinator.stop();
-      disableFeedbackExportFlushes();
+      await feedbackExportWorker?.stop();
       unsubscribeChatPublicationSignals();
       chatReconciliation.stop();
       if (chatPublicationTimer) {

server/src/index.ts

@@ -1,8 +1,8 @@
+import { createDeliveryQueueWorker, DELIVERY_QUEUES } from "./services/delivery-queue-worker.js";
 import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js";
 import { markIdleStartupComplete } from "./services/idle-local-work.js";
 import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js";
 import { createCloudWarmStandby } from "./services/cloud-warm-standby.js";
-import { subscribeAllCompanyLiveEvents } from "./services/live-events.js";
 import { chatCompletionDeliveryService } from "./services/chat-completion-delivery.js";
 /// <reference path="./types/express.d.ts" />
 // Kicks off the OTel bootstrap as early as possible (no-op unless
@@ -1241,20 +1241,26 @@ async function startServerWithDatabaseTeardown(
   const ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS = 5 * 60 * 1000;
   const environmentLeaseCleanupHeartbeat = accountingHeartbeat;
   const chatCompletionDeliveries = chatCompletionDeliveryService(db as any, environmentLeaseCleanupHeartbeat);
-  // Activity publication happens after the status transaction commits. This is
-  // a best-effort fast path; the durable outbox and sweeps remain authoritative.
-  const unsubscribeChatCompletions = subscribeAllCompanyLiveEvents(event => {
-    if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || event.type !== "activity.logged" ||
-      event.payload.action !== "issue.updated" || typeof event.payload.entityId !== "string") return;
-    trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending({ companyId: event.companyId, taskId: event.payload.entityId })
-      .catch(err => logger.error({ err }, "post-commit chat completion delivery failed")));
-  });
-  server.on("close", unsubscribeChatCompletions);
   const connectionDeliveries = connectionIntentDeliveryService(db as any, environmentLeaseCleanupHeartbeat);
   const questionResponseDeliveries = questionResponseDeliveryService(db as any, {
     heartbeat: environmentLeaseCleanupHeartbeat,
     resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction),
   });
+  const deliveryWorkers = ([
+    [DELIVERY_QUEUES.chatCompletion, chatCompletionDeliveries],
+    [DELIVERY_QUEUES.connection, connectionDeliveries],
+    [DELIVERY_QUEUES.question, questionResponseDeliveries],
+    [DELIVERY_QUEUES.toolAction, app.locals.toolActionDeliveries as ReturnType<typeof import("./services/tool-action-delivery.js").toolActionDeliveryService>],
+  ] as const).map(([topic, service]) => createDeliveryQueueWorker({
+    db,
+    topic,
+    retryMs: config.heartbeatSchedulerIntervalMs,
+    run: () => service.sweepPending(),
+    hasPending: () => service.hasPending(),
+    canRun: () => !heartbeatSchedulerStopped && !isWarmStandby() && !isIdleTaskDrainActive(),
+    onError: (err) => logger.error({ err, queue: topic }, "pending delivery recovery failed"),
+  }));
+  server.on("close", () => { void Promise.all(deliveryWorkers.map(worker => worker.stop())); });
   const runEnvironmentLeaseCleanupSweep = (backoffMs: number) =>
     environmentLeaseCleanupHeartbeat
       .sweepPendingCleanupLeases({ backoffMs })
@@ -1305,18 +1311,9 @@ async function startServerWithDatabaseTeardown(
       }));
   };
 
-  await chatCompletionDeliveries.sweepPending().catch((err) => logger.error({ err }, "startup chat completion delivery recovery failed"));
-  await connectionDeliveries.sweepPending();
+  await Promise.all(deliveryWorkers.map(worker => worker.ready));
   await app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed"));
   await app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed"));
-  await app.locals.toolActionDeliveries.sweepPending().catch((err: unknown) => logger.error({ err }, "startup tool review delivery sweep failed"));
-  await questionResponseDeliveries.sweepPending().then((result) => {
-    if (result.scanned > 0) {
-      logger.info(result, "startup question-response delivery sweep completed");
-    }
-  }).catch((err) => {
-    logger.error({ err }, "startup question-response delivery sweep failed");
-  });
   scheduleGitHubConnectionEventPoll();
   scheduleGitHubConnectionContinuitySweep();
 
@@ -1770,23 +1767,10 @@ async function startServerWithDatabaseTeardown(
             logger.error({ err }, "periodic secret proposal expiry sweep failed");
           }));
 
-        trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending().catch((err) => logger.error({ err }, "chat completion delivery failed")));
-
         trackHeartbeatSchedulerWork(accountingHeartbeat.reconcileCostAccounting().catch((err) => logger.error({ err }, "Cost accounting recovery failed")));
 
-        trackHeartbeatSchedulerWork(connectionDeliveries.sweepPending().catch((err) => logger.error({ err }, "connection continuation delivery failed")));
         trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed")));
         trackHeartbeatSchedulerWork(app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "gateway token cleanup failed")));
-        trackHeartbeatSchedulerWork(app.locals.toolActionDeliveries.sweepPending().catch((err: unknown) => logger.error({ err }, "tool review delivery sweep failed")));
-        trackHeartbeatSchedulerWork(questionResponseDeliveries.sweepPending()
-          .then((result) => {
-            if (result.scanned > 0) {
-              logger.info(result, "periodic question-response delivery sweep completed");
-            }
-          })
-          .catch((err) => {
-            logger.error({ err }, "periodic question-response delivery sweep failed");
-          }));
 
         if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
         if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
@@ -1956,7 +1940,7 @@ async function startServerWithDatabaseTeardown(
   ) => {
     await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]);
     heartbeatSchedulerStopped = true;
-    unsubscribeChatCompletions();
+    await Promise.all(deliveryWorkers.map(worker => worker.stop()));
     clearInterval(executionControlInterval);
     if (heartbeatSchedulerInterval) {
       clearInterval(heartbeatSchedulerInterval);

server/src/services/chat-completion-delivery.ts

@@ -1,3 +1,5 @@
+import { signalDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "./delivery-queue-worker.js";
 import { and, asc, eq, inArray, lte, sql } from "drizzle-orm";
 import { agents, agentWakeupRequests, chatCompletionDeliveries as deliveries, chatTaskHandoffs as handoffs,
   heartbeatRuns, issueComments, issueDocuments, issues, type Db } from "@paperclipai/db";
@@ -33,6 +35,7 @@ export async function recordChatHandoff(tx: Connection, task: Issue, actorRunId:
 /** Must run on the same transaction as status projection (including native arbitration). */
 export async function recordChatCompletion(tx: Connection, before: Issue, after: Issue) {
   if (before.status === after.status) return;
+  signalDatabaseWork(tx, DELIVERY_QUEUES.chatCompletion);
   await tx.update(deliveries).set({ status: "superseded" }).where(and(eq(deliveries.taskId, after.id),
     eq(deliveries.companyId, after.companyId), inArray(deliveries.status, [...pending])));
   if (after.status !== "done") return;
@@ -243,5 +246,9 @@ export function chatCompletionDeliveryService(db: Db, heartbeat: { wakeup(agentI
       .orderBy(asc(deliveries.nextAttemptAt)).limit(100);
     for (const row of due) await deliver(row.id);
   }
-  return { deliver, sweepPending };
+  async function hasPending() {
+    return (await db.select({ id: deliveries.id }).from(deliveries)
+      .where(inArray(deliveries.status, [...pending])).limit(1)).length > 0;
+  }
+  return { deliver, sweepPending, hasPending };
 }

server/src/services/connection-intent-delivery.ts

@@ -167,7 +167,11 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
       await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(eq(connectionIntentDeliveries.interactionId, interactionId));
     }
   }
-  return { deliver, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => {
+  async function hasPending() {
+    return (await db.select({ id: connectionIntentDeliveries.interactionId }).from(connectionIntentDeliveries)
+      .where(isNull(connectionIntentDeliveries.deliveredAt)).limit(1)).length > 0;
+  }
+  return { deliver, hasPending, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => {
     const rows = await db.select().from(connectionIntentDeliveries).where(and(isNull(connectionIntentDeliveries.deliveredAt), lte(connectionIntentDeliveries.nextAttemptAt, new Date())))
       .orderBy(asc(connectionIntentDeliveries.nextAttemptAt)).limit(50);
     let failed = 0;

server/src/services/delivery-queue-worker.ts

@@ -0,0 +1,88 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { beginIdleTrackedWork } from "./task-admission.js";
+
+export const DELIVERY_QUEUES = {
+  feedback: "feedback-exports",
+  chatCompletion: "chat-completions",
+  connection: "connection-continuations",
+  question: "question-responses",
+  toolAction: "tool-action-receipts",
+} as const;
+
+/**
+ * Reconcile once at startup and on committed writes. Only outstanding work,
+ * a failed reconciliation, or a deferred admission owns a retry timer. A
+ * successful empty reconciliation disarms the worker until the next signal.
+ */
+export function createDeliveryQueueWorker(input: {
+  db: object;
+  topic: string;
+  retryMs: number;
+  run: () => Promise<unknown>;
+  hasPending: () => Promise<boolean>;
+  canRun: () => boolean;
+  onError: (error: unknown) => void;
+}) {
+  let stopped = false;
+  let dirty = false;
+  let uncertainCommit = false;
+  let running: Promise<void> | null = null;
+  let timer: ReturnType<typeof setTimeout> | null = null;
+
+  function arm(delay: number) {
+    if (stopped) return;
+    if (timer) clearTimeout(timer);
+    timer = setTimeout(() => {
+      timer = null;
+      start();
+    }, delay);
+    timer.unref?.();
+  }
+
+  function start() {
+    if (stopped || running) return;
+    if (timer) clearTimeout(timer);
+    timer = null;
+    if (!input.canRun()) { arm(input.retryMs); return; }
+    dirty = false;
+    const finish = beginIdleTrackedWork();
+    running = Promise.resolve().then(async () => {
+      await input.run();
+      return input.hasPending();
+    }).then(pending => {
+      if (pending || uncertainCommit) arm(input.retryMs);
+    }).catch(error => {
+      arm(input.retryMs);
+      input.onError(error);
+    }).finally(() => {
+      running = null;
+      finish();
+      // A commit during either the sweep or its empty check must not be lost.
+      if (dirty) arm(0);
+    });
+  }
+
+  function wake(uncertain = false) {
+    if (stopped) return;
+    // A lost COMMIT acknowledgement can precede visibility on another
+    // connection. Even observing other pending work cannot resolve that
+    // ambiguity. Retain a conservative recovery backstop until restart.
+    uncertainCommit ||= uncertain;
+    dirty = true;
+    if (!running) arm(0);
+  }
+  const unsubscribe = subscribeDatabaseWork(input.db, input.topic, wake);
+  // Callers can await this first reconciliation before marking startup ready.
+  const ready = Promise.resolve().then(() => { start(); return running; });
+  return {
+    ready,
+    wake,
+    async stop() {
+      stopped = true;
+      unsubscribe();
+      if (timer) clearTimeout(timer);
+      timer = null;
+      await running;
+    },
+  };
+}

server/src/services/feedback.ts

@@ -1,3 +1,5 @@
+import { signalDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "./delivery-queue-worker.js";
 import { readFile, readdir } from "node:fs/promises";
 import path from "node:path";
 import { and, asc, desc, eq, getTableColumns, gte, isNull, lte, ne, or } from "drizzle-orm";
@@ -1715,6 +1717,15 @@ async function buildFeedbackTraceBundleFromRow(
 }
 
 export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
+  // The route's immediate flush and the queue worker share this service.
+  // Serialize them so an enqueue signal cannot upload the same trace twice.
+  let exportFlush: Promise<unknown> = Promise.resolve();
+  function serializeExportFlush<T>(operation: () => Promise<T>): Promise<T> {
+    const result = exportFlush.then(operation, operation);
+    exportFlush = result.catch(() => undefined);
+    return result;
+  }
+
   return {
     listIssueVotesForUser: async (issueId: string, authorUserId: string) =>
       db
@@ -1786,12 +1797,20 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
       return row ? buildFeedbackTraceBundleFromRow(db, row) : null;
     },
 
+    hasPendingFeedbackTraces: async () => {
+      const pending = options.shareClient
+        ? or(eq(feedbackExports.status, "pending"), eq(feedbackExports.status, "failed"))
+        : eq(feedbackExports.status, "pending");
+      return (await db.select({ id: feedbackExports.id }).from(feedbackExports)
+        .where(pending).limit(1)).length > 0;
+    },
+
     flushPendingFeedbackTraces: async (input?: {
       companyId?: string;
       traceId?: string;
       limit?: number;
       now?: Date;
-    }) => {
+    }) => serializeExportFlush(async () => {
       const shareClient = options.shareClient;
       if (!shareClient) {
         const filters = [eq(feedbackExports.status, "pending")];
@@ -1900,7 +1919,7 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
         sent,
         failed,
       };
-    },
+    }),
 
     saveIssueVote: async (input: {
       issueId: string;
@@ -2071,6 +2090,7 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
           })
           .where(eq(feedbackVotes.id, savedVote.id));
 
+        signalDatabaseWork(tx, DELIVERY_QUEUES.feedback);
         const [savedTrace] = await tx
           .insert(feedbackExports)
           .values({

server/src/services/issue-thread-interactions.ts

@@ -1,3 +1,5 @@
+import { signalDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "./delivery-queue-worker.js";
 import { activeIssueInteractionCondition, historicalQuestionCondition } from "./issue-question-context.js";
 import {
   currentContinuationOrigins,
@@ -2823,6 +2825,7 @@ export function issueThreadInteractionService(
         .returning();
         if (!row) throw interactionAlreadyResolvedError();
         if (status === "accepted" || status === "rejected") {
+          signalDatabaseWork(tx, DELIVERY_QUEUES.connection);
           await tx.insert(connectionIntentDeliveries).values({ interactionId, companyId: issue.companyId }).onConflictDoNothing();
         }
         return row;
@@ -5017,6 +5020,7 @@ export function issueThreadInteractionService(
         // This answer updates conversation history only. It must not resume
         // the completed source run or enqueue new work for the closed task.
         if (!historicalAnswer) {
+          signalDatabaseWork(tx, DELIVERY_QUEUES.question);
           await tx
             .insert(issueQuestionResponseDeliveries)
             .values(questionResponseDeliveryValues(answered));

server/src/services/question-response-delivery.ts

@@ -1340,5 +1340,9 @@ export function questionResponseDeliveryService(
     return counts;
   }
 
-  return { deliver, sweepPending };
+  async function hasPending() {
+    return (await db.select({ id: issueQuestionResponseDeliveries.id }).from(issueQuestionResponseDeliveries)
+      .where(inArray(issueQuestionResponseDeliveries.status, ["pending", "delivering"])).limit(1)).length > 0;
+  }
+  return { deliver, sweepPending, hasPending };
 }

server/src/services/tool-action-delivery.ts

@@ -345,6 +345,12 @@ export function toolActionDeliveryService(
   }
   return {
     deliver,
+    async hasPending() {
+      // Include receipts awaiting review/execution: their later terminal write
+      // must remain recoverable even if its optional fast-path callback fails.
+      return (await db.select({ id: toolActionDeliveries.actionRequestId }).from(toolActionDeliveries)
+        .where(isNull(toolActionDeliveries.deliveredAt)).limit(1)).length > 0;
+    },
     async deliverForRun(input: { companyId: string; runId: string }) {
       // A review can settle before its source run yields. Retry that run's
       // receipts as soon as execution cleanup finishes; the periodic sweep

server/src/services/tool-action-review.ts

@@ -1,3 +1,5 @@
+import { signalDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "./delivery-queue-worker.js";
 import { and, eq } from "drizzle-orm";
 import {
   issues,
@@ -206,6 +208,7 @@ export async function commitToolActionReview(
               : { version: 1, outcome: "rejected", reason: input.reason },
         })
         .where(eq(issueThreadInteractions.id, interaction.id));
+      signalDatabaseWork(tx, DELIVERY_QUEUES.toolAction);
       await tx
         .insert(toolActionDeliveries)
         .values({

server/src/services/tool-gateway.ts

@@ -1,3 +1,5 @@
+import { signalDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "./delivery-queue-worker.js";
 import { composeConnectionInstructions } from "./connection-instructions.js";
 import { isInsufficientConnectionScope, INSUFFICIENT_CONNECTION_SCOPE_MESSAGE } from "./connection-permission-errors.js";
 import { boundedMcpToolName } from "./mcp-tool-names.js";
@@ -2607,7 +2609,11 @@ export function createToolGatewayService(
         issueId: input.session.issueId,
         interactionId: interaction.id,
       })
-      .onConflictDoNothing();
+      .onConflictDoNothing()
+      .then(
+        () => signalDatabaseWork(db, DELIVERY_QUEUES.toolAction),
+        (error) => { signalDatabaseWork(db, DELIVERY_QUEUES.toolAction, true); throw error; },
+      );
 
     await writeToolCallEvent({
       invocationId: input.invocation.id,