diff --git a/doc/muse-personal-agent.md b/doc/muse-personal-agent.md index 8143614643b..a6008923a51 100644 --- a/doc/muse-personal-agent.md +++ b/doc/muse-personal-agent.md @@ -25,6 +25,10 @@ Asset6 setup records the operator’s standing authorization to receive, validat Approve Paperclip's hostname through Muse's permission UI. Read the permission scope shown by Muse: its standing approval may cover a parent domain and all its subdomains. Paperclip cannot grant this permission on Muse's behalf. Setup distinguishes paired, receiver detected, and an authenticated background reply; the detector alone cannot finish setup. Hiring approval and native lifecycle readiness remain additional gates. +After receiver contact has committed to Paperclip, a temporary failure to create the readiness challenge is retried automatically, including after a server restart. Contact uses the existing thirty-second persistence schedule; an uncommitted check is not durable recovery evidence. You do not need to repeat contact or select Verify to repair an ordinary transient server failure. + +Verification still checks the current company, agent, authorizing operator and credential. It waits while the feature is disabled or the agent is paused or access is unavailable; restoring eligibility schedules another check. Expired or revoked credentials require reconnecting and are never revived by a role or status change. Automatic retry does not establish readiness: the authenticated background reply must still confirm the challenge. While that reply is pending, the native lifecycle saves an event-owned verification wait instead of polling readiness. Authenticated confirmation records its lifecycle recheck in the same transaction, including when a verification attempt is active. On startup, Paperclip rechecks saved waits once and parks them again if the reply is still absent. No repeated contact or manual Verify action is needed for this recovery. With no eligible pending verification, this recovery does not keep issuing periodic verification queries. Qualification deadlines and unresolved native work remain separate obligations, and the connected receiver's current tenant-sleep restriction still applies. + Muse’s Secure Vault/authd owns access and refresh credentials. The Python client stores only nonsecret connection metadata and stable operation receipts. The no-argument Bash detector reads the public receiver profile and produces opaque wake references. Empty checks and failures must not wake the model. Access expires after fifteen minutes, refresh rotates automatically, and thirty days of authenticated inactivity requires reconnecting. Signal checks do not extend credential activity. Assigned work claims an existing native run. Submitted progress, questions, task documents and completion use Paperclip's native authority and receipts. Paperclip displays reported progress; it cannot inspect every private Muse conversation/tool call. When idle, Muse can read permitted company work, create tasks and comment as the named agent. Other Paperclip execution operations require the current assignment. diff --git a/packages/shared/src/types/agent-lifecycle.ts b/packages/shared/src/types/agent-lifecycle.ts index bfb9b6533a9..8eece70c235 100644 --- a/packages/shared/src/types/agent-lifecycle.ts +++ b/packages/shared/src/types/agent-lifecycle.ts @@ -13,6 +13,10 @@ export interface AgentLifecycleOperation { leaseOwner?: string; leaseUntil?: string; retryAt?: string; + /** Healthy host verification awaits a durable external producer, not a retry timer. */ + waitingForSignal?: boolean; + /** A committed recheck raced an active host attempt; its defer must not park it. */ + recheckRequested?: boolean; /** A failed configuration check waits for configuration repair or an explicit retry. */ retryRequired?: boolean; /** Failed generic host verification tests for this saved configuration. */ diff --git a/server/src/__tests__/auth-session-route.test.ts b/server/src/__tests__/auth-session-route.test.ts index 1ab75bf346f..0ad9fe91a57 100644 --- a/server/src/__tests__/auth-session-route.test.ts +++ b/server/src/__tests__/auth-session-route.test.ts @@ -11,17 +11,18 @@ import { errorHandler } from "../middleware/error-handler.js"; import { assertCompanyAccess } from "../routes/authz.js"; function queryResult(rows: T) { - return Object.assign(Promise.resolve(rows), { for: () => Promise.resolve(rows) }); + return Object.assign(Promise.resolve(rows), { + for: () => Promise.resolve(rows), + orderBy: () => ({ limit: () => Promise.resolve(rows) }), + }); } -function createSelectChain(rows: unknown[]) { +function createSelectChain(rows: unknown[] | ((table: unknown) => unknown[])) { return { - from() { - return { - where() { - return queryResult(rows); - }, - }; + from(table: unknown) { + const selected = typeof rows === "function" ? rows(table) : rows; + const where = () => queryResult(selected); + return { where, innerJoin: () => ({ where }) }; }, }; } @@ -255,16 +256,8 @@ describe("actorMiddleware authenticated session profile", () => { }, }; const db = { - select: vi.fn(() => ({ - from: (table: unknown) => ({ - where: () => - queryResult( - table === companies - ? [{ name: "paperclip-stack-purple-rain" }] - : [], - ), - }), - })), + select: vi.fn(() => createSelectChain(table => table === companies + ? [{ name: "paperclip-stack-purple-rain" }] : [])), insert: vi.fn((table: unknown) => { if (table !== activityLog) return insertChain; return { @@ -352,14 +345,8 @@ describe("actorMiddleware authenticated session profile", () => { }, }; const db = { - select: vi.fn(() => ({ - from: (table: unknown) => ({ - where: () => - queryResult( - table === instanceUserRoles && state.staleInstanceAdminRow ? [{ id: "stale-role-row" }] : [], - ), - }), - })), + select: vi.fn(() => createSelectChain(table => + table === instanceUserRoles && state.staleInstanceAdminRow ? [{ id: "stale-role-row" }] : [])), insert: vi.fn(() => insertChain), delete: vi.fn((table: unknown) => ({ where: () => { diff --git a/server/src/__tests__/muse-contact-normal-app.postgres.test.ts b/server/src/__tests__/muse-contact-normal-app.postgres.test.ts new file mode 100644 index 00000000000..1d57f713211 --- /dev/null +++ b/server/src/__tests__/muse-contact-normal-app.postgres.test.ts @@ -0,0 +1,383 @@ +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import type { Express } from "express"; +import request from "supertest"; +import { and, eq } from "drizzle-orm"; +import { z } from "zod"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, authUsers, checkpointDb, closeRegisteredClients, + companies, createDb, fileStorageDb, heartbeatRuns, livenessDb, mcpOauthGrants, + museAgentBindings, museMailboxItems, museReceiverContactBuckets, subscribeDatabaseWork } from "@paperclipai/db"; +import { MUSE_OAUTH_SCOPES, MUSE_RESOURCE_PATH } from "@paperclipai/shared"; +import { resolveExternalTestDatabaseUrl } from "../lib/dedicated-test-database-url.js"; +import { accessService } from "../services/access.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { migrateCloudConfiguration } from "../services/cloud-configuration-migration.js"; +import { initializeAdapterSettings } from "../services/adapter-settings.js"; +import { initializeCloudRuntimeConfig } from "../services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { museRunnerBroker } from "../services/muse-runner-broker.js"; +import { getTaskDrainStatus } from "../services/heartbeat.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { removeTestHome } from "./remove-test-home.js"; + +const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MUSE_CONTACT_NORMAL_TEST_DATABASE_URL"); +const origin = "https://muse-contact.paperclip.test"; +const host = new URL(origin).host; +const resource = origin + MUSE_RESOURCE_PATH; +const redirect = "https://muse-contact-client.example.test/oauth/callback"; +const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +const pairingSchema = z.object({ bindingId: z.uuid(), generation: z.number().int().positive(), ticket: z.string().min(20) }); +const tokenSchema = z.object({ access_token: z.string().min(1), refresh_token: z.string().min(1) }); +const mailboxSchema = z.object({ bindingId: z.uuid(), generation: z.number().int().positive(), + items: z.array(z.object({ id: z.number().int(), kind: z.string(), references: z.object({ nonce: z.string() }).passthrough() })), + nextCursor: z.number().int() }); + +describe.skipIf(!databaseUrl)("normal Muse contact verification failure, durable retry and owner reconstruction", () => { + let db: ReturnType | undefined; + let app: Express | undefined; + let home: string | undefined; + let artifactPath: string | undefined; + let unsubscribe: (() => void) | undefined; + let restoreVerify: (() => void) | undefined; + let passed = false; + let registeredClientsClosed = false; + let stage = "not-started"; + let queryCount = 0; + const poolQueryCounts = { main: 0, liveness: 0, checkpoint: 0, fileStorage: 0 }; + let callbacks = 0; + let sessions = 0; + let verifyCalls = 0; + let unexpectedDecisionWakes = 0; + let invitationFailure: { status: number; code: string; message: string } | undefined; + let oldOwnerFailures = 0; + let reconstructedCommits = 0; + let firstAttemptAt: number | undefined; + let retryObservedAt: number | undefined; + let attemptOverflow = 0; + const attempts: Array<{ owner: "original" | "reconstructed"; atMs: number; outcome: "injected-failure" | "committed" | "error" }> = []; + const userId = randomUUID(), companyId = randomUUID(); + let agentId: string | undefined; + let bindingId: string | undefined; + let generation: number | undefined; + let clientId: string | undefined; + let accessToken: string | undefined; + let refreshToken: string | undefined; + let grantId: string | undefined; + let firstFailureState: { contactPersisted: boolean; challengeAbsent: boolean; mailboxCount: number } | undefined; + + function classifyInvitationFailure(status: number, body: unknown) { + const parsed = z.object({ error: z.string().optional() }).safeParse(body); + const fixed = [ + { message: "Board access required", code: "board_access_required" }, + { message: "User does not have access to this company", code: "company_access_denied" }, + { message: "User does not have active company access", code: "active_company_access_required" }, + { message: "Viewer access is read-only", code: "viewer_read_only" }, + { message: "Sign in as a company operator to configure Muse.", code: "company_operator_required" }, + { message: "Missing permission: agents:create.", code: "agents_create_permission_missing" }, + { message: "Permission agents:create does not cover the requested scope.", code: "agents_create_scope_denied" }, + { message: "Enable Muse and Paperclip Runner in experimental settings.", code: "muse_runner_disabled" }, + ].find(value => parsed.success && value.message === parsed.data.error); + return fixed ? { status, ...fixed } + : { status, code: "unclassified_invitation_response", message: "Invitation response details withheld" }; + } + + function currentDb() { if (!db) throw new Error("Missing owned Muse database"); return db; } + function currentApp() { if (!app) throw new Error("Missing normal Muse App"); return app; } + function record(owner: "original" | "reconstructed", outcome: "injected-failure" | "committed" | "error") { + const at = performance.now(); + firstAttemptAt ??= at; + if (attempts.length < 16) attempts.push({ owner, atMs: at - firstAttemptAt, outcome }); + else attemptOverflow++; + } + async function openNormalApp(observeBroker?: (owner: ReturnType) => void) { + if (!databaseUrl || !home) throw new Error("Explicit owned Muse fixture inputs required"); + const owner = createDb(databaseUrl, { maxConnections: 2, applicationName: "muse-contact-normal-app" }); + db = owner; + registeredClientsClosed = false; + for (const [name, pool] of [["main", owner], ["liveness", livenessDb(owner)], + ["checkpoint", checkpointDb(owner)], ["fileStorage", fileStorageDb(owner)]] as const) { + pool.$client.options.debug = () => { queryCount++; poolQueryCounts[name]++; }; + } + unsubscribe = subscribeDatabaseWork(owner, DELIVERY_QUEUES.museMaintenance, () => { callbacks++; }); + await migrateCloudConfiguration(owner); + await initializeAdapterSettings(owner); + await initializeCloudRuntimeConfig(owner); + observeBroker?.(owner); + const { createApp } = await import("../app.js"); + app = await createApp(owner, { + uiMode: "none", serverPort: 3100, + storageService: createStorageService(createLocalDiskStorageProvider(path.join(home, "objects"))), + deploymentMode: "authenticated", deploymentExposure: "private", bindHost: "127.0.0.1", + allowedHostnames: [host], authPublicBaseUrl: origin, authReady: true, + companyDeletionEnabled: false, managedPluginAutoInstall: [], + decisionServiceOptions: { wakeOriginAgent: async () => { + unexpectedDecisionWakes++; + throw new Error("This Muse contact fixture does not admit Decision continuation work"); + } }, + resolveSession: async req => { + sessions++; + if (req.get("x-muse-test-operator") !== userId) return null; + const [user] = await owner.select().from(authUsers).where(eq(authUsers.id, userId)); + return user ? { session: { id: "muse-contact-normal-session", userId }, user } : null; + }, + }); + await app.locals.bundledPluginsStartup; + } + async function closeNormalApp() { + if (app) { await app.locals.paperclipShutdown(); app = undefined; } + restoreVerify?.(); restoreVerify = undefined; + unsubscribe?.(); unsubscribe = undefined; + if (db) { await db.$client.end({ timeout: 1 }); db = undefined; } + if (databaseUrl) { + await closeRegisteredClients(databaseUrl, { scope: "database" }); + registeredClientsClosed = true; + } + } + + beforeAll(async () => { + const output = process.env.PAPERCLIP_MUSE_CONTACT_ACCEPTANCE_PATH; + if (!output || !path.isAbsolute(output)) throw new Error("Explicit absolute owned Muse acceptance artifact required"); + artifactPath = output; + expect(process.env.PAPERCLIP_CLOUD_STACK_ID).toBeUndefined(); + home = await mkdtemp(path.join(tmpdir(), "muse-contact-normal-")); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_CONFIG", path.join(home, "runtime-config.json")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); + vi.stubEnv("PAPERCLIP_PUBLIC_URL", origin); + vi.stubEnv("PAPERCLIP_AUTH_PUBLIC_BASE_URL", origin); + vi.stubEnv("PAPERCLIP_API_URL", origin); + if (!databaseUrl) throw new Error("Explicit Muse test database required"); + await openNormalApp(); + const owner = currentDb(); + await owner.insert(companies).values({ id: companyId, name: "Muse contact recovery", issuePrefix: "MCR0", requireBoardApprovalForNewAgents: false }); + await owner.insert(authUsers).values({ id: userId, name: "Muse fixture operator", email: userId + "@example.test", createdAt: new Date(), updatedAt: new Date() }); + await accessService(owner).ensureMembership(companyId, "user", userId, "owner", "active"); + await accessService(owner).ensureRoleDefaultGrants(companyId, userId, "owner", userId); + await instanceSettingsService(owner).updateExperimental({ enableMuse: true, enableNativeRunner: true }); + }, 90_000); + + afterAll(async () => { + const errors: unknown[] = []; + try { await closeNormalApp(); } catch (error) { errors.push(error); } + if (!errors.length && passed && home) { + try { removeTestHome(home); } catch (error) { errors.push(error); } + } + setCloudRuntimeConfigSnapshot(null); + vi.unstubAllEnvs(); + if (artifactPath) { + try { await writeFile(artifactPath + ".cleanup.json", JSON.stringify({ stage, passed, + oldOwnerFailures, reconstructedCommits, firstFailureState, invitationFailure, attempts, attemptOverflow, + queryCount, poolQueryCounts, callbacks, sessions, verifyCalls, unexpectedDecisionWakes, cleanupErrors: errors.length, + appOwnerJoined: app === undefined, rootDbEnded: db === undefined, registeredClientsClosed, + retainedHome: passed && !errors.length ? false : home ?? null }, null, 2) + "\n", { mode: 0o600, flag: "wx" }); } + catch (error) { errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Muse normal App owned cleanup failed"); + }, 30_000); + + it("recovers the sole failed contact verification through normal retry and a new Db owner, then stays quiet", async () => { + stage = "normal-invitation"; + const invite = await request(currentApp()).post(`/api/companies/${companyId}/muse-invitations`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", userId) + .send({ name: "Contact recovery Muse", role: "Contact verification regression" }); + if (invite.status !== 200) invitationFailure = classifyInvitationFailure(invite.status, invite.body); + expect(invite.status, invitationFailure ? JSON.stringify(invitationFailure) : undefined).toBe(200); + const createdAgentId = z.object({ agent: z.object({ id: z.uuid() }) }).parse(invite.body).agent.id; + agentId = createdAgentId; + const pairing = await request(currentApp()).post(`/api/companies/${companyId}/agents/${agentId}/muse-binding`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", userId).send({}); + expect(pairing.status).toBe(201); + const paired = pairingSchema.parse(pairing.body); + bindingId = paired.bindingId; generation = paired.generation; + stage = "normal-oauth"; + const registration = await request(currentApp()).post("/api/muse/v1/oauth/register").set("Host", host) + .send({ client_name: "Normal Muse contact recovery", redirect_uris: [redirect], + grant_types: ["authorization_code", "refresh_token"], response_types: ["code"], token_endpoint_auth_method: "none" }); + expect(registration.status).toBe(201); + clientId = z.object({ client_id: z.string().min(1) }).parse(registration.body).client_id; + const verifier = randomBytes(48).toString("base64url"); + const authorization = await request(currentApp()).get("/api/muse/v1/oauth/authorize").set("Host", host) + .query({ client_id: clientId, redirect_uri: redirect, response_type: "code", resource, + scope: MUSE_OAUTH_SCOPES.join(" "), code_challenge: createHash("sha256").update(verifier).digest("base64url"), + code_challenge_method: "S256", pairing_ticket: paired.ticket, state: "muse-contact-normal-state" }); + expect(authorization.status).toBe(303); + const codeLocation = new URL(z.string().parse(authorization.headers.location)); + expect(codeLocation.origin + codeLocation.pathname).toBe(redirect); + expect(codeLocation.searchParams.get("state")).toBe("muse-contact-normal-state"); + const tokens = await request(currentApp()).post("/api/muse/v1/oauth/token").set("Host", host) + .send({ grant_type: "authorization_code", client_id: clientId, redirect_uri: redirect, resource, + code: codeLocation.searchParams.get("code"), code_verifier: verifier }); + expect(tokens.status).toBe(200); + const credentials = tokenSchema.parse(tokens.body); + accessToken = credentials.access_token; refreshToken = credentials.refresh_token; + const owner = currentDb(); + const [authorized] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(authorized).toMatchObject({ companyId, agentId, operatorId: userId, generation: paired.generation, + status: "connected", receiverContactAt: null, verifiedReplyAt: null, challengeHash: null }); + expect(authorized.pairedAt).not.toBeNull(); expect(authorized.grantId).not.toBeNull(); + grantId = z.uuid().parse(authorized.grantId); + const [grant] = await owner.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, grantId)); + expect(grant).toMatchObject({ companyId, agentId, userId, clientId, resource, purpose: "agent", revokedAt: null }); + expect(grant.scopes).toEqual([...MUSE_OAUTH_SCOPES]); + expect((await request(currentApp()).post("/api/muse/v1/pair").set("Host", host).send({})).status).toBe(404); + + stage = "sole-contact-failure"; + const originalBroker = museRunnerBroker(owner); + const fault = vi.spyOn(originalBroker, "verify").mockImplementation(async (...args) => { + if (args[2].bindingId !== paired.bindingId) throw new Error("Unexpected Muse verification scope in isolated fixture"); + verifyCalls++; oldOwnerFailures++; record("original", "injected-failure"); + if (oldOwnerFailures === 1) { + const [saved] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + const existing = await owner.select({ id: museMailboxItems.id }).from(museMailboxItems) + .where(and(eq(museMailboxItems.bindingId, paired.bindingId), eq(museMailboxItems.kind, "readiness_challenge"))); + firstFailureState = { contactPersisted: saved.receiverContactAt !== null, + challengeAbsent: saved.challengeHash === null, mailboxCount: existing.length }; + } + if (oldOwnerFailures >= 2 && firstAttemptAt !== undefined && performance.now() - firstAttemptAt >= 5_000) { + retryObservedAt ??= performance.now() - firstAttemptAt; + } + throw Object.assign(new Error("Injected Muse verification infrastructure failure"), { code: "MUSE_TEST_VERIFICATION_UNAVAILABLE" }); + }); + restoreVerify = () => fault.mockRestore(); + const signal = await request(currentApp()).get("/api/muse/v1/signal").set("Host", host).set("X-Paperclip-Muse-Binding", paired.bindingId); + expect(signal.status).toBe(200); expect(signal.body.version).toBe(1); expect(signal.body.signal === null).toBe(true); + await vi.waitFor(() => { + expect(firstFailureState).toEqual({ contactPersisted: true, challengeAbsent: true, mailboxCount: 0 }); + }, { timeout: 45_000, interval: 25 }); + await vi.waitFor(() => { + expect(oldOwnerFailures).toBeGreaterThanOrEqual(2); + expect(retryObservedAt).toBeGreaterThanOrEqual(5_000); + }, { timeout: 10_000, interval: 25 }); + stage = "durable-retry-observed"; + const [beforeRestart] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(beforeRestart).toMatchObject({ generation: paired.generation, grantId, verifiedReplyAt: null, challengeHash: null }); + expect(beforeRestart.receiverContactAt).not.toBeNull(); + const contactsBefore = await owner.select().from(museReceiverContactBuckets).where(eq(museReceiverContactBuckets.bindingId, paired.bindingId)); + expect(contactsBefore).toHaveLength(1); + expect(contactsBefore[0]).toMatchObject({ contacts: 1, bindingGeneration: paired.generation, incomplete: false }); + expect(contactsBefore[0].timestamps).toHaveLength(1); + await closeNormalApp(); + + stage = "new-db-owner-reconstruction"; + await openNormalApp(reconstructed => { + const reconstructedBroker = museRunnerBroker(reconstructed); + expect(reconstructedBroker).not.toBe(originalBroker); + const realVerify = reconstructedBroker.verify; + const successful = vi.spyOn(reconstructedBroker, "verify").mockImplementation(async (...args) => { + verifyCalls++; + try { + const result = await realVerify(...args); + record("reconstructed", "committed"); reconstructedCommits++; return result; + } catch (error) { record("reconstructed", "error"); throw error; } + }); + restoreVerify = () => successful.mockRestore(); + }); + const reconstructed = currentDb(); + expect(reconstructed).not.toBe(owner); + await vi.waitFor(() => { expect(reconstructedCommits).toBe(1); }, { timeout: 15_000, interval: 25 }); + const [recovered] = await reconstructed.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(recovered).toMatchObject({ companyId, agentId, operatorId: userId, generation: paired.generation, + grantId, status: "connected", verifiedReplyAt: null, revision: beforeRestart.revision + 1 }); + expect(recovered.pairedAt).toEqual(beforeRestart.pairedAt); + expect(recovered.receiverContactAt).toEqual(beforeRestart.receiverContactAt); + expect(recovered.challengeHash).not.toBeNull(); + if (!recovered.challengeExpiresAt) throw new Error("Real Muse verification did not persist its challenge deadline"); + expect(recovered.challengeExpiresAt.getTime()).toBeGreaterThan(Date.now()); + expect(await reconstructed.select().from(museReceiverContactBuckets).where(eq(museReceiverContactBuckets.bindingId, paired.bindingId))).toEqual(contactsBefore); + const challengeRows = await reconstructed.select().from(museMailboxItems).where(and( + eq(museMailboxItems.bindingId, paired.bindingId), eq(museMailboxItems.kind, "readiness_challenge"))); + expect(challengeRows.length).toBe(1); + expect({ companyId: challengeRows[0].companyId, bindingGeneration: challengeRows[0].bindingGeneration, + assignmentId: challengeRows[0].assignmentId }).toEqual({ companyId, bindingGeneration: paired.generation, assignmentId: null }); + expect(await reconstructed.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, companyId), eq(activityLog.entityId, createdAgentId), eq(activityLog.action, "muse.readiness_challenge_created")))).toHaveLength(1); + stage = "healthy-lifecycle-reply-wait"; + await vi.waitFor(() => { + expect(getTaskDrainStatus()).toMatchObject({ activeRuns: 0, pendingWakes: 0, quiescent: true }); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 15_000, interval: 25 }); + const beforeWaitRead = { total: queryCount, main: poolQueryCounts.main }; + const [waitingAgent] = await reconstructed.select().from(agents).where(eq(agents.id, createdAgentId)); + expect(waitingAgent).toMatchObject({ lifecycleState: "verifying", status: "paused", + lifecycleOperation: { waitingForSignal: true, hostComplete: false } }); + expect(waitingAgent.lifecycleOperation).not.toHaveProperty("retryAt"); + expect(queryCount).toBeGreaterThan(beforeWaitRead.total); + expect(poolQueryCounts.main).toBeGreaterThan(beforeWaitRead.main); + const waitingBefore = { queryCount, poolQueryCounts: { ...poolQueryCounts }, callbacks, sessions, verifyCalls, unexpectedDecisionWakes }; + const waitingStarted = performance.now(); + await pause(15_000); + const waitingElapsedMs = performance.now() - waitingStarted; + const waitingAfter = { queryCount, poolQueryCounts: { ...poolQueryCounts }, callbacks, sessions, verifyCalls, unexpectedDecisionWakes }; + expect(waitingAfter).toEqual(waitingBefore); + stage = "genuine-challenge-confirmation"; + const inbox = await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: 0 }); + expect(inbox.status).toBe(200); + const mailbox = mailboxSchema.parse(inbox.body); + expect({ bindingId: mailbox.bindingId, generation: mailbox.generation }).toEqual({ bindingId: paired.bindingId, generation: paired.generation }); + expect(mailbox.items.length).toBe(1); + expect(mailbox.items[0].kind).toBe("readiness_challenge"); + const confirm = await request(currentApp()).post("/api/muse/v1/commands").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, command: "challenge.confirm", nonce: mailbox.items[0].references.nonce, requestId: randomUUID() }); + expect(confirm.status).toBe(200); + const confirmed = z.object({ status: z.literal("ready"), bindingId: z.uuid(), generation: z.number().int().positive() }).parse(confirm.body); + expect(confirmed).toEqual({ status: "ready", bindingId: paired.bindingId, generation: paired.generation }); + const acknowledge = await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: mailbox.nextCursor }); + expect(acknowledge.status).toBe(200); expect(z.array(z.unknown()).parse(acknowledge.body.items).length).toBe(0); + await vi.waitFor(async () => { + const [agent] = await reconstructed.select().from(agents).where(eq(agents.id, createdAgentId)); + expect(agent.lifecycleState).toBe("ready"); + expect(getTaskDrainStatus()).toMatchObject({ activeRuns: 0, pendingWakes: 0, quiescent: true }); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 15_000, interval: 25 }); + const beforeRequiredRead = { total: queryCount, main: poolQueryCounts.main }; + expect(await reconstructed.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(0); + const afterRequiredRead = { total: queryCount, main: poolQueryCounts.main }; + expect(afterRequiredRead.total).toBeGreaterThan(beforeRequiredRead.total); + expect(afterRequiredRead.main).toBeGreaterThan(beforeRequiredRead.main); + expect(poolQueryCounts.fileStorage).toBeGreaterThan(0); + expect(unexpectedDecisionWakes).toBe(0); + const quietBefore = { queryCount, poolQueryCounts: { ...poolQueryCounts }, callbacks, sessions, verifyCalls, unexpectedDecisionWakes }; + stage = "strict-fifteen-second-quiet"; + const quietStarted = performance.now(); + await pause(15_000); + const quietElapsedMs = performance.now() - quietStarted; + const quietAfter = { queryCount, poolQueryCounts: { ...poolQueryCounts }, callbacks, sessions, verifyCalls, unexpectedDecisionWakes }; + expect(quietAfter).toEqual(quietBefore); + stage = "revocation-and-negatives"; + const revoke = await request(currentApp()).post("/api/muse/v1/oauth/revoke").set("Host", host) + .send({ client_id: clientId, resource, token: refreshToken }); + expect(revoke.status).toBe(200); + expect((await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: mailbox.nextCursor })).status).toBe(401); + expect((await request(currentApp()).get("/api/muse/v1/signal").set("Host", host).set("X-Paperclip-Muse-Binding", paired.bindingId)).status).toBe(410); + const [revokedGrant] = await reconstructed.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, grantId)); + expect(revokedGrant.revokedAt).not.toBeNull(); + const [finalBinding] = await reconstructed.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(finalBinding.revokedAt).not.toBeNull(); expect(finalBinding.verifiedReplyAt).not.toBeNull(); + expect(unexpectedDecisionWakes).toBe(0); + if (!artifactPath) throw new Error("Missing owned Muse acceptance path"); + await writeFile(artifactPath, JSON.stringify({ contract: "normal Muse contact verification recovery", + companyId, userId, agentId, bindingId, generation, clientId, grantId, normalOAuth: true, + receiverSignals: 1, oldOwnerFailures, retryObservedAtMs: retryObservedAt, firstFailureState, + reconstructedDbObject: true, challengeCount: challengeRows.length, challengeConfirmed: true, + healthyLifecycleWait: { elapsedMs: waitingElapsedMs, before: waitingBefore, after: waitingAfter, + positiveReadControl: true, waitingForSignal: true }, + contacts: contactsBefore[0].contacts, contactTimestampCount: contactsBefore[0].timestamps.length, + beforeRestartRevision: beforeRestart.revision, recoveredRevision: recovered.revision, + queryInstrumentation: { beforeRequiredRead, afterRequiredRead }, unexpectedDecisionWakes, + quiet: { elapsedMs: quietElapsedMs, before: quietBefore, after: quietAfter }, + revoked: true, oldAccessRejected: true, oldSignalRejected: true, nativeRuns: 0, + cost: null, usage: null, liveMuseBackgroundReply: false, + attempts, attemptOverflow }, null, 2) + "\n", { mode: 0o600, flag: "wx" }); + stage = "complete"; passed = true; + }, 90_000); +}); diff --git a/server/src/__tests__/muse-contact-restoration-app.postgres.test.ts b/server/src/__tests__/muse-contact-restoration-app.postgres.test.ts new file mode 100644 index 00000000000..25b2565abbb --- /dev/null +++ b/server/src/__tests__/muse-contact-restoration-app.postgres.test.ts @@ -0,0 +1,390 @@ +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import type { Express } from "express"; +import request from "supertest"; +import { and, eq } from "drizzle-orm"; +import { z } from "zod"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, authUsers, checkpointDb, closeRegisteredClients, + companies, createDb, fileStorageDb, heartbeatRuns, livenessDb, mcpOauthGrants, + museAgentBindings, museMailboxItems, museReceiverContactBuckets, subscribeDatabaseWork } from "@paperclipai/db"; +import { MUSE_OAUTH_SCOPES, MUSE_RESOURCE_PATH } from "@paperclipai/shared"; +import { resolveExternalTestDatabaseUrl } from "../lib/dedicated-test-database-url.js"; +import { accessService } from "../services/access.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { migrateCloudConfiguration } from "../services/cloud-configuration-migration.js"; +import { initializeAdapterSettings } from "../services/adapter-settings.js"; +import { initializeCloudRuntimeConfig } from "../services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { museRunnerBroker } from "../services/muse-runner-broker.js"; +import { getTaskDrainStatus } from "../services/heartbeat.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { parseMuseVerificationHint } from "../modules/external-agents/index.js"; +import { HttpError } from "../errors.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { removeTestHome } from "./remove-test-home.js"; + +const databaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_MUSE_CONTACT_NORMAL_TEST_DATABASE_URL"); +const origin = "https://muse-restoration.paperclip.test"; +const host = new URL(origin).host; +const resource = origin + MUSE_RESOURCE_PATH; +const redirect = "https://muse-restoration-client.example.test/oauth/callback"; +const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +const pairingSchema = z.object({ bindingId: z.uuid(), generation: z.number().int().positive(), ticket: z.string().min(20) }); +const tokenSchema = z.object({ access_token: z.string().min(1), refresh_token: z.string().min(1) }); +const mailboxSchema = z.object({ bindingId: z.uuid(), generation: z.number().int().positive(), + items: z.array(z.object({ id: z.number().int(), kind: z.string(), references: z.object({ nonce: z.string() }).passthrough() })), + nextCursor: z.number().int() }); + +describe.skipIf(!databaseUrl).each(["permission", "lifecycle"] as const)("normal Muse %s restoration", mode => { + let db: ReturnType | undefined; + let app: Express | undefined; + let home: string | undefined; + let artifactPath: string | undefined; + let unsubscribe: (() => void) | undefined; + let restoreVerify: (() => void) | undefined; + let passed = false; + let registeredClientsClosed = false; + let stage = "not-started"; + let queryCount = 0; + const poolQueryCounts = { main: 0, liveness: 0, checkpoint: 0, fileStorage: 0 }; + let callbacks = 0; + let sessions = 0; + let verifyCalls = 0; + let unexpectedDecisionWakes = 0; + let invitationFailure: { status: number; code: string; message: string } | undefined; + const userId = randomUUID(), restorerId = randomUUID(), companyId = randomUUID(); + let contactSettled = 0; + let deniedVerifications = 0; + let committedVerifications = 0; + let otherVerifyCalls = 0; + let otherDeniedVerifications = 0; + let otherCommittedVerifications = 0; + let expectedResumeState: string | undefined; + let agentId: string | undefined; + let bindingId: string | undefined; + let generation: number | undefined; + let clientId: string | undefined; + let accessToken: string | undefined; + let refreshToken: string | undefined; + let grantId: string | undefined; + + function classifyInvitationFailure(status: number, body: unknown) { + const parsed = z.object({ error: z.string().optional() }).safeParse(body); + const fixed = [ + { message: "Board access required", code: "board_access_required" }, + { message: "User does not have access to this company", code: "company_access_denied" }, + { message: "User does not have active company access", code: "active_company_access_required" }, + { message: "Viewer access is read-only", code: "viewer_read_only" }, + { message: "Sign in as a company operator to configure Muse.", code: "company_operator_required" }, + { message: "Missing permission: agents:create.", code: "agents_create_permission_missing" }, + { message: "Permission agents:create does not cover the requested scope.", code: "agents_create_scope_denied" }, + { message: "Enable Muse and Paperclip Runner in experimental settings.", code: "muse_runner_disabled" }, + ].find(value => parsed.success && value.message === parsed.data.error); + return fixed ? { status, ...fixed } + : { status, code: "unclassified_invitation_response", message: "Invitation response details withheld" }; + } + + function currentDb() { if (!db) throw new Error("Missing owned Muse database"); return db; } + function currentApp() { if (!app) throw new Error("Missing normal Muse App"); return app; } + async function openNormalApp(observeBroker?: (owner: ReturnType) => void) { + if (!databaseUrl || !home) throw new Error("Explicit owned Muse fixture inputs required"); + const owner = createDb(databaseUrl, { maxConnections: 2, applicationName: "muse-contact-normal-app" }); + db = owner; + registeredClientsClosed = false; + for (const [name, pool] of [["main", owner], ["liveness", livenessDb(owner)], + ["checkpoint", checkpointDb(owner)], ["fileStorage", fileStorageDb(owner)]] as const) { + pool.$client.options.debug = () => { queryCount++; poolQueryCounts[name]++; }; + } + unsubscribe = subscribeDatabaseWork(owner, DELIVERY_QUEUES.museMaintenance, (event, hints) => { + callbacks++; + if (event === "settled" && Array.from(hints ?? []).some(raw => { + const hint = parseMuseVerificationHint(raw); + return hint !== undefined && hint.bindingId === bindingId && hint.generation === generation && hint.companyId === companyId; + })) contactSettled++; + }); + await migrateCloudConfiguration(owner); + await initializeAdapterSettings(owner); + await initializeCloudRuntimeConfig(owner); + observeBroker?.(owner); + const { createApp } = await import("../app.js"); + app = await createApp(owner, { + uiMode: "none", serverPort: 3100, + storageService: createStorageService(createLocalDiskStorageProvider(path.join(home, "objects"))), + deploymentMode: "authenticated", deploymentExposure: "private", bindHost: "127.0.0.1", + allowedHostnames: [host], authPublicBaseUrl: origin, authReady: true, + companyDeletionEnabled: false, managedPluginAutoInstall: [], + decisionServiceOptions: { wakeOriginAgent: async () => { + unexpectedDecisionWakes++; + throw new Error("This Muse contact fixture does not admit Decision continuation work"); + } }, + resolveSession: async req => { + sessions++; + const sessionUserId = req.get("x-muse-test-operator"); + if (sessionUserId !== userId && sessionUserId !== restorerId) return null; + const [user] = await owner.select().from(authUsers).where(eq(authUsers.id, sessionUserId)); + return user ? { session: { id: "muse-restoration-session-" + sessionUserId, userId: sessionUserId }, user } : null; + }, + }); + await app.locals.bundledPluginsStartup; + } + async function closeNormalApp() { + if (app) { await app.locals.paperclipShutdown(); app = undefined; } + restoreVerify?.(); restoreVerify = undefined; + unsubscribe?.(); unsubscribe = undefined; + if (db) { await db.$client.end({ timeout: 1 }); db = undefined; } + if (databaseUrl) { + await closeRegisteredClients(databaseUrl, { scope: "database" }); + registeredClientsClosed = true; + } + } + + beforeAll(async () => { + const output = process.env.PAPERCLIP_MUSE_CONTACT_ACCEPTANCE_PATH; + if (!output || !path.isAbsolute(output)) throw new Error("Explicit absolute owned Muse acceptance artifact required"); + artifactPath = output; + expect(process.env.PAPERCLIP_CLOUD_STACK_ID).toBeUndefined(); + home = await mkdtemp(path.join(tmpdir(), "muse-contact-normal-")); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_CONFIG", path.join(home, "runtime-config.json")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", randomBytes(32).toString("base64")); + vi.stubEnv("PAPERCLIP_PUBLIC_URL", origin); + vi.stubEnv("PAPERCLIP_AUTH_PUBLIC_BASE_URL", origin); + vi.stubEnv("PAPERCLIP_API_URL", origin); + if (!databaseUrl) throw new Error("Explicit Muse test database required"); + await openNormalApp(); + const owner = currentDb(); + await owner.insert(companies).values({ id: companyId, name: "Muse contact recovery", issuePrefix: mode === "permission" ? "MRP0" : "MRL0", requireBoardApprovalForNewAgents: false }); + await owner.insert(authUsers).values({ id: userId, name: "Muse fixture operator", email: userId + "@example.test", createdAt: new Date(), updatedAt: new Date() }); + await accessService(owner).ensureMembership(companyId, "user", userId, "operator", "active"); + await accessService(owner).ensureRoleDefaultGrants(companyId, userId, "operator", userId); + await owner.insert(authUsers).values({ id: restorerId, name: "Restoration owner", email: restorerId + "@example.test", createdAt: new Date(), updatedAt: new Date() }); + await accessService(owner).ensureMembership(companyId, "user", restorerId, "owner", "active"); + await accessService(owner).ensureRoleDefaultGrants(companyId, restorerId, "owner", restorerId); + await instanceSettingsService(owner).updateExperimental({ enableMuse: true, enableNativeRunner: true }); + }, 90_000); + + afterAll(async () => { + const errors: unknown[] = []; + try { await closeNormalApp(); } catch (error) { errors.push(error); } + if (!errors.length && passed && home) { + try { removeTestHome(home); } catch (error) { errors.push(error); } + } + setCloudRuntimeConfigSnapshot(null); + vi.unstubAllEnvs(); + if (artifactPath) { + try { await writeFile(artifactPath + "." + mode + ".cleanup.json", JSON.stringify({ stage, passed, + mode, deniedVerifications, committedVerifications, otherVerifyCalls, otherDeniedVerifications, otherCommittedVerifications, contactSettled, expectedResumeState, invitationFailure, + queryCount, poolQueryCounts, callbacks, sessions, verifyCalls, unexpectedDecisionWakes, cleanupErrors: errors.length, + appOwnerJoined: app === undefined, rootDbEnded: db === undefined, registeredClientsClosed, + retainedHome: passed && !errors.length ? false : home ?? null }, null, 2) + "\n", { mode: 0o600, flag: "wx" }); } + catch (error) { errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Muse normal App owned cleanup failed"); + }, 30_000); + + it("restores the same current binding without another contact or manual verification, then remains quiet and revoked", async () => { + stage = "normal-invitation"; + const invite = await request(currentApp()).post(`/api/companies/${companyId}/muse-invitations`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", restorerId) + .send({ name: "Contact recovery Muse", role: "Contact verification regression" }); + if (invite.status !== 200) invitationFailure = classifyInvitationFailure(invite.status, invite.body); + expect(invite.status, invitationFailure ? JSON.stringify(invitationFailure) : undefined).toBe(200); + const createdAgentId = z.object({ agent: z.object({ id: z.uuid() }) }).parse(invite.body).agent.id; + agentId = createdAgentId; + const pairing = await request(currentApp()).post(`/api/companies/${companyId}/agents/${agentId}/muse-binding`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", userId).send({}); + expect(pairing.status).toBe(201); + const paired = pairingSchema.parse(pairing.body); + bindingId = paired.bindingId; generation = paired.generation; + stage = "normal-oauth"; + const registration = await request(currentApp()).post("/api/muse/v1/oauth/register").set("Host", host) + .send({ client_name: "Normal Muse contact recovery", redirect_uris: [redirect], + grant_types: ["authorization_code", "refresh_token"], response_types: ["code"], token_endpoint_auth_method: "none" }); + expect(registration.status).toBe(201); + clientId = z.object({ client_id: z.string().min(1) }).parse(registration.body).client_id; + const verifier = randomBytes(48).toString("base64url"); + const authorization = await request(currentApp()).get("/api/muse/v1/oauth/authorize").set("Host", host) + .query({ client_id: clientId, redirect_uri: redirect, response_type: "code", resource, + scope: MUSE_OAUTH_SCOPES.join(" "), code_challenge: createHash("sha256").update(verifier).digest("base64url"), + code_challenge_method: "S256", pairing_ticket: paired.ticket, state: "muse-contact-normal-state" }); + expect(authorization.status).toBe(303); + const codeLocation = new URL(z.string().parse(authorization.headers.location)); + expect(codeLocation.origin + codeLocation.pathname).toBe(redirect); + expect(codeLocation.searchParams.get("state")).toBe("muse-contact-normal-state"); + const tokens = await request(currentApp()).post("/api/muse/v1/oauth/token").set("Host", host) + .send({ grant_type: "authorization_code", client_id: clientId, redirect_uri: redirect, resource, + code: codeLocation.searchParams.get("code"), code_verifier: verifier }); + expect(tokens.status).toBe(200); + const credentials = tokenSchema.parse(tokens.body); + accessToken = credentials.access_token; refreshToken = credentials.refresh_token; + const owner = currentDb(); + const [authorized] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(authorized).toMatchObject({ companyId, agentId, operatorId: userId, generation: paired.generation, + status: "connected", receiverContactAt: null, verifiedReplyAt: null, challengeHash: null }); + expect(authorized.pairedAt).not.toBeNull(); expect(authorized.grantId).not.toBeNull(); + grantId = z.uuid().parse(authorized.grantId); + const [grant] = await owner.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, grantId)); + expect(grant).toMatchObject({ companyId, agentId, userId, clientId, resource, purpose: "agent", revokedAt: null }); + expect(grant.scopes).toEqual([...MUSE_OAUTH_SCOPES]); + expect((await request(currentApp()).post("/api/muse/v1/pair").set("Host", host).send({})).status).toBe(404); + + stage = "sole-contact-before-parking"; + const broker = museRunnerBroker(owner); + const realVerify = broker.verify; + const observed = vi.spyOn(broker, "verify").mockImplementation(async (...args) => { + verifyCalls++; + const current = args[0] === companyId && args[1] === createdAgentId + && args[2].bindingId === paired.bindingId && args[2].generation === paired.generation; + if (!current) otherVerifyCalls++; + try { + const result = await realVerify(...args); + if (current) committedVerifications++; + else otherCommittedVerifications++; + return result; + } catch (error) { + if (error instanceof HttpError && error.status === 403) { + if (current) deniedVerifications++; + else otherDeniedVerifications++; + } + throw error; + } + }); + restoreVerify = () => observed.mockRestore(); + const signal = await request(currentApp()).get("/api/muse/v1/signal").set("Host", host).set("X-Paperclip-Muse-Binding", paired.bindingId); + expect(signal.status).toBe(200); expect(signal.body.version).toBe(1); expect(signal.body.signal).toBeNull(); + const member = await accessService(owner).getMembership(companyId, "user", userId); + if (!member) throw new Error("Missing normally authorized Muse operator membership"); + if (mode === "permission") { + const demote = await request(currentApp()).patch(`/api/companies/${companyId}/members/${member.id}`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", restorerId) + .send({ membershipRole: "viewer", status: "active" }); + expect(demote.status).toBe(200); + expect(demote.body).toMatchObject({ principalId: userId, membershipRole: "viewer", status: "active" }); + } else { + const pauseAgent = await request(currentApp()).post(`/api/agents/${createdAgentId}/pause`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", restorerId).send({}); + expect(pauseAgent.status).toBe(200); + } + // Wait only on the real commit-bound contact notification; no flush/Verify kick. + await vi.waitFor(() => { expect(contactSettled).toBeGreaterThan(0); }, { timeout: 45_000, interval: 25 }); + if (mode === "permission") { + await vi.waitFor(() => { expect(deniedVerifications).toBeGreaterThan(0); }, { timeout: 10_000, interval: 25 }); + } + async function settled() { + await vi.waitFor(() => { + expect(getTaskDrainStatus()).toMatchObject({ activeRuns: 0, pendingWakes: 0, quiescent: true }); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 15_000, interval: 25 }); + } + const counters = () => ({ queryCount, pools: { ...poolQueryCounts }, callbacks, sessions, verifyCalls, unexpectedDecisionWakes }); + async function quiet() { + await settled(); + const beforeRequiredRead = { total: queryCount, main: poolQueryCounts.main }; + expect(await owner.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(0); + expect(queryCount).toBeGreaterThan(beforeRequiredRead.total); + expect(poolQueryCounts.main).toBeGreaterThan(beforeRequiredRead.main); + expect(poolQueryCounts.fileStorage).toBeGreaterThan(0); + const before = counters(), at = performance.now(); + await pause(15_000); + const elapsedMs = performance.now() - at, after = counters(); + expect(after).toEqual(before); + return { before, after, elapsedMs, positiveReadControl: true }; + } + await settled(); + const [parked] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(parked).toMatchObject({ companyId, agentId, operatorId: userId, generation: paired.generation, + grantId, status: "connected", challengeHash: null, verifiedReplyAt: null, revokedAt: null }); + expect(parked.receiverContactAt).not.toBeNull(); + const [parkedGrant] = await owner.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, grantId)); + expect(parkedGrant).toMatchObject({ userId, companyId, agentId, clientId, revokedAt: null }); + const contactsBefore = await owner.select().from(museReceiverContactBuckets).where(eq(museReceiverContactBuckets.bindingId, paired.bindingId)); + expect(contactsBefore).toHaveLength(1); expect(contactsBefore[0]).toMatchObject({ contacts: 1, bindingGeneration: paired.generation, incomplete: false }); + expect(contactsBefore[0].timestamps).toHaveLength(1); expect(committedVerifications).toBe(0); + if (mode === "lifecycle") { + const [pausedAgent] = await owner.select().from(agents).where(eq(agents.id, createdAgentId)); + expect(pausedAgent).toMatchObject({ status: "paused", lifecycleState: "paused" }); + expect(pausedAgent.lifecycleHolds).not.toContain("setup"); + expectedResumeState = pausedAgent.lifecycleOperation?.resumeState; + expect(["preparing", "verifying"]).toContain(expectedResumeState); + // A snapshot taken before Pause can reach the real verifier and be denied. + // The strict window below still accounts for ALL owner-wide calls. + expect(otherCommittedVerifications).toBe(0); + } + stage = "parked-current-binding"; + const parkedQuiet = await quiet(); + stage = "normal-authorized-restoration"; + if (mode === "permission") { + const restore = await request(currentApp()).patch(`/api/companies/${companyId}/members/${member.id}`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", restorerId) + .send({ membershipRole: "operator", status: "active" }); + expect(restore.status).toBe(200); + expect(restore.body).toMatchObject({ principalId: userId, membershipRole: "operator", status: "active" }); + } else { + const resume = await request(currentApp()).post(`/api/agents/${createdAgentId}/resume`) + .set("Host", host).set("Origin", origin).set("x-muse-test-operator", restorerId).send({}); + expect(resume.status).toBe(200); expect(resume.body.status).toBe("paused"); + } + await vi.waitFor(() => { expect(committedVerifications).toBe(1); }, { timeout: 15_000, interval: 25 }); + const [recovered] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(recovered).toMatchObject({ companyId, agentId, operatorId: userId, generation: paired.generation, + grantId, status: "connected", revision: parked.revision + 1, verifiedReplyAt: null, revokedAt: null }); + expect(recovered.pairedAt).toEqual(parked.pairedAt); expect(recovered.receiverContactAt).toEqual(parked.receiverContactAt); + expect(recovered.challengeHash).not.toBeNull(); + expect(await owner.select().from(museReceiverContactBuckets).where(eq(museReceiverContactBuckets.bindingId, paired.bindingId))).toEqual(contactsBefore); + const challengeRows = await owner.select().from(museMailboxItems).where(and( + eq(museMailboxItems.bindingId, paired.bindingId), eq(museMailboxItems.kind, "readiness_challenge"))); + expect(challengeRows).toHaveLength(1); + expect(challengeRows[0]).toMatchObject({ companyId, bindingGeneration: paired.generation, assignmentId: null }); + expect(await owner.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, companyId), eq(activityLog.entityId, createdAgentId), eq(activityLog.action, "muse.readiness_challenge_created")))).toHaveLength(1); + const [restoredAgent] = await owner.select().from(agents).where(eq(agents.id, createdAgentId)); + expect(restoredAgent.status).toBe("paused"); + expect(["preparing", "verifying"]).toContain(restoredAgent.lifecycleState); + const inbox = await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: 0 }); + expect(inbox.status).toBe(200); + const mailbox = mailboxSchema.parse(inbox.body); + expect(mailbox).toMatchObject({ bindingId: paired.bindingId, generation: paired.generation }); + expect(mailbox.items).toHaveLength(1); expect(mailbox.items[0].kind).toBe("readiness_challenge"); + const confirm = await request(currentApp()).post("/api/muse/v1/commands").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, command: "challenge.confirm", nonce: mailbox.items[0].references.nonce, requestId: randomUUID() }); + expect(confirm.status).toBe(200); expect(confirm.body).toMatchObject({ status: "ready", bindingId: paired.bindingId, generation: paired.generation }); + const acknowledge = await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: mailbox.nextCursor }); + expect(acknowledge.status).toBe(200); expect(acknowledge.body.items).toEqual([]); + await vi.waitFor(async () => { + const [ready] = await owner.select().from(agents).where(eq(agents.id, createdAgentId)); + expect(ready.lifecycleState).toBe("ready"); + }, { timeout: 15_000, interval: 25 }); + stage = "restored-ready-quiet"; + const restoredQuiet = await quiet(); + const revoke = await request(currentApp()).post("/api/muse/v1/oauth/revoke").set("Host", host) + .send({ client_id: clientId, token: refreshToken, resource }); + expect(revoke.status).toBe(200); + expect((await request(currentApp()).post("/api/muse/v1/queries").set("Host", host).set("Authorization", "Bearer " + accessToken) + .send({ version: 1, query: "mailbox", after: 0 })).status).toBe(401); + expect((await request(currentApp()).get("/api/muse/v1/signal").set("Host", host).set("X-Paperclip-Muse-Binding", paired.bindingId)).status).toBe(410); + await instanceSettingsService(owner).updateExperimental({ enableMuse: false }); + stage = "revoked-and-disabled-quiet"; + const revokedDisabledQuiet = await quiet(); + const [finalGrant] = await owner.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, grantId)); + const [finalBinding] = await owner.select().from(museAgentBindings).where(eq(museAgentBindings.id, paired.bindingId)); + expect(finalGrant.revokedAt).not.toBeNull(); expect(finalBinding.revokedAt).not.toBeNull(); + expect(committedVerifications).toBe(1); expect(unexpectedDecisionWakes).toBe(0); + if (!artifactPath) throw new Error("Missing owned restoration evidence path"); + await writeFile(artifactPath + "." + mode + ".json", JSON.stringify({ mode, companyId, userId, restorerId, + agentId, bindingId, generation, grantId, receiverSignals: 1, contacts: 1, timestampCount: 1, + parkedRevision: parked.revision, recoveredRevision: recovered.revision, expectedResumeState, + actualRestoredLifecycle: restoredAgent.lifecycleState, compatibilityStatusUnchanged: true, + normalOAuth: true, deniedVerifications, committedVerifications, otherVerifyCalls, otherDeniedVerifications, otherCommittedVerifications, noManualVerify: true, + parkedQuiet, restoredQuiet, revokedDisabledQuiet, nativeRuns: 0, cost: null, usage: null, + liveMuseBackgroundReply: false, revoked: true, disabled: true }, null, 2) + "\n", { mode: 0o600, flag: "wx" }); + stage = "complete"; passed = true; + }, 90_000); +}); diff --git a/server/src/board-claim.ts b/server/src/board-claim.ts index a5a2cbdf20e..816b9911eeb 100644 --- a/server/src/board-claim.ts +++ b/server/src/board-claim.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "./modules/external-agents/index.js"; import { randomBytes } from "node:crypto"; import { and, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -126,6 +127,7 @@ export async function claimBoardOwnership( if (!existing) { await notifyChatCompanyAuthorityChange(tx, { companyId: company.id }); + await notifyMuseContactVerification(tx,{companyId:company.id,operatorId:opts.userId}); await tx.insert(companyMemberships).values({ companyId: company.id, principalType: "user", @@ -138,6 +140,7 @@ export async function claimBoardOwnership( if (existing.status !== "active") { await notifyChatCompanyAuthorityChange(tx, { companyId: company.id }); + await notifyMuseContactVerification(tx,{companyId:company.id,operatorId:opts.userId}); await tx .update(companyMemberships) .set({ status: "active", membershipRole: "owner", updatedAt: new Date() }) diff --git a/server/src/middleware/auth.ts b/server/src/middleware/auth.ts index bf8a05f22d0..92ccddcf508 100644 --- a/server/src/middleware/auth.ts +++ b/server/src/middleware/auth.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { idleApiActivity } from "../services/idle-api-activity.js"; import { createHash, timingSafeEqual } from "node:crypto"; import type { Request, RequestHandler } from "express"; @@ -672,6 +673,7 @@ async function resolveCloudTenantActorOnce( // guarded upsert that leaves unchanged memberships untouched. if (!existing || existing.status !== "active" || existing.membershipRole === "viewer") { await notifyChatCompanyAuthorityChange(tx, { companyId }); + await notifyMuseContactVerification(tx, { companyId, operatorId: userId }); } return tx .insert(companyMemberships) diff --git a/server/src/middleware/cloud-tenant-actor.test.ts b/server/src/middleware/cloud-tenant-actor.test.ts index 4f9e12c97ad..8aaca4e5990 100644 --- a/server/src/middleware/cloud-tenant-actor.test.ts +++ b/server/src/middleware/cloud-tenant-actor.test.ts @@ -75,6 +75,8 @@ function createFakeDb(options: { select: () => { return { from: (table: unknown) => ({ + // Pending Muse bindings use a joined read; this fixture has none. + innerJoin() { return this; }, where: (condition: unknown) => { // Only settings lookup is degradable. Membership locking and // permission seeding must still work in the fail-closed flag test. @@ -91,6 +93,7 @@ function createFakeDb(options: { : []; let locked = false; const result = { + orderBy: () => ({ limit: () => result }), for: () => { locked = true; selected.locked = true; return result; }, then: (resolve: (v: unknown) => unknown, reject?: (error: unknown) => unknown) => { // The optional cross-company read can fail independently of diff --git a/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts b/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts index fe9410c111c..0d295e2b37e 100644 --- a/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/diagnostics.test.ts @@ -23,6 +23,17 @@ function fixture() { afterEach(() => { vi.restoreAllMocks(); vi.useRealTimers(); }); describe("lifecycle readiness handoff", () => { + it("defers event-owned host verification without marking host or readiness complete", async () => { + const { store, driver, worker } = fixture(); + driver.runHost.mockResolvedValue("waiting"); + await worker.process(agent.id); + expect(store.defer).toHaveBeenCalledWith(agent, expect.any(String), null, expect.any(Date), { waitForSignal: true }); + expect(store.completeHost).not.toHaveBeenCalled(); + expect(store.change).not.toHaveBeenCalled(); + expect(driver.onReady).not.toHaveBeenCalled(); + await worker.stop(); + }); + it("notifies the scheduler only after host, plugins, and readiness have committed", async () => { const { store, driver, worker } = fixture(); let commit!: (ready: LifecycleAgent) => void; diff --git a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts index 53ab1392803..e55f627582a 100644 --- a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts @@ -26,7 +26,7 @@ import { createAgentLifecycle, configureAgentLifecycle, startAgentLifecycle } fr import { projectCompanyHolds } from "../adapters/company-policy.js"; import { createLifecycleStore as createStore } from "../adapters/postgres.js"; import { transition } from "../domain/policy.js"; -import type { LifecycleAgent } from "../application/ports.js"; +import type { LifecycleAgent, LifecycleDriver } from "../application/ports.js"; import { budgetService } from "../../../services/budgets.js"; import { agentConfigurationRevision } from "../../../lib/agent-configuration-revision.js"; @@ -53,7 +53,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu companyId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Lifecycle test", issuePrefix: `L${companyId.slice(0, 7)}` }); }); - function worker(runHost: (agent: LifecycleAgent) => Promise<"complete" | "pending"> = async () => "complete") { + function worker(runHost: LifecycleDriver["runHost"] = async () => "complete") { const value = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runPlugin: async () => "complete", runHost }); workers.push(value); return value; } diff --git a/server/src/modules/agent-lifecycle/__tests__/setup.test.ts b/server/src/modules/agent-lifecycle/__tests__/setup.test.ts index fd558df179e..8c4ded0a840 100644 --- a/server/src/modules/agent-lifecycle/__tests__/setup.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/setup.test.ts @@ -11,7 +11,7 @@ import { createLifecycleDriver } from "../../../services/agent-lifecycle-driver. import { agentRoutes } from "../../../routes/agents.js"; import { errorHandler } from "../../../middleware/error-handler.js"; import { createLifecycleStore as createStore } from "../adapters/postgres.js"; -import type { LifecycleAgent } from "../application/ports.js"; +import type { LifecycleAgent, LifecycleDriver } from "../application/ports.js"; const createLifecycleStore = (db: Db) => createStore(db, createAgentLifecycleEffects()); const support = await getEmbeddedPostgresTestSupport(); @@ -27,7 +27,7 @@ const support = await getEmbeddedPostgresTestSupport(); companyId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Setup test", issuePrefix: `S${companyId.slice(0, 7)}` }); }); - function worker(runHost: (agent: LifecycleAgent) => Promise<"complete" | "pending"> = async () => "complete") { + function worker(runHost: LifecycleDriver["runHost"] = async () => "complete") { const work = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runPlugin: async () => "complete", runHost }); workers.push(work); return work; } diff --git a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts index 5ff5f3f4faa..3976bf71e34 100644 --- a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts @@ -5,10 +5,11 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; import { createAgentLifecycle, configureAgentLifecycle } from "../../../services/agent-lifecycle.js"; import { createAgentLifecycleEffects } from "../../../services/agent-lifecycle.js"; -import { startAgentLifecycle } from "../index.js"; +import { requestAgentLifecycleRecheckInTransaction, startAgentLifecycle } from "../index.js"; import { createLifecycleStore } from "../adapters/postgres.js"; import { createLifecycleWorker } from "../application/worker.js"; import type { LifecycleAgent } from "../application/ports.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LIFECYCLE_VERIFICATION_RETRY_TEST_DATABASE_URL"); @@ -19,8 +20,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let companyId: string; const workers: Array<{ stop(): Promise }> = []; beforeAll(async () => { - if (external) { await applyPendingMigrations(external); db = createDb(external, { maxConnections: 6 }); } - else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString, { maxConnections: 6 }); } + if (external) { await applyPendingMigrations(external); db = createDb(external, { maxConnections: 4 }); } + else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString, { maxConnections: 4 }); } }, 90_000); afterAll(async () => { if (external) await closeRegisteredClients(external); else await database?.cleanup(); }); beforeEach(async () => { @@ -208,4 +209,150 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await current(agent.id)).lifecycleState).toBe("terminated"); expect(runHost.mock.calls.map(([a]) => a.lifecycleState)).toContain("cleaning_up"); }); + + async function eventWaitFixture() { + let complete = false; + const runHost = vi.fn(async (agent: LifecycleAgent) => agent.lifecycleState !== "verifying" || complete + ? "complete" as const : "waiting" as const); + const driver = { requiredPluginIds: async () => [], runPlugin: async () => "complete" as const, runHost }; + const work = configureAgentLifecycle(db, driver); workers.push(work); + const agent = await createAgentLifecycle(db).requestHire(companyId, { name: "External reply wait", adapterType: "process" }); + await work.process(agent.id); + return { agent: await current(agent.id), work, driver, runHost, complete: () => { complete = true; } }; + } + function recheck(tx: Pick, id: string, scopeCompanyId = companyId, version?: number) { + return requestAgentLifecycleRecheckInTransaction(tx, + { companyId: scopeCompanyId, agentId: id, expectedVersion: version }, + () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle)); + } + + it("parks healthy external verification without a retry deadline, claim or readiness bypass", async () => { + const { agent, work, runHost } = await eventWaitFixture(); + expect(agent).toMatchObject({ lifecycleState: "verifying", status: "paused", lifecycleError: null, + lifecycleOperation: { waitingForSignal: true, hostComplete: false } }); + expect(agent.lifecycleOperation).not.toHaveProperty("retryAt"); + expect(agent.lifecycleOperation).not.toHaveProperty("retryRequired"); + expect(agent.lifecycleOperation).not.toHaveProperty("leaseOwner"); + const store = createLifecycleStore(db, createAgentLifecycleEffects()), calls = runHost.mock.calls.length; + const later = new Date(Date.now() + 86_400_000); + expect(await store.pending(100, later)).not.toContain(agent.id); + expect(await store.claim(agent.id, "unnotified", later)).toBeNull(); + await work.process(agent.id); + expect(runHost).toHaveBeenCalledTimes(calls); + expect((await current(agent.id)).lifecycleState).toBe("verifying"); + }); + + it.each(["during", "after"] as const)("retains a committed recheck %s the active attempt parks", async ordering => { + let entered!: () => void, release!: () => void, ready = false; + const inside = new Promise(resolve => { entered = resolve; }); + const hold = new Promise(resolve => { release = resolve; }); + const runHost = vi.fn(async (agent: LifecycleAgent) => { + if (agent.lifecycleState !== "verifying" || ready) return "complete" as const; + entered(); await hold; return "waiting" as const; + }); + const work = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runPlugin: async () => "complete", runHost }); + workers.push(work); + const agent = await createAgentLifecycle(db).requestHire(companyId, { name: "Commit race", adapterType: "process" }); + const processing = work.process(agent.id); + await inside; + try { + if (ordering === "after") { release(); await processing; } + await db.transaction(async tx => { + expect(await recheck(tx, agent.id)).toBe(true); + }); + ready = true; + } finally { release(); await processing; } + const signalled = await current(agent.id); + expect(signalled.lifecycleOperation?.waitingForSignal).not.toBe(true); + expect(signalled.lifecycleOperation).not.toHaveProperty("retryAt"); + await work.process(agent.id); + expect(await current(agent.id)).toMatchObject({ lifecycleState: "ready", status: "idle" }); + expect(runHost.mock.calls.filter(([a]) => a.lifecycleState === "verifying")).toHaveLength(2); + }); + + it("retains a committed recheck when an expired attempt returns its earlier wait", async () => { + const { agent, work, complete, runHost } = await eventWaitFixture(); + await db.transaction(async tx => { expect(await recheck(tx, agent.id)).toBe(true); }); + const store = createLifecycleStore(db, createAgentLifecycleEffects()); + const owner = "expired-verification-owner"; + // Use the store's normal clock input: the real claim is already expired when + // confirmation commits, without changing timers or waiting two minutes. + const claimed = await store.claim(agent.id, owner, new Date(Date.now() - 121_000)); + if (!claimed) throw new Error("Expected a real verification claim"); + expect(Date.parse(claimed.lifecycleOperation!.leaseUntil!)).toBeLessThan(Date.now()); + await db.transaction(async tx => { expect(await recheck(tx, agent.id)).toBe(true); }); + const afterConfirmation = await current(agent.id); + complete(); + expect(await store.defer(claimed, owner, null, new Date(), { waitForSignal: true })).toBe(true); + const signalled = await current(agent.id); + expect(signalled.lifecycleOperation).not.toHaveProperty("waitingForSignal"); + expect(afterConfirmation.lifecycleOperation).toMatchObject({ leaseOwner: owner, recheckRequested: true }); + expect(signalled.lifecycleOperation).not.toHaveProperty("leaseOwner"); + expect(signalled.lifecycleOperation).not.toHaveProperty("retryAt"); + expect(await store.pending(100, new Date())).toContain(agent.id); + await work.process(agent.id); + expect(await current(agent.id)).toMatchObject({ lifecycleState: "ready", status: "idle" }); + expect(runHost.mock.calls.filter(([a]) => a.lifecycleState === "verifying")).toHaveLength(2); + }); + + it("rolls back the recheck with its producer and coalesces redundant committed signals", async () => { + const { agent, work, complete, runHost } = await eventWaitFixture(); + await expect(db.transaction(async tx => { + expect(await recheck(tx, agent.id)).toBe(true); + throw new Error("Rollback this producer"); + })).rejects.toThrow("Rollback this producer"); + expect((await current(agent.id)).lifecycleOperation?.waitingForSignal).toBe(true); + await db.transaction(async tx => { + expect(await recheck(tx, agent.id)).toBe(true); + expect(await recheck(tx, agent.id)).toBe(true); + }); + complete(); await work.process(agent.id); + const calls = runHost.mock.calls.length; + await work.process(agent.id); + expect(runHost).toHaveBeenCalledTimes(calls); + expect((await current(agent.id)).lifecycleState).toBe("ready"); + }); + + it("finishes a committed recheck on a new owner without a post-commit scheduling call", async () => { + const { agent, work, driver, complete } = await eventWaitFixture(); + await work.stop(); + await db.transaction(async tx => { expect(await recheck(tx, agent.id)).toBe(true); }); + complete(); + // No process/Retry/wake callback follows the committed producer here. + const restarted = startAgentLifecycle(db, createAgentLifecycleEffects(), driver, () => true); workers.push(restarted); + await restarted.sweep(); + expect(await current(agent.id)).toMatchObject({ lifecycleState: "ready", status: "idle" }); + expect(await restarted.hasPending()).toBe(false); + }); + + it("rechecks a saved wait once at startup and remains parked without recurring startup scans", async () => { + const { agent, work, driver, runHost } = await eventWaitFixture(); + await work.stop(); + const restarted = startAgentLifecycle(db, createAgentLifecycleEffects(), driver, () => true); workers.push(restarted); + const calls = runHost.mock.calls.length; + expect(await restarted.hasPending()).toBe(true); + await restarted.sweep(); + expect(runHost).toHaveBeenCalledTimes(calls + 1); + expect((await current(agent.id)).lifecycleOperation?.waitingForSignal).toBe(true); + expect(await restarted.hasPending()).toBe(false); + expect(restarted.nextRunAt()).toBeNull(); + await restarted.sweep(); + expect(runHost).toHaveBeenCalledTimes(calls + 1); + }); + + it("does not rearm a parked wait after a company or version mismatch, pause or required manual repair", async () => { + const { agent, work } = await eventWaitFixture(); + await db.transaction(async tx => { + expect(await recheck(tx, agent.id, randomUUID())).toBe(false); + expect(await recheck(tx, agent.id, companyId, agent.lifecycleVersion - 1)).toBe(false); + }); + expect((await current(agent.id)).lifecycleOperation).toEqual(agent.lifecycleOperation); + await createAgentLifecycle(db).pauseAgent(agent.id); + await work.process(agent.id); + await db.transaction(async tx => { expect(await recheck(tx, agent.id)).toBe(false); }); + expect(await current(agent.id)).toMatchObject({ lifecycleState: "paused", status: "paused" }); + const manual = await failed(); + await db.transaction(async tx => { expect(await recheck(tx, manual.agent.id)).toBe(false); }); + expect((await current(manual.agent.id)).lifecycleOperation?.retryRequired).toBe(true); + }); }); diff --git a/server/src/modules/agent-lifecycle/adapters/configuration.ts b/server/src/modules/agent-lifecycle/adapters/configuration.ts index 908cffe6236..cec045f49a2 100644 --- a/server/src/modules/agent-lifecycle/adapters/configuration.ts +++ b/server/src/modules/agent-lifecycle/adapters/configuration.ts @@ -6,7 +6,7 @@ import { eq } from "drizzle-orm"; import { agents, type Db } from "@paperclipai/db"; import { agentRecordQueries, AGENT_CONFIGURATION_FIELDS, jsonEqual, type AgentConfigurationPatch, type UpdateAgentOptions } from "../../../lib/agent-records.js"; import type { ActivityPublication } from "../../../types/activity-publication.js"; -import { AgentLifecycleConflict } from "../domain/policy.js"; +import { AgentLifecycleConflict, canConfigureAgentConnection } from "../domain/policy.js"; import type { LifecycleEffects } from "./effects.js"; export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects, id: string, data: AgentConfigurationPatch, options?: UpdateAgentOptions, publications: ActivityPublication[] = []) { @@ -40,6 +40,7 @@ export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects { adapterType: patch.adapterType ?? current.adapterType, adapterConfig: Object.prototype.hasOwnProperty.call(patch, "adapterConfig") ? patch.adapterConfig : current.adapterConfig }); const invalidate = changed && (["preparing", "verifying", "resuming"].includes(current.lifecycleState) || prepareSetup); + if(prepareSetup&&!canConfigureAgentConnection(current))await effects.beforeConnectionEligibilityRestore?.(tx,current.companyId,current.id); if (invalidate) await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ ...patch, updatedAt: new Date(), ...(invalidate ? { lifecycleVersion: current.lifecycleVersion + 1, lifecycleError: null, @@ -48,7 +49,8 @@ export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects ? { lifecycleState: "pausing" as const } : {}), lifecycleOperation: { ...current.lifecycleOperation!, id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 0, leaseOwner: undefined, leaseUntil: undefined, - retryAt: undefined, retryRequired: undefined, verificationFailures: undefined }, + retryAt: undefined, retryRequired: undefined, verificationFailures: undefined, + waitingForSignal: undefined, recheckRequested: undefined }, } : {}) }).where(eq(agents.id, id)).returning(); await effects.completeConfiguration(tx, current, updated, patch, options, publications); return agentRecordQueries(tx).getById(id); diff --git a/server/src/modules/agent-lifecycle/adapters/effects.ts b/server/src/modules/agent-lifecycle/adapters/effects.ts index a77ef6eb77a..43eec359774 100644 --- a/server/src/modules/agent-lifecycle/adapters/effects.ts +++ b/server/src/modules/agent-lifecycle/adapters/effects.ts @@ -11,6 +11,7 @@ export interface LifecycleEffects { environmentRequiresNewPlugin(db: Db, environmentId: string | null | undefined, selectedPluginIds: string[] | null): Promise; notifyWork(db: Db): Promise; beforeStatusChange(db: Db, companyId: string, agentId: string, before: string, after: string): Promise; + beforeConnectionEligibilityRestore?(db: Db, companyId: string, agentId: string): Promise; reportFailure(context: LifecycleFailure, error: unknown): void; deleteDependencies(db: Db, companyId: string, agentId: string): Promise; prepareConfiguration(db: Db, current: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise; diff --git a/server/src/modules/agent-lifecycle/adapters/postgres.ts b/server/src/modules/agent-lifecycle/adapters/postgres.ts index 4ca50b7edcc..8d5f9b1f576 100644 --- a/server/src/modules/agent-lifecycle/adapters/postgres.ts +++ b/server/src/modules/agent-lifecycle/adapters/postgres.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { and, asc, eq, inArray, sql } from "drizzle-orm"; import { agents, agentApiKeys, companies, budgetPolicies, type Db } from "@paperclipai/db"; import type { AgentLifecycleOperation } from "@paperclipai/shared"; -import { AgentLifecycleConflict, compatibilityStatus, isLifecycleWorkPending, transition, type LifecycleCommand } from "../domain/policy.js"; +import { AgentLifecycleConflict, canConfigureAgentConnection, compatibilityStatus, isLifecycleWorkPending, transition, type LifecycleCommand } from "../domain/policy.js"; import type { LifecycleEffects } from "./effects.js"; import type { LifecycleStore } from "../application/ports.js"; @@ -18,8 +18,43 @@ export const pendingLifecycleWork = and( inArray(agents.lifecycleState, ["preparing", "verifying", "pausing", "resuming", "terminating", "cleaning_up"]), sql`not (${agents.lifecycleState} = 'verifying' and ${agents.lifecycleHolds} @> '["setup"]'::jsonb)`, sql`not (${agents.lifecycleState} = 'verifying' and coalesce(${agents.lifecycleOperation}->>'retryRequired', 'false') = 'true')`, + sql`not (${agents.lifecycleState} = 'verifying' and coalesce(${agents.lifecycleOperation}->>'waitingForSignal', 'false') = 'true')`, ); +/** A bounded startup pass rechecks this saved wait once, without a timer. */ +export const lifecycleSignalWait = and( + eq(agents.lifecycleState, "verifying"), + sql`coalesce(${agents.lifecycleOperation}->>'waitingForSignal', 'false') = 'true'`, + sql`coalesce(${agents.lifecycleOperation}->>'retryRequired', 'false') <> 'true'`, +); + +/** The producer owns this outer transaction. Register its intent before either + * the recheck write or the corresponding external fact commits. */ +export async function requestAgentLifecycleRecheckInTransaction( + tx: Pick, + input: { companyId: string; agentId: string; expectedVersion?: number; onlyIfWaiting?: boolean }, + notify: () => Promise, +): Promise { + if (!("nestedIndex" in tx)) throw new AgentLifecycleConflict("Lifecycle rechecks require the producer transaction"); + const [agent] = await tx.select().from(agents).where(and( + eq(agents.id, input.agentId), eq(agents.companyId, input.companyId), + )).for("update"); + if (!agent || !["preparing", "verifying"].includes(agent.lifecycleState) + || agent.lifecycleOperation?.retryRequired + || input.expectedVersion !== undefined && agent.lifecycleVersion !== input.expectedVersion + || input.onlyIfWaiting && !agent.lifecycleOperation?.waitingForSignal) return false; + await notify(); + // Expiry permits a replacement claim; only an owner change fences a late park. + const owned = Boolean(agent.lifecycleOperation?.leaseOwner); + await tx.update(agents).set({ lifecycleError: null, + lifecycleOperation: owned + ? sql`${agents.lifecycleOperation} || '{"recheckRequested":true}'::jsonb` + : sql`${agents.lifecycleOperation} - 'waitingForSignal' - 'recheckRequested' - 'retryAt'`, + updatedAt: new Date(), + }).where(and(eq(agents.id, agent.id), eq(agents.companyId, input.companyId), eq(agents.lifecycleVersion, agent.lifecycleVersion))); + return true; +} + /** The caller owns the accounting transaction and has locked this agent. */ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: LifecycleRow, command: LifecycleCommand, options: ChangeOptions = {}, policyHolds?: string[], restartPause = false) { @@ -31,10 +66,16 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec agent.lifecycleRequiredPluginIds.some(id => !agent.lifecycleOperation!.completedPluginIds.includes(id)))) return null; if (command === "retry") { transition(agent.lifecycleState, command); - if (agent.lifecycleOperation?.leaseUntil && Date.parse(agent.lifecycleOperation.leaseUntil) > Date.now()) return agent; + if (agent.lifecycleOperation?.leaseUntil && Date.parse(agent.lifecycleOperation.leaseUntil) > Date.now()) { + await effects.notifyWork(tx); + const [requested] = await tx.update(agents).set({ lifecycleOperation: + sql`${agents.lifecycleOperation} || '{"recheckRequested":true}'::jsonb`, updatedAt: new Date(), + }).where(eq(agents.id, agent.id)).returning(); + return requested; + } await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ lifecycleError: null, - lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt' - 'retryRequired' - 'verificationFailures' - 'leaseOwner' - 'leaseUntil'`, updatedAt: new Date() }).where(eq(agents.id, agent.id)).returning(); + lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt' - 'retryRequired' - 'verificationFailures' - 'leaseOwner' - 'leaseUntil' - 'waitingForSignal' - 'recheckRequested'`, updatedAt: new Date() }).where(eq(agents.id, agent.id)).returning(); return updated; } if (command === "resume" && !["ready", "paused", "pausing", "resuming"].includes(agent.lifecycleState)) transition(agent.lifecycleState, command); @@ -81,6 +122,7 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec }; if (next === "ready") operation.resumeState = "ready"; const status = compatibilityStatus(next, agent.status); + if(!canConfigureAgentConnection(agent)&&canConfigureAgentConnection({lifecycleState:next,status}))await effects.beforeConnectionEligibilityRestore?.(tx,agent.companyId,agent.id); if (status !== agent.status) await effects.beforeStatusChange(tx, agent.companyId, agent.id, agent.status, status); if (isLifecycleWorkPending(next)) await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ @@ -139,7 +181,7 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc }, async claim(id, owner, now) { const [agent] = await db.update(agents).set({ - lifecycleOperation: sql`coalesce(${agents.lifecycleOperation}, '{"id":"", "hostComplete":false, "completedPluginIds":[], "attempts":0}'::jsonb) + lifecycleOperation: sql`(coalesce(${agents.lifecycleOperation}, '{"id":"", "hostComplete":false, "completedPluginIds":[], "attempts":0}'::jsonb) - 'recheckRequested') || jsonb_build_object('leaseOwner', ${owner}::text, 'leaseUntil', ${(new Date(now.getTime() + 120_000)).toISOString()}::text, 'attempts', coalesce((${agents.lifecycleOperation}->>'attempts')::integer, 0) + 1)`, }).where(and(eq(agents.id, id), pendingLifecycleWork, @@ -180,6 +222,16 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc return rows.length > 0; }, async defer(agent, owner, error, now, options) { + if (options?.waitForSignal && agent.lifecycleState === "verifying") { + const rows = await db.update(agents).set({ lifecycleError: null, + lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'retryAt' - 'waitingForSignal' - 'recheckRequested') + || case when coalesce(${agents.lifecycleOperation}->>'recheckRequested', 'false') = 'true' + then '{}'::jsonb else '{"waitingForSignal":true}'::jsonb end`, + updatedAt: now, + }).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion), + sql`${agents.lifecycleOperation}->>'leaseOwner' = ${owner}`)).returning({ id: agents.id }); + return rows.length > 0; + } const retry: Partial = options?.retryRequired && agent.lifecycleState === "verifying" ? { retryRequired: true } : { retryAt: new Date(now.getTime() + (error ? 60_000 : 2_000)).toISOString() }; @@ -188,7 +240,7 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc } const rows = await db.update(agents).set({ lifecycleError: error, - lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'retryAt' - 'retryRequired') + lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'retryAt' - 'retryRequired' - 'waitingForSignal' - 'recheckRequested') || ${JSON.stringify(retry)}::jsonb`, updatedAt: now, }).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion), diff --git a/server/src/modules/agent-lifecycle/adapters/records.ts b/server/src/modules/agent-lifecycle/adapters/records.ts index 64e10144c01..cdc85e7297a 100644 --- a/server/src/modules/agent-lifecycle/adapters/records.ts +++ b/server/src/modules/agent-lifecycle/adapters/records.ts @@ -4,7 +4,7 @@ import { and, eq, sql } from "drizzle-orm"; import { agents, agentApiKeys, companies, type Db } from "@paperclipai/db"; import { conflict, notFound } from "../../../errors.js"; import { agentRecordQueries, type CreateAgentOptions, type CreateAgentData } from "../../../lib/agent-records.js"; -import { isAgentSetupDraft } from "../domain/policy.js"; +import { isAgentSetupDraft, canConfigureAgentConnection } from "../domain/policy.js"; export function agentRecords(db: Db, effects: LifecycleEffects) { const { getById } = agentRecordQueries(db); @@ -40,6 +40,7 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { const lifecycleState = status === "pending_approval" ? "pending_approval" : status === "terminated" ? "terminated" : prepareSetup ? "preparing" : status === "paused" ? "paused" : companyHolds.length ? "pausing" : "preparing"; if (lifecycleState === "preparing" || lifecycleState === "pausing") await effects.notifyWork(tx); + const createdStatus=status === "pending_approval" ? "pending_approval" : status === "terminated" ? "terminated" : "paused"; const values: typeof agents.$inferInsert = { ...prepared, createdByUserId: setup?.createdByUserId ?? options?.createdByUserId ?? null, @@ -49,9 +50,10 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { pauseReason: holds[0] ?? null, pausedAt: holds.length ? new Date() : null, lifecycleOperation: { id: randomUUID(), hostComplete: false, completedPluginIds: [], attempts: 0, resumeState: "preparing" as const, responsibleUserId: options?.responsibleUserId ?? options?.createdByUserId }, - status: status === "pending_approval" ? "pending_approval" : status === "terminated" ? "terminated" : "paused", + status: createdStatus, companyId, }; + if(setup&&!canConfigureAgentConnection(setup)&&canConfigureAgentConnection({lifecycleState,status:createdStatus}))await effects.beforeConnectionEligibilityRestore?.(txDb,companyId,setup.id); const created = await (options?.setupAgentId ? tx.update(agents).set({ ...values, updatedAt: new Date() }).where(eq(agents.id, options.setupAgentId)) : tx.insert(agents).values(values)) @@ -123,6 +125,7 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { const holds = existing.lifecycleHolds.filter(hold => hold !== "company_paused" && hold !== "company_archived"); if (company?.status === "paused") holds.push("company_paused"); if (company?.status === "archived") holds.push("company_archived"); + if(!holds.length)await effects.beforeConnectionEligibilityRestore?.(txDb,existing.companyId,existing.id); await effects.notifyWork(txDb); const updated = await tx .update(agents) diff --git a/server/src/modules/agent-lifecycle/application/ports.ts b/server/src/modules/agent-lifecycle/application/ports.ts index 2fbcd33c8cc..ade8e539011 100644 --- a/server/src/modules/agent-lifecycle/application/ports.ts +++ b/server/src/modules/agent-lifecycle/application/ports.ts @@ -20,13 +20,14 @@ export interface LifecycleStore { setRequiredPlugins(agent: LifecycleAgent, owner: string, pluginIds: string[]): Promise; completeHost(agent: LifecycleAgent, owner: string): Promise; completePlugin(agent: LifecycleAgent, owner: string, pluginId: string): Promise; - defer(agent: LifecycleAgent, owner: string, error: string | null, now: Date, options?: { retryRequired: boolean; verificationFailures?: number }): Promise; + defer(agent: LifecycleAgent, owner: string, error: string | null, now: Date, options?: { retryRequired?: boolean; verificationFailures?: number; waitForSignal?: boolean }): Promise; pending(limit: number, now: Date): Promise; } export interface LifecycleDriver { requiredPluginIds(agent: LifecycleAgent): Promise; - runHost(agent: LifecycleAgent): Promise<"complete" | "pending">; + /** waiting is only a healthy verifying phase owned by an external event. */ + runHost(agent: LifecycleAgent): Promise<"complete" | "pending" | "waiting">; runPlugin(agent: LifecycleAgent, pluginId: string): Promise<"complete" | "pending">; /** Notify the scheduler only after readiness has committed. */ onReady?(agent: LifecycleAgent): Promise; diff --git a/server/src/modules/agent-lifecycle/application/worker.ts b/server/src/modules/agent-lifecycle/application/worker.ts index 27c533329b3..5a83aec531e 100644 --- a/server/src/modules/agent-lifecycle/application/worker.ts +++ b/server/src/modules/agent-lifecycle/application/worker.ts @@ -45,7 +45,13 @@ export function createLifecycleWorker(store: LifecycleStore, driver: LifecycleDr } if (!operation.hostComplete) { if (!await runStep("renew", () => store.renew(agent, owner, new Date()))) continue; - if (await runStep("host", () => driver.runHost(agent)) === "pending") { + const host = await runStep("host", () => driver.runHost(agent)); + if (host === "waiting") { + if (agent.lifecycleState !== "verifying") throw new Error("Only host verification may await an external event"); + if (!await runStep("defer", () => store.defer(agent, owner, null, new Date(), { waitForSignal: true }))) continue; + return; + } + if (host === "pending") { if (!await runStep("defer", () => store.defer(agent, owner, null, new Date()))) continue; return; } diff --git a/server/src/modules/agent-lifecycle/index.ts b/server/src/modules/agent-lifecycle/index.ts index 429fab07f88..8fc083de80b 100644 --- a/server/src/modules/agent-lifecycle/index.ts +++ b/server/src/modules/agent-lifecycle/index.ts @@ -1,6 +1,6 @@ import { hireApprovalService as approvalRecords, type HireDecisionTarget } from "./adapters/approvals.js"; -import { agents } from "@paperclipai/db"; -import { inArray, sql } from "drizzle-orm"; +import { agents, companies } from "@paperclipai/db"; +import { and, asc, desc, eq, gt, inArray, lte, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import type { LifecycleEffects } from "./adapters/effects.js"; export type { LifecycleEffects } from "./adapters/effects.js"; @@ -8,12 +8,12 @@ import { updateAgentConfiguration } from "./adapters/configuration.js"; import type { AgentConfigurationPatch, UpdateAgentOptions } from "../../lib/agent-records.js"; import { purgeAgent } from "./adapters/deletion.js"; import { agentRecords } from "./adapters/records.js"; -import { assertRootDatabase, createLifecycleStore, pendingLifecycleWork } from "./adapters/postgres.js"; +import { assertRootDatabase, createLifecycleStore, pendingLifecycleWork, lifecycleSignalWait, requestAgentLifecycleRecheckInTransaction } from "./adapters/postgres.js"; import { hasCompanyHoldMismatch, projectCompanyHolds } from "./adapters/company-policy.js"; import { createLifecycleWorker } from "./application/worker.js"; import type { LifecycleDriver } from "./application/ports.js"; -export { applyAgentPolicyHoldsInTransaction } from "./adapters/postgres.js"; +export { applyAgentPolicyHoldsInTransaction, requestAgentLifecycleRecheckInTransaction } from "./adapters/postgres.js"; export { companyHoldMismatch, fenceActiveCompanyAgentStopsInTransaction } from "./adapters/company-policy.js"; export { AgentLifecycleConflict, canConfigureAgentConnection, isAgentAwaitingSetup, isAgentSetupDraft } from "./domain/policy.js"; @@ -105,6 +105,35 @@ export function startAgentLifecycle(db: Db, effects: LifecycleEffects, driver: L let running: Promise | undefined; let stopped = false; let nextDeadline: number | null = null; + const recoveryStartedAt = new Date(); + let recoveryInitialized = false, recoveryDone = false; + let recoveryThrough: string | undefined, recoveryAfter: string | undefined; + let recoveryInitialization: Promise | undefined; + const recoveryCondition = and(lifecycleSignalWait, lte(agents.createdAt, recoveryStartedAt)); + async function initializeRecovery() { + if (recoveryInitialized) return; + return recoveryInitialization ??= (async () => { + const [last] = await db.select({ id: agents.id }).from(agents).where(recoveryCondition) + .orderBy(desc(agents.id)).limit(1); + recoveryThrough = last?.id; recoveryDone = !last; recoveryInitialized = true; + })().catch(error => { recoveryInitialization = undefined; throw error; }); + } + async function recoverPage() { + await initializeRecovery(); + if (recoveryDone || !recoveryThrough) return; + const page = await db.select({ id: agents.id, companyId: agents.companyId, version: agents.lifecycleVersion }) + .from(agents).innerJoin(companies, eq(companies.id, agents.companyId)) + .where(and(recoveryCondition, eq(companies.status, "active"), lte(agents.id, recoveryThrough), + recoveryAfter ? gt(agents.id, recoveryAfter) : undefined)).orderBy(asc(agents.id)).limit(100); + for (const agent of page) { + if (stopped || !canRun()) return; + await effects.transaction(db, agent.companyId, tx => requestAgentLifecycleRecheckInTransaction(tx, + { companyId: agent.companyId, agentId: agent.id, expectedVersion: agent.version, onlyIfWaiting: true }, + () => effects.notifyWork(tx))); + recoveryAfter = agent.id; + } + if (page.length < 100) recoveryDone = true; + } // The coordinator supplies the wake after construction. No direct work escapes it. wakes.set(db, () => {}); return { @@ -112,6 +141,10 @@ export function startAgentLifecycle(db: Db, effects: LifecycleEffects, driver: L nextRunAt: () => nextDeadline, async hasPending() { if (stopped) { nextDeadline = null; return false; } + if (canRun()) { + await initializeRecovery(); + if (!recoveryDone) return true; + } const [pending] = await db.select({ due: sql`min(greatest( coalesce(extract(epoch from (${agents.lifecycleOperation}->>'leaseUntil')::timestamptz) * 1000, 0), coalesce(extract(epoch from (${agents.lifecycleOperation}->>'retryAt')::timestamptz) * 1000, 0)))::double precision` }) @@ -128,6 +161,8 @@ export function startAgentLifecycle(db: Db, effects: LifecycleEffects, driver: L return running ??= (async () => { try { await projectCompanyHolds(db, effects); } catch (error) { effects.reportFailure({ stage: "policy_scan" }, error); throw error; } + try { await recoverPage(); } + catch (error) { effects.reportFailure({ stage: "scan" }, error); throw error; } await worker.sweep(); })().finally(() => { running = undefined; }); }, diff --git a/server/src/modules/external-agents/index.ts b/server/src/modules/external-agents/index.ts index 97c76af8105..ac854b0b9bc 100644 --- a/server/src/modules/external-agents/index.ts +++ b/server/src/modules/external-agents/index.ts @@ -19,3 +19,5 @@ export function createDotPrincipalHttp(oauth: Parameters, + scope?: MuseVerificationScope, + after?: string, + contactWillPersist = false, +): Promise { + return db.select({ binding: bindings, agent: { status: agents.status, lifecycleState: agents.lifecycleState } }) + .from(bindings) + .innerJoin(agents, and(eq(agents.id, bindings.agentId), eq(agents.companyId, bindings.companyId))) + .where(and( + isNull(bindings.revokedAt), isNotNull(bindings.pairedAt), + contactWillPersist ? undefined : isNotNull(bindings.receiverContactAt), + isNull(bindings.verifiedReplyAt), isNull(bindings.challengeHash), + scope?.companyId ? eq(bindings.companyId, scope.companyId) : undefined, + scope?.bindingId ? eq(bindings.id, scope.bindingId) : undefined, + scope?.generation !== undefined ? eq(bindings.generation, scope.generation) : undefined, + scope?.operatorId ? eq(bindings.operatorId, scope.operatorId) : undefined, + scope?.agentId ? eq(bindings.agentId, scope.agentId) : undefined, + after ? gt(bindings.id, after) : undefined, + )).orderBy(asc(bindings.id)).limit(100); +} + +export async function notifyMuseContactVerification( + tx: Pick, + scope?: MuseVerificationScope, + trigger: "authority" | "contact-persist" = "authority", +): Promise { + if (trigger === "contact-persist" && (!scope?.bindingId || scope.generation === undefined)) { + throw new Error("Muse contact intent requires its exact binding generation"); + } + let after: string | undefined; + let count = 0; + while (true) { + const rows = await pendingMuseContactVerifications(tx, scope, after, trigger === "contact-persist"); + for (const { binding } of rows) { + const hint: MuseVerificationHint = { + kind: "muse-contact-verification", companyId: binding.companyId, + bindingId: binding.id, generation: binding.generation, + }; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, JSON.stringify(hint)); + count++; + } + if (rows.length < 100) return count; + after = rows.at(-1)!.binding.id; + } +} + +export function parseMuseVerificationHint(raw: string): MuseVerificationHint | undefined { + let value: unknown; + try { value = JSON.parse(raw); } catch { return; } + if (!value || typeof value !== "object" || Array.isArray(value) + || !("kind" in value) || value.kind !== "muse-contact-verification" + || !("companyId" in value) || typeof value.companyId !== "string" + || !("bindingId" in value) || typeof value.bindingId !== "string" + || !("generation" in value) || typeof value.generation !== "number" + || !Number.isSafeInteger(value.generation) || value.generation < 1) return; + return { kind: "muse-contact-verification", companyId: value.companyId, + bindingId: value.bindingId, generation: value.generation }; +} diff --git a/server/src/modules/external-agents/muse.ts b/server/src/modules/external-agents/muse.ts index c4d7fddfff2..58875a6000e 100644 --- a/server/src/modules/external-agents/muse.ts +++ b/server/src/modules/external-agents/muse.ts @@ -15,8 +15,7 @@ import { conflict, forbidden, notFound, unprocessable } from "../../errors.js"; import { museIdentity, museCredentialHash, assertMuseCurrentAuthority, assertMuseSubjectCredential, inspectMuseConnectionCredential, assertMuseCleanupCredential, museCleanupBoundaryIds, assertMuseDisabledStopControlCredential, museStopControlBoundaryIds } from "../../services/muse-identity.js"; import { instanceSettingsService } from "../../services/instance-settings.js"; import { updateAgentConfigurationInTransaction } from "../../services/agent-configuration-transaction.js"; -import { AgentLifecycleConflict, canConfigureAgentConnection } from "../agent-lifecycle/index.js"; -import { createAgentLifecycle } from "../../services/agent-lifecycle.js"; +import { canConfigureAgentConnection, requestAgentLifecycleRecheckInTransaction } from "../agent-lifecycle/index.js"; import { authorizationService, issueReadSqlCondition, type AuthorizationActor } from "../../services/authorization.js"; import { logActivity, publishActivity, type ActivityPublication } from "../../services/activity-log.js"; import { issueService } from "../../services/issues.js"; @@ -387,23 +386,17 @@ function createMuseBroker(db:Db, publicOrigin?:string) { if(!b.receiverContactAt)throw conflict("No persisted receiver contact yet."); const publications: ActivityPublication[] = []; await db.transaction(async tx => { + // Match reconnect's agent-before-binding lock order. Both readiness + // and its durable recheck commit together, before any callback runs. + await requestAgentLifecycleRecheckInTransaction(tx, + { companyId: b.companyId, agentId: b.agentId }, + () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle)); + await assertMuseCurrentAuthority(tx, b, true); const [ready]=await tx.update(bindings).set({status:"ready",verifiedReplyAt:new Date(),challengeHash:null,challengeExpiresAt:null,revision:b.revision+1,updatedAt:new Date()}).where(and(eq(bindings.id,b.id),eq(bindings.generation,b.generation),eq(bindings.challengeHash,museCredentialHash(c.nonce)),gt(bindings.challengeExpiresAt,new Date()),isNull(bindings.revokedAt))).returning(); if(!ready)throw conflict("Challenge expired or does not match."); await logActivity(tx as unknown as Db,{companyId:b.companyId,actorType:"agent",actorId:b.agentId,agentId:b.agentId,responsibleUserIdOverride:b.operatorId,action:"muse.background_reply_verified",entityType:"agent",entityId:b.agentId,details:{bindingId:b.id,generation:b.generation,authorizingUserId:b.operatorId,requestId:c.requestId}},publications); }); publications.forEach(publishActivity); - // The completed challenge changes this hire's verification facts. Clear - // its deferred retry durably; a wake alone cannot bypass that deadline. - const lifecycle=createAgentLifecycle(db),agent=await lifecycle.get(b.agentId); - if(agent&&["preparing","verifying"].includes(agent.lifecycleState)) { - try { await lifecycle.retry(b.agentId); } - catch(error) { - // A concurrent lifecycle completion or Stop owns the newer phase. - // Confirmation remains accepted; it must not restart that phase. - const current=error instanceof AgentLifecycleConflict?await lifecycle.get(b.agentId):null; - if(!current||["preparing","verifying"].includes(current.lifecycleState))throw error; - } - } return {status:"ready",bindingId:b.id,generation:b.generation}; } if(c.command==="task.create"||c.command==="task.comment")return idleMutation(s,c); diff --git a/server/src/services/access.ts b/server/src/services/access.ts index bcc1a719b71..a010fe13f72 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -560,6 +561,10 @@ export function accessService(db: Db) { data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole; const nextStatus = data.status ?? existing.status; + if(nextStatus==="active"&&(existing.principalType==="agent"||nextMembershipRole!=="viewer") + &&(existing.status!=="active"||existing.principalType==="user"&&existing.membershipRole==="viewer")) + await notifyMuseContactVerification(tx,{companyId,...(existing.principalType==="user"?{operatorId:existing.principalId}:{agentId:existing.principalId})}); + if (existing.principalType === "user" && nextStatus === "active" && nextMembershipRole !== "viewer" && (existing.status !== "active" || existing.membershipRole === "viewer")) { await notifyChatCompanyAuthorityChange(tx, { companyId }); @@ -903,7 +908,10 @@ export function accessService(db: Db) { const existingMembership = existingByCompany.get(companyId); if (existingMembership) { if (existingMembership.status !== "active") { - if (existingMembership.membershipRole !== "viewer") await notifyChatCompanyAuthorityChange(tx, { companyId }); + if (existingMembership.membershipRole !== "viewer") { + await notifyChatCompanyAuthorityChange(tx, { companyId }); + await notifyMuseContactVerification(tx,{companyId,operatorId:userId}); + } await tx .update(companyMemberships) .set({ @@ -916,6 +924,7 @@ export function accessService(db: Db) { continue; } await notifyChatCompanyAuthorityChange(tx, { companyId }); + await notifyMuseContactVerification(tx,{companyId,operatorId:userId}); await tx.insert(companyMemberships).values({ companyId, principalType: "user", @@ -943,6 +952,9 @@ export function accessService(db: Db) { )).for("update").then(rows => rows[0] ?? null); if (existing) { if (existing.status !== status || existing.membershipRole !== membershipRole) { + if(status==="active"&&(principalType==="agent"||membershipRole!=="viewer") + &&(existing.status!=="active"||principalType==="user"&&existing.membershipRole==="viewer")) + await notifyMuseContactVerification(tx,{companyId,...(principalType==="user"?{operatorId:principalId}:{agentId:principalId})}); if (principalType === "user" && status === "active" && membershipRole !== "viewer" && (existing.status !== "active" || existing.membershipRole === "viewer")) { await notifyChatCompanyAuthorityChange(tx, { companyId }); @@ -958,6 +970,7 @@ export function accessService(db: Db) { return existing; } + if(status==="active"&&(principalType==="agent"||membershipRole!=="viewer"))await notifyMuseContactVerification(tx,{companyId,...(principalType==="user"?{operatorId:principalId}:{agentId:principalId})}); if (principalType === "user" && status === "active" && membershipRole !== "viewer") { await notifyChatCompanyAuthorityChange(tx, { companyId }); } @@ -1153,6 +1166,10 @@ export function accessService(db: Db) { data.membershipRole !== undefined ? data.membershipRole : existing.membershipRole; const nextStatus = data.status ?? existing.status; + if(nextStatus==="active"&&(existing.principalType==="agent"||nextMembershipRole!=="viewer") + &&(existing.status!=="active"||existing.principalType==="user"&&existing.membershipRole==="viewer")) + await notifyMuseContactVerification(tx,{companyId,...(existing.principalType==="user"?{operatorId:existing.principalId}:{agentId:existing.principalId})}); + if (existing.principalType === "user" && nextStatus === "active" && nextMembershipRole !== "viewer" && (existing.status !== "active" || existing.membershipRole === "viewer")) { await notifyChatCompanyAuthorityChange(tx, { companyId }); diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index c46abea410c..43feffbd4fa 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { assertEphemeralAgentInstructions } from "./agent-instructions.js"; import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; @@ -12,7 +13,7 @@ import { normalizeAgentUrlKey } from "@paperclipai/shared"; import { normalizePaperclipRunnerAdapterConfig } from "@paperclipai/adapter-utils/server-utils"; import { conflict, notFound, unprocessable } from "../errors.js"; import { agentRecordQueries, type UpdateAgentOptions, type AgentConfigurationPatch, type AgentConfigurationRecord, isPlainRecord, buildConfigSnapshot, - hasConfigPatchFields, changedPendingApprovalConfigFields, diffConfigSnapshot } from "../lib/agent-records.js"; + hasConfigPatchFields, changedPendingApprovalConfigFields, diffConfigSnapshot, jsonEqual } from "../lib/agent-records.js"; import { normalizeAgentPermissions } from "../lib/agent-permissions.js"; import type { ActivityPublication } from "../types/activity-publication.js"; import { secretService, assertClaudeOAuthBindingInvariant } from "./secrets.js"; @@ -89,6 +90,9 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura existing.adapterConfig, ); } + if(normalizedPatch.adapterType!==undefined&&normalizedPatch.adapterType!==existing.adapterType + ||normalizedPatch.adapterConfig!==undefined&&!jsonEqual(normalizedPatch.adapterConfig,existing.adapterConfig)) + await notifyMuseContactVerification(db,{companyId:existing.companyId,agentId:existing.id}); await notifyStrandedAgentStatus(db, { companyId: existing.companyId, agentId: existing.id }); await notifyRunDispatch(db, { companyId: existing.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: existing.companyId }); diff --git a/server/src/services/agent-harness-verification.ts b/server/src/services/agent-harness-verification.ts index 6f011562275..4dd8b43e1b0 100644 --- a/server/src/services/agent-harness-verification.ts +++ b/server/src/services/agent-harness-verification.ts @@ -46,7 +46,7 @@ export function agentHarnessVerificationService(db: Db, manager: PluginWorkerMan const tests = agentEnvironmentTestService(db, manager); const secrets = secretService(db); const environments = environmentService(db); - async function verify(snapshot: LifecycleAgent): Promise<"complete" | "pending"> { + async function verify(snapshot: LifecycleAgent): Promise<"complete" | "pending" | "waiting"> { const [agent] = await db.select().from(agents).where(and(eq(agents.id, snapshot.id), eq(agents.lifecycleVersion, snapshot.lifecycleVersion))); if (!agent) return "pending"; if (agent.adapterType === "paperclip_runner" && ["openai_dot", "grok_bot"].includes(String(agent.adapterConfig.provider))) { @@ -55,7 +55,9 @@ export function agentHarnessVerificationService(db: Db, manager: PluginWorkerMan } if (agent.adapterType === "paperclip_runner" && agent.adapterConfig.provider === "muse") { const binding = await museRunnerBroker(db).bindingForAgent(agent.companyId,agent.id); - return binding?.status === "ready" && binding.paired && binding.receiverDetected && binding.backgroundReplyVerified && binding.id === agent.adapterConfig.museBindingId ? "complete" : "pending"; + // A successful read of these external facts has no retry deadline. The + // real confirmation/configuration producer rechecks this phase durably. + return binding?.status === "ready" && binding.paired && binding.receiverDetected && binding.backgroundReplyVerified && binding.id === agent.adapterConfig.museBindingId ? "complete" : "waiting"; } const providerApi = resolveProviderApiPlacement({ agent }); if (providerApi) assertProviderApiPlacementPolicy(await instanceSettingsService(db).getGeneral()); diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 74549a387b5..53c9081e247 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; @@ -49,6 +50,9 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li return plugin !== null && !selectedPluginIds?.includes(plugin.id); }, notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), + beforeConnectionEligibilityRestore: async (tx, companyId, agentId) => { + await notifyMuseContactVerification(tx, { companyId, agentId }); + }, async beforeStatusChange(tx, companyId, agentId, before, after) { await notifyAttentionRetentionWork(tx, { companyId }); if (before !== after) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 57a55550021..1c1762b1736 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; @@ -475,6 +476,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; + await notifyMuseContactVerification(db,{companyId:policy.companyId}); await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index e22e35b6790..768525e50aa 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; @@ -342,6 +343,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch); if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { + if(willReactivate)await notifyMuseContactVerification(tx,{companyId:id}); await notifyRunDispatch(tx, { companyId: id }); await notifyGitHubCompanyChange(tx, id); await notifyDeferredExecutionEligibility(tx, { companyId: id }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 7a3b576db7f..1719878f28a 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -5,7 +5,7 @@ import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyStatusCardEvaluation } from "./status-card-notifications.js"; import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; -import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; +import { notifyMuseContactVerification, notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -658,6 +658,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + if (toExperimentalView(current.experimental).enableMuse !== toExperimentalView(nextExperimental).enableMuse || toExperimentalView(current.experimental).enableNativeRunner !== toExperimentalView(nextExperimental).enableNativeRunner) await notifyMuseContactVerification(tx); if (patch.enableExternalObjects !== undefined) await notifyExternalObjectRefresh(tx); if (chatBecameEnabled) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAssignments, "*"); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); diff --git a/server/src/services/muse-maintenance.test.ts b/server/src/services/muse-maintenance.test.ts index bb7477ab089..b58b3f282ed 100644 --- a/server/src/services/muse-maintenance.test.ts +++ b/server/src/services/muse-maintenance.test.ts @@ -1,8 +1,9 @@ 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, applyPendingMigrations, heartbeatRuns, museRunnerAssignments, museAgentBindings } from "@paperclipai/db"; +import { agents, companies, createDb, applyPendingMigrations, heartbeatRuns, museRunnerAssignments, museAgentBindings, subscribeDatabaseWork } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { notifyMuseContactVerification, parseMuseVerificationHint } from "../modules/external-agents/index.js"; import { startMuseMaintenance } from "./muse-maintenance.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -84,6 +85,74 @@ describeDatabase("Muse commit and deadline maintenance", () => { }); } + it("registers first-contact intent before its write, including a currently paused identity", async () => { + const binding = await fixture(); + await db.update(agents).set({ status: "paused", lifecycleState: "paused" }).where(eq(agents.id, binding.agentId)); + await db.update(museAgentBindings).set({ pairedAt: new Date() }).where(eq(museAgentBindings.id, binding.id)); + const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.museMaintenance, (event, values) => { + if (event === "intent") hints.push(...values ?? []); + }); + try { + await expect(db.transaction(async tx => { + expect(await notifyMuseContactVerification(tx, { + companyId: binding.companyId, bindingId: binding.id, generation: binding.generation, + }, "contact-persist")).toBe(1); + expect(hints.map(parseMuseVerificationHint)).toEqual([{ + kind: "muse-contact-verification", companyId: binding.companyId, + bindingId: binding.id, generation: binding.generation, + }]); + const [before] = await tx.select().from(museAgentBindings).where(eq(museAgentBindings.id, binding.id)); + expect(before.receiverContactAt).toBeNull(); + await tx.update(museAgentBindings).set({ receiverContactAt: new Date() }).where(eq(museAgentBindings.id, binding.id)); + throw new Error("first contact rollback"); + })).rejects.toThrow("first contact rollback"); + const [after] = await db.select().from(museAgentBindings).where(eq(museAgentBindings.id, binding.id)); + expect(after.receiverContactAt).toBeNull(); + expect(after.challengeHash).toBeNull(); + } finally { unsubscribe(); } + }); + + it("targets only pending contact bindings for the affected company and operator", async () => { + const binding = await fixture(); + const foreign = await fixture(); + const otherOperator = await fixture(); + for (const pending of [binding, foreign, otherOperator]) { + await db.update(museAgentBindings).set({ pairedAt: new Date(), receiverContactAt: new Date() }) + .where(eq(museAgentBindings.id, pending.id)); + } + const [otherAgent] = await db.insert(agents).values({ companyId: binding.companyId, name: "Other Muse agent" }).returning(); + await db.update(museAgentBindings).set({ companyId: binding.companyId, agentId: otherAgent.id, + operatorId: "another-operator" }).where(eq(museAgentBindings.id, otherOperator.id)); + const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.museMaintenance, (event, values) => { + if (event === "intent") hints.push(...values ?? []); + }); + try { + await db.transaction(async tx => { + expect(await notifyMuseContactVerification(tx, { + companyId: binding.companyId, operatorId: binding.operatorId, + })).toBe(1); + }); + expect(hints.map(parseMuseVerificationHint)).toEqual([{ + kind: "muse-contact-verification", companyId: binding.companyId, + bindingId: binding.id, generation: binding.generation, + }]); + expect(hints.some(hint => hint.includes(foreign.id) || hint.includes(otherOperator.id))).toBe(false); + await db.transaction(async tx => { + expect(await notifyMuseContactVerification(tx, { companyId: binding.companyId, + bindingId: binding.id, generation: binding.generation + 1 })).toBe(0); + }); + expect(hints).toHaveLength(1); + await db.update(museAgentBindings).set({ verifiedReplyAt: new Date() }).where(eq(museAgentBindings.id, binding.id)); + await db.transaction(async tx => { + expect(await notifyMuseContactVerification(tx, { companyId: binding.companyId, + operatorId: binding.operatorId })).toBe(0); + }); + expect(hints).toHaveLength(1); + } finally { unsubscribe(); } + }); + it("stays SQL-quiet after startup with no bindings, including during drain", async () => { await start(); const baseline = [...calls]; diff --git a/server/src/services/muse-maintenance.ts b/server/src/services/muse-maintenance.ts index 9dc622d1463..c9352689a98 100644 --- a/server/src/services/muse-maintenance.ts +++ b/server/src/services/muse-maintenance.ts @@ -1,4 +1,7 @@ -import { pendingMuseTerminalBindings, type ExternalAgentSettlementScope } from "../modules/external-agents/index.js"; +import { HttpError } from "../errors.js"; +import { museIdentity, inspectMuseConnectionCredential } from "./muse-identity.js"; +import { canConfigureAgentConnection } from "../modules/agent-lifecycle/index.js"; +import { pendingMuseContactVerifications, parseMuseVerificationHint, type MuseVerificationScope, pendingMuseTerminalBindings, type ExternalAgentSettlementScope } from "../modules/external-agents/index.js"; 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"; @@ -35,6 +38,7 @@ export function startMuseMaintenance( startupMarked = true; } let failed = false; + let verificationEnabled: boolean | undefined; let after: string | undefined; if (global) { while (!signal.aborted) { @@ -87,6 +91,36 @@ export function startMuseMaintenance( after = active.at(-1)!.id; } } + for (const hint of global ? ["*"] : hints) { + const scope: MuseVerificationScope | undefined = hint === "*" ? undefined : parseMuseVerificationHint(hint); + if (hint !== "*" && !scope) continue; + let after: string | undefined; + while (!signal.aborted) { + const rows = await pendingMuseContactVerifications(db, scope, after); + for (const { binding: b, agent } of rows) { + if (signal.aborted) throw new Error("Muse verification recovery stopped"); + if (!canConfigureAgentConnection(agent)) continue; + verificationEnabled ??= await museIdentity(db).enabled(); + if (!verificationEnabled) break; + if (!settlementAllowed) { requested.add(hint); continue; } + try { + const credential = await inspectMuseConnectionCredential(db, b); + if (credential.state !== "active" && credential.state !== "pending") continue; + await broker.verify(b.companyId, b.agentId, { + bindingId: b.id, generation: b.generation, expectedRevision: b.revision, + }); + } catch (error) { + if (!(error instanceof HttpError && [401, 403, 404, 410].includes(error.status))) { + failed = true; + requested.add(hint); + } + } + } + if (rows.length < 100 || verificationEnabled === false) break; + after = rows.at(-1)!.binding.id; + } + if (verificationEnabled === false) break; + } if (settlementAllowed) { for (const hint of global ? ["*"] : hints) { const scope: ExternalAgentSettlementScope | undefined = hint === "*" ? undefined : JSON.parse(hint); diff --git a/server/src/services/muse-receiver.ts b/server/src/services/muse-receiver.ts index d014a172814..e56a79c7120 100644 --- a/server/src/services/muse-receiver.ts +++ b/server/src/services/muse-receiver.ts @@ -1,3 +1,4 @@ +import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; import { museAgentBindings as bindings, museMailboxItems as mailbox, museRunnerAssignments as assignments, museReceiverContactBuckets as buckets, externalAgentHolds as holds, type Db } from "@paperclipai/db"; @@ -18,11 +19,13 @@ function createReceiver(db:Db) { const batch=[...pending.entries()];pending.clear(); for(const [key,bucket]of batch) { try { - await db.transaction(async tx=>{ + const current=await db.transaction(async tx=>{ + const [current]=await tx.select().from(bindings).where(and(eq(bindings.id,bucket.bindingId),eq(bindings.generation,bucket.bindingGeneration),isNull(bindings.revokedAt))); + if(current?.pairedAt&&!current.verifiedReplyAt&&!current.challengeHash)await notifyMuseContactVerification(tx,{companyId:current.companyId,bindingId:current.id,generation:current.generation},"contact-persist"); await tx.insert(buckets).values({...bucket,replicaId}).onConflictDoUpdate({target:[buckets.bindingId,buckets.bindingGeneration,buckets.replicaId,buckets.bucketAt],set:{lastAt:bucket.lastAt,contacts:sql`${buckets.contacts}+${bucket.contacts}`,gapsAtMostSevenSeconds:sql`${buckets.gapsAtMostSevenSeconds}+${bucket.gapsAtMostSevenSeconds}`,maxGapMs:sql`greatest(${buckets.maxGapMs},${bucket.maxGapMs})`,timestamps:sql`${buckets.timestamps} || ${JSON.stringify(bucket.timestamps)}::jsonb`,incomplete:sql`${buckets.incomplete} or ${bucket.incomplete}`}}); await tx.update(bindings).set({receiverContactAt:sql`greatest(${bindings.receiverContactAt},${bucket.lastAt.toISOString()}::timestamptz)`}).where(and(eq(bindings.id,bucket.bindingId),eq(bindings.generation,bucket.bindingGeneration))); + return current; }); - const [current]=await db.select().from(bindings).where(and(eq(bindings.id,bucket.bindingId),eq(bindings.generation,bucket.bindingGeneration),isNull(bindings.revokedAt))); if(current?.pairedAt&&!current.verifiedReplyAt&&!current.challengeHash) { const {museRunnerBroker}=await import("./muse-runner-broker.js"); await museRunnerBroker(db).verify(current.companyId,current.agentId,{bindingId:current.id,generation:current.generation,expectedRevision:current.revision}).catch(()=>{});