diff --git a/server/src/modules/external-agents/broker-feature-owner.postgres.test.ts b/server/src/modules/external-agents/broker-feature-owner.postgres.test.ts new file mode 100644 index 00000000000..b7c08cac439 --- /dev/null +++ b/server/src/modules/external-agents/broker-feature-owner.postgres.test.ts @@ -0,0 +1,117 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { + agents, applyPendingMigrations, authUsers, companies, companyMemberships, createDb, + dotAgentBindings, dotRunnerAssignments, heartbeatRuns, issues, museAgentBindings, + museRunnerAssignments, type Db, +} from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "../../lib/dedicated-test-database-url.js"; +import { accessService } from "../../services/access.js"; +import * as settings from "../../services/instance-settings.js"; +import * as admission from "./admission.js"; +import { dotRunnerBroker } from "./dot.js"; +import { museExternalAgents } from "./muse.js"; + +const url = resolveExternalTestDatabaseUrl("PAPERCLIP_BROKER_FEATURE_OWNER_TEST_DATABASE_URL"); +const providers = ["openai_dot", "grok_bot", "muse"] as const; + +describe.skipIf(!url)("external broker forwards its admission owner to feature policy", () => { + let db: Db; + beforeAll(async () => { await applyPendingMigrations(url!); }, 90_000); + beforeEach(() => { + db = createDb(url!, { maxConnections: 2, applicationName: "broker-feature-owner" }); + }); + afterEach(async () => { + vi.restoreAllMocks(); + await db.$client.end({ timeout: 5 }); + }, 30_000); + + for (const provider of providers) it(`${provider} reads disabled policy on the exact dispatch transaction`, async () => { + await settings.instanceSettingsService(db).updateExperimental({ + enablePublicMcp: true, enableNativeRunner: true, + enableOpenAiDot: true, enableGrokBot: true, enableMuse: true, + }); + const operatorId = randomUUID(); + await db.insert(authUsers).values({ + id: operatorId, name: "Feature owner operator", email: `${operatorId}@example.test`, + createdAt: new Date(), updatedAt: new Date(), + }); + const [company] = await db.insert(companies).values({ + name: "Broker feature owner", issuePrefix: `BF${operatorId.replaceAll("-", "")}`, + }).returning(); + await db.insert(companyMemberships).values({ + companyId: company.id, principalType: "user", principalId: operatorId, + membershipRole: "owner", status: "active", + }); + await accessService(db).setPrincipalPermission(company.id, "user", operatorId, "agents:create", true, operatorId); + const [agent] = await db.insert(agents).values({ + companyId: company.id, name: provider, adapterType: "paperclip_runner", status: "active", + adapterConfig: { provider, allowUnmeteredProvider: true }, + }).returning(); + const broker = provider === "muse" + ? museExternalAgents(db, { publicOrigin: "https://paperclip.example" }) + : dotRunnerBroker(db); + const pairingInput = { companyId: company.id, agentId: agent.id, operatorId }; + const pairing = provider === "muse" + ? await museExternalAgents(db, { publicOrigin: "https://paperclip.example" }).createPairing(pairingInput) + : await dotRunnerBroker(db).createPairing({ ...pairingInput, provider }); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Denied broker dispatch" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, issueId: issue.id, nativeIssueId: issue.id, + status: "queued", runtimeMode: "legacy", + }).returning(); + await settings.instanceSettingsService(db).updateExperimental({ + enableOpenAiDot: provider !== "openai_dot", enableGrokBot: provider !== "grok_bot", + enableMuse: provider !== "muse", + }); + + const admitted = vi.spyOn(admission, "withExternalExecutionAdmission"); + const readOwners: Array | undefined> = []; + const readValues: boolean[] = []; + const originalSettings = settings.instanceSettingsService; + vi.spyOn(settings, "instanceSettingsService").mockImplementation((...args) => { + const service = originalSettings(...args); + return { + ...service, + getExperimentalReadOnly: async (owner?: Pick) => { + readOwners.push(owner); + const value = await service.getExperimentalReadOnly(owner); + readValues.push(provider === "openai_dot" ? value.enableOpenAiDot + : provider === "grok_bot" ? value.enableGrokBot : value.enableMuse); + return value; + }, + }; + }); + const snapshot = { + companyId: company.id, agentId: agent.id, bindingId: pairing.bindingId, bindingGeneration: 1, + acceptByUnixMs: Date.now() + 60_000, expiresAtUnixMs: Date.now() + 120_000, + }; + const binding = { companyId: company.id, agentId: agent.id, runId: run.id }; + await expect(broker.port({ binding, provider: { binding: snapshot } }).dispatch({ + sourceEventId: randomUUID(), + payload: { + kind: "assignment", binding: { ...binding, bindingId: snapshot.bindingId, bindingGeneration: snapshot.bindingGeneration, + normalizedSessionId: randomUUID(), turnId: "denied-feature", assignmentRevision: 1 }, + tools: [], acceptByUnixMs: snapshot.acceptByUnixMs, expiresAtUnixMs: snapshot.expiresAtUnixMs, + }, + })).rejects.toMatchObject({ status: 403, message: "External binding is unavailable for new effects." }); + expect(admitted).toHaveBeenCalledTimes(1); + expect(admitted.mock.calls[0].slice(1, 4)).toEqual([company.id, agent.id, run.id]); + expect(readOwners).toHaveLength(1); + expect(readOwners[0]).toBe(admitted.mock.calls[0][0]); + expect(readOwners[0]).not.toBe(db); + expect(readValues).toEqual([false]); + expect(await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.runId, run.id))).toEqual([]); + expect(await db.select().from(museRunnerAssignments).where(eq(museRunnerAssignments.runId, run.id))).toEqual([]); + const [unchanged] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(unchanged).toMatchObject({ status: "queued", runtimeMode: "legacy", nativeSessionId: null }); + if (provider === "muse") { + const [pending] = await db.select().from(museAgentBindings).where(eq(museAgentBindings.id, pairing.bindingId)); + expect(pending).toMatchObject({ grantId: null, revokedAt: null }); + } else { + const [pending] = await db.select().from(dotAgentBindings).where(eq(dotAgentBindings.id, pairing.bindingId)); + expect(pending).toMatchObject({ status: "pairing", grantId: null, revokedAt: null }); + } + }, 10_000); +}); diff --git a/server/src/modules/external-agents/dot.ts b/server/src/modules/external-agents/dot.ts index 6e60ace57dd..7aa744825e7 100644 --- a/server/src/modules/external-agents/dot.ts +++ b/server/src/modules/external-agents/dot.ts @@ -89,7 +89,7 @@ function createBroker(db: Db) { return { binding, assignment: a }; } - const nativeBroker = createNativeExternalBroker(db, "openai_dot", { enabled, authorize: authorizeAssignment }); + const nativeBroker = createNativeExternalBroker(db, "openai_dot", { enabled: (database, provider) => externalBotEnabled(db, provider ?? "openai_dot", database), authorize: authorizeAssignment }); return { enabled, diff --git a/server/src/modules/external-agents/feature-reader-owner.postgres.test.ts b/server/src/modules/external-agents/feature-reader-owner.postgres.test.ts new file mode 100644 index 00000000000..d9f545db7e3 --- /dev/null +++ b/server/src/modules/external-agents/feature-reader-owner.postgres.test.ts @@ -0,0 +1,100 @@ +import {appendFile} from "node:fs/promises"; +import {performance} from "node:perf_hooks"; +import {eq,sql} from "drizzle-orm"; +import {afterEach,beforeAll,beforeEach,describe,expect,it} from "vitest"; +import {applyPendingMigrations,createDb,instanceSettings,type Db} from "@paperclipai/db"; +import {resolveExternalTestDatabaseUrl} from "../../lib/dedicated-test-database-url.js"; +import {externalBotEnabled} from "../../services/external-bot-authority.js"; +import {instanceSettingsService} from "../../services/instance-settings.js"; +import {museIdentity} from "../../services/muse-identity.js"; + +const url=resolveExternalTestDatabaseUrl("PAPERCLIP_FEATURE_OWNER_TEST_DATABASE_URL"); +type QueryOwner=Pick; +type Provider="openai_dot"|"grok_bot"|"muse"; +const cases=[ + {provider:"openai_dot",enabled:true}, + {provider:"openai_dot",enabled:false}, + {provider:"grok_bot",enabled:true}, + {provider:"grok_bot",enabled:false}, + {provider:"muse",enabled:true}, + {provider:"muse",enabled:false}, +] satisfies Array<{provider:Provider;enabled:boolean}>; +const states=new Set(["57014","40P01","55P03","08006","25P02"]); +function sqlState(error:unknown){ + let current=error; + for(let depth=0;depth<3&¤t&&typeof current==="object";depth++){ + if("code"in current&&typeof current.code==="string"&&states.has(current.code))return current.code; + current="cause"in current?current.cause:undefined; + } + return "other"; +} + +describe.skipIf(!url)("existing external feature readers keep the admitted query owner",()=>{ + let db:Db; + beforeAll(async()=>{await applyPendingMigrations(url!);},90_000); + beforeEach(()=>{db=createDb(url!,{maxConnections:2,applicationName:"feature-owner-pool2"});}); + afterEach(async()=>{await db.$client.end({timeout:5});},30_000); + for(const item of cases)it(`reads ${item.provider} enabled=${item.enabled} with both pool2 slots owned`,async()=>{ + const artifact=process.env.PAPERCLIP_FEATURE_OWNER_TRACE; + if(!artifact)throw new Error("Explicit owned feature trace path required"); + const row=await instanceSettingsService(db).updateExperimental({ + enablePublicMcp:true,enableNativeRunner:true, + enableOpenAiDot:item.provider==="openai_dot"?item.enabled:true, + enableGrokBot:item.provider==="grok_bot"?item.enabled:true, + enableMuse:item.provider==="muse"?item.enabled:true, + }); + const started=performance.now(),events:Array<{phase:string;atMs:number}>=[]; + function mark(phase:string){ + if(events.length>=16)throw new Error("Feature owner trace overflow"); + events.push({phase,atMs:performance.now()-started}); + console.log(JSON.stringify({kind:"feature-owner-phase",provider:item.provider,enabled:item.enabled,phase})); + } + let competing:Promise|undefined,competingError:string|null=null,observed:boolean|undefined; + try{ + await db.transaction(async tx=>{ + mark("admission-transaction-entered"); + expect(await tx.select({id:instanceSettings.id}).from(instanceSettings).where(eq(instanceSettings.id,row.id)).for("update")).toHaveLength(1); + mark("admission-row-locked"); + let occupiedResolve!:()=>void,occupiedReject!:(error:unknown)=>void; + const occupied=new Promise((resolve,reject)=>{occupiedResolve=resolve;occupiedReject=reject;}); + mark("competing-connection-requested"); + competing=db.transaction(async other=>{ + mark("competing-transaction-entered"); + await other.execute(sql`set local statement_timeout = '2000ms'`); + mark("competing-timeout-installed"); + await other.execute(sql`select pg_backend_pid()`); + mark("second-connection-owned");occupiedResolve(); + mark("competing-row-lock-requested"); + expect(await other.select({id:instanceSettings.id}).from(instanceSettings).where(eq(instanceSettings.id,row.id)).for("update")).toHaveLength(1); + mark("competing-row-lock-acquired"); + }).then(()=>{mark("competing-committed");},error=>{ + competingError=sqlState(error);mark("competing-failed");occupiedReject(error); + }); + await occupied; + mark("feature-read-start"); + if(item.provider==="muse"){ + const read:(owner:QueryOwner)=>Promise=museIdentity(db).enabled; + observed=await read(tx); + }else{ + const read:(root:Db,provider:"openai_dot"|"grok_bot",owner:QueryOwner)=>Promise=externalBotEnabled; + observed=await read(db,item.provider,tx); + } + mark("feature-read-finished"); + }); + mark("admission-committed");await competing; + expect(competingError,JSON.stringify({events,competingError,observed})).toBeNull(); + expect(observed).toBe(item.enabled); + const names=events.map(event=>event.phase); + expect(names.indexOf("feature-read-finished")).toBeLessThan(names.indexOf("competing-row-lock-acquired")); + expect(names).toContain("competing-committed"); + expect(await instanceSettingsService(db).getExperimentalReadOnly()).toMatchObject({ + enableOpenAiDot:item.provider==="openai_dot"?item.enabled:true, + enableGrokBot:item.provider==="grok_bot"?item.enabled:true, + enableMuse:item.provider==="muse"?item.enabled:true, + }); + }finally{ + await competing; + await appendFile(artifact,JSON.stringify({kind:"feature-owner-result",provider:item.provider,enabled:item.enabled,events,competingError,observed})+"\n",{mode:0o600}); + } + },10_000); +}); diff --git a/server/src/modules/external-agents/muse.ts b/server/src/modules/external-agents/muse.ts index 247fe83d8c6..c4d7fddfff2 100644 --- a/server/src/modules/external-agents/muse.ts +++ b/server/src/modules/external-agents/muse.ts @@ -133,7 +133,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { await logActivity(tx as unknown as Db,{companyId:b.companyId,actorType,actorId:operatorId,action:"muse.binding_revoked",entityType:"agent",entityId:b.agentId,details:{bindingId:b.id,generation:b.generation,authorizingUserId:b.operatorId,source:reason,externalStopConfirmed:false}}); return runs; } - const nativeBroker=createNativeExternalBroker(db,"muse",{enabled:identity.enabled,expectedResource:connectionResource, + const nativeBroker=createNativeExternalBroker(db,"muse",{enabled:database=>identity.enabled(database),expectedResource:connectionResource, assertSubjectCredential:async(tx,s)=>{await assertMuseSubjectCredential(tx,s,s.credentialSource==="oauth"?connectionResource():undefined);}, authorize:async(s,id,settled)=>{const {b,a}=await authorizeAssignment(s,id,settled);return {binding:b,assignment:a};}}); const operation=nativeBroker.operation; diff --git a/server/src/modules/external-agents/native-broker.ts b/server/src/modules/external-agents/native-broker.ts index 4ccbcda72b3..f6aa485491e 100644 --- a/server/src/modules/external-agents/native-broker.ts +++ b/server/src/modules/external-agents/native-broker.ts @@ -31,7 +31,7 @@ interface Receipt extends Record {id:string;companyId:string;ass interface Authority {binding:{id:string;companyId:string;agentId:string;generation:number};assignment:Pick & {expiresAt:Date;createdAt:Date;claimedAt?:Date|null}} interface Identity { authorize(subject:S,assignmentId:string,settled?:boolean):Promise; - enabled(provider?:"openai_dot"|"grok_bot"):Promise; + enabled(database:Pick,provider?:"openai_dot"|"grok_bot"):Promise; expectedResource?():string; assertSubjectCredential?(tx:ExternalAdmissionTransaction,subject:S):Promise; } @@ -60,7 +60,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I async function assertCurrentAuthority(database:Database,b:Binding,allowFence=false) { if(allowFence)return; const actualProvider=provider==="muse"?"muse":b.providerKind; - if(!actualProvider||!["openai_dot","grok_bot","muse"].includes(actualProvider)||!await identity.enabled(actualProvider==="muse"?undefined:actualProvider)||b.status!=="ready"||b.revokedAt)throw forbidden("External binding is unavailable for new effects."); + if(!actualProvider||!["openai_dot","grok_bot","muse"].includes(actualProvider)||!await identity.enabled(database,actualProvider==="muse"?undefined:actualProvider)||b.status!=="ready"||b.revokedAt)throw forbidden("External binding is unavailable for new effects."); const [agent]=await database.select().from(agents).where(and(eq(agents.id,b.agentId),eq(agents.companyId,b.companyId))); const [company]=await database.select().from(companies).where(eq(companies.id,b.companyId)); const [user]=await database.select({id:authUsers.id}).from(authUsers).where(eq(authUsers.id,b.operatorId)); diff --git a/server/src/services/external-bot-authority.ts b/server/src/services/external-bot-authority.ts index 9f7bfbd4724..79d1000818f 100644 --- a/server/src/services/external-bot-authority.ts +++ b/server/src/services/external-bot-authority.ts @@ -6,8 +6,11 @@ import { boardAuthService } from "./board-auth.js"; import { instanceSettingsService } from "./instance-settings.js"; import { McpOAuthError, type McpPrincipal } from "./public-mcp/oauth.js"; -export async function externalBotEnabled(db: Db, provider: "openai_dot" | "grok_bot") { - const experimental = await instanceSettingsService(db).getExperimental(); +export async function externalBotEnabled(db: Db, provider: "openai_dot" | "grok_bot", queryOwner?: Pick) { + const settings = instanceSettingsService(db); + const experimental = queryOwner + ? await settings.getExperimentalReadOnly(queryOwner) + : await settings.getExperimental(); return experimental.enablePublicMcp && (provider === "openai_dot" ? experimental.enableOpenAiDot : experimental.enableGrokBot && experimental.enableNativeRunner); } diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 73d9d9645b6..24356a502ca 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -583,6 +583,12 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService return toGeneralView(row.general); }, + getExperimentalReadOnly: async (runner: Pick = db): Promise => { + const [row] = await runner.select({ experimental: instanceSettings.experimental }).from(instanceSettings) + .where(eq(instanceSettings.singletonKey, DEFAULT_SINGLETON_KEY)); + return toExperimentalView(row?.experimental); + }, + getExperimental: async (): Promise => { const row = await getOrCreateRow(); return toExperimentalView(row.experimental); diff --git a/server/src/services/muse-identity.ts b/server/src/services/muse-identity.ts index 5e991e96b2c..b1955c9057f 100644 --- a/server/src/services/muse-identity.ts +++ b/server/src/services/muse-identity.ts @@ -156,7 +156,11 @@ export async function assertMuseCleanupCredential(db:MuseIdentityDatabase,s:Agen } export function museIdentity(db: Db) { - async function enabled() { const s=await instanceSettingsService(db).getExperimental(); return s.enableMuse && s.enableNativeRunner; } + async function enabled(queryOwner?: Pick) { + const settings = instanceSettingsService(db); + const s = queryOwner ? await settings.getExperimentalReadOnly(queryOwner) : await settings.getExperimental(); + return s.enableMuse && s.enableNativeRunner; + } async function issue(tx: ExternalAdmissionTransaction, b: typeof bindings.$inferSelect, familyId: string=randomUUID()): Promise { const accessToken=token(), refreshToken=token(), cleanupToken=token(), detectorCleanupToken=token(); const accessExpiresAt=new Date(Date.now()+MUSE_ACCESS_TTL_MS), refreshExpiresAt=new Date(Date.now()+MUSE_REFRESH_INACTIVITY_MS);