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)