mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
perf: stop polling idle Muse maintenance (#17000)
## Thinking Path > - Paperclip manages AI agents and their work. > - Muse connects an external worker to a Paperclip agent. > - Its maintenance loop scans the database every five seconds, even without bindings. > - These empty scans prevent an otherwise idle database from sleeping. > - The shared delivery coordinator can react to committed writes and saved deadlines. > - This change uses that coordinator and retains recovery for unfinished assignments. > - Idle Muse maintenance can now stop querying the database. ## Linked Issues or Issue Description **What existing behavior does this improve?** Muse qualification, OAuth authority, and terminal assignment maintenance. **Current behavior** A claimed instance runs two empty scans every five seconds even when Muse has no bindings. This produces about 34,560 unnecessary queries per day. Warm standby suppression from #16974 does not cover claimed instances. **Proposed behavior** Recover once at startup. Wake after relevant transactions settle. Schedule qualification expiry at its saved deadline. Retain five-second recovery only for failures and unfinished assignments. **Reason and benefit** Remove recurring database traffic from idle Muse maintenance while preserving authority checks and recovery. Reuse the database work signal foundation from #15625 without adding tables. ## What Changed - Register Muse maintenance with the shared delivery coordinator and its idle admission rules. - Signal pairing, OAuth attachment and revocation, qualification changes, and new assignments before transaction writes. The existing transaction layer waits for outer commit. - Restore qualification deadlines from persisted bindings. Retry failed sweeps and unfinished assignments. - Add PostgreSQL tests for idle quiet, nested commit, rollback, deadlines, standby, drain, and failure recovery. Verify signals from actual Muse producers. Preserve startup, newly committed work, and failed sweep retries across idle drain without SQL polling. - Document the remaining recovery cadence and sleep limitations. ## Verification - `pnpm -r typecheck`: passed before rebase. Server typechecking also passed after rebase and the drain fix. - `pnpm build`: passed before rebase. - Final change: 33 maintenance, delivery coordinator, and scheduler tests passed. Maintenance used a fresh disposable PostgreSQL 17 database. - Muse control-plane tests: 254 passed against the same disposable database. - `pnpm test:run`: attempted, then stopped after repeated embedded PostgreSQL startup failures from this host's semaphore limit. Native-session, Muse, and AI-connection suites could not create local clusters. The complete suite runs in CI. - `git diff --check`: passed. - Final commit `e106cbf20d`: all CI gates passed, including complete test suites, E2E, build, production bundle boot, public-install smoke test, and canary dry run. Greptile and Paperclip source reviews both rated 5/5, with no unresolved findings. The focused suites accept `PAPERCLIP_MUSE_MAINTENANCE_TEST_DATABASE_URL` and `PAPERCLIP_MUSE_CONTROL_TEST_DATABASE_URL`. Use a disposable test database; fixtures remove test data. ## Risks - Unfinished assignments retain five-second recovery. This change removes idle polling, not all Muse polling. - Signals are process-local. Independent database writers need explicit integration. Startup recovers persisted state. - Qualification deadlines depend on the running process. Existing durable Muse sleep blockers remain unchanged. - The receiver keeps its batching timer. An empty buffer already makes no database calls. - No schema migration or API change. ## Model Used OpenAI Codex, GPT-6 family, with code execution and repository tools. The exact model version is not exposed in this session. ## 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 (focused suites; full local run has the environment limitation above) - [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 - [x] All Paperclip CI gates are green - [x] 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
2b16262280
commit
cb2d6e80d3
9 files changed
+364
-131
No files matched your search
@@ -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.
|
||||
@@ -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<void>) | 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 };
|
||||
|
||||
+1
-3
@@ -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) {
|
||||
|
||||
@@ -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};
|
||||
|
||||
@@ -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<S>(db:Db,provider:Provider,identity:I
|
||||
const [old]=await tx.execute<Assignment>(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<Assignment>(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`);
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
let stop: ReturnType<typeof startMuseMaintenance>;
|
||||
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<void>(resolve => { release = resolve; });
|
||||
let written!: () => void;
|
||||
const ready = new Promise<void>(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);
|
||||
});
|
||||
});
|
||||
@@ -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<ReturnType<typeof createDeliveryWorkCoordinator>, "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 });
|
||||
}
|
||||
@@ -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({
|
||||
|
||||
Reference in new issue
Block a user