From 2adcde162e33fe5898f654f6955e7e7e016bbfa6 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:07:53 -0500 Subject: [PATCH] fix(server): wake Grok dispatch from committed reservations (#17490) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages agent work and should become idle when no work is owed. > - Grok routine dispatch already persists each outbound intent before delivery. > - Its dispatcher starts a one-second timer whenever public MCP is configured. > - That timer queries settings even when Grok is disabled, and scans dispatches when enabled but idle. > - The reservation transaction and feature-setting transaction already define when work can become available. > - This PR connects those writes to the shared coordinator and retains the Grok service's claim and delivery policy. > - The worker becomes quiet without adding a queue table or replaying uncertain sends. ## Linked Issues or Issue Description Stacked on #17486. This PR owns only Grok routine dispatch. It does not include decision retention or other remaining maintenance owners. Related open Grok PR #17278 concerns Runner credential-lease fixtures, not this dispatcher. The duplicate search found no overlapping dispatcher migration. **What existing behavior does this improve?** Grok routine dispatch recovery in the server. **Current behavior** With public MCP configured, the dispatcher reads instance settings every second, including when Grok is disabled. With Grok enabled, it scans for prepared dispatches every second. The baseline regression observed three queries in 3.2 seconds with Grok disabled. **Proposed behavior** Reconcile once at startup. Wake from committed dispatch reservations and experimental-setting changes. Schedule another pass only while prepared dispatches remain or reconciliation failed. Disabled and settled workers perform no recurring SQL or provider calls. **Reason and benefit** Remove unnecessary tenant database activity while preserving dispatch recovery, current authorization, provider outcome semantics, and shutdown ownership. This follows the existing sleep-aware maintenance direction; ROADMAP review found no competing dispatcher redesign. **Breaking changes** No public API or provider payload change. The internal dispatcher start method receives the shared coordinator; that coordinator now owns shutdown. Sending, accepted, unknown, and terminal dispatches remain non-replayable. This does not extend Cloud wake coverage. ## What Changed - Register the Grok worker with the existing coordinator; remove its independent timer. - Signal reservation and experimental-setting writes before commit. Nested writes follow their outer transaction. - Preserve ten-row batches, durable send claims, current authority checks, and existing cancellation/uncertainty policy. - Join admitted work during shutdown and leave later rows for restart. - Add a concurrent, replay-safe partial index for prepared dispatch lookup. Keep only the five newest generated snapshots. - Reuse the existing real OAuth/agent/routine fixture for lifecycle tests and add a validated external PostgreSQL test option. - Document producers, recovery, quiet conditions, and remaining host-sleep limitations. ## Verification Test revision: `6d81746fe260ee879442e7e0c879cfe88c250c8e`, followed by docs-only head `679f523d78`, which received both 5/5 reviews and passing CI. This candidate normally integrates landed target `8dfed893d8`. Its entire runtime and acceptance-test sources match that reviewed/tested tree; only four independently merged Runner E2E accounting files and the ledger differ, preserved exactly. Migration 0391 and the parent journal prefix are unchanged. Local checks were not repeated on unchanged runtime; fresh final-head CI and reviews are required. - **273 cases pass** in nine complete suites: real-PostgreSQL dispatch lifecycle, Grok qualification, server startup, scoped deadlines and all five instance-settings suites. Server typecheck and production build pass with the updated Runner vendor. - The baseline disabled-feature regression failed with three SQL reads. Existing lifecycle cases cover first reservation, restart, non-replayable history, re-enable, outer commit/rollback, drain, failure/retry, concurrent enqueue, paging, foreground send race, expiry, settings commit/rollback, and shutdown. - Generated migration `0391_third_roulette.sql` adds only the concurrent/idempotent prepared-dispatch index. All 389 parent journal entries are unchanged; five generated snapshots remain (0387–0391). Migration safety/numbering, schema drift and actual fresh apply/replay on two isolated PostgreSQL databases pass. - The earlier 259-case selection and 40,000-row generic-plan probe remain historical evidence: both unchanged queries used the partial index, six shared buffers including the routine join, 0.032ms / 0.025ms local execution. No new deployed suspension or cost claim is made. - Earlier browser CI failed during agent-chat fixture restoration (lock timeout/HTTP 500), followed by a test seeing the experiment left enabled. Those failures remain recorded in the owning audit; they were not dispatcher assertions. Later pre-reconciliation `a74305e2b4` checks were terminal and green. Fresh CI and both 5/5 bot reviews are required for this head. Commands: server Vitest for `grok-dispatch-work`, `grok-bot-routine`, `server-startup-feedback-export`, `scoped-deadline-work` and the five `instance-settings-*` suites; server typecheck/build; DB migration safety/numbering, snapshot drift and fresh migration apply/replay. ## Risks - Signals are local to the shared database owner and its child pools. Independent SQL writers and other replicas need explicit wake integration. Startup recovers committed prepared rows. - Disabled Grok retains prepared rows and resumes after a committed settings change. This does not relax Cloud's binding/assignment/hold guards. - Existing send-error cancellation policy is preserved. Sending or unknown outcomes are not reset or automatically replayed; this is not an exactly-once delivery change. - Shutdown awaits the current send within existing transport timeouts and stops before the next row. - The index adds write/storage cost but bounds queries independently of completed history. Existing migration execution runs concurrent index creation outside a transaction. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-grok-dispatch.md | 121 ++++++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 40 +++- packages/db/src/client.ts | 1 + .../db/src/migrations/0391_third_roulette.sql | 1 + ...{0386_snapshot.json => 0391_snapshot.json} | 114 +++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/grok_bot.ts | 2 + server/src/__tests__/grok-bot-routine.test.ts | 41 +---- .../src/__tests__/grok-dispatch-work.test.ts | 174 ++++++++++++++++++ .../__tests__/helpers/grok-routine-fixture.ts | 33 ++++ server/src/app.ts | 4 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/grok-bot/dispatcher.ts | 54 +++--- server/src/services/grok-bot/routine.ts | 2 + server/src/services/instance-settings.ts | 1 + 15 files changed, 535 insertions(+), 61 deletions(-) create mode 100644 doc/idle-grok-dispatch.md create mode 100644 packages/db/src/migrations/0391_third_roulette.sql rename packages/db/src/migrations/meta/{0386_snapshot.json => 0391_snapshot.json} (99%) create mode 100644 server/src/__tests__/grok-dispatch-work.test.ts create mode 100644 server/src/__tests__/helpers/grok-routine-fixture.ts 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({