diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 3a05b5d4dbf..16b46f9cb4b 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -400,7 +400,7 @@ keep the application fully active. A failed database check fails startup. An empty unclaimed application keeps its HTTP server and sandbox plugins ready, but skips recurring database work: chat/email delivery, plugin jobs, browser -cleanup, feedback export, import cleanup, execution reconciliation, heartbeat +cleanup, feedback export, import cleanup, execution reconciliation, Muse maintenance, heartbeat schedules, and automatic backups. Startup migrations, plugin installation, and other one-time preparation still run. Normal anonymous health probes return `warmStandby: true` without SQL or session lookups. This is application liveness, @@ -408,6 +408,11 @@ not a current database connectivity check. Other API requests and WebSocket upgr the signed claim succeeds. Page and asset requests serve only the static UI router, bypassing session, bearer-key, tenant, and other dynamic handlers. +Muse defers its startup qualification-evidence update and recurring database +sweeps until standby ends. The first enabled sweep marks interrupted evidence +before it reconciles qualification deadlines or terminal assignments. Claimed +workspaces retain ordinary Muse maintenance even when idle task drain is active. + The existing signed claim on `GET /api/health` writes the identity durably before normal requests and polling resume. No polling discovers claims and no process restart is required. Timers resume on their next normal tick; request-driven diff --git a/server/src/app.ts b/server/src/app.ts index 71ea02af042..208634d147c 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -624,7 +624,9 @@ export async function createApp( app.use(cloudWarmStandbyMiddleware(isWarmStandby, health, staticUi)); app.use("/api/customer-success/v1", customerSuccessRoutes(db, opts.storageService)); app.use(museTransportRoutes(db, opts.authPublicBaseUrl)); - const stopMuseMaintenance = startMuseMaintenance(db); + const stopMuseMaintenance = startMuseMaintenance(db, { + isBackgroundWorkEnabled: () => !isWarmStandby(), + }); app.use(publicMcpIngress); // Connection-intent tools carry their own short-lived, run-bound bearer and // must be reachable by remote adapters that intentionally do not receive an diff --git a/server/src/services/muse-maintenance.test.ts b/server/src/services/muse-maintenance.test.ts new file mode 100644 index 00000000000..b0ae3cb6b01 --- /dev/null +++ b/server/src/services/muse-maintenance.test.ts @@ -0,0 +1,133 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { agents, companies, createDb, museAgentBindings } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { startMuseMaintenance } from "./muse-maintenance.js"; + +const effects = vi.hoisted(() => ({ + endQualification: vi.fn().mockResolvedValue(undefined), + revokeOAuthGrant: vi.fn().mockResolvedValue(undefined), + reconcileTerminalAssignments: vi.fn().mockResolvedValue(undefined), + start: vi.fn(), + stop: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("./muse-runner-broker.js", () => ({ museRunnerBroker: () => effects })); +vi.mock("./muse-receiver.js", () => ({ museReceiver: () => effects })); + +const support = await getEmbeddedPostgresTestSupport(); +const describeDatabase = support.supported ? describe : describe.skip; + +describeDatabase("Muse maintenance background admission", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("muse-maintenance-"); + db = createDb(database.connectionString, { maxConnections: 1 }); + }); + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + }); + afterEach(async () => { + vi.useRealTimers(); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + }); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await database?.cleanup(); + }); + + async function fixture(expired = false) { + const [company] = await db.insert(companies).values({ name: "Maintenance", issuePrefix: `MT${randomUUID().slice(0, 6)}` }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Muse agent" }).returning(); + const [binding] = await db.insert(museAgentBindings).values({ + companyId: company.id, + agentId: agent.id, + operatorId: "synthetic-operator", + qualificationId: randomUUID(), + qualificationExpiresAt: new Date(Date.now() + (expired ? -1 : 1) * 3_600_000), + }).returning(); + return binding; + } + + function observeDatabase(failFirstUpdate = false) { + const calls: string[] = []; + const observed = new Proxy(db, { get(target, key) { + if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) { + return Reflect.get(target, key); + } + return (...args: unknown[]) => { + calls.push(String(key)); + if (key === "update" && failFirstUpdate) { + failFirstUpdate = false; + throw new Error("temporary database failure"); + } + return Reflect.apply(Reflect.get(target, key), target, args); + }; + } }); + return { observed, calls }; + } + + it("leaves PostgreSQL untouched in standby and resumes maintenance after claim", async () => { + const binding = await fixture(); + const { observed, calls } = observeDatabase(); + let enabled = false; + const stop = startMuseMaintenance(observed, { isBackgroundWorkEnabled: () => enabled }); + try { + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(calls).toEqual([]); + expect(effects.reconcileTerminalAssignments).not.toHaveBeenCalled(); + enabled = true; + await vi.advanceTimersByTimeAsync(5000); + await vi.waitFor(() => expect(effects.reconcileTerminalAssignments).toHaveBeenCalledWith(binding.companyId, binding.agentId, binding.id)); + const [saved] = await db.select().from(museAgentBindings).where(eq(museAgentBindings.id, binding.id)); + expect(saved.cadenceEvidenceIncompleteAt).toBeInstanceOf(Date); + expect(calls.filter(call => call === "update")).toHaveLength(1); + enabled = false; + const settledCalls = [...calls]; + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(calls).toEqual(settledCalls); + enabled = true; + await vi.advanceTimersByTimeAsync(5000); + await vi.waitFor(() => expect(effects.reconcileTerminalAssignments).toHaveBeenCalledTimes(2)); + expect(calls.filter(call => call === "update")).toHaveLength(1); + } finally { + await stop(); + } + }); + + it("keeps ordinary startup and qualification deadline enforcement enabled by default", async () => { + const binding = await fixture(true); + const stop = startMuseMaintenance(db); + try { + await vi.waitFor(() => expect(effects.endQualification).toHaveBeenCalledWith({ + companyId: binding.companyId, + agentId: binding.agentId, + bindingId: binding.id, + qualificationId: binding.qualificationId, + operatorId: binding.operatorId, + })); + await vi.waitFor(() => expect(effects.reconcileTerminalAssignments).toHaveBeenCalled()); + } finally { + await stop(); + } + }); + + it("retries startup evidence marking before reconciliation after a database failure", async () => { + const binding = await fixture(); + const { observed, calls } = observeDatabase(true); + const stop = startMuseMaintenance(observed); + try { + await Promise.resolve(); + expect(effects.reconcileTerminalAssignments).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(5000); + await vi.waitFor(() => expect(effects.reconcileTerminalAssignments).toHaveBeenCalled()); + const [saved] = await db.select().from(museAgentBindings).where(eq(museAgentBindings.id, binding.id)); + expect(saved.cadenceEvidenceIncompleteAt).toBeInstanceOf(Date); + expect(calls.filter(call => call === "update")).toHaveLength(2); + } finally { + await stop(); + } + }); +}); diff --git a/server/src/services/muse-maintenance.ts b/server/src/services/muse-maintenance.ts index e6a977d52b1..be6f5b04c79 100644 --- a/server/src/services/muse-maintenance.ts +++ b/server/src/services/muse-maintenance.ts @@ -4,12 +4,17 @@ import { museAgentBindings as bindings, mcpOauthGrants, type Db } from "@papercl import { museRunnerBroker } from "./muse-runner-broker.js"; import { museReceiver } from "./muse-receiver.js"; /** Deadline authority lives in the database; every fresh process recovers it. */ -export function startMuseMaintenance(db:Db) { +export function startMuseMaintenance(db:Db, options: { isBackgroundWorkEnabled?: () => boolean } = {}) { const broker=museRunnerBroker(db),receiver=museReceiver(db); - let running=false; + const isBackgroundWorkEnabled=options.isBackgroundWorkEnabled??(()=>true); + let running=false, startupMarked=false; const sweep=async()=>{ - if(running)return;running=true; + if(running||!isBackgroundWorkEnabled())return;running=true; try{ + if(!startupMarked){ + await db.update(bindings).set({cadenceEvidenceIncompleteAt:new Date()}).where(and(isNull(bindings.revokedAt),gt(bindings.qualificationExpiresAt,new Date()))); + startupMarked=true; + } let after:string|undefined; while(true) { const expired=await db.select().from(bindings).where(and(isNull(bindings.revokedAt),lte(bindings.qualificationExpiresAt,new Date()),after?gt(bindings.id,after):undefined)).orderBy(asc(bindings.id)).limit(100); @@ -30,9 +35,7 @@ export function startMuseMaintenance(db:Db) { } }finally{running=false;} }; - // A process restart loses any not-yet-persisted ingress tail. It is explicit - // incomplete qualification evidence even when subsequent traffic is healthy. - void db.update(bindings).set({cadenceEvidenceIncompleteAt:new Date()}).where(and(isNull(bindings.revokedAt),gt(bindings.qualificationExpiresAt,new Date()))).then(()=>sweep()).catch(()=>{logger.warn("Muse startup qualification evidence marking failed.");}); + void sweep().catch(()=>{logger.warn("Muse startup qualification evidence marking failed.");}); const timer=setInterval(()=>void sweep().catch(()=>{logger.warn("Muse maintenance sweep failed; persisted authority checks remain active.");}),5000);timer.unref();receiver.start(); return async()=>{clearInterval(timer);await receiver.stop();}; }