diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index 74629708703..96f0d2cabe1 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -382,3 +382,31 @@ does not make existing in-memory-only debt survive an unplanned process crash. Related public PR #13413 adds runtime-service controller requirements and uses the same `purpose: "idle"` / `ownerId` vocabulary. This report conservatively blocks all retained runtime-service state instead of approving such services. + +## Muse maintenance + +Muse maintenance uses the shared delivery coordinator. Startup reconstructs +existing work and marks incomplete receiver evidence once. Pairing, OAuth grant +attachment/revocation/replay, qualification changes and new assignments notify +the coordinator before their transaction writes. Nested writes wake it only +after the outer transaction settles; uncertain outcomes retain the normal hold. + +Unrevoked qualifications schedule their saved expiration time. An instance with +no Muse bindings, or with idle bindings and no qualification deadline, performs +no recurring maintenance SQL after startup. Failed reconciliation retries in +five seconds. Offered, claimed, accepted and not-yet-finalized settled assignments +retain that five-second recovery cadence until settlement; this covers terminal +run writers even when their direct reconciliation path fails. Retained finalized +assignment history does not keep that cadence running. Direct SQL or independent +process writers require explicit integration; startup still reconstructs state. + +Warm standby prevents maintenance SQL. An idle drain pauses ordinary dispatch +and passive deadlines through the shared coordinator. Startup and notifications +received during drain retain a memory-only wake until a sweep can run. The wait +does not query the database. Producer uncertainty and +outstanding assignment recovery retain its existing admission rules. Shutdown +awaits the worker before flushing the receiver. The receiver's separate batching +timer remains; an empty buffer performs no database work. Existing Muse durable +sleep blockers, external worker stop evidence, and ingress requirements remain +unchanged. This change establishes database quiet for inactive Muse, not host +sleep eligibility for a configured Muse connection. diff --git a/server/src/__tests__/muse-control.test.ts b/server/src/__tests__/muse-control.test.ts index 5957ade1d4b..9e9b3f3a33f 100644 --- a/server/src/__tests__/muse-control.test.ts +++ b/server/src/__tests__/muse-control.test.ts @@ -1,3 +1,5 @@ +import { subscribeDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import express from "express"; import { execFileSync } from "node:child_process"; import { fileURLToPath } from "node:url"; @@ -78,6 +80,32 @@ describe("personal Muse control plane",()=>{ const [binding]=await db.select().from(bindings).where(eq(bindings.id,f.binding.id)); return {...f,binding,ticket,oauth,client,verifier,input,authorizationUrl,tokenInput,tokens,subject}; } + it("signals maintenance from real qualification, assignment, and OAuth revocation producers", async () => { + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.museMaintenance, event => events.push(event)); + let detach: (() => Promise) | undefined; + try { + const f = await oauthFixture(true); + expect(events).toContain("intent"); + expect(events.at(-1)).toBe("settled"); + events.length = 0; + const [binding] = await db.select().from(bindings).where(eq(bindings.id, f.binding.id)); + await f.broker.beginQualification({ companyId: f.company.id, agentId: f.agent.id, + bindingId: binding.id, generation: binding.generation, expectedRevision: binding.revision, + qualificationId: randomUUID(), expiresAt: new Date(Date.now() + 60_000), operatorId: f.operatorId }); + expect(events).toEqual(["intent", "settled"]); + events.length = 0; + const running = await active(f); + detach = running.detach; + expect(events).toContain("intent"); + expect(events.at(-1)).toBe("settled"); + events.length = 0; + await f.oauth.revokeConnection(binding.grantId!, f.operatorId); + expect(events).toContain("intent"); + expect(events.at(-1)).toBe("settled"); + expect((await db.select().from(bindings).where(eq(bindings.id, binding.id)))[0].revokedAt).not.toBeNull(); + } finally { unsubscribe(); await detach?.(); } + }); it("permits a paused handoff recipient authority read while retaining connection revocation checks", async () => { const f = await oauthFixture(true); const ref = { companyId: f.company.id, agentId: f.agent.id, bindingId: f.binding.id, bindingGeneration: f.binding.generation }; diff --git a/server/src/app.ts b/server/src/app.ts index 77e5c4543a1..8c74cbd548c 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -626,9 +626,6 @@ 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, { - 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 @@ -1279,6 +1276,7 @@ export async function createApp( onError: (err, queue) => logger.error({ err, queue }, "Delivery reconciliation failed"), }); app.locals.deliveryWork = deliveryWork; + const stopMuseMaintenance = startMuseMaintenance(db, deliveryWork); const resourceLifecycleDelivery = registerPluginLifecycleDelivery(deliveryWork, db, workerManager, lifecycle); await resourceLifecycleDelivery.ready; if (opts.feedbackExportService) { diff --git a/server/src/modules/external-agents/muse.ts b/server/src/modules/external-agents/muse.ts index da1607f5b8d..ad0ece3f0a5 100644 --- a/server/src/modules/external-agents/muse.ts +++ b/server/src/modules/external-agents/muse.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; import { createHash, randomBytes, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm"; import { z } from "zod"; @@ -120,6 +121,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { await tx.update(mcpOauthTokens).set({usedAt:new Date(),expiresAt:new Date()}).where(eq(mcpOauthTokens.grantId,b.grantId)); } async function revokeBinding(tx:ExternalAdmissionTransaction,b:typeof bindings.$inferSelect,operatorId:string,reason:string,actorType:"user"|"system"="user") { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); const runs=await fence(tx,b,reason); await tx.update(bindings).set({status:"revoked",revokedAt:new Date(),ticketHash:null,challengeHash:null,cleanupExpiresAt:new Date(Date.now()+MUSE_CLEANUP_TTL_MS),deadlineEnforcedAt:b.deadlineEnforcedAt??(b.qualificationExpiresAt&&b.qualificationExpiresAt<=new Date()?new Date():null),revision:b.revision+1,updatedAt:new Date()}).where(eq(bindings.id,b.id)); await tx.update(credentials).set({revokedAt:new Date()}).where(and(eq(credentials.bindingId,b.id),inArray(credentials.kind,["access","refresh","signal"]))); @@ -177,6 +179,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { await tx.update(credentials).set({expiresAt:new Date(Date.now()+MUSE_CLEANUP_TTL_MS)}).where(and(eq(credentials.bindingId,old.id),inArray(credentials.kind,["cleanup","detector_cleanup"]),isNull(credentials.revokedAt))); } else if(input.replaceBindingId)throw conflict("Invitation changed. Refresh before replacing its ticket."); const [latest]=await tx.select({generation:bindings.generation}).from(bindings).where(and(eq(bindings.companyId,input.companyId),eq(bindings.agentId,input.agentId))).orderBy(desc(bindings.generation)).limit(1); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); const [created]=await tx.insert(bindings).values({id:bindingId,companyId:input.companyId,agentId:input.agentId,operatorId:input.operatorId,generation:(latest?.generation??0)+1,ticketHash:museCredentialHash(ticket),ticketExpiresAt:new Date(Date.now()+MUSE_TICKET_TTL_MS)}).returning(); await logActivity(tx as unknown as Db,{companyId:input.companyId,actorType:"user",actorId:input.operatorId,action:"muse.pairing_created",entityType:"agent",entityId:input.agentId,details:{bindingId:created!.id,generation:created!.generation}}); await updateAgentConfigurationInTransaction(tx as unknown as Db,agent.id,{adapterConfig:{...agent.adapterConfig,museBindingId:created!.id}},{recordRevision:{createdByUserId:input.operatorId,source:"muse-pairing"}}); @@ -312,6 +315,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { if(b.qualificationId)throw conflict("Another qualification already exists."); await assertMuseCurrentAuthority(tx,b); const startedAt=new Date(input.expiresAt.getTime()-24*60*60_000); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); await tx.update(bindings).set({qualificationId:input.qualificationId,qualificationStartedAt:startedAt,qualificationExpiresAt:input.expiresAt,revision:b.revision+1,updatedAt:new Date()}).where(eq(bindings.id,b.id)); await logActivity(tx as unknown as Db,{companyId:input.companyId,actorType:"user",actorId:input.operatorId,agentId:input.agentId,action:"muse.qualification_started",entityType:"agent",entityId:input.agentId,details:{bindingId:b.id,generation:b.generation,revision:b.revision+1,authorizingUserId:b.operatorId,qualificationId:input.qualificationId,startedAt:startedAt.toISOString(),expiresAt:input.expiresAt.toISOString()}},publications); return {qualificationId:input.qualificationId,startedAt:startedAt.toISOString(),expiresAt:input.expiresAt.toISOString(),revision:b.revision+1}; diff --git a/server/src/modules/external-agents/native-broker.ts b/server/src/modules/external-agents/native-broker.ts index 38d579696b8..fe89c303672 100644 --- a/server/src/modules/external-agents/native-broker.ts +++ b/server/src/modules/external-agents/native-broker.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { z } from "zod"; import { sql, and, eq, isNull, gt, inArray } from "drizzle-orm"; @@ -267,6 +268,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I const [old]=await tx.execute(sql`select ${assignmentColumns} from ${assignmentTable} where run_id=${runId}::uuid and turn_id=${b.turnId}`); const projection={...p,binding:b}; if(old&&externalOperationDigest("tool",old.projection)!==externalOperationDigest("tool",projection))throw conflict("Assignment event changed its immutable projection."); + if (!old && provider === "muse") await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); const [a]=old?[old]:await tx.execute(sql`insert into ${assignmentTable} (company_id,binding_id,binding_generation,run_id,agent_id,normalized_session_id,turn_id,revision,controller_generation,catalog_digest,status,projection,accept_by,expires_at) values (${b.companyId}::uuid,${b.bindingId}::uuid,${b.bindingGeneration},${runId}::uuid,${b.agentId}::uuid,${b.normalizedSessionId},${b.turnId},${b.assignmentRevision},${owner.controllerGeneration},${externalOperationDigest("tool",{tools:p.tools})},'offered',${JSON.stringify(projection)}::jsonb,${new Date(Number(p.acceptByUnixMs)).toISOString()}::timestamptz,${new Date(Number(p.expiresAtUnixMs)).toISOString()}::timestamptz) returning ${assignmentColumns}`); if(!a)throw conflict("Assignment creation failed."); await tx.execute(sql`insert into ${mailboxTable} (company_id,binding_id,binding_generation,assignment_id,kind,source_event_id,"references") values (${a.companyId}::uuid,${a.bindingId}::uuid,${a.bindingGeneration},${a.id}::uuid,'assignment',${event.sourceEventId},${JSON.stringify({assignmentId:a.id,runId,revision:a.revision})}::jsonb) on conflict do nothing`); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index e3720b27937..4e4f9302876 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -19,6 +19,7 @@ export const DELIVERY_QUEUES = { chatReceipts: "chat-slack-receipts", browser: "browser-use", email: "email-channels", + museMaintenance: "muse-maintenance", fastResponse: "fast-responses", feedback: "feedback-exports", chatCompletion: "chat-completions", diff --git a/server/src/services/muse-maintenance.test.ts b/server/src/services/muse-maintenance.test.ts index b0ae3cb6b01..be6f2cfa519 100644 --- a/server/src/services/muse-maintenance.test.ts +++ b/server/src/services/muse-maintenance.test.ts @@ -1,133 +1,243 @@ 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 { agents, companies, createDb, applyPendingMigrations, heartbeatRuns, museRunnerAssignments, museAgentBindings } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { startMuseMaintenance } from "./muse-maintenance.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { idleWorkSnapshot } from "./task-admission.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), + endQualification: vi.fn(), revokeOAuthGrant: vi.fn(), reconcileTerminalAssignments: vi.fn(), + start: vi.fn(), stop: vi.fn(), })); vi.mock("./muse-runner-broker.js", () => ({ museRunnerBroker: () => effects })); vi.mock("./muse-receiver.js", () => ({ museReceiver: () => effects })); - -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = process.env.PAPERCLIP_MUSE_MAINTENANCE_TEST_DATABASE_URL; +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeDatabase = support.supported ? describe : describe.skip; -describeDatabase("Muse maintenance background admission", () => { +describeDatabase("Muse commit and deadline maintenance", () => { let database: Awaited>; let db: ReturnType; + let coordinator: ReturnType; + let stop: ReturnType; + let enabled: boolean; + let draining: boolean; + let calls: string[]; + let errors: unknown[]; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("muse-maintenance-"); - db = createDb(database.connectionString, { maxConnections: 1 }); + if (externalUrl) await applyPendingMigrations(externalUrl); + else database = await startEmbeddedPostgresTestDatabase("muse-maintenance-"); + db = createDb(externalUrl ?? database.connectionString, { maxConnections: 3 }); }); beforeEach(() => { - vi.clearAllMocks(); - vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + vi.resetAllMocks(); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"], shouldAdvanceTime: true }); + enabled = true; draining = false; calls = []; errors = []; + effects.stop.mockResolvedValue(undefined); + effects.reconcileTerminalAssignments.mockResolvedValue(undefined); + effects.revokeOAuthGrant.mockResolvedValue(undefined); + effects.endQualification.mockImplementation(async ({ bindingId }) => { + await db.update(museAgentBindings).set({ revokedAt: new Date() }).where(eq(museAgentBindings.id, bindingId)); + }); }); afterEach(async () => { + await coordinator?.stop(); + await stop?.(); vi.useRealTimers(); await db.execute(sql`TRUNCATE TABLE companies CASCADE`); }); - afterAll(async () => { - await db?.$client.end({ timeout: 1 }); - await database?.cleanup(); - }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); - async function fixture(expired = false) { + async function fixture(expiresAt: Date | null = null) { 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(); + const [binding] = await db.insert(museAgentBindings).values({ companyId: company.id, agentId: agent.id, + operatorId: "synthetic-operator", qualificationId: expiresAt ? randomUUID() : null, qualificationExpiresAt: expiresAt }).returning(); return binding; } - - function observeDatabase(failFirstUpdate = false) { - const calls: string[] = []; + async function start(failFirstUpdate = false) { const observed = new Proxy(db, { get(target, key) { - if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) { - return Reflect.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"); - } + if (key === "update" && failFirstUpdate) { failFirstUpdate = false; throw new Error("temporary DB failure"); } return Reflect.apply(Reflect.get(target, key), target, args); }; } }); - return { observed, calls }; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled && !draining, + canReconcile: () => enabled, onError: error => errors.push(error) }); + stop = startMuseMaintenance(observed, coordinator); + await stop.ready; + } + async function settle() { + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + } + async function signalWrite(bindingId: string, at: Date) { + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await tx.update(museAgentBindings).set({ qualificationId: randomUUID(), qualificationExpiresAt: at }) + .where(eq(museAgentBindings.id, bindingId)); + }); } - 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("stays SQL-quiet after startup with no bindings, including during drain", async () => { + await start(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(600_000); + draining = true; + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual(baseline); + expect(coordinator.nextWakeAt()).toBeNull(); }); - - 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("does not poll an idle configured binding", async () => { + await fixture(); await start(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual(baseline); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledTimes(1); }); - - 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(); - } + it("waits for outer commit, then enforces a newly saved deadline without polling", async () => { + const binding = await fixture(); await start(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let written!: () => void; + const ready = new Promise(resolve => { written = resolve; }); + const at = new Date(Date.now() + 60_000); + const transaction = db.transaction(async tx => { + await tx.transaction(async nested => { + await notifyDeliveryWork(nested, DELIVERY_QUEUES.museMaintenance); + await nested.update(museAgentBindings).set({ qualificationId: randomUUID(), qualificationExpiresAt: at }) + .where(eq(museAgentBindings.id, binding.id)); + }); + written(); await gate; + }); + await ready; + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledTimes(1); + release(); await transaction; await settle(); + expect(coordinator.nextWakeAt()).toBe(at.getTime()); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(59_000); + expect(calls).toEqual(baseline); + await vi.advanceTimersByTimeAsync(1000); await settle(); + expect(effects.endQualification).toHaveBeenCalledWith(expect.objectContaining({ bindingId: binding.id })); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("recovers a rolled-back intent without leaving an empty timer", async () => { + const binding = await fixture(); await start(); + await expect(db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await tx.update(museAgentBindings).set({ qualificationExpiresAt: new Date(), qualificationId: randomUUID() }).where(eq(museAgentBindings.id, binding.id)); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await settle(); + expect(effects.endQualification).not.toHaveBeenCalled(); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("retains recovery only until an outstanding assignment settles", async () => { + const b = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: b.companyId, agentId: b.agentId, + invocationSource: "automation", status: "running" }).returning(); + const [assignment] = await db.insert(museRunnerAssignments).values({ companyId: b.companyId, + agentId: b.agentId, bindingId: b.id, bindingGeneration: b.generation, runId: run.id, + normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, + catalogDigest: "test", status: "accepted", projection: {}, acceptBy: new Date(), expiresAt: new Date(Date.now() + 60_000) }).returning(); + await start(); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + effects.reconcileTerminalAssignments.mockImplementationOnce(async () => { + await db.update(museRunnerAssignments).set({ status: "settled", acceptedResultAt: new Date(), finalizedAt: new Date() }) + .where(eq(museRunnerAssignments.id, assignment.id)); + }); + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(coordinator.nextWakeAt()).toBeNull(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual(baseline); + }); + it("moves a scheduled deadline earlier on a committed change", async () => { + const binding = await fixture(new Date(Date.now() + 600_000)); await start(); + const at = new Date(Date.now() + 60_000); + await signalWrite(binding.id, at); await settle(); + expect(coordinator.nextWakeAt()).toBe(at.getTime()); + }); + it("recovers overdue qualifications at startup and retries failed reconciliation", async () => { + await fixture(new Date(Date.now() - 1000)); + effects.endQualification.mockRejectedValueOnce(new Error("transient")); + await start(); + expect(errors).toHaveLength(1); + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.endQualification).toHaveBeenCalledTimes(2); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("does no SQL in standby and recovers after claim", async () => { + const binding = await fixture(new Date(Date.now() + 3_600_000)); + enabled = false; await start(); + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual([]); + enabled = true; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledWith(binding.companyId, binding.agentId, binding.id); + expect(calls.filter(c => c === "update")).toHaveLength(1); + }); + it("pauses deadline maintenance during drain and catches up after release", async () => { + await fixture(new Date(Date.now() + 1000)); await start(); + const baseline = [...calls]; draining = true; + await vi.advanceTimersByTimeAsync(20_000); + expect(calls).toEqual(baseline); + draining = false; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.endQualification).toHaveBeenCalledTimes(1); + }); + it("retains a first qualification committed during drain until dispatch resumes", async () => { + const binding = await fixture(); await start(); + expect(coordinator.nextWakeAt()).toBeNull(); + draining = true; + await signalWrite(binding.id, new Date(Date.now() + 1000)); await settle(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(20_000); + expect(calls).toEqual(baseline); + expect(effects.endQualification).not.toHaveBeenCalled(); + draining = false; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.endQualification).toHaveBeenCalledTimes(1); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("retains startup recovery when the worker starts during drain", async () => { + await fixture(new Date(Date.now() - 1000)); + draining = true; await start(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(20_000); + expect(calls).toEqual(baseline); + draining = false; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.endQualification).toHaveBeenCalledTimes(1); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("preserves a failed sweep retry across drain", async () => { + await fixture(); await start(true); + expect(errors).toHaveLength(1); + draining = true; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(20_000); + expect(calls).toEqual(baseline); + draining = false; + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledTimes(1); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + it("retries startup evidence marking before reconciliation", async () => { + await fixture(); await start(true); + expect(effects.reconcileTerminalAssignments).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledTimes(1); + expect(calls.filter(c => c === "update")).toHaveLength(2); }); }); diff --git a/server/src/services/muse-maintenance.ts b/server/src/services/muse-maintenance.ts index be6f5b04c79..08180b8ec54 100644 --- a/server/src/services/muse-maintenance.ts +++ b/server/src/services/muse-maintenance.ts @@ -1,41 +1,99 @@ import { logger } from "../middleware/logger.js"; -import { and, asc, eq, gt, isNull, lte } from "drizzle-orm"; -import { museAgentBindings as bindings, mcpOauthGrants, type Db } from "@paperclipai/db"; +import { and, asc, eq, gt, inArray, isNull, lte, or } from "drizzle-orm"; +import { museAgentBindings as bindings, museRunnerAssignments as assignments, mcpOauthGrants, type Db } from "@paperclipai/db"; 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, options: { isBackgroundWorkEnabled?: () => boolean } = {}) { - const broker=museRunnerBroker(db),receiver=museReceiver(db); - const isBackgroundWorkEnabled=options.isBackgroundWorkEnabled??(()=>true); - let running=false, startupMarked=false; - const sweep=async()=>{ - 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; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; + +export function startMuseMaintenance( + db: Db, + coordinator: Pick, "register">, +) { + const broker = museRunnerBroker(db); + const receiver = museReceiver(db); + let startupMarked = false; + let requestedSweep = 0; + let completedSweep = -1; + let nextDeadline: number | null = null; + const worker = coordinator.register(DELIVERY_QUEUES.museMaintenance, { + retryMs: 5_000, + pausePassiveMaintenance: true, + onSettled: () => { requestedSweep++; }, + async run(signal) { + const sweep = requestedSweep; + 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); - for(const b of expired)if(b.qualificationId)try{await broker.endQualification({companyId:b.companyId,agentId:b.agentId,bindingId:b.id,qualificationId:b.qualificationId,operatorId:b.operatorId});}catch{logger.warn({bindingId:b.id},"Muse qualification deadline maintenance failed; authority remains denied and will retry.");} - if(expired.length<100)break;after=expired.at(-1)!.id; - } - after=undefined; - while(true) { - const active=await db.select().from(bindings).where(and(isNull(bindings.revokedAt),after?gt(bindings.id,after):undefined)).orderBy(asc(bindings.id)).limit(100); - for(const b of active)try{ - if(b.grantId){ - const [grant]=await db.select({revokedAt:mcpOauthGrants.revokedAt}).from(mcpOauthGrants).where(and(eq(mcpOauthGrants.id,b.grantId),eq(mcpOauthGrants.companyId,b.companyId),eq(mcpOauthGrants.agentId,b.agentId),eq(mcpOauthGrants.userId,b.operatorId))); - if(!grant||grant.revokedAt){await broker.revokeOAuthGrant({grantId:b.grantId,companyId:b.companyId,agentId:b.agentId,authorizingUserId:b.operatorId});continue;} + let failed = false; + let after: string | undefined; + while (!signal.aborted) { + 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); + for (const b of expired) { + if (signal.aborted) return; + if (!b.qualificationId) throw new Error("Muse qualification is missing its identity"); + try { + await broker.endQualification({ companyId: b.companyId, agentId: b.agentId, + bindingId: b.id, qualificationId: b.qualificationId, operatorId: b.operatorId }); + } catch { + failed = true; + logger.warn({ bindingId: b.id }, "Muse qualification deadline maintenance failed; will retry."); } - await broker.reconcileTerminalAssignments(b.companyId,b.agentId,b.id); - }catch{logger.warn({bindingId:b.id},"Muse authority or terminal reconciliation failed; unresolved holds remain and will retry.");} - if(active.length<100)break;after=active.at(-1)!.id; + } + if (expired.length < 100) break; + after = expired.at(-1)!.id; } - }finally{running=false;} - }; - 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();}; + after = undefined; + while (!signal.aborted) { + const active = await db.select().from(bindings).where(and( + isNull(bindings.revokedAt), after ? gt(bindings.id, after) : undefined, + )).orderBy(asc(bindings.id)).limit(100); + for (const b of active) { + if (signal.aborted) return; + try { + if (b.grantId) { + const [grant] = await db.select({ revokedAt: mcpOauthGrants.revokedAt }).from(mcpOauthGrants) + .where(and(eq(mcpOauthGrants.id, b.grantId), eq(mcpOauthGrants.companyId, b.companyId), + eq(mcpOauthGrants.agentId, b.agentId), eq(mcpOauthGrants.userId, b.operatorId))); + if (!grant || grant.revokedAt) { + await broker.revokeOAuthGrant({ grantId: b.grantId, companyId: b.companyId, + agentId: b.agentId, authorizingUserId: b.operatorId }); + continue; + } + } + await broker.reconcileTerminalAssignments(b.companyId, b.agentId, b.id); + } catch { + failed = true; + logger.warn({ bindingId: b.id }, "Muse authority or terminal reconciliation failed; will retry."); + } + } + if (active.length < 100) break; + after = active.at(-1)!.id; + } + if (failed) throw new Error("Muse maintenance has unfinished recovery"); + const [next] = await db.select({ at: bindings.qualificationExpiresAt }).from(bindings) + .where(isNull(bindings.revokedAt)) + .orderBy(asc(bindings.qualificationExpiresAt)).limit(1); + nextDeadline = next?.at?.getTime() ?? null; + completedSweep = sweep; + }, + async hasPending() { + const [pending] = await db.select({ id: assignments.id }).from(assignments) + .innerJoin(bindings, and(eq(bindings.id, assignments.bindingId), + eq(bindings.companyId, assignments.companyId), eq(bindings.agentId, assignments.agentId), + eq(bindings.generation, assignments.bindingGeneration), isNull(bindings.revokedAt))) + .where(or(inArray(assignments.status, ["offered", "claimed", "accepted"]), + and(eq(assignments.status, "settled"), isNull(assignments.finalizedAt)))) + .limit(1); + return !!pending; + }, + nextRunAt: () => completedSweep !== requestedSweep ? Date.now() + 5_000 : nextDeadline, + }); + receiver.start(); + return Object.assign(() => receiver.stop(), { ready: worker.ready }); } diff --git a/server/src/services/public-mcp/oauth.ts b/server/src/services/public-mcp/oauth.ts index f9f32f89856..4f5cdad55db 100644 --- a/server/src/services/public-mcp/oauth.ts +++ b/server/src/services/public-mcp/oauth.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { authorizeExternalBotOperator, externalBotEnabled } from "../external-bot-authority.js"; import { updateAgentConfigurationInTransaction } from "../agent-configuration-transaction.js"; import { instanceSettingsService } from "../instance-settings.js"; @@ -319,6 +320,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { if(!candidate)return db.transaction(tx=>operation(tx as unknown as Db)); return db.transaction(tx=>withExternalAdmissionGuard(tx,candidate.companyId,candidate.agentId,async()=>{ await tx.select({id:museAgentBindings.id}).from(museAgentBindings).where(and(eq(museAgentBindings.id,candidate.id),eq(museAgentBindings.grantId,grant.id))).for("update"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); return operation(tx as unknown as Db); })); } @@ -506,6 +508,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { await tx.insert(mcpOauthRequests).values({ id, clientId: client.id, redirectUri: p.redirect_uri, resource: config.resource, scopes, requestedCompanyId: binding.companyId, state: p.state ?? null, challenge: p.code_challenge, grantId: grant.id, codeHash: hashMcpSecret(code), decidedAt: now, expiresAt: new Date(now.getTime() + minute) }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); await tx.update(museAgentBindings).set({ grantId: grant.id, status: "connected", pairedAt: now, workerActivityAt: now, ticketHash: null, ticketExpiresAt: null, clientVersion: null, revision: binding.revision + 1, updatedAt: now }).where(eq(museAgentBindings.id, binding.id)); await logActivity(tx as unknown as Db, { companyId: binding.companyId, actorType: "user", actorId: binding.operatorId, @@ -657,6 +660,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { const [grant] = await tx.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, token.grantId)).for("update"); if (!grant || grant.clientId !== clientId || grant.resource !== config.resource || grant.revokedAt) return null; if (token.usedAt) { + if (museAgentConnection) await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); await tx.update(mcpOauthGrants).set({ revokedAt: new Date() }).where(eq(mcpOauthGrants.id, grant.id)); revokedGrant = grant; await tx.insert(activityLog).values({