diff --git a/doc/idle-grok-dispatch.md b/doc/idle-grok-dispatch.md new file mode 100644 index 00000000000..799dcce7855 --- /dev/null +++ b/doc/idle-grok-dispatch.md @@ -0,0 +1,121 @@ +# Idle Grok routine dispatch + +The Grok dispatcher previously started a one-second timer whenever public MCP +was configured. Each tick read experimental settings, even when Grok was +otherwise disabled. Enabled ticks scanned prepared dispatches regardless of +whether any work existed. The disabled baseline regression observed three +settings queries in 3.2 seconds. + +The dispatcher now registers with the existing delivery coordinator and shared +scheduler. It performs one startup reconciliation. The worker becomes quiet +when Grok is disabled or when no prepared dispatch remains. Quiet means no +recurring worker SQL or provider calls, including with retained terminal history. +This does not establish whole-instance or database suspension. + +## Ownership and producers + +`grokRoutineService.reserve` is the only supported creator of prepared dispatches. +It signals intent before its insert in the existing transaction. Route verification, +manual retry of a confirmed rejection, and native-broker assignment admission all +use this service. The native broker passes its transaction, so the wake follows +the outer commit. Savepoint release is not publication; rollback does not send. + +`instanceSettingsService.updateExperimental` signals the same queue before its +existing transactional update. This covers enablePublicMcp, enableNativeRunner, +and enableGrokBot changes, including caller-owned outer transactions. It permits +prepared work retained while disabled to resume when enabled. Managed settings +and environment overlays are process configuration and are reconstructed on restart. +Other settings writers update general/default-environment fields, not these flags. + +All other dispatch writers move rows away from prepared: send claims, delivery +settlement, acknowledgement, explicit recovery, assignment completion, and +revocation. None reopens a row as prepared. They cannot introduce hidden new work. +A previously scheduled backlog pass may observe cancellation and disarm. + +Signals cover the current process's shared database owner and child pools. +Independent SQL writers or another process need explicit ingress integration; +this change does not claim cross-process notifications. Startup reconstructs +committed prepared rows from the existing dispatch table. No queue table is added. + +## Delivery and recovery + +The Grok service continues to own current authority checks, binding locks, +prepared-to-sending compare-and-set, secret access, provider delivery, and outcome +settlement. The worker does not interpret provider payloads or relocate policy +into app wiring. A foreground send and background recovery share the durable +claim; only the winner contacts the provider. + +Each pass processes at most ten prepared rows. The next lookup uses the same +predicate and schedules the existing one-second cadence only while prepared work +remains, or after a reconciliation failure. New commits wake the shared worker; +its deadline adapter preserves requested generations through concurrent changes +and drain. The partial created_at/id index keeps both selection and pending-work +lookup independent of retained completed history. The migration builds it +concurrently, is replay-safe, and uses the normal snapshot-retention workflow. + +The existing send-error policy is preserved: if sending throws and the row is +still prepared, the dispatcher cancels it. A row already claimed as sending is +never reset or replayed. Accepted, sending, unknown, acknowledged, rejected, and +cancelled rows do not become automatic retry work. Explicit operator retry still +creates a new intent only for confirmed rejection. This migration does not claim +exactly-once delivery or resolve uncertain provider outcomes automatically. + +## Lifecycle and sleep boundary + +The app registers only when public MCP is configured, after the shared coordinator +exists. Registration does not delay readiness for a provider send. The coordinator +tracks startup work and joins admitted sends on shutdown. Abort prevents the next +row from starting; existing HTTP timeouts bound a current transport operation. +Later prepared rows remain durable for restart. + +Drain and warm standby retain pending requests without passive worker queries. +Resume uses the same coordinator retry path; a first write or failed sweep cannot +be forgotten merely because dispatch was paused. A disabled feature may retain +prepared rows without scheduling a scan; a committed settings change reactivates +it. Disabled feature state is not permission to discard durable work or relax +Cloud sleep guards. Existing binding/assignment/hold blockers remain unchanged. + +## Verification + +- The baseline failed the disabled-feature quiet assertion with three SQL reads. +- 21 real-PostgreSQL lifecycle cases cover disabled/configured idle, first real + reservation, prepared startup recovery, all six non-prepared states, re-enable, + outer commit/rollback, drain/startup, failed scan through drain, concurrent + enqueue, two-page draining, foreground claim overlap, expired unsent work, + settings transaction rollback/commit, and shutdown joining a send. +- The existing 45 Grok qualification cases and 39 startup cases pass together + with the new suite: 105 cases total. +- The four affected settings suites pass 154 cases (259 selected cases total). +- Workspace typecheck, production build, migration checks and replay pass. A + generic-plan probe with 40,000 completed dispatches and one prepared row uses + the partial index for both queries (six shared buffers including the routine + join; 0.032ms batch and 0.025ms next-work execution in this local probe). +- Remote PR review and CI must pass before the slice is ready. + +The tests use the existing real OAuth/agent/routine fixture and a controlled +provider response. A validated disposable external PostgreSQL option permits +local execution where embedded PostgreSQL is unavailable; CI retains embedded +PostgreSQL by default. Provider-host suspension is outside this local proof. + + +## Current parent reconciliation + +The normal merge of target parent `85ce6545bf` preserves its complete tree outside +this duty's explicit files. Grok dispatch, reservation, settings and app wiring +are unchanged from the preceding candidate. The index is regenerated as +`0391_third_roulette.sql`, with concurrent/idempotent execution. All 389 parent +journal entries are unchanged; five generated snapshots remain (0387–0391). +Schema drift, migration safety/numbering and fresh apply/replay on two isolated +PostgreSQL databases pass. + +All 273 cases in nine complete suites pass at `6d81746fe2`: dispatch lifecycle, +Grok qualification, server startup, scoped deadlines and all five instance-settings +suites. Server typecheck and production build pass with the current Runner vendor. +The earlier 259-case and populated-plan measurements above remain historical +evidence; fresh exact-head reviews and CI are required. + +An earlier browser CI run failed during agent-chat fixture restoration with a +lock timeout/HTTP 500; the subsequent case observed the experiment left enabled. +Those failures were not dispatcher assertions and are retained in the evidence. +The later pre-reconciliation `a74305e2b4` checks were terminal and green. Neither +that older result nor local acceptance substitutes for this candidate's CI. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index f81d138e2d9..8d5a064676e 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -16,7 +16,7 @@ Do not mark an item done from a timer removal or a green check alone. | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged after exact-head 5/5 reviews and passing current CI | -| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) await current-head review/CI gates | +| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) merged | @@ -32,6 +32,7 @@ Do not mark an item done from a timer removal or a green check alone. | 14 | Skill-version retention and native-harness backup expiration | Separate retention owners and existing retention guarantees | Pending | | 15 | Automatic database backups | Shared scheduling, actual backup deadlines, existing checkpoint contract | Pending | | 16 | Aggregate maintenance tick | Retire enabled and disabled branches after all constituent owners migrate | Pending | +| 17 | Grok routine dispatch | Committed reservations and feature changes; prepared-only recovery | Slice 13 implemented, local verification in progress | ## Stack and verification contract @@ -768,3 +769,40 @@ that landed dependency and preserves four independent Runner E2E accounting files. All runtime and acceptance test sources are byte-identical to its tested `85ce6545bf` tree (248 cases, server types/build). Fresh final-head CI and reviews remain required after retargeting to master. + +## Slice 13: Grok routine dispatch + +The one-second idle scan is replaced with a shared-coordinator worker. The existing +reservation and experimental-settings transactions notify it; the Grok service +retains all authority, claims, delivery, and uncertainty handling. Prepared-only +selection uses a concurrent partial index. The 21 lifecycle cases, 45 existing +Grok cases, and 39 startup cases pass together (105). Workspace typecheck passes. +See [Grok dispatch](../idle-grok-dispatch.md). Build, populated query proof, PR +publication, and fresh review/CI remain required. Decision retention and every +other pending inventory owner remain in scope. + + +Slice 13 final local verification also passes all 154 affected settings cases +(259 selected tests total), workspace typecheck, production build, migration +checks/replay, and a generic-plan query probe with 40,000 completed dispatches. +Both queries use the prepared-only partial index with six shared buffers, +including the routine join. Remote CI/review remain required. + + +Slice 13 normally integrates target parent `85ce6545bf`. Its Grok implementation +is unchanged; migration 0391 is regenerated after the current parent's history. +All 389 parent journal entries are preserved and five snapshots remain. All 273 +cases across nine complete dispatch/qualification/startup/deadline/settings +suites, server types/build, schema drift, migration safety and fresh apply/replay +pass. Prior browser fixture-cleanup failure evidence remains in its owning audit. +Fresh current-head CI and Greptile/Ponytail are required before merge. + + +Slice 12 merged `e6dff4b06f` → `8dfed893d8` after exact-head Greptile/Ponytail 5/5 +and all 58 checks passed or were neutral/skipped. One failed-job retry passed; +the original CLI startup/ZIP timeouts and local startup limitation remain in its +PR evidence. Slice 13 normally integrates that landed dependency and four +independent Runner E2E accounting files. All runtime and acceptance sources match +its tested/reviewed `679f523d78` tree; the 273-case, type/build and migration +receipts retain their original revisions. Migration 0391 and all parent journal +entries are unchanged. Fresh final-head CI and reviews required against master. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index ff714b6919d..0b743db6fce 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -509,6 +509,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0388_ambitious_micromax.sql", "0389_brave_hulk.sql", "0390_salty_killraven.sql", + "0391_third_roulette.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0391_third_roulette.sql b/packages/db/src/migrations/0391_third_roulette.sql new file mode 100644 index 00000000000..473dde35901 --- /dev/null +++ b/packages/db/src/migrations/0391_third_roulette.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "grok_dispatch_prepared_idx" ON "grok_bot_dispatches" USING btree ("created_at","id") WHERE "grok_bot_dispatches"."status" = 'prepared'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0386_snapshot.json b/packages/db/src/migrations/meta/0391_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0386_snapshot.json rename to packages/db/src/migrations/meta/0391_snapshot.json index 8dd26eef4fb..29b3a477203 100644 --- a/packages/db/src/migrations/meta/0386_snapshot.json +++ b/packages/db/src/migrations/meta/0391_snapshot.json @@ -1,6 +1,6 @@ { - "id": "da3229b5-4eab-4d8b-ba6b-d756e6c42430", - "prevId": "9965d300-e355-4f47-ba0a-951ed3b4b8a2", + "id": "e080fc1b-af8e-4012-8ce5-3b5065b65143", + "prevId": "f38ffbf3-c0ce-435e-aad8-9e5afa6b9417", "version": "7", "dialect": "postgresql", "tables": { @@ -22110,6 +22110,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "decision_archive_notification_outbox_delivery_idx": { + "name": "decision_archive_notification_outbox_delivery_idx", + "columns": [ + { + "expression": "last_attempt_at", + "isExpression": false, + "asc": true, + "nulls": "first" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decision_archive_notification_outbox\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -24030,6 +24058,66 @@ "method": "btree", "with": {} }, + "decisions_running_recovery_idx": { + "name": "decisions_running_recovery_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "decisions_open_expiry_idx": { + "name": "decisions_open_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'open'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "decisions_pending_continuation_idx": { + "name": "decisions_pending_continuation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"continuation_policy\" = 'wake_origin_agent' AND \"decisions\".\"metadata\" ->> 'continuationPending' = 'true' AND (\"decisions\".\"status\" IN ('expired', 'cancelled') OR (\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" IN ('succeeded', 'partial', 'failed')))", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_bundle_idx": { "name": "decisions_bundle_idx", "columns": [ @@ -32019,6 +32107,28 @@ "method": "btree", "with": {} }, + "grok_dispatch_prepared_idx": { + "name": "grok_dispatch_prepared_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'", + "concurrently": false, + "method": "btree", + "with": {} + }, "grok_dispatch_company_status_idx": { "name": "grok_dispatch_company_status_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 9db984a3918..c75fc030ee5 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2724,6 +2724,13 @@ "when": 1793139659776, "tag": "0390_salty_killraven", "breakpoints": true + }, + { + "idx": 391, + "version": "7", + "when": 1793142971791, + "tag": "0391_third_roulette", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/grok_bot.ts b/packages/db/src/schema/grok_bot.ts index da978d3a629..1ad24895a18 100644 --- a/packages/db/src/schema/grok_bot.ts +++ b/packages/db/src/schema/grok_bot.ts @@ -1,3 +1,4 @@ +import { sql } from "drizzle-orm"; import { index, integer, jsonb, pgTable, text, timestamp, uniqueIndex, uuid } from "drizzle-orm/pg-core"; import { companies } from "./companies.js"; import { agents } from "./agents.js"; @@ -49,6 +50,7 @@ export const grokBotDispatches = pgTable("grok_bot_dispatches", { updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, t => [ uniqueIndex("grok_dispatch_event_uq").on(t.bindingId, t.bindingGeneration, t.routineRevision, t.sourceEventId), + index("grok_dispatch_prepared_idx").on(t.createdAt, t.id).where(sql`${t.status} = 'prepared'`), index("grok_dispatch_company_status_idx").on(t.companyId, t.status), index("grok_dispatch_assignment_kind_idx").on(t.assignmentId, t.kind), ]); diff --git a/server/src/__tests__/grok-bot-routine.test.ts b/server/src/__tests__/grok-bot-routine.test.ts index 1bc93d8ff41..434da7cfcbb 100644 --- a/server/src/__tests__/grok-bot-routine.test.ts +++ b/server/src/__tests__/grok-bot-routine.test.ts @@ -4,7 +4,7 @@ import { grokBotRoutes } from "../routes/grok-bot.js"; import { publicMcpIngressRoutes } from "../routes/public-mcp.js"; import { createPublicMcpExecutor } from "../services/public-mcp/capabilities.js"; import { createDotRunnerMcpTools } from "../services/dot-runner-broker.js"; -import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { randomBytes, randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { beforeAll, afterAll, afterEach, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, companyMemberships, createDb, dotAgentBindings, grokBotDispatches, @@ -12,7 +12,8 @@ import { agents, authUsers, companies, companyMemberships, createDb, dotAgentBin import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { accessService } from "../services/access.js"; import { instanceSettingsService } from "../services/instance-settings.js"; -import { createPublicMcpOAuth } from "../services/public-mcp/oauth.js"; +import { grokRoutineFixture } from "./helpers/grok-routine-fixture.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { dotRunnerBroker } from "../services/dot-runner-broker.js"; import { createGrokRoutineDispatcher } from "../services/grok-bot/dispatcher.js"; import { grokRoutineService } from "../services/grok-bot/routine.js"; @@ -26,44 +27,22 @@ import { hasRequiredWorkspaceRecovery } from "../services/workspace-restore-reco import { heartbeatService } from "../services/heartbeat.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; -let temporary: Awaited>; +let temporary: Awaited> | undefined; let db: ReturnType; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-grok-routine-"); - db = createDb(temporary.connectionString); + const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GROK_ROUTINE_TEST_DATABASE_URL"); + if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("paperclip-grok-routine-"); + db = createDb(externalUrl ?? temporary!.connectionString); vi.stubEnv("PAPERCLIP_ISSUE_PRIVACY_MODE", "enforce"); vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true, enableNativeRunner: true, enableGrokBot: true, enableOpenAiDot: true }); }, 120000); -afterAll(async () => { try { await temporary?.cleanup(); } finally { vi.unstubAllEnvs(); } }); +afterAll(async () => { try { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); } finally { vi.unstubAllEnvs(); } }); afterEach(() => vi.restoreAllMocks()); -async function fixture() { - const userId = randomUUID(); - await db.insert(authUsers).values({ id: userId, name: "Grok operator", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); - const [company] = await db.insert(companies).values({ name: "Grok test", issuePrefix: "GK" + randomBytes(3).toString("hex") }).returning(); - await db.insert(companyMemberships).values({ companyId: company!.id, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" }); - await accessService(db).setPrincipalPermission(company!.id, "user", userId, "agents:create", true, userId); - const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Grok test", adapterType: "paperclip_runner", status: "active", adapterConfig: { provider: "grok_bot", allowUnmeteredProvider: true, lifecycleMode: "per_turn" } }).returning(); - const oauth = createPublicMcpOAuth(db, { origin: "https://paperclip.example", resource: "https://paperclip.example/mcp/runner" }); - const client = await oauth.register({ client_name: "Synthetic Grok client", redirect_uris: ["https://client.example/callback"] }, randomUUID()); - const verifier = randomBytes(32).toString("base64url"); - const input = { client_id: client.client_id, redirect_uri: "https://client.example/callback", response_type: "code", resource: oauth.config.resource, - scope: "paperclip:agent offline_access", code_challenge: createHash("sha256").update(verifier).digest("base64url"), code_challenge_method: "S256" }; - const requestId = (await oauth.authorize(input)).split("/").at(-1)!; - const pairing = await dotRunnerBroker(db).createPairing({ companyId: company!.id, agentId: agent!.id, operatorId: userId, provider: "grok_bot" }); - const consent = await oauth.consentDotPairing(requestId, pairing.pairingCode); - const tokens = await oauth.token({ grant_type: "authorization_code", client_id: client.client_id, redirect_uri: input.redirect_uri, - resource: oauth.config.resource, code: new URL(consent.redirectUrl).searchParams.get("code"), code_verifier: verifier }); - const principal = await oauth.authenticate(tokens.access_token); - const delivery = vi.fn().mockResolvedValue({ delivery: "accepted", httpStatus: 200 }); - const service = grokRoutineService(db, delivery); - const config = { companyId: company!.id, agentId: agent!.id, bindingId: pairing.bindingId, operatorId: userId, - expectedRevision: 0, botName: "Synthetic Bot", webhookUrl: "https://bot.example.test/routines/synthetic", webhookKey: "synthetic-sender-key" }; - await service.configure(config); - return { company: company!, agent: agent!, principal, pairing, service, delivery, config, oauth, tokens }; -} +const fixture = () => grokRoutineFixture(db); + async function verification(f: Awaited>, sourceEventId = randomUUID()) { return f.service.reserve({ companyId: f.company.id, agentId: f.agent.id, bindingId: f.pairing.bindingId, sourceEventId, kind: "verify" }); } diff --git a/server/src/__tests__/grok-dispatch-work.test.ts b/server/src/__tests__/grok-dispatch-work.test.ts new file mode 100644 index 00000000000..6d1c243dc5a --- /dev/null +++ b/server/src/__tests__/grok-dispatch-work.test.ts @@ -0,0 +1,174 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDb, grokBotDispatches } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { grokRoutineFixture } from "./helpers/grok-routine-fixture.js"; +import { createGrokRoutineDispatcher } from "../services/grok-bot/dispatcher.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { grokRoutineService } from "../services/grok-bot/routine.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_GROK_DISPATCH_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("Grok dispatch worker", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let coordinator: ReturnType | undefined; + let worker: ReturnType; + let draining: boolean; + const errors: unknown[] = []; + const delivery = vi.fn(async () => ({ delivery: "accepted" as const, httpStatus: 200 })); + const enable = (enableGrokBot = true) => instanceSettingsService(db).updateExperimental({ enablePublicMcp: true, enableNativeRunner: true, enableGrokBot }); + const reserve = (f: Awaited>, database = db) => grokRoutineService(database, delivery).reserve({ + companyId: f.company.id, agentId: f.agent.id, bindingId: f.pairing.bindingId, sourceEventId: randomUUID(), kind: "verify", + }); + const status = async (id: string) => (await db.select().from(grokBotDispatches).where(eq(grokBotDispatches.id, id)))[0]?.status; + + beforeAll(async () => { + if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("grok-dispatch-"); + db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 120_000); + beforeEach(async () => { + vi.stubEnv("PAPERCLIP_ISSUE_PRIVACY_MODE", "enforce"); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); + await db.delete(grokBotDispatches); + await enable(); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + draining = false; errors.length = 0; delivery.mockReset(); + delivery.mockResolvedValue({ delivery: "accepted", httpStatus: 200 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs(); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); + worker = createGrokRoutineDispatcher(db, () => !draining, delivery); + await worker.start(coordinator).ready; + } + async function quiet() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + async function delivered(id: string) { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await status(id)).toBe("accepted"); }, { timeout: 5000 }); + } + + it("does not query every second with Grok disabled", async () => { + await enable(false); await start(); await quiet(); expect(delivery).not.toHaveBeenCalled(); + }); + it("is quiet when configured and wakes from the real reservation", async () => { + const f = await grokRoutineFixture(db); await start(); await quiet(); + const row = await reserve(f); await delivered(row.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(1); + }); + it("reconstructs prepared work on restart", async () => { + const f = await grokRoutineFixture(db); const row = await reserve(f); + await start(); await delivered(row.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(1); + }); + it.each(["sending", "accepted", "unknown", "acknowledged", "rejected", "cancelled"] as const)("does not replay %s history", async state => { + const f = await grokRoutineFixture(db); const row = await reserve(f); + await db.update(grokBotDispatches).set({ status: state }).where(eq(grokBotDispatches.id, row.id)); + await start(); await quiet(); expect(await status(row.id)).toBe(state); expect(delivery).not.toHaveBeenCalled(); + }); + it("keeps disabled prepared work quiet and wakes from committed settings", async () => { + const f = await grokRoutineFixture(db); const row = await reserve(f); await enable(false); + await start(); await quiet(); expect(await status(row.id)).toBe("prepared"); + await enable(); await delivered(row.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(1); + }); + it.each([true, false])("waits for the outer transaction, commit=%s", async commit => { + const f = await grokRoutineFixture(db); await start(); await quiet(); + let release!: () => void; const held = new Promise(resolve => { release = resolve; }); + let created!: Awaited>; + const transaction = db.transaction(async tx => { + created = await reserve(f, tx as unknown as typeof db); await held; + if (!commit) throw new Error("rollback fixture"); + }).catch(error => { if (commit || error.message !== "rollback fixture") throw error; }); + await vi.waitFor(() => expect(created).toBeDefined()); + await vi.advanceTimersByTimeAsync(2000); expect(delivery).not.toHaveBeenCalled(); + release(); await transaction; + if (commit) await delivered(created.id); + else expect(await status(created.id)).toBeUndefined(); + await quiet(); expect(delivery).toHaveBeenCalledTimes(commit ? 1 : 0); + }); + it.each(["startup", "first write"] as const)("preserves %s during drain without passive SQL", async when => { + const f = await grokRoutineFixture(db); + if (when === "first write") { await start(); await quiet(); } + draining = true; const row = await reserve(f); if (when === "startup") await start(); + await vi.advanceTimersByTimeAsync(2000); await vi.waitFor(() => expect(idleWorkSnapshot().active).toBeLessThanOrEqual(1)); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(delivery).not.toHaveBeenCalled(); draining = false; + await delivered(row.id); await quiet(); + }); + it("retains a failed scan through drain and resumes its retry", async () => { + const f = await grokRoutineFixture(db); await start(); await quiet(); + draining = true; const row = await reserve(f); + const read = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("scan unavailable"); }); + draining = false; await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); }); + read.mockRestore(); draining = true; errors.length = 0; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await delivered(row.id); await quiet(); + }); + it("preserves an enqueue during provider delivery", async () => { + const first = await grokRoutineFixture(db), second = await grokRoutineFixture(db); + await start(); await quiet(); let next!: Awaited>; + delivery.mockImplementationOnce(async () => { next = await reserve(second); return { delivery: "accepted", httpStatus: 200 }; }); + const row = await reserve(first); await delivered(row.id); await vi.waitFor(() => expect(next).toBeDefined()); + await delivered(next.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(2); + }); + it("drains more than one page then stops with retained history", async () => { + const rows = []; + for (let i = 0; i < 11; i++) rows.push(await reserve(await grokRoutineFixture(db))); + await start(); await delivered(rows.at(-1)!.id); await quiet(); + expect(delivery).toHaveBeenCalledTimes(11); + expect((await db.select().from(grokBotDispatches)).every(row => row.status === "accepted")).toBe(true); + }); + it("shares the durable claim with a simultaneous foreground send", async () => { + const f = await grokRoutineFixture(db); await start(); await quiet(); + const row = await reserve(f); + await Promise.all([grokRoutineService(db, delivery).send(f.company.id, f.agent.id, row.id), worker.tick()]); + await delivered(row.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(1); + }); + it("cancels an expired unsent dispatch without contacting the provider", async () => { + const f = await grokRoutineFixture(db); const row = await reserve(f); + await db.update(grokBotDispatches).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(grokBotDispatches.id, row.id)); + await start(); await quiet(); expect(await status(row.id)).toBe("cancelled"); expect(delivery).not.toHaveBeenCalled(); + }); + it("does not enable pending dispatches before the settings transaction commits", async () => { + const f = await grokRoutineFixture(db); const row = await reserve(f); await enable(false); + await start(); await quiet(); + await expect(db.transaction(async tx => { + await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableGrokBot: true }); + await vi.advanceTimersByTimeAsync(2000); expect(delivery).not.toHaveBeenCalled(); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect(await status(row.id)).toBe("prepared"); + await db.transaction(async tx => { + await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableGrokBot: true }); + await vi.advanceTimersByTimeAsync(2000); expect(delivery).not.toHaveBeenCalled(); + }); + await delivered(row.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(1); + }); + it("joins an admitted send on shutdown and leaves later rows for restart", async () => { + const first = await grokRoutineFixture(db), second = await grokRoutineFixture(db); + let release!: () => void; const held = new Promise(resolve => { release = resolve; }); + delivery.mockImplementationOnce(async () => { await held; return { delivery: "accepted", httpStatus: 200 }; }); + const one = await reserve(first); const two = await reserve(second); + const starting = start(); await vi.waitFor(() => expect(delivery).toHaveBeenCalledTimes(1)); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(2000); expect(stopped).toBe(false); + release(); await stopping; await starting; + expect(await status(one.id)).toBe("accepted"); expect(await status(two.id)).toBe("prepared"); + await start(); await delivered(two.id); await quiet(); expect(delivery).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/__tests__/helpers/grok-routine-fixture.ts b/server/src/__tests__/helpers/grok-routine-fixture.ts new file mode 100644 index 00000000000..b955a5bbd92 --- /dev/null +++ b/server/src/__tests__/helpers/grok-routine-fixture.ts @@ -0,0 +1,33 @@ +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { vi } from "vitest"; +import { agents, authUsers, companies, companyMemberships, type createDb } from "@paperclipai/db"; +import { accessService } from "../../services/access.js"; +import { createPublicMcpOAuth } from "../../services/public-mcp/oauth.js"; +import { dotRunnerBroker } from "../../services/dot-runner-broker.js"; +import { grokRoutineService } from "../../services/grok-bot/routine.js"; + +export async function grokRoutineFixture(db: ReturnType) { + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Grok operator", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + const [company] = await db.insert(companies).values({ name: "Grok test", issuePrefix: "GK" + randomBytes(3).toString("hex") }).returning(); + await db.insert(companyMemberships).values({ companyId: company!.id, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" }); + await accessService(db).setPrincipalPermission(company!.id, "user", userId, "agents:create", true, userId); + const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Grok test", adapterType: "paperclip_runner", status: "active", adapterConfig: { provider: "grok_bot", allowUnmeteredProvider: true, lifecycleMode: "per_turn" } }).returning(); + const oauth = createPublicMcpOAuth(db, { origin: "https://paperclip.example", resource: "https://paperclip.example/mcp/runner" }); + const client = await oauth.register({ client_name: "Synthetic Grok client", redirect_uris: ["https://client.example/callback"] }, randomUUID()); + const verifier = randomBytes(32).toString("base64url"); + const input = { client_id: client.client_id, redirect_uri: "https://client.example/callback", response_type: "code", resource: oauth.config.resource, + scope: "paperclip:agent offline_access", code_challenge: createHash("sha256").update(verifier).digest("base64url"), code_challenge_method: "S256" }; + const requestId = (await oauth.authorize(input)).split("/").at(-1)!; + const pairing = await dotRunnerBroker(db).createPairing({ companyId: company!.id, agentId: agent!.id, operatorId: userId, provider: "grok_bot" }); + const consent = await oauth.consentDotPairing(requestId, pairing.pairingCode); + const tokens = await oauth.token({ grant_type: "authorization_code", client_id: client.client_id, redirect_uri: input.redirect_uri, + resource: oauth.config.resource, code: new URL(consent.redirectUrl).searchParams.get("code"), code_verifier: verifier }); + const principal = await oauth.authenticate(tokens.access_token); + const delivery = vi.fn().mockResolvedValue({ delivery: "accepted", httpStatus: 200 }); + const service = grokRoutineService(db, delivery); + const config = { companyId: company!.id, agentId: agent!.id, bindingId: pairing.bindingId, operatorId: userId, + expectedRevision: 0, botName: "Synthetic Bot", webhookUrl: "https://bot.example.test/routines/synthetic", webhookKey: "synthetic-sender-key" }; + await service.configure(config); + return { company: company!, agent: agent!, principal, pairing, service, delivery, config, oauth, tokens }; +} diff --git a/server/src/app.ts b/server/src/app.ts index d7d0e4ee9ca..f5f083ebf38 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1094,7 +1094,6 @@ export async function createApp( api.use(publicMcpManagementRoutes(publicMcpOAuth, dotOAuth)); api.use(dotRunnerRoutes(db, publicMcpOAuth.config.origin + "/mcp/runner")); api.use(grokBotRoutes(db, publicMcpOAuth.config.origin + "/mcp/runner")); - grokDispatcher.start(); } api.use(museBoardRoutes(db, opts.authPublicBaseUrl)); @@ -1285,6 +1284,7 @@ export async function createApp( agentLifecycle.setWake(agentLifecycleDelivery.wake); publicMcpEvents?.start(deliveryWork); dotMcpEvents?.start(deliveryWork); + if (publicMcpOAuth) grokDispatcher.start(deliveryWork); const stopMuseMaintenance = startMuseMaintenance(db, deliveryWork); const resourceLifecycleDelivery = registerPluginLifecycleDelivery(deliveryWork, db, workerManager, lifecycle); await resourceLifecycleDelivery.ready; @@ -1461,7 +1461,7 @@ export async function createApp( scheduler.stop(); await agentLifecycle.stop(); await publicMcpEvents?.stop(); - await dotMcpEvents?.stop(); grokDispatcher.stop(); + await dotMcpEvents?.stop(); jobCoordinator.stop(); resourceLifecycleDelivery.dispose(); await deliveryWork.stop(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 899ca920654..59389821c89 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -50,6 +50,7 @@ export const DELIVERY_QUEUES = { museMaintenance: "muse-maintenance", mcpEvents: "mcp-events", dotEvents: "dot-events", + grokDispatch: "grok-dispatch", skillSourceSync: "skill-source-sync", fastResponse: "fast-responses", feedback: "feedback-exports", diff --git a/server/src/services/grok-bot/dispatcher.ts b/server/src/services/grok-bot/dispatcher.ts index 864e3707114..b27d77d7a38 100644 --- a/server/src/services/grok-bot/dispatcher.ts +++ b/server/src/services/grok-bot/dispatcher.ts @@ -1,40 +1,44 @@ -import { and, asc, eq } from "drizzle-orm"; +import { and, asc, eq, sql } from "drizzle-orm"; import { grokBotDispatches, grokBotRoutines, type Db } from "@paperclipai/db"; import { grokRoutineService } from "./routine.js"; import type { deliverGrokRoutineWake } from "./webhook.js"; import { externalBotEnabled } from "../external-bot-authority.js"; +import { registerDeadlineWork } from "../deadline-work.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; -/** Drain only unsent durable intents. Sending/accepted/unknown work never retries. */ export function createGrokRoutineDispatcher(db: Db, isBackgroundWorkEnabled: () => boolean = () => true, deliver?: typeof deliverGrokRoutineWake) { - let timer: ReturnType | undefined; - let running = false; - let stopped = false; - async function tick() { - if (running || stopped || !isBackgroundWorkEnabled() || !await externalBotEnabled(db, "grok_bot")) return; - running = true; - try { + let running: Promise | null = null; + let enabled = false; + const prepared = () => db.select({ dispatch: grokBotDispatches, agentId: grokBotRoutines.agentId }).from(grokBotDispatches) + .innerJoin(grokBotRoutines, and(eq(grokBotRoutines.id, grokBotDispatches.routineId), eq(grokBotRoutines.companyId, grokBotDispatches.companyId))) + .where(sql`${grokBotDispatches.status} = 'prepared'`) + .orderBy(asc(grokBotDispatches.createdAt), asc(grokBotDispatches.id)); + function tick(signal?: AbortSignal): Promise { + if (running) return running; + running = (async () => { + if (!isBackgroundWorkEnabled() || signal?.aborted) return; + enabled = await externalBotEnabled(db, "grok_bot"); + if (!enabled) return; const service = grokRoutineService(db, deliver); - // Bot-initiated callers poll their durable request while admission is - // pending. Delay does not prove that the originating conversation ended. - // Starting a webhook here would create a competing Grok interaction. - // An accepted session already has a routine handling it. Quiet MCP - // traffic does not prove that routine ended. Keep feedback and receipts - // durable for the Bot to read instead of starting overlapping work. - const rows = await db.select({ dispatch: grokBotDispatches, agentId: grokBotRoutines.agentId }).from(grokBotDispatches) - .innerJoin(grokBotRoutines, and(eq(grokBotRoutines.id, grokBotDispatches.routineId), eq(grokBotRoutines.companyId, grokBotDispatches.companyId))) - .where(eq(grokBotDispatches.status, "prepared")).orderBy(asc(grokBotDispatches.createdAt)).limit(10); - for (const { dispatch, agentId } of rows) { - if (stopped || !isBackgroundWorkEnabled()) break; + for (const { dispatch, agentId } of await prepared().limit(10)) { + if (signal?.aborted || !isBackgroundWorkEnabled()) break; try { await service.send(dispatch.companyId, agentId, dispatch.id); } catch { - // No HTTP effect started while status is still prepared. Rejected - // authority is not a reason to keep attempting a stale wake. await db.update(grokBotDispatches).set({ status: "cancelled", updatedAt: new Date() }) .where(and(eq(grokBotDispatches.id, dispatch.id), eq(grokBotDispatches.status, "prepared"))); } } - } finally { running = false; } + })().finally(() => { running = null; }); + return running; } - return { tick, start() { stopped = false; timer ??= setInterval(() => { void tick().catch(() => {}); }, 1000); timer.unref(); }, - stop() { stopped = true; if (timer) clearInterval(timer); timer = undefined; } }; + return { + tick, + start: (coordinator: ReturnType) => registerDeadlineWork(coordinator, DELIVERY_QUEUES.grokDispatch, { + run: tick, + next: async () => enabled && (await prepared().limit(1)).length > 0 ? Date.now() : null, + canRun: isBackgroundWorkEnabled, + retryMs: 1000, + }), + }; } diff --git a/server/src/services/grok-bot/routine.ts b/server/src/services/grok-bot/routine.ts index 6e6acb5ac62..144936ae652 100644 --- a/server/src/services/grok-bot/routine.ts +++ b/server/src/services/grok-bot/routine.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto"; import { and, asc, desc, eq, inArray, isNull, isNotNull, lte, or, sql } from "drizzle-orm"; import { z } from "zod"; @@ -155,6 +156,7 @@ export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutin await tx.update(routines).set({ verifiedAt: null, verifiedGrantId: null, updatedAt: new Date() }).where(eq(routines.id, routine.id)); await tx.update(bindings).set({ status: "connected", readyAt: null, updatedAt: new Date() }).where(eq(bindings.id, binding.id)); } + await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); const [row] = await tx.insert(dispatches).values({ id: correlationId, companyId: input.companyId, routineId: routine.id, bindingId: binding.id, bindingGeneration: binding.generation, grantId: grant.id, routineRevision: routine.revision, sourceEventId: z.string().min(1).max(240).parse(input.sourceEventId), kind: input.kind, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 1d2395ffebe..d13920e8b9f 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -612,6 +612,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService const [updated] = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); return tx .update(instanceSettings) .set({