mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 04:25:52 +01:00
fix(server): wake Grok dispatch from committed reservations (#17490)
## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
8dfed893d8
commit
2adcde162e
15 files changed
+535
-61
No files matched your search
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
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<ReturnType<typeof fixture>>, sourceEventId = randomUUID()) {
|
||||
return f.service.reserve({ companyId: f.company.id, agentId: f.agent.id, bindingId: f.pairing.bindingId, sourceEventId, kind: "verify" });
|
||||
}
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let worker: ReturnType<typeof createGrokRoutineDispatcher>;
|
||||
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<ReturnType<typeof grokRoutineFixture>>, 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<void>(resolve => { release = resolve; });
|
||||
let created!: Awaited<ReturnType<typeof reserve>>;
|
||||
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<ReturnType<typeof reserve>>;
|
||||
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<void>(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);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof createDb>) {
|
||||
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 };
|
||||
}
|
||||
+2
-2
@@ -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();
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<typeof setInterval> | 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<void> | 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<void> {
|
||||
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<typeof createDeliveryWorkCoordinator>) => registerDeadlineWork(coordinator, DELIVERY_QUEUES.grokDispatch, {
|
||||
run: tick,
|
||||
next: async () => enabled && (await prepared().limit(1)).length > 0 ? Date.now() : null,
|
||||
canRun: isBackgroundWorkEnabled,
|
||||
retryMs: 1000,
|
||||
}),
|
||||
};
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in new issue
Block a user