From c6ee96a3df7d5614e9db6990f678ef0c60089622 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 21 Oct 2026 07:55:09 -0500 Subject: [PATCH] fix: stop idle MCP event polling (#17036) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - MCP task subscriptions and Dot mailboxes notify external clients when work changes. > - Their delivery workers queried SQL every two seconds, including when no work existed. > - Those queries prevent idle databases from becoming quiet across many hosted instances. > - Existing activity, mailbox, subscription, and delivery records already preserve the work. > - This pull request uses committed-write signals and saved retry deadlines to drive delivery. > - Idle subscriptions become quiet while startup, retry, drain, and shutdown recovery remain supported. ## Linked Issues or Issue Description Related sleep migrations: #15625, #17000, #17008. Searched open PRs and issues for MCP event sleep/polling work; no duplicate found. **What happened?** Each public MCP and Dot event service started a two-second interval. It checked settings, scanned subscriptions, created delivery receipts, cleaned expired records, and attempted delivery even without changes. **Expected behavior** An empty or subscribed-but-idle service produces no recurring SQL or network traffic. Committed events trigger prompt delivery. Only unfinished work retains a retry deadline. **Steps to reproduce** Configure a public MCP origin and start the server. Observe recurring settings queries with MCP disabled, or subscription/queue queries with it enabled. No incoming task changes are required. **Paperclip version or commit** Initially inspected master at `8b4ab65b1c3d863d282752e83307943720a5c85c`; rebased onto `c2daf906d9` after master advanced. The only conflict was overlapping imports in the heartbeat queue; both were preserved. ## What Changed - Register separate MCP and Dot workers with the existing delivery coordinator. Remove both unconditional intervals. - Notify from committed activity, comment, mailbox, subscription, and experimental-settings writes. Centralize comment insertion in one small helper so normal callers do not manage recovery state. - Preserve outer transaction ownership, uncertainty reconciliation, and requests deferred during drain or standby. - Restore delivery retry/abandoned-claim deadlines on startup. Scan subscriptions in pages of 20; drain full source and delivery batches without another external event. Scope subscription refreshes to their subscription or shared Dot binding, issue activity to its task, comment discovery to its company, and mailbox/assignment changes to their binding. Retry deadlines do not rescan unchanged sources. - Run expired-record cleanup opportunistically. Preserve on-use expiry checks and the existing one-time automatic Dot challenge behavior. - Add real PostgreSQL lifecycle coverage and guarded external test database options for existing MCP/Dot integrations. Document producers and the supported writer model. ## Verification - Final head `e54fb34485921016ef988f7df9cca173a9ebdfc8`: all CI checks passed (55 successful checks/statuses, one neutral review, two intentionally skipped Storybook checks). Mergeable with no unresolved review threads. Greptile: 5/5 on this exact commit; security scan passed. - All 284 affected tests passed in one local run after rebasing: 16 real-PostgreSQL lifecycle tests, 24 coordinator/scheduler/standby tests, 84 public MCP tests, 54 Dot runner tests, and 106 interaction-service tests. - Server TypeScript passed after refreshing generated Runner/plugin SDK dependencies from master. Full workspace typecheck and build passed before the rebase; final CI repeated typecheck and build successfully on the rebased head. - Coverage includes SQL/network quiet with configuration and history, committed writes, outer commit/rollback, drain/standby, retry restoration, reenable/unsubscribe, concurrent writes, shutdown, source/delivery/subscription batch boundaries, Dot challenges and reconnects, native broker fences, comment discovery, and company-isolated scans. - CI exposed an existing lock-order test whose insert proxy missed nested transactions and waited indefinitely. The fixture now observes nested inserts, preserves the lock assertion, and fails explicitly if interception is missed. The complete 106-test interaction suite and the formerly hanging CI shard pass. The warm-standby/settings test doubles were also updated for the new coordinator/transaction boundaries. - Earlier fixture failures (required timestamps, cleanup, external database selection, async waits) were corrected. One existing OAuth first-poll timing assertion and one E2E company-cleanup race passed on single reruns; the final full CI run passed without a rerun. No OAuth or company-deletion behavior was changed. - The local full `pnpm test:run` attempt could not initialize embedded PostgreSQL after five attempts. It was stopped; affected integrations were verified with separate disposable PostgreSQL 17 databases, and final CI supplied the full-suite result. Embedded-only local skips are not counted as passes. ## Risks - Notifications use one live server/database scope. Independent SQL writers and separate live replicas do not trigger these process-local workers. Startup recovers durable state; continuous multi-writer discovery needs a notification/ingress contract. - Idle physical cleanup waits for startup or another enabled sweep. Subscription and authorization expiry remain enforced on use. - Existing delivery claims, retry bounds, signatures, and receipt deduplication remain in place. No new tables or migrations. - Host sleep-safety checks remain conservative. MCP/Dot retry deadlines are not yet covered by Paperclip Cloud's scheduled-wake contract. This proves worker quiet, not deployed host suspension or fleet savings. ## Model Used OpenAI Codex (GPT-6), with reasoning, local code execution, and tool use. The runtime does not expose a more specific model version or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/DEVELOPING.md | 50 +++ .../src/__tests__/cloud-warm-standby.test.ts | 12 +- server/src/__tests__/dot-runner.test.ts | 10 +- .../instance-settings-managed-overlay.test.ts | 1 + .../issue-thread-interactions-service.test.ts | 19 +- server/src/__tests__/public-mcp.test.ts | 14 +- server/src/app.ts | 4 +- server/src/modules/external-agents/dot.ts | 3 + .../modules/external-agents/native-broker.ts | 3 + .../modules/wake-queue/adapters/postgres.ts | 2 + server/src/services/activity-log.ts | 9 +- server/src/services/activity.ts | 12 +- .../services/delivery-work-notifications.ts | 2 + .../src/services/dot-assignment-follow-up.ts | 3 + server/src/services/fast-responses.ts | 28 +- server/src/services/heartbeat/queue.ts | 3 +- server/src/services/heartbeat/scheduling.ts | 3 +- server/src/services/instance-settings.ts | 21 +- server/src/services/issue-comment-writes.ts | 12 + server/src/services/issues.ts | 55 ++-- .../services/public-mcp/events-work.test.ts | 291 ++++++++++++++++++ server/src/services/public-mcp/events.ts | 99 +++++- server/src/services/workspace-runtime.ts | 43 ++- 23 files changed, 585 insertions(+), 114 deletions(-) create mode 100644 server/src/services/issue-comment-writes.ts create mode 100644 server/src/services/public-mcp/events-work.test.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index f4d2b902062..4b500837668 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3165,3 +3165,53 @@ unavailable. Use a separate, disposable database whose name ends in `_test` for `PAPERCLIP_PROJECT_SKILLS_TEST_DATABASE_URL`, and `PAPERCLIP_SKILL_SOURCES_TEST_DATABASE_URL`. The suites apply migrations and create test data; the synchronization suite truncates company data between tests. + +### Public MCP and Dot event delivery + +Public MCP task events and Dot mailbox events use separate topics in the shared +transaction-aware delivery coordinator. They no longer scan SQL every two seconds. +An enabled subscription without new events or unfinished deliveries is quiet, +including when delivery history remains in the database. + +| Obligation | Source and wake path | +| --- | --- | +| Task status, comments, documents, prototype work | Existing activity rows; issue activity writes in `persistActivity`, the activity service, and wake recovery notify before their transaction commits | +| Dot task follow-up discovery | Existing comments and assignments; `insertIssueComment` centralizes committed comment notifications, and assignment acceptance wakes the scanner | +| Dot mailbox delivery | Existing mailbox rows; native broker events, explicit challenges, fencing, follow-up publication/consumption, and subscription reconnection notify through their owning transactions | +| Subscription creation, renewal, removal | The subscription transaction wakes its worker; renewal can recover unreceived Dot assignment references | +| Feature disable/reenable | The experimental settings transaction wakes both workers; disabled workers remain quiet until another change | +| Delivery retries and abandoned claims | Existing `mcp_event_deliveries.next_attempt_at` drives the next deadline, reconstructed on startup | +| Admission and old subscription cleanup | Opportunistic on startup and subsequent enabled sweeps; expiry remains enforced on use | + +The scanner preserves durable delivery receipts instead of advancing a timestamp +cursor that could skip a late commit. Subscription scans use pages of 20; full source/delivery batches schedule another +sweep until caught up. Subscription creation, refresh, and removal target their +subscription ID (or the shared Dot binding for reconnect mailbox entries). Issue +activity targets its task; comment discovery targets its company; mailbox and +assignment changes target their Dot binding. Delivery retries do not rescan +unchanged subscriptions. Only startup and instance-wide settings changes request +a full scan. Settings changes retain an explicit wildcard when a transaction also +contains scoped hints. A fresh request or failed sweep survives warm standby and +idle drain; skipped dispatch never marks it complete. Shutdown stops admission of +new batches and waits for the current bounded webhook attempt to settle. + +Automatic Dot readiness challenges are issued once for a connected subscription. +A pending or expired challenge does not imply periodic renewal; an explicit retest +creates and signals another challenge. The worker keeps the existing challenge and +authorization checks. No schema, OAuth, or webhook contract changes are required. + +Notifications require supported writers in the same live server/database scope. +Independent SQL writers and separate live server replicas do not wake this process; +startup reconstructs their durable records, but continuous multi-writer discovery +would require an ingress or notification contract before removing its backstop. +The host sleep-safety checks remain conservative for MCP/Dot state. These retry +deadlines are not yet exported to Paperclip Cloud's scheduled-wake contract. + +`events-work.test.ts` exercises real PostgreSQL commits, rollback, startup/drain, +retry restoration, producer reactivation, large batches, and SQL/network quiet. +When embedded PostgreSQL is unavailable, use separate disposable databases ending +in `_test` for `PAPERCLIP_MCP_SLEEP_TEST_DATABASE_URL`, +`PAPERCLIP_PUBLIC_MCP_TEST_DATABASE_URL`, and +`PAPERCLIP_DOT_RUNNER_TEST_DATABASE_URL`. The interaction-service regression suite +also accepts `PAPERCLIP_INTERACTIONS_TEST_DATABASE_URL`. The suites migrate and write to those +URLs; the sleep suite truncates company data between tests. diff --git a/server/src/__tests__/cloud-warm-standby.test.ts b/server/src/__tests__/cloud-warm-standby.test.ts index 60cdc331114..e5d8bbfdcd5 100644 --- a/server/src/__tests__/cloud-warm-standby.test.ts +++ b/server/src/__tests__/cloud-warm-standby.test.ts @@ -1,3 +1,4 @@ +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { createServer, request as httpRequest } from "node:http"; import express from "express"; import request from "supertest"; @@ -101,23 +102,28 @@ describe("unclaimed Cloud background work", () => { } }); - it("MCP event timers do not read the persisted setting until claim", async () => { + it("MCP event work does not read the persisted setting until claim", async () => { vi.useFakeTimers(); let standby = true; const isEnabled = vi.fn().mockResolvedValue(false); - const events = createPublicMcpEvents({} as Db, { isEnabled } as PublicMcpOAuth, vi.fn(), { + const db = installDatabaseWorkSignals({ execute: vi.fn(), transaction: vi.fn() }) as unknown as Db; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !standby, onError: vi.fn() }); + const events = createPublicMcpEvents(db, { isEnabled } as PublicMcpOAuth, vi.fn(), { isBackgroundWorkEnabled: () => !standby, }); try { - events.start(); + await events.start(coordinator).ready; await vi.advanceTimersByTimeAsync(10 * 60_000); await events.tick(); expect(isEnabled).not.toHaveBeenCalled(); standby = false; await vi.advanceTimersByTimeAsync(2000); expect(isEnabled).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(60_000); + expect(isEnabled).toHaveBeenCalledOnce(); } finally { await events.stop(); + await coordinator.stop(); } }); diff --git a/server/src/__tests__/dot-runner.test.ts b/server/src/__tests__/dot-runner.test.ts index 5d70180a010..ff99b1f8cbd 100644 --- a/server/src/__tests__/dot-runner.test.ts +++ b/server/src/__tests__/dot-runner.test.ts @@ -1,3 +1,5 @@ +import { applyPendingMigrations } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createGrokRoutineDispatcher } from "../services/grok-bot/dispatcher.js"; import { grokRoutineService } from "../services/grok-bot/routine.js"; import { assertNoExternalOverlap } from "../modules/external-agents/admission.js"; @@ -42,6 +44,8 @@ import { consumeDotHistoryReceipt, publishActiveDotComment } from "../services/d import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js"; import { createReleaseIssueExecution } from "../modules/wake-queue/application/use-cases.js"; +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DOT_RUNNER_TEST_DATABASE_URL"); + describe("durable Dot Runner integration", () => { let temporary: Awaited>; let db: ReturnType; @@ -49,8 +53,9 @@ describe("durable Dot Runner integration", () => { beforeAll(async () => { const runnerRoot = fileURLToPath(new URL("../../../packages/paperclip-runner/", import.meta.url)); execFileSync("cargo", ["build", "--release", "--locked", "--manifest-path", join(runnerRoot, "runner/Cargo.toml"), "-p", "paperclip-runner-core", "--bin", "paperclip-runnerd"], { cwd: runnerRoot, stdio: "inherit", timeout: 300000 }); - temporary = await startEmbeddedPostgresTestDatabase("paperclip-dot-runner-"); - db = createDb(temporary.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else temporary = await startEmbeddedPostgresTestDatabase("paperclip-dot-runner-"); + db = createDb(externalUrl ?? temporary.connectionString); root = await mkdtemp(join(tmpdir(), "paperclip-dot-state-")); vi.stubEnv("PAPERCLIP_IN_WORKTREE", "false"); vi.stubEnv("PAPERCLIP_RUNNER_BINARY", join(runnerRoot, "runner/target/release", process.platform === "win32" ? "paperclip-runnerd.exe" : "paperclip-runnerd")); @@ -65,6 +70,7 @@ describe("durable Dot Runner integration", () => { }); afterAll(async () => { try { + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); } finally { try { diff --git a/server/src/__tests__/instance-settings-managed-overlay.test.ts b/server/src/__tests__/instance-settings-managed-overlay.test.ts index 5d29183a3d0..5285f5b4694 100644 --- a/server/src/__tests__/instance-settings-managed-overlay.test.ts +++ b/server/src/__tests__/instance-settings-managed-overlay.test.ts @@ -31,6 +31,7 @@ function managedEnv(raw: string | undefined = MANAGED_RAW) { function stubDb(row: Record) { const persistedSets: Array> = []; const db = { + transaction: (fn: (tx: Db) => Promise) => fn(db), select: () => ({ from: () => ({ where: () => Promise.resolve([row]) }) }), insert: () => { throw new Error("unexpected insert in test"); diff --git a/server/src/__tests__/issue-thread-interactions-service.test.ts b/server/src/__tests__/issue-thread-interactions-service.test.ts index 04225ce2c88..97ef4f5be2f 100644 --- a/server/src/__tests__/issue-thread-interactions-service.test.ts +++ b/server/src/__tests__/issue-thread-interactions-service.test.ts @@ -4,6 +4,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { + applyPendingMigrations, activityLog, agents, authUsers, @@ -31,13 +32,15 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { ONBOARDING_FIRST_TASK_ORIGIN_KIND } from "@paperclipai/shared"; import { instanceSettingsService } from "../services/instance-settings.js"; import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { agentService } from "../services/agents.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_INTERACTIONS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("issueThreadInteractionService", () => { @@ -47,8 +50,9 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-thread-interactions-"); - db = createDb(tempDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-thread-interactions-"); + db = createDb(externalUrl ?? tempDb!.connectionString); issuesSvc = issueService(db); interactionsSvc = issueThreadInteractionService(db); }, 20_000); @@ -78,6 +82,7 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -717,8 +722,10 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { const release = new Promise((resolve) => { releaseInsert = resolve; }); await db.transaction(async (tx) => { - const lockedTx = new Proxy(tx as any, { + const observeInserts = (runner: typeof tx): typeof tx => new Proxy(runner as any, { get(target, property, receiver) { + if (property === "transaction") return (callback: (nested: typeof tx) => Promise) => + target.transaction((nested: typeof tx) => callback(observeInserts(nested))); if (property !== "insert") return Reflect.get(target, property, receiver); return (table: unknown) => { const builder = target.insert(table); @@ -752,10 +759,10 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { "Comment inserted under a caller-owned transaction.", { userId: "board-user" }, { authorType: "user" }, - lockedTx, + observeInserts(tx), ); - await reached; try { + await Promise.race([reached, commentPromise.then(() => { throw new Error("Comment insert was not intercepted"); })]); await expect(db.transaction(async (observer) => { await observer.execute(sql` select id from issues diff --git a/server/src/__tests__/public-mcp.test.ts b/server/src/__tests__/public-mcp.test.ts index 30cb12009a7..3d15a28f6e8 100644 --- a/server/src/__tests__/public-mcp.test.ts +++ b/server/src/__tests__/public-mcp.test.ts @@ -1,3 +1,5 @@ +import { applyPendingMigrations } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { mkdtemp, rm } from "node:fs/promises"; import { join } from "node:path"; import { tmpdir } from "node:os"; @@ -36,7 +38,8 @@ import { DotHarnessDriver } from "../../../packages/paperclip-runner/src/drivers import { HarnessDriverBackend } from "../../../packages/paperclip-runner/src/backends/harness-driver-backend.js"; import { eventFetch, signingKey, type EventFetch } from "../services/public-mcp/event-webhooks.js"; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_PUBLIC_MCP_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Public MCP database checks unavailable: ${support.reason}`); const config = { origin: "https://paperclip.example", resource: "https://paperclip.example/mcp/paperclip" }; const redirectUri = "https://client.example/callback"; @@ -60,13 +63,14 @@ describe.skipIf(!support.supported)("public MCP OAuth and tool boundary", () => let oauth: ReturnType; beforeAll(async () => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); - temp = await startEmbeddedPostgresTestDatabase("paperclip-public-mcp-"); - db = createDb(temp.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else temp = await startEmbeddedPostgresTestDatabase("paperclip-public-mcp-"); + db = createDb(externalUrl ?? temp.connectionString); oauth = createPublicMcpOAuth(db, config); expect(await oauth.isEnabled()).toBe(false); await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true }); }, 90000); - afterAll(async () => { await temp?.cleanup(); vi.unstubAllEnvs(); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temp?.cleanup(); vi.unstubAllEnvs(); }); it("keeps public MCP and discovery out of SQL and SPA fallback until Cloud claim", async () => { let standby = true; @@ -853,7 +857,7 @@ describe.skipIf(!support.supported)("public MCP OAuth and tool boundary", () => it("completes OAuth and concurrent upload retries with a one-connection pool", async () => { const originalDb = db, originalOauth = oauth; - db = createDb(temp.connectionString, { maxConnections: 1 }); + db = createDb(externalUrl ?? temp.connectionString, { maxConnections: 1 }); oauth = createPublicMcpOAuth(db, config); let cleanup: (() => Promise) | undefined; try { diff --git a/server/src/app.ts b/server/src/app.ts index 0f120634d19..d5983a03f0d 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1080,13 +1080,11 @@ export async function createApp( publicMcpEvents = createPublicMcpEvents(db, publicMcpOAuth, dispatch, { isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(), }); - publicMcpEvents.start(); const transfers = createPublicMcpTransfers(db, publicMcpOAuth, dispatch, opts.storageService); publicMcpIngress.use(transfers.router); publicMcpIngress.use(publicMcpIngressRoutes(publicMcpOAuth, createPublicMcpExecutor(db, publicMcpOAuth, dispatch, transfers), publicMcpEvents)); const dotOAuth = createPublicMcpOAuth(db, { ...publicMcpOAuth.config, resource: publicMcpOAuth.config.origin + "/mcp/runner" }); dotMcpEvents = createPublicMcpEvents(db, dotOAuth, dispatch, { enableDotRunner: true, isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive() }); - dotMcpEvents.start(); const grokTransfers = createPublicMcpTransfers(db, dotOAuth, dispatch, opts.storageService, { pathPrefix: "/mcp/runner/files", agentSession: { authorize: (principal, sessionId, taskId, write, database) => dotRunnerBroker(database ?? db).authorizeFileSession(principal, sessionId, taskId, write) } }); publicMcpIngress.use(grokTransfers.router); @@ -1283,6 +1281,8 @@ export async function createApp( nextRunAt: () => agentLifecycle.nextRunAt(), }); agentLifecycle.setWake(agentLifecycleDelivery.wake); + publicMcpEvents?.start(deliveryWork); + dotMcpEvents?.start(deliveryWork); const stopMuseMaintenance = startMuseMaintenance(db, deliveryWork); const resourceLifecycleDelivery = registerPluginLifecycleDelivery(deliveryWork, db, workerManager, lifecycle); await resourceLifecycleDelivery.ready; diff --git a/server/src/modules/external-agents/dot.ts b/server/src/modules/external-agents/dot.ts index 83840280596..2295c71ea37 100644 --- a/server/src/modules/external-agents/dot.ts +++ b/server/src/modules/external-agents/dot.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; import { withExternalAdmissionGuard, withExternalExecutionAdmission } from "./admission.js"; import { createNativeExternalBroker } from "./native-broker.js"; import { uploadSchema, type PublicMcpTransfers } from "../../services/public-mcp/file-transfers.js"; @@ -177,6 +178,7 @@ function createBroker(db: Db) { for (const { assignment } of terminal) { await tx.update(assignments).set({ status: "fenced" }).where(eq(assignments.id, assignment.id)); if (binding.providerKind === "grok_bot") await tx.update(grokBotRoutines).set({ recoveryRequiredAt: new Date(), recoveryReason: "work_session", updatedAt: new Date() }).where(eq(grokBotRoutines.bindingId, binding.id)); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${bindingId}`); await tx.insert(mailbox).values({ companyId, bindingId, bindingGeneration: binding.generation, assignmentId: assignment.id, kind: "authority_revoked", sourceEventId: randomUUID(), references: { assignmentId: assignment.id, runId: assignment.runId, externalStopConfirmed: false } }); @@ -465,6 +467,7 @@ function createBroker(db: Db) { throw fail("The event subscription is no longer verified."); } await tx.update(bindings).set({ challengeHash: hash(nonce), challengeExpiresAt: new Date(Date.now() + 10 * 60_000), updatedAt: new Date() }).where(eq(bindings.id, b.id)); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${b.id}`); await tx.insert(mailbox).values({ companyId, bindingId: b.id, bindingGeneration: b.generation, kind: "readiness_challenge", sourceEventId: "challenge_" + randomUUID(), references: { challenge: nonce } }); await logActivity(tx as unknown as Db, { companyId, actorType: "user", actorId: b.operatorId, diff --git a/server/src/modules/external-agents/native-broker.ts b/server/src/modules/external-agents/native-broker.ts index fe89c303672..f67181b347d 100644 --- a/server/src/modules/external-agents/native-broker.ts +++ b/server/src/modules/external-agents/native-broker.ts @@ -186,6 +186,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I return (await tx.execute(sql`select ${assignmentColumns} from ${assignmentTable} where run_id=${b.runId}::uuid and company_id=${b.companyId}::uuid and agent_id=${b.agentId}::uuid and binding_id=${b.bindingId}::uuid and binding_generation=${b.bindingGeneration} and normalized_session_id=${b.normalizedSessionId} and turn_id=${b.turnId} and revision=${b.assignmentRevision} for update`))[0]; } async function reserveRevocationEvent(tx:ExternalAdmissionTransaction,a:Assignment,sourceEventId:string,references:Record) { + if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); const [created]=await tx.execute(sql`insert into ${mailboxTable} (company_id,binding_id,binding_generation,assignment_id,kind,source_event_id,"references") values (${a.companyId}::uuid,${a.bindingId}::uuid,${a.bindingGeneration},${a.id}::uuid,'authority_revoked',${sourceEventId},${JSON.stringify(references)}::jsonb) on conflict do nothing returning id`); if(created)return true; const [old]=await tx.execute<{companyId:string;assignmentId:string|null;kind:string}>(sql`select company_id as "companyId",assignment_id as "assignmentId",kind from ${mailboxTable} where binding_id=${a.bindingId}::uuid and binding_generation=${a.bindingGeneration} and source_event_id=${sourceEventId}`); @@ -271,6 +272,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I if (!old && provider === "muse") await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); const [a]=old?[old]:await tx.execute(sql`insert into ${assignmentTable} (company_id,binding_id,binding_generation,run_id,agent_id,normalized_session_id,turn_id,revision,controller_generation,catalog_digest,status,projection,accept_by,expires_at) values (${b.companyId}::uuid,${b.bindingId}::uuid,${b.bindingGeneration},${runId}::uuid,${b.agentId}::uuid,${b.normalizedSessionId},${b.turnId},${b.assignmentRevision},${owner.controllerGeneration},${externalOperationDigest("tool",{tools:p.tools})},'offered',${JSON.stringify(projection)}::jsonb,${new Date(Number(p.acceptByUnixMs)).toISOString()}::timestamptz,${new Date(Number(p.expiresAtUnixMs)).toISOString()}::timestamptz) returning ${assignmentColumns}`); if(!a)throw conflict("Assignment creation failed."); + if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); await tx.execute(sql`insert into ${mailboxTable} (company_id,binding_id,binding_generation,assignment_id,kind,source_event_id,"references") values (${a.companyId}::uuid,${a.bindingId}::uuid,${a.bindingGeneration},${a.id}::uuid,'assignment',${event.sourceEventId},${JSON.stringify({assignmentId:a.id,runId,revision:a.revision})}::jsonb) on conflict do nothing`); if(binding.providerKind==="grok_bot"){ const [run]=await tx.select({contextSnapshot:heartbeatRuns.contextSnapshot}).from(heartbeatRuns).where(eq(heartbeatRuns.id,runId)); @@ -296,6 +298,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I // the same fact must preserve later work and its original evidence. return; } + if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); await tx.execute(sql`update ${operationTable} set status=${provider==="muse"&&outcome.status==="rejected"?"rejected":store.completed},outcome=${JSON.stringify(outcome)}::jsonb,source_event_id=${event.sourceEventId},updated_at=now() where id=${op.id}::uuid`); const action=op.command.action; await tx.execute(sql`update ${assignmentTable} set last_activity_at=now() where id=${a.id}::uuid`); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index a74aaf3ecc6..c4929f482ba 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; import { readRunCancellation } from "../../../services/run-cancellation.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; @@ -768,6 +769,7 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow const projected = await issueService(tx).update(issue.id, { status: "blocked" }, tx); if (projected) { nativeFailureBlock = { runId: run.id, statusVersion: projected.statusVersion }; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${issue.id}`); await tx.insert(activityLog).values({ companyId: issue.companyId, actorType: "system", actorId: "execution-recovery", action: "issue.updated", entityType: "issue", entityId: issue.id, runId: run.id, diff --git a/server/src/services/activity-log.ts b/server/src/services/activity-log.ts index 4657c4eb426..72b18795541 100644 --- a/server/src/services/activity-log.ts +++ b/server/src/services/activity-log.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { beginRunPhase } from "./execution-preparation-timing.js"; import type { ActivityPublication } from "../types/activity-publication.js"; export type { ActivityPublication } from "../types/activity-publication.js"; @@ -164,7 +165,7 @@ export function publishActivity(publication: ActivityPublication) { export async function persistActivity(db: Db, input: LogActivityInput) { const redactedDetails = await redactActivityDetails(db, input.details ?? null); const responsibleUserId = await resolveResponsibleUserIdForActivity(db, input); - const [activity] = await db.insert(activityLog).values({ + const insert = (runner: Db) => runner.insert(activityLog).values({ companyId: input.companyId, actorType: input.actorType, actorId: input.actorId, @@ -176,6 +177,12 @@ export async function persistActivity(db: Db, input: LogActivityInput) { responsibleUserId, details: redactedDetails, }).returning({ id: activityLog.id }); + const [activity] = input.entityType === "issue" + ? await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${input.entityId}`); + return insert(tx as unknown as Db); + }) + : await insert(db); const payload = { issueId: input.issueId ?? (input.entityType === "issue" ? input.entityId : null), diff --git a/server/src/services/activity.ts b/server/src/services/activity.ts index 70f2fab7c68..d4f6b03983e 100644 --- a/server/src/services/activity.ts +++ b/server/src/services/activity.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { skillAccessService } from "./skill-access.js"; import { credentialAccessConnectionNameSql } from "./credential-access-visibility.js"; import { executionProjectionsForRuns } from "./execution-projection.js"; @@ -649,11 +650,10 @@ export function activityService(db: Db) { return [fromContext, ...fromActivity]; }, - create: (data: typeof activityLog.$inferInsert) => - db - .insert(activityLog) - .values(data) - .returning() - .then((rows) => rows[0]), + create: (data: typeof activityLog.$inferInsert) => db.transaction(async tx => { + if (data.entityType === "issue") await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${data.entityId}`); + const [row] = await tx.insert(activityLog).values(data).returning(); + return row; + }), }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 4466035fd8b..b1c7c62062a 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -23,6 +23,8 @@ export const DELIVERY_QUEUES = { browser: "browser-use", email: "email-channels", museMaintenance: "muse-maintenance", + mcpEvents: "mcp-events", + dotEvents: "dot-events", skillSourceSync: "skill-source-sync", fastResponse: "fast-responses", feedback: "feedback-exports", diff --git a/server/src/services/dot-assignment-follow-up.ts b/server/src/services/dot-assignment-follow-up.ts index 194dcab6eb1..7b9182770ac 100644 --- a/server/src/services/dot-assignment-follow-up.ts +++ b/server/src/services/dot-assignment-follow-up.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { withExternalAdmissionGuard } from "../modules/external-agents/index.js"; import { and, eq, gt, gte, inArray, isNull, sql } from "drizzle-orm"; import { dotAgentBindings, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations, agentWakeupRequests, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db"; @@ -32,6 +33,7 @@ export async function publishActiveDotComment(db: Db, input: { gt(dotRunnerAssignments.expiresAt, new Date()))).limit(1); if (!assignment) return false; // Same key as the event scanner: either path may discover the comment first. + await notifyDeliveryWork(db, DELIVERY_QUEUES.dotEvents, `binding:${binding.id}`); await db.insert(dotMailboxItems).values({ companyId: input.companyId, bindingId: binding.id, bindingGeneration: binding.generation, assignmentId: assignment.id, kind: "follow_up", sourceEventId: `dot-follow-up:${assignment.id}:${input.commentId}`, @@ -89,6 +91,7 @@ export async function consumeDotHistoryReceipt(db: Db, operation: typeof dotRunn if (!comments.length) return; // Materialize consumption even if the comment wake has not yet been admitted. // Admission, event scanning and receipt reads all serialize on this binding. + await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${binding.id}`); for (const comment of comments) await tx.insert(dotMailboxItems).values({ companyId: source.companyId, bindingId: binding.id, bindingGeneration: binding.generation, assignmentId: source.id, kind: "follow_up", sourceEventId: `dot-follow-up:${source.id}:${comment.id}`, diff --git a/server/src/services/fast-responses.ts b/server/src/services/fast-responses.ts index abbd6cde874..8ba6665dea6 100644 --- a/server/src/services/fast-responses.ts +++ b/server/src/services/fast-responses.ts @@ -1,3 +1,4 @@ +import { insertIssueComment } from "./issue-comment-writes.js"; import { randomUUID } from "node:crypto"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; import { @@ -961,21 +962,18 @@ export function fastResponseService( ), ) : []; - const [comment] = await tx - .insert(issueComments) - .values({ - companyId: request.companyId, - issueId: request.issueId!, - authorAgentId: request.agentId, - authorUserId: request.agentId ? null : "board-concierge", - authorType: request.agentId ? "agent" : "system", - body: safe.text, - sourceTrust: source?.sourceTrust ?? null, - conversationSessionGeneration: request.sessionGeneration, - origin: "fast_response", - fastResponseRequestId: request.id, - }) - .returning(); + const [comment] = await insertIssueComment(tx, { + companyId: request.companyId, + issueId: request.issueId!, + authorAgentId: request.agentId, + authorUserId: request.agentId ? null : "board-concierge", + authorType: request.agentId ? "agent" : "system", + body: safe.text, + sourceTrust: source?.sourceTrust ?? null, + conversationSessionGeneration: request.sessionGeneration, + origin: "fast_response", + fastResponseRequestId: request.id, + }); if ( request.endpointId && request.conversationId && diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 4a1df79a5b1..73df5a8b722 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,5 +1,6 @@ import { authorizationService } from "../authorization.js"; import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry } from "../task-retry.js"; +import { insertIssueComment } from "../issue-comment-writes.js"; import { withExternalAdmissionGuard, assertNoExternalOverlap } from "../../modules/external-agents/index.js"; import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js"; import { normalizeAgentNameKey } from "./retries.js"; @@ -3774,7 +3775,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende updatedAt: now, }) .where(eq(issues.id, issue.id)); - await tx.insert(issueComments).values({ + await insertIssueComment(tx, { companyId: issue.companyId, issueId: issue.id, body: blockedComment, diff --git a/server/src/services/heartbeat/scheduling.ts b/server/src/services/heartbeat/scheduling.ts index d233e6fe56c..c1714de7b55 100644 --- a/server/src/services/heartbeat/scheduling.ts +++ b/server/src/services/heartbeat/scheduling.ts @@ -1,3 +1,4 @@ +import { insertIssueComment } from "../issue-comment-writes.js"; import { issues, issueComments, @@ -383,7 +384,7 @@ export function createHeartbeatScheduling( } if (input.recoveryPolicy === "escalate_to_board") { - await db.insert(issueComments).values({ + await insertIssueComment(db, { companyId: input.claimed.companyId, issueId: input.claimed.id, body: monitorRecoveryComment({ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 77e6e32dd43..c3c5b01b505 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { Db } from "@paperclipai/db"; import { companies, instanceSettings } from "@paperclipai/db"; @@ -604,14 +605,18 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService } } const now = new Date(); - const [updated] = await db - .update(instanceSettings) - .set({ - experimental: { ...nextExperimental }, - updatedAt: now, - }) - .where(eq(instanceSettings.id, current.id)) - .returning(); + const [updated] = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); + return tx + .update(instanceSettings) + .set({ + experimental: { ...nextExperimental }, + updatedAt: now, + }) + .where(eq(instanceSettings.id, current.id)) + .returning(); + }); return toInstanceSettings(updated ?? current); }, diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts new file mode 100644 index 00000000000..e24e78e9855 --- /dev/null +++ b/server/src/services/issue-comment-writes.ts @@ -0,0 +1,12 @@ +import type { PgInsertValue } from "drizzle-orm/pg-core"; +import { issueComments, type Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +type WriteDb = Db | Parameters[0]>[0]; + +export function insertIssueComment(db: WriteDb, input: PgInsertValue & { companyId: string }) { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); + return tx.insert(issueComments).values(input).returning(); + }); +} diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1f5f1bb33d3..9acfdf86074 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { insertIssueComment } from "./issue-comment-writes.js"; import { measureRunTransaction } from "./execution-transaction-timing.js"; import { readNativeMonitorSource } from "./native-runtime/native-monitor-delivery.js"; import { readNativeContinuationSource } from "./native-runtime/native-continuation-delivery.js"; @@ -7911,15 +7912,12 @@ export function issueService(db: Db) { ); if (!parent) return null; - const [comment] = await dbOrTx - .insert(issueComments) - .values({ - companyId: child.companyId, - issueId: parent.id, - authorType: "system", - body, - }) - .returning(); + const [comment] = await insertIssueComment(dbOrTx, { + companyId: child.companyId, + issueId: parent.id, + authorType: "system", + body, + }); await dbOrTx .update(issues) .set({ updatedAt: new Date() }) @@ -12804,27 +12802,24 @@ export function issueService(db: Db) { } } if (!comment) { - [comment] = await measureRunPhase("response.write", () => dbOrTx - .insert(issueComments) - .values({ - companyId: issue.companyId, - issueId, - authorAgentId: actor.agentId ?? null, - authorUserId: actor.userId ?? null, - onBehalfOfUserId, - authorType, - createdByRunId, - body: redactedBody, - clientRequestId: options?.clientRequestId ?? null, - presentation, - metadata, - sourceTrust: options?.sourceTrust ?? null, - createdAt: validCreatedAt ?? currentInsertTimestamp!, - ...(currentInsertTimestamp - ? { updatedAt: currentInsertTimestamp } - : {}), - }) - .returning()); + [comment] = await measureRunPhase("response.write", () => insertIssueComment(dbOrTx, { + companyId: issue.companyId, + issueId, + authorAgentId: actor.agentId ?? null, + authorUserId: actor.userId ?? null, + onBehalfOfUserId, + authorType, + createdByRunId, + body: redactedBody, + clientRequestId: options?.clientRequestId ?? null, + presentation, + metadata, + sourceTrust: options?.sourceTrust ?? null, + createdAt: validCreatedAt ?? currentInsertTimestamp!, + ...(currentInsertTimestamp + ? { updatedAt: currentInsertTimestamp } + : {}), + })); } if (!comment) throw new Error("Failed to create issue comment"); if (options?.completionReply && actor.agentId && createdByRunId) await acknowledgeChatCompletionReply(dbOrTx, createdByRunId, comment.id); diff --git a/server/src/services/public-mcp/events-work.test.ts b/server/src/services/public-mcp/events-work.test.ts new file mode 100644 index 00000000000..5d3f8d62592 --- /dev/null +++ b/server/src/services/public-mcp/events-work.test.ts @@ -0,0 +1,291 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { applyPendingMigrations, createDb, companies, authUsers, companyMemberships, mcpOauthClients, mcpOauthGrants, issues, agents, dotAgentBindings, dotMailboxItems, dotRunnerAssignments, heartbeatRuns, mcpEventDeliveries, mcpEventSubscriptions } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { createPublicMcpEvents } from "./events.js"; +import { createPublicMcpOAuth } from "./oauth.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { logActivity } from "../activity-log.js"; +import { insertIssueComment } from "../issue-comment-writes.js"; +import { dotRunnerBroker } from "../dot-runner-broker.js"; +import { idleWorkSnapshot } from "../task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MCP_SLEEP_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("MCP event committed work", () => { + let temporary: Awaited>; + let db: ReturnType; + let coordinator: ReturnType; + let service: ReturnType; + let companyId: string, userId: string, taskId: string; + let draining: boolean, standby: boolean; + let calls: string[], received: Array<{ eventId: string; data: Record }>; + let errors: unknown[], status: number; + let failRead: boolean; + let duringDelivery: (() => Promise) | undefined; + beforeAll(async () => { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); + if (externalUrl) await applyPendingMigrations(externalUrl); + else temporary = await startEmbeddedPostgresTestDatabase("mcp-sleep-"); + db = createDb(externalUrl ?? temporary.connectionString, { maxConnections: 6 }); + }, 90000); + beforeEach(async () => { + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + draining = false; standby = false; calls = []; received = []; errors = []; status = 204; failRead = false; duringDelivery = undefined; + companyId = randomUUID(); userId = randomUUID(); taskId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "MCP idle", issuePrefix: "MCP" }); + await db.insert(authUsers).values({ id: userId, name: "MCP test", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalId: userId, principalType: "user", membershipRole: "owner", status: "active" }); + await db.insert(issues).values({ id: taskId, companyId, title: "Events" }); + await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true, enableOpenAiDot: true }); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval", "Date"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await service?.stop(); await coordinator?.stop(); + vi.useRealTimers(); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + await db.delete(authUsers).where(eq(authUsers.id, userId)); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); vi.unstubAllEnvs(); }); + + async function fixture(dot = false, scope = { companyId, taskId }) { + const { companyId, taskId } = scope; + const config = { origin: "https://paperclip.example", resource: `https://paperclip.example/mcp/${dot ? "runner" : "paperclip"}` }; + const clientId = randomUUID(); + await db.insert(mcpOauthClients).values({ id: clientId, name: "Sleep fixture", redirectUris: ["https://receiver.example/oauth"] }); + const agentId = dot ? randomUUID() : null; + if (agentId) await db.insert(agents).values({ id: agentId, companyId, name: "Dot", status: "active", adapterType: "paperclip_runner", adapterConfig: { provider: "openai_dot" } }); + const [grant] = await db.insert(mcpOauthGrants).values({ companyId, userId, clientId, resource: config.resource, purpose: dot ? "agent" : "personal", agentId, scopes: [dot ? "paperclip:agent" : "paperclip:read"] }).returning(); + const bindingId = dot ? randomUUID() : null; + if (bindingId && agentId) { + await db.insert(dotAgentBindings).values({ id: bindingId, companyId, agentId, operatorId: userId, grantId: grant.id, status: "connected" }); + await db.update(agents).set({ adapterConfig: { provider: "openai_dot", dotBindingId: bindingId } }).where(eq(agents.id, agentId)); + } + const observed = new Proxy(db, { get(target, key) { + if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) return Reflect.get(target, key); + return (...args: unknown[]) => { + calls.push(String(key)); + if (failRead && key === "select") { failRead = false; throw new Error("injected read failure"); } + return Reflect.apply(Reflect.get(target, key), target, args); + }; + } }); + const oauth = createPublicMcpOAuth(observed, config); + const principal = await oauth.authorizeGrant(grant.id); + const input = { name: dot ? "paperclip.dot.mailbox_updated" : "paperclip.task.status_changed", arguments: { companyId, ...(dot ? { bindingId } : { taskId }) }, + delivery: { mode: "webhook", url: "https://receiver.example/events", secret: "whsec_" + randomBytes(32).toString("base64") } }; + const makeService = () => createPublicMcpEvents(observed, oauth, async () => ({ id: taskId, companyId }), { enableDotRunner: dot, + isBackgroundWorkEnabled: () => !draining && !standby, + fetch: async (_url, init) => { + const body = JSON.parse(String(init.body)); + if (body.type === "verification") return Response.json({ challenge: body.challenge }); + received.push(body); + const callback = duringDelivery; duringDelivery = undefined; await callback?.(); + return new Response(null, { status }); + }, + }); + service = makeService(); + const start = async () => { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining && !standby, canReconcile: () => !standby, onError: error => errors.push(error) }); + await service.start(coordinator).ready; + await settle(); + }; + const restart = async () => { await service.stop(); await coordinator.stop(); service = makeService(); await start(); }; + return { input, principal, start, restart, bindingId: bindingId!, agentId: agentId!, subscribe: () => service.subscribe(principal, input) }; + } + async function settle() { + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + } + async function quiet(ms = 60_000) { + await settle(); const baseline = [...calls], deliveries = received.length; + await vi.advanceTimersByTimeAsync(ms); + expect(calls).toEqual(baseline); expect(received).toHaveLength(deliveries); + } + async function activity(database = db, matching = true) { + return logActivity(database, { companyId, actorType: "user", actorId: userId, entityType: "issue", entityId: taskId, action: matching ? "issue.updated" : "issue.unrelated", details: { status: "done", _previous: { status: "todo" } } }); + } + async function delivered(count: number) { await vi.waitFor(() => expect(received).toHaveLength(count), { timeout: 8_000 }); await settle(); } + + it("is quiet empty, subscribed without events, and with delivered history", async () => { + const f = await fixture(); await f.start(); await quiet(); + vi.setSystemTime(vi.getRealSystemTime()); + await f.subscribe(); await settle(); await quiet(); + await activity(); + await delivered(1); await quiet(600_000); + expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); + }); + + it("waits for outer commit and rolls back nested producer notifications", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { written = resolve; }); + const transaction = db.transaction(async tx => { await activity(tx as unknown as typeof db); written(); await gate; }); + await ready; await vi.advanceTimersByTimeAsync(500); + expect(received).toEqual([]); expect(idleWorkSnapshot().active).toBeGreaterThan(0); + release(); await transaction; await delivered(1); + await expect(db.transaction(async tx => { await activity(tx as unknown as typeof db); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settle(); await quiet(); expect(received).toHaveLength(1); + }); + + it("retains first writes and failed sweeps through drain without passive SQL", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); + draining = true; await activity(); await settle(); await quiet(10_000); expect(received).toEqual([]); + draining = false; await vi.advanceTimersByTimeAsync(2_000); await delivered(1); + failRead = true; await activity(); await vi.waitFor(() => expect(errors).toHaveLength(1)); + draining = true; await vi.advanceTimersByTimeAsync(2_000); await settle(); await quiet(10_000); + draining = false; await vi.advanceTimersByTimeAsync(2_000); await delivered(2); await quiet(); + }); + + it("recovers startup during warm standby and drain", async () => { + const f = await fixture(); await f.subscribe(); await activity(); + standby = true; const before = [...calls]; await f.start(); await vi.advanceTimersByTimeAsync(10_000); expect(calls).toEqual(before); + standby = false; draining = true; await vi.advanceTimersByTimeAsync(2_000); await settle(); await quiet(10_000); + draining = false; await vi.advanceTimersByTimeAsync(2_000); await delivered(1); await quiet(); + }); + + it("restores retry and abandoned claim deadlines on restart", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); status = 503; + await activity(); await delivered(1); + const [pending] = await db.select().from(mcpEventDeliveries); + expect(pending.finishedAt).toBeNull(); expect(pending.attempts).toBe(1); + const claimAt = Date.now() + 30_000; + await db.update(mcpEventDeliveries).set({ nextAttemptAt: new Date(claimAt) }).where(eq(mcpEventDeliveries.id, pending.id)); + await f.restart(); expect(coordinator.nextWakeAt()).toBe(claimAt); + await quiet(20_000); status = 204; await vi.advanceTimersByTimeAsync(10_000); await delivered(2); + expect((await db.select().from(mcpEventDeliveries))[0].outcome).toBe("delivered"); await quiet(); + }); + + it("wakes on reenable and cancels retries when unsubscribed", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); + await instanceSettingsService(db).updateExperimental({ enablePublicMcp: false }); await settle(); + await activity(); await settle(); await quiet(); expect(received).toEqual([]); + status = 503; await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true }); await delivered(1); + await service.unsubscribe(f.principal, f.input); await settle(); await quiet(); expect(coordinator.nextWakeAt()).toBeNull(); + }); + + it("drains more than 100 source rows and 20 deliveries without a new write", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); + await db.transaction(async tx => { for (let i = 0; i < 105; i++) await activity(tx as unknown as typeof db, i >= 80); }); + await delivered(25); expect(new Set(received.map(event => event.eventId)).size).toBe(25); + expect(await db.select().from(mcpEventDeliveries)).toHaveLength(105); await quiet(); + }); + + it("scans more than 20 subscriptions after a single committed event", async () => { + const f = await fixture(); + for (let i = 0; i < 20; i++) await service.subscribe(f.principal, { ...f.input, delivery: { ...f.input.delivery, url: `https://receiver.example/${i}` } }); + const second = await fixture(); + for (let i = 0; i < 5; i++) await service.subscribe(second.principal, { ...second.input, delivery: { ...second.input.delivery, url: `https://receiver.example/second/${i}` } }); + await second.start(); await activity(); await delivered(25); await quiet(); + }); + + it("scopes subscription refreshes and does not rescan sources on delivery retries", async () => { + const first = await fixture(); const firstSubscription = await first.subscribe(); + const second = await fixture(); const secondSubscription = await second.subscribe(); await second.start(); + const untouched = new Date(0); + await db.update(mcpEventSubscriptions).set({ scannedAt: untouched }); + await second.subscribe(); await settle(); + const subscriptions = await db.select().from(mcpEventSubscriptions); + expect(subscriptions.find(s => s.id === firstSubscription.id)?.scannedAt).toEqual(untouched); + expect(subscriptions.find(s => s.id === secondSubscription.id)?.scannedAt.getTime()).toBeGreaterThan(0); + status = 503; await activity(); await delivered(2); + await db.update(mcpEventSubscriptions).set({ scannedAt: untouched }); + status = 204; await vi.advanceTimersByTimeAsync(4_000); await delivered(4); + expect((await db.select().from(mcpEventSubscriptions)).every(s => s.scannedAt.getTime() === 0)).toBe(true); + await quiet(); + }); + + it.each([false, true])("keeps tenant content wakes scoped with Dot=%s", async dot => { + const first = await fixture(dot); const firstSubscription = await first.subscribe(); + const other = { companyId: randomUUID(), taskId: randomUUID() }; + await db.insert(companies).values({ id: other.companyId, name: "Other tenant", issuePrefix: "OTHER" }); + await db.insert(companyMemberships).values({ companyId: other.companyId, principalId: userId, principalType: "user", membershipRole: "owner", status: "active" }); + await db.insert(issues).values({ id: other.taskId, companyId: other.companyId, title: "Other task" }); + const second = await fixture(dot, other); const secondSubscription = await second.subscribe(); await second.start(); + if (dot) await delivered(2); + await db.update(mcpEventSubscriptions).set({ scannedAt: new Date(0) }); + if (dot) await insertIssueComment(db, { companyId, issueId: taskId, authorUserId: userId, body: "Tenant-local comment" }); + else await activity(); + await vi.waitFor(async () => { + const subscriptions = await db.select().from(mcpEventSubscriptions); + expect(subscriptions.find(s => s.id === firstSubscription.id)?.scannedAt.getTime()).toBeGreaterThan(0); + expect(subscriptions.find(s => s.id === secondSubscription.id)?.scannedAt.getTime()).toBe(0); + }); + if (!dot) await delivered(1); + await quiet(); + }); + + it("keeps changes arriving during delivery and waits for in-flight shutdown", async () => { + const f = await fixture(); await f.subscribe(); await f.start(); + duringDelivery = async () => { await activity(); }; + await activity(); await delivered(2); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + duringDelivery = async () => { entered(); await gate; }; + await activity(); await ready; + let stopped = false; const stop = service.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); release(); await stop; await coordinator.stop(); + await quiet(); await f.restart(); await quiet(); expect(received).toHaveLength(3); + }); + + it("delivers automatic and explicit Dot challenges and becomes quiet after expiry", async () => { + const f = await fixture(true); await f.start(); await f.subscribe(); await delivered(1); + expect(received[0].data.kind).toBe("readiness_challenge"); + const [challenge] = await db.select().from(dotMailboxItems); + await dotRunnerBroker(db).confirmChallenge(f.principal, String(challenge.references.challenge)); + await quiet(11 * 60_000); + await dotRunnerBroker(db).challenge(companyId, f.agentId); await delivered(2); await quiet(); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + + it("scopes Dot reconnect scans to the binding and delivers to all its subscriptions", async () => { + const f = await fixture(true); await f.subscribe(); await f.start(); await delivered(1); + const [challenge] = await db.select().from(dotMailboxItems); + await dotRunnerBroker(db).confirmChallenge(f.principal, String(challenge.references.challenge)); + await service.subscribe(f.principal, { ...f.input, delivery: { ...f.input.delivery, url: "https://receiver.example/second" } }); + await settle(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: f.agentId, invocationSource: "assignment", status: "queued" }).returning(); + await db.insert(dotRunnerAssignments).values({ companyId, agentId: f.agentId, bindingId: f.bindingId, bindingGeneration: 1, + runId: run.id, normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, catalogDigest: "test", status: "offered", projection: {}, + acceptBy: new Date(Date.now() + 600_000), expiresAt: new Date(Date.now() + 3600_000) }); + await quiet(); + await f.subscribe(); await delivered(3); + expect(received.slice(1).map(event => event.data.kind)).toEqual(["assignment", "assignment"]); + await quiet(); + }); + + it("wakes for a native broker authority fence without another comment or subscription write", async () => { + const f = await fixture(true); await f.subscribe(); await f.start(); await delivered(1); + const broker = dotRunnerBroker(db); + const [challenge] = await db.select().from(dotMailboxItems); + await broker.confirmChallenge(f.principal, String(challenge.references.challenge)); + const snapshot = await broker.snapshot(companyId, f.agentId, f.bindingId); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: f.agentId, invocationSource: "assignment", status: "queued", nativeIssueId: taskId }).returning(); + const [assignment] = await db.insert(dotRunnerAssignments).values({ companyId, agentId: f.agentId, bindingId: f.bindingId, bindingGeneration: 1, + runId: run.id, normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, catalogDigest: "test", status: "offered", projection: {}, + acceptBy: new Date(Date.now() + 60_000), expiresAt: new Date(Date.now() + 3600_000) }).returning(); + await quiet(); + const port = broker.port({ binding: { companyId, agentId: f.agentId, runId: run.id }, provider: { binding: snapshot } }); + await port.dispatch({ sourceEventId: randomUUID(), payload: { kind: "authority_revoked", binding: { ...snapshot, runId: run.id, + normalizedSessionId: assignment.normalizedSessionId, turnId: assignment.turnId, assignmentRevision: assignment.revision } } }); + await delivered(2); expect(received[1].data.kind).toBe("authority_revoked"); await quiet(); + }); + + it("discovers Dot follow-up comments through their committed insert", async () => { + const f = await fixture(true); await f.subscribe(); await f.start(); await delivered(1); + const [challenge] = await db.select().from(dotMailboxItems); + await dotRunnerBroker(db).confirmChallenge(f.principal, String(challenge.references.challenge)); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: f.agentId, invocationSource: "on_demand", status: "running", nativeIssueId: taskId }).returning(); + await db.insert(dotRunnerAssignments).values({ companyId, agentId: f.agentId, bindingId: f.bindingId, bindingGeneration: 1, runId: run.id, normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, catalogDigest: "test", status: "accepted", projection: {}, acceptBy: new Date(Date.now() + 60_000), expiresAt: new Date(Date.now() + 3600_000) }); + await quiet(); + await insertIssueComment(db, { companyId, issueId: taskId, authorUserId: userId, body: "Follow up" }); + await delivered(2); expect(received[1].data.kind).toBe("follow_up"); await quiet(); + }); +}); diff --git a/server/src/services/public-mcp/events.ts b/server/src/services/public-mcp/events.ts index fad08bda65a..8164ab5cc3e 100644 --- a/server/src/services/public-mcp/events.ts +++ b/server/src/services/public-mcp/events.ts @@ -6,7 +6,8 @@ import { dotRunnerBroker } from "../dot-runner-broker.js"; import { ISSUE_STATUSES } from "@paperclipai/shared"; import { localEncryptedProvider } from "../../secrets/local-encrypted-provider.js"; import { logActivity } from "../activity-log.js"; -import { logger } from "../../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { type ApiDispatch } from "./capabilities.js"; import { PublicMcpDisabledError, type McpPrincipal, type PublicMcpOAuth } from "./oauth.js"; import { boundedJson, callbackUrl, eventFetch, McpEventError, postEvent, signingKey, verifyCallback, type EventFetch } from "./event-webhooks.js"; @@ -48,6 +49,12 @@ const dotEventDefinition = { export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDispatch, options: { fetch?: EventFetch; now?: () => number; cloudOrigin?: string; enableDotPrototype?: boolean; enableDotRunner?: boolean; isBackgroundWorkEnabled?: () => boolean } = {}) { const fetcher = options.fetch ?? eventFetch; + const queue = options.enableDotRunner ? DELIVERY_QUEUES.dotEvents : DELIVERY_QUEUES.mcpEvents; + let stopped = false; + let more = false; + let moreSources = false; + let nextPage: string | null = null; + let enabled = false; const now = options.now ?? Date.now; // Dedicated tenant Dot grants are authenticated locally, including background // deliveries. They never use the personal connection's Cloud broker proof. @@ -174,6 +181,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis const value = { companyId: principal.grant.companyId, grantId: principal.grant.id, name: input.name, taskId: input.arguments.taskId ?? null, bindingId: input.arguments.bindingId ?? null, arguments: input.arguments, deliveryMaterial: material, expiresAt, stoppedAt: null, verifiedAt: verify ? new Date(now()) : existing!.verifiedAt, startsAt: existing?.startsAt ?? requestedAt, scannedAt: new Date(now()) }; + await notifyDeliveryWork(tx, queue, input.arguments.bindingId ? `binding:${input.arguments.bindingId}` : id); await tx.insert(subscriptions).values({ id, ...value }).onConflictDoUpdate({ target: subscriptions.id, set: value }); if (input.arguments.bindingId && lease) { const [binding] = await tx.select().from(dotAgentBindings).where(and( @@ -204,6 +212,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis await db.transaction(async tx => { await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${id}, 0))`); await tx.update(admissions).set({ finishedAt: new Date(now()) }).where(and(eq(admissions.subscriptionId, id), isNull(admissions.finishedAt))); + await notifyDeliveryWork(tx, queue, id); await tx.delete(subscriptions).where(and(eq(subscriptions.id, id), eq(subscriptions.grantId, principal.grant.id))); await logActivity(tx as unknown as Db, { companyId: principal.grant.companyId, actorType: "user", actorId: principal.grant.userId, action: "mcp.event_unsubscribed", entityType: "mcp_subscription", entityId: id, details: { name: input.name, taskId: input.arguments.taskId } }); @@ -232,6 +241,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis .where(and(eq(dotRunnerAssignments.companyId, b.companyId), eq(dotRunnerAssignments.bindingId, b.id), eq(dotRunnerAssignments.bindingGeneration, b.generation), eq(dotRunnerAssignments.status, "accepted"), gt(dotRunnerAssignments.expiresAt, new Date(now())), gte(issueComments.createdAt, dotRunnerAssignments.createdAt), sql`not exists (select 1 from ${dotMailboxItems} previous where previous.source_event_id = 'dot-follow-up:' || ${dotRunnerAssignments.id}::text || ':' || ${issueComments.id}::text)`, sql`(${issueComments.authorAgentId} is null or ${issueComments.authorAgentId} <> ${b.agentId})`)).limit(100); + if (incoming.length === 100) moreSources = true; for (const input of incoming) await tx.insert(dotMailboxItems).values({ companyId: b.companyId, bindingId: b.id, bindingGeneration: b.generation, assignmentId: input.assignmentId, kind: "follow_up", sourceEventId: `dot-follow-up:${input.assignmentId}:${input.commentId}`, references: { assignmentId: input.assignmentId, commentId: input.commentId } }).onConflictDoNothing(); }); @@ -240,6 +250,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis .leftJoin(deliveries, and(eq(deliveries.subscriptionId, s.id), eq(deliveries.mailboxItemId, dotMailboxItems.id))) .where(and(eq(dotMailboxItems.bindingId, b.id), eq(dotMailboxItems.bindingGeneration, b.generation), isNull(deliveries.id))) .orderBy(asc(dotMailboxItems.id)).limit(100); + if (rows.length === 100) moreSources = true; for (const { item, assignment } of rows) { const wanted = item.kind === "readiness_challenge" ? !!b.challengeHash && !!b.challengeExpiresAt && b.challengeExpiresAt > new Date(now()) : item.kind === "authority_revoked" || (item.kind !== "follow_up" || item.references.consumed !== true) @@ -262,6 +273,7 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis .where(and(eq(activityLog.companyId, s.companyId), eq(activityLog.entityType, "issue"), eq(activityLog.entityId, s.taskId), gte(activityLog.createdAt, s.startsAt), lt(activityLog.createdAt, s.expiresAt), isNull(deliveries.id))) .orderBy(asc(activityLog.createdAt), asc(activityLog.id)).limit(100); + if (rows.length === 100) moreSources = true; for (const { activity } of rows) { const details = activity.details ?? {}; const changes = details.changes && typeof details.changes === "object" ? details.changes as Record : null; @@ -334,23 +346,90 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis return true; } let running: Promise | null = null; - const tick = () => running ?? (running = (async () => { - // The experimental setting is stored in SQL; check warm standby first. - if (options.isBackgroundWorkEnabled?.() === false || !await oauth.isEnabled()) return; + type Scan = { after: string | null; ids: string[] | null }; + const tick = (scan: Scan | null = { after: null, ids: null }) => running ?? (running = (async () => { + if (stopped || options.isBackgroundWorkEnabled?.() === false) return; + more = false; moreSources = false; nextPage = null; + enabled = await oauth.isEnabled(); + if (!enabled) return; await db.delete(admissions).where(lte(admissions.expiresAt, new Date(now()))); await db.delete(subscriptions).where(lt(subscriptions.expiresAt, new Date(now() - 7 * 24 * hour))); - const active = await db.select().from(subscriptions).where(and(isNull(subscriptions.stoppedAt), gt(subscriptions.expiresAt, new Date(now())), options.enableDotRunner ? isNotNull(subscriptions.bindingId) : isNotNull(subscriptions.taskId))).orderBy(asc(subscriptions.scannedAt)).limit(20); - for (const s of active) await enqueue(s); - for (let i = 0; i < 20; i++) if (!await deliverOne()) break; + if (scan) { + const active = await db.select().from(subscriptions).where(and(isNull(subscriptions.stoppedAt), gt(subscriptions.expiresAt, new Date(now())), + options.enableDotRunner ? isNotNull(subscriptions.bindingId) : isNotNull(subscriptions.taskId), + scan.after ? gt(subscriptions.id, scan.after) : undefined, + scan.ids ? or( + inArray(subscriptions.id, scan.ids.filter(id => !id.includes(":"))), + inArray(subscriptions.bindingId, scan.ids.filter(id => id.startsWith("binding:")).map(id => id.slice(8))), + inArray(subscriptions.taskId, scan.ids.filter(id => id.startsWith("task:")).map(id => id.slice(5))), + inArray(subscriptions.companyId, scan.ids.filter(id => id.startsWith("company:")).map(id => id.slice(8))), + ) : undefined)).orderBy(asc(subscriptions.id)).limit(20); + for (const s of active) { + if (stopped || options.isBackgroundWorkEnabled?.() === false) return; + await enqueue(s); + } + if (active.length === 20) nextPage = active[active.length - 1].id; + } + for (let i = 0; i < 20; i++) { + if (stopped || options.isBackgroundWorkEnabled?.() === false || !await deliverOne()) break; + if (i === 19) more = true; + } })().finally(() => { running = null; })); - let timer: NodeJS.Timeout | undefined; + function start(coordinator: Pick, "register">) { + let requested = 0, completed = -1; + let nextDeadline: number | null = null; + let requestedIds: Set | null = null; + let scan: (Scan & { generation: number }) | null = null; + return coordinator.register(queue, { + retryMs: 2_000, + pausePassiveMaintenance: true, + onSettled: hints => { + requested++; + if (!hints || hints.has("*")) requestedIds = null; + else if (requestedIds) for (const id of hints) requestedIds.add(id); + }, + async run() { + if (stopped || options.isBackgroundWorkEnabled?.() === false) return; + const generation = requested; + if (!scan && completed !== requested) { + scan = { generation: requested, after: null, ids: requestedIds ? [...requestedIds] : null }; + requestedIds = new Set(); + } + await tick(scan); + if (stopped || options.isBackgroundWorkEnabled?.() === false) return; + nextDeadline = null; + if (enabled) { + const [row] = await db.select({ nextAt: sql`min(${deliveries.nextAttemptAt})` }) + .from(deliveries).innerJoin(subscriptions, eq(subscriptions.id, deliveries.subscriptionId)) + .where(and(isNull(deliveries.finishedAt), options.enableDotRunner ? isNotNull(subscriptions.bindingId) : isNotNull(subscriptions.taskId))); + if (!row || row.nextAt === undefined) throw new Error("MCP event deadline query returned no result"); + if (row.nextAt !== null) { + nextDeadline = new Date(row.nextAt).getTime(); + if (!Number.isFinite(nextDeadline)) throw new Error("Invalid MCP event deadline"); + if (nextDeadline <= Date.now()) nextDeadline = Date.now() + 2_000; + } + if (scan && !moreSources) { + if (nextPage) scan.after = nextPage; + else { completed = scan.generation; scan = null; } + } + if (more || scan || completed !== requested) nextDeadline = Date.now(); + } else { + completed = generation; scan = null; + if (completed === requested) requestedIds = new Set(); + } + }, + hasPending: async () => false, + nextRunAt: () => stopped ? null : completed !== requested + ? Math.min(nextDeadline ?? Infinity, Date.now() + 2_000) : nextDeadline, + }); + } return { subscribe, unsubscribe, tick, definitions: options.enableDotRunner ? [{ name: names[4], description: "The bound Dot mailbox changed. Drain paperclip_dot_inbox, read and accept current work. Duplicate/out-of-order events do not imply new assignments. Readiness challenges require paperclip_dot_confirm_event.", delivery: ["webhook"], inputSchema: z.toJSONSchema(z.object({ companyId: z.uuid(), bindingId: z.uuid() }).strict()), payloadSchema: z.toJSONSchema(z.object({ companyId: z.uuid(), bindingId: z.uuid(), bindingGeneration: z.number().int(), mailboxItemId: z.number().int(), kind: z.string() }).passthrough()), }] : [...publicMcpEventDefinitions, ...(options.enableDotPrototype ? [dotEventDefinition] : [])], - start() { if (!timer) { timer = setInterval(() => { void tick().catch(() => logger.warn("Public MCP event delivery tick failed")); }, 2000); timer.unref(); } }, - async stop() { if (timer) clearInterval(timer); timer = undefined; await running?.catch(() => {}); }, + start, + async stop() { stopped = true; await running?.catch(() => {}); }, }; } export type PublicMcpEvents = ReturnType; diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 4fd0db5226f..737dddedc68 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -1,3 +1,4 @@ +import { insertIssueComment } from "./issue-comment-writes.js"; import { restoreGitWorkspaceContents } from "./git-workspace-contents.js"; import { savedGitWorkspaceRestoration, ensureSavedGitCommit, GitCacheRestorationFailure } from "./git-workspace-restoration.js"; import { sanitizeRuntimeServiceBaseEnv } from "./runtime-service-env.js"; @@ -1632,18 +1633,15 @@ async function writeDirtyQuarantineAuditComments(input: { let claimantAuditCommentId: string | null = null; const now = new Date(); if (input.evidence.sourceIssueId) { - const [sourceComment] = await input.db - .insert(issueComments) - .values({ - companyId: input.companyId, - issueId: input.evidence.sourceIssueId, - authorAgentId: null, - authorUserId: null, - authorType: "system", - createdByRunId: input.heartbeatRunId, - body, - }) - .returning({ id: issueComments.id }); + const [sourceComment] = await insertIssueComment(input.db, { + companyId: input.companyId, + issueId: input.evidence.sourceIssueId, + authorAgentId: null, + authorUserId: null, + authorType: "system", + createdByRunId: input.heartbeatRunId, + body, + }); sourceAuditCommentId = sourceComment?.id ?? null; await input.db .update(issues) @@ -1653,18 +1651,15 @@ async function writeDirtyQuarantineAuditComments(input: { const claimantIssueId = input.evidence.contention?.claimedByIssueId ?? null; if (claimantIssueId && claimantIssueId !== input.evidence.sourceIssueId) { - const [claimantComment] = await input.db - .insert(issueComments) - .values({ - companyId: input.companyId, - issueId: claimantIssueId, - authorAgentId: null, - authorUserId: null, - authorType: "system", - createdByRunId: input.heartbeatRunId, - body, - }) - .returning({ id: issueComments.id }); + const [claimantComment] = await insertIssueComment(input.db, { + companyId: input.companyId, + issueId: claimantIssueId, + authorAgentId: null, + authorUserId: null, + authorType: "system", + createdByRunId: input.heartbeatRunId, + body, + }); claimantAuditCommentId = claimantComment?.id ?? null; await input.db .update(issues)