From 46a1e4a4344e682ed583fd2034975045710409df Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 8 Oct 2026 14:57:50 -0500 Subject: [PATCH] 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 --- doc/DATABASE.md | 15 +++ ...-instance-sleep-and-scheduler-inventory.md | 60 ++++++++++- packages/db/src/client.ts | 3 +- packages/db/src/index.ts | 2 + packages/db/src/work-signals.test.ts | 70 ++++++++++++ packages/db/src/work-signals.ts | 67 ++++++++++++ .../chat-completion-delivery.test.ts | 29 +++++ .../connection-intents-service.test.ts | 6 ++ .../__tests__/delivery-queue-worker.test.ts | 100 ++++++++++++++++++ server/src/__tests__/feedback-service.test.ts | 11 +- .../issue-thread-interactions-service.test.ts | 6 ++ .../server-startup-feedback-export.test.ts | 8 +- .../__tests__/tool-gateway-service.test.ts | 7 ++ server/src/app.ts | 48 +++------ server/src/index.ts | 52 ++++----- .../src/services/chat-completion-delivery.ts | 9 +- .../services/connection-intent-delivery.ts | 6 +- server/src/services/delivery-queue-worker.ts | 88 +++++++++++++++ server/src/services/feedback.ts | 24 ++++- .../src/services/issue-thread-interactions.ts | 4 + .../services/question-response-delivery.ts | 6 +- server/src/services/tool-action-delivery.ts | 6 ++ server/src/services/tool-action-review.ts | 3 + server/src/services/tool-gateway.ts | 8 +- 24 files changed, 558 insertions(+), 80 deletions(-) create mode 100644 packages/db/src/work-signals.test.ts create mode 100644 packages/db/src/work-signals.ts create mode 100644 server/src/__tests__/delivery-queue-worker.test.ts create mode 100644 server/src/services/delivery-queue-worker.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 7799ead39ed..7653be473c8 100644 --- a/doc/DATABASE.md +++ b/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). diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index cb0f6bf1122..072f87d1ab2 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/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 | diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index c1b34d21eb6..e9e23569a08 100644 --- a/packages/db/src/client.ts +++ b/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", })); diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 378c9bd37b8..a53a8d68227 100644 --- a/packages/db/src/index.ts +++ b/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"; diff --git a/packages/db/src/work-signals.test.ts b/packages/db/src/work-signals.test.ts new file mode 100644 index 00000000000..c5f929ebc9e --- /dev/null +++ b/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(callback: (tx: Connection) => Promise): Promise }; +function connection(commit?: () => void): Connection { + return { async transaction(callback: (tx: ReturnType) => Promise): Promise { + 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(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); + }); +}); diff --git a/packages/db/src/work-signals.ts b/packages/db/src/work-signals.ts new file mode 100644 index 00000000000..01bd0900173 --- /dev/null +++ b/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> }; +type Context = { scope: Scope; pending: Set | null }; +const contexts = new WeakMap(); + +// 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 }; +export function installDatabaseWorkSignals(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(); + 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); }; +} diff --git a/server/src/__tests__/chat-completion-delivery.test.ts b/server/src/__tests__/chat-completion-delivery.test.ts index bb8e7f076db..7a73180eeda 100644 --- a/server/src/__tests__/chat-completion-delivery.test.ts +++ b/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 }]); diff --git a/server/src/__tests__/connection-intents-service.test.ts b/server/src/__tests__/connection-intents-service.test.ts index c6c318c372d..058b61ad2ba 100644 --- a/server/src/__tests__/connection-intents-service.test.ts +++ b/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"); diff --git a/server/src/__tests__/delivery-queue-worker.test.ts b/server/src/__tests__/delivery-queue-worker.test.ts new file mode 100644 index 00000000000..eaa7371f75c --- /dev/null +++ b/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[] = []; +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); + }); +}); diff --git a/server/src/__tests__/feedback-service.test.ts b/server/src/__tests__/feedback-service.test.ts index ed3c7f21898..22f7a096aec 100644 --- a/server/src/__tests__/feedback-service.test.ts +++ b/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 | undefined; diff --git a/server/src/__tests__/issue-thread-interactions-service.test.ts b/server/src/__tests__/issue-thread-interactions-service.test.ts index 6e909fc8b3a..7744fae5230 100644 --- a/server/src/__tests__/issue-thread-interactions-service.test.ts +++ b/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"); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 7b89d44dd9a..8b4bc743a88 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/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, diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 2f0774be90c..795cf76ee03 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/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", diff --git a/server/src/app.ts b/server/src/app.ts index 6c9dd7e32ec..bd921edcb5b 100644 --- a/server/src/app.ts +++ b/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; flushPendingFeedbackTraces(input?: { companyId?: string; traceId?: string; @@ -1187,41 +1189,17 @@ export async function createApp( jobCoordinator.start(); scheduler.start(); - let feedbackExportShuttingDown = false; - let feedbackExportTimer: ReturnType | 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) { diff --git a/server/src/index.ts b/server/src/index.ts index 0c79e723e62..91b09b76c7c 100644 --- a/server/src/index.ts +++ b/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"; /// // 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], + ] 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); diff --git a/server/src/services/chat-completion-delivery.ts b/server/src/services/chat-completion-delivery.ts index 1cbd95b5009..51d3a544fe5 100644 --- a/server/src/services/chat-completion-delivery.ts +++ b/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 }; } diff --git a/server/src/services/connection-intent-delivery.ts b/server/src/services/connection-intent-delivery.ts index 540524bf4ea..c128d309fcd 100644 --- a/server/src/services/connection-intent-delivery.ts +++ b/server/src/services/connection-intent-delivery.ts @@ -167,7 +167,11 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick { 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; diff --git a/server/src/services/delivery-queue-worker.ts b/server/src/services/delivery-queue-worker.ts new file mode 100644 index 00000000000..8ee487b5e60 --- /dev/null +++ b/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; + hasPending: () => Promise; + canRun: () => boolean; + onError: (error: unknown) => void; +}) { + let stopped = false; + let dirty = false; + let uncertainCommit = false; + let running: Promise | null = null; + let timer: ReturnType | 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; + }, + }; +} diff --git a/server/src/services/feedback.ts b/server/src/services/feedback.ts index 2bbc0361a3c..789d16c5acb 100644 --- a/server/src/services/feedback.ts +++ b/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 = Promise.resolve(); + function serializeExportFlush(operation: () => Promise): Promise { + 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({ diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 3b4c78bc075..60a92e4ecf3 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/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)); diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 5db1e051508..dad35c77d7e 100644 --- a/server/src/services/question-response-delivery.ts +++ b/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 }; } diff --git a/server/src/services/tool-action-delivery.ts b/server/src/services/tool-action-delivery.ts index 63f217de9a0..0718ed2d449 100644 --- a/server/src/services/tool-action-delivery.ts +++ b/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 diff --git a/server/src/services/tool-action-review.ts b/server/src/services/tool-action-review.ts index b429b35ce69..3bf3a0fd3de 100644 --- a/server/src/services/tool-action-review.ts +++ b/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({ diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 3e56275d205..7f7c7079642 100644 --- a/server/src/services/tool-gateway.ts +++ b/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,