perf: batch retained-reader authorization reads (#16424)

Batch retained-reader project checks through the existing multi-actor read API. Share raw database reads within one call, retain actor-specific privacy and supplied membership data, and avoid loading large saved-run payloads for identity checks.

Keep current-reader writes, distinct saved-run policies, company root fences and later-call revocation checks. Add query-count and access-control regressions. All 255 affected cases and full repository typecheck/build pass. Hosted response-time acceptance remains separate.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-15-retained-reader-query-batches.md

@@ -0,0 +1,42 @@
+# Bound retained-reader authorization database work
+
+## Measured problem
+
+A synthetic local service fixture uses 383 workspace paths and 64 retained runs
+with the same agent and responsible user. It exercises the production publication
+guard against disposable PostgreSQL. It does not represent a live tenant or a
+Product E2E evaluation.
+
+The original guard made 1,431 queries. A declared 4ms round-trip delay in a
+loopback proxy increased one admission to 15.19 seconds and three concurrent
+admissions to 54.69 seconds. The company root fence serializes those admissions.
+With this change, one admission makes 680 queries and takes 7.13 seconds under
+the same proxy; three take 20.93 seconds. The zero-injected-delay samples took
+359ms before and 146ms after. Timing samples are controlled local evidence, not
+an agent-response or hosted latency guarantee.
+
+## Change boundary
+
+Retained readers still need individual saved-run identity and execution-policy
+checks. Current readers still need the existing write checks. Existing company
+root fences, row locks, source privacy and publication ordering stay in place.
+
+Batch retained-reader project reads through the existing multi-actor read API.
+Share only database read promises within one call. Keep actor-supplied membership
+snapshots and per-actor privacy results separate. Key raw reads by their company,
+principal and run as applicable. Each later call reads current grants again.
+Select only the run identity and saved policy fields needed by the guard,
+including both native and legacy issue-binding fallbacks.
+
+## Verification and remaining proof
+
+The query-count regression must fail on the original code without changing its
+800-query bound. Distinct reader identities and stored policies must still fail
+closed. Grant revocation must affect the next call. Supplied membership snapshots
+must match individual decisions in both actor orders without leaking into database
+snapshots. Run the complete affected authorization and source-trust suites.
+
+Full repository typecheck and build plus fresh current-head CI and review precede
+merge. A real hosted warm agent reply remains the acceptance test for the separate
+under-30-second objective. Neither this service benchmark nor a fresh local
+provider run establishes that hosted result.

server/src/__tests__/repository-root-reader-queries.test.ts

@@ -0,0 +1,115 @@
+import { createHash, randomUUID } from "node:crypto";
+import { mkdtemp, realpath, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { and, eq, inArray } from "drizzle-orm";
+import { drizzle } from "drizzle-orm/postgres-js";
+import { afterAll, beforeAll, expect, it, vi } from "vitest";
+import { agents, authUsers, companies, companyMemberships, createDb, executionWorkspaceRepositoryAccess,
+  executionWorkspaces, heartbeatRuns, issues, projects, projectAccessMembers, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
+import { executionWorkspaceRepositoryService } from "../services/execution-workspace-repositories.js";
+import { authorizationService, type AuthorizationActor } from "../services/authorization.js";
+import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+let db: ReturnType<typeof createDb>;
+let cwd: string;
+const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), issueId = randomUUID(), projectId = randomUUID();
+const runIds = Array.from({ length: 64 }, () => randomUUID());
+let workspaceId: string;
+
+beforeAll(async () => {
+  vi.stubEnv("PAPERCLIP_ISSUE_PRIVACY_MODE", "enforce");
+  database = await startEmbeddedPostgresTestDatabase("root-reader-query-");
+  db = createDb(database.connectionString);
+  cwd = await realpath(await mkdtemp(path.join(tmpdir(), "root-reader-query-")));
+  await db.insert(companies).values({ id: companyId, name: "Retained readers", issuePrefix: "RRQ" });
+  await db.insert(authUsers).values({ id: userId, name: "Sponsor", email: `${userId}@example.test`, emailVerified: true, createdAt: new Date(), updatedAt: new Date() });
+  await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "process" });
+  await db.insert(companyMemberships).values([
+    { companyId, principalType: "user", principalId: userId, membershipRole: "operator", status: "active" },
+    { companyId, principalType: "agent", principalId: agentId, membershipRole: "member", status: "active" },
+  ]);
+  await db.insert(projects).values({ id: projectId, companyId, name: "Source" });
+  const [workspace] = await db.insert(executionWorkspaces).values({ companyId, projectId, cwd, name: "Task source", mode: "shared_workspace", strategyType: "operator_directory" }).returning();
+  workspaceId = workspace.id;
+  await db.insert(issues).values({ id: issueId, companyId, projectId, title: "Task", assigneeAgentId: agentId, executionWorkspaceId: workspaceId });
+  await db.insert(heartbeatRuns).values(runIds.map(id => ({ id, companyId, agentId, responsibleUserId: userId,
+    status: "succeeded", contextSnapshot: { issueId }, resultJson: { unusedPayload: "x".repeat(65536) } })));
+  const rootKey = createHash("sha256").update(cwd).digest("hex");
+  await db.insert(executionWorkspaceRepositoryAccess).values(runIds.map(scopeId => ({ companyId, rootKey,
+    canonicalCwd: cwd, kind: "reader", scopeId, issueId, agentId, responsibleUserId: userId })));
+}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+
+afterAll(async () => {
+  vi.unstubAllEnvs();
+  await db?.$client.end({ timeout: 0 });
+  await database?.cleanup();
+  if (cwd) await rm(cwd, { recursive: true, force: true });
+});
+
+const input = () => ({ companyId, issueId, workspaceId, cwd, agentId, runId: runIds[0] });
+
+it("bounds repeated-principal queries without loading retained run payloads", async () => {
+  const queries: string[] = [];
+  const recorded = drizzle(db.$client, { schema: db._.fullSchema, logger: { logQuery: query => queries.push(query) } }) as unknown as typeof db;
+  await executionWorkspaceRepositoryService(recorded).guardForPublication(input());
+  expect(queries.length).toBeLessThan(800);
+  const runs = queries.filter(query => query.includes('from "heartbeat_runs"'));
+  expect(runs.length).toBeGreaterThan(0);
+  expect(runs.every(query => !query.includes('"result_json"') && !query.includes('"runner_profile_json"'))).toBe(true);
+});
+
+it("keeps each retained run identity and policy distinct for the same principals", async () => {
+  const service = executionWorkspaceRepositoryService(db);
+  const invalidId = runIds[31];
+  try {
+    await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: randomUUID() }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId));
+    await expect(service.guardForPublication(input())).rejects.toThrow("identity");
+    await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: userId }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId));
+    await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, executionPolicy: { trustPreset: "unsupported" } } }).where(eq(heartbeatRuns.id, invalidId));
+    await expect(service.guardForPublication(input())).rejects.toThrow();
+  } finally {
+    await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: userId }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId));
+    await db.update(heartbeatRuns).set({ contextSnapshot: { issueId } }).where(eq(heartbeatRuns.id, invalidId));
+  }
+  await expect(service.guardForPublication(input())).resolves.toBeUndefined();
+});
+
+it("reloads project grants on every publication attempt", async () => {
+  const service = executionWorkspaceRepositoryService(db);
+  await db.update(projects).set({ visibility: "private" }).where(eq(projects.id, projectId));
+  const grants = await db.insert(projectAccessMembers).values([
+    { companyId, projectId, subjectType: "agent", subjectId: agentId },
+    { companyId, projectId, subjectType: "user", subjectId: userId },
+  ]).returning();
+  try {
+    await expect(service.guardForPublication(input())).resolves.toBeUndefined();
+    await db.delete(projectAccessMembers).where(and(eq(projectAccessMembers.projectId, projectId), eq(projectAccessMembers.subjectType, "user")));
+    await expect(service.guardForPublication(input())).rejects.toThrow();
+  } finally {
+    await db.delete(projectAccessMembers).where(inArray(projectAccessMembers.id, grants.map(grant => grant.id)));
+    await db.update(projects).set({ visibility: "open" }).where(eq(projects.id, projectId));
+  }
+});
+
+it("keeps supplied membership snapshots separate from fresh database snapshots", async () => {
+  const actor: AuthorizationActor = { type: "agent", source: "agent_jwt", companyId, agentId,
+    runId: runIds[0], onBehalfOfUserId: userId };
+  const supplied = { ...actor, onBehalfOfMemberships: [{ companyId, membershipRole: "operator", status: "active" }] };
+  const service = authorizationService(db);
+  const resource = { type: "company" as const, companyId };
+  await db.update(companyMemberships).set({ status: "inactive" }).where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, userId)));
+  try {
+    for (const candidates of [[supplied, actor], [actor, supplied]]) {
+      const decisions = await service.decideReadsForActors(candidates, [resource]);
+      for (let index = 0; index < candidates.length; index++) {
+        expect(decisions[index][0]).toEqual(await service.decide({ actor: { ...candidates[index] }, action: "company_scope:read", resource }));
+        expect(decisions[index][0].code).toBe(candidates[index] === supplied
+          ? "RESPONSIBLE_USER_UNAUTHORIZED" : "RESPONSIBLE_USER_UNAVAILABLE");
+      }
+    }
+  } finally {
+    await db.update(companyMemberships).set({ status: "active" }).where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, userId)));
+  }
+});

server/src/services/authorization.ts

@@ -755,7 +755,8 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA
 
   async function decideReadsForActors(actors: AuthorizationActor[], resources: ReadAuthorizationResource[]): Promise<AuthorizationDecision[][]> {
     if (!actors.length) return [];
-    const snapshots: ReadAuthorizationSnapshot[] = actors.map(() => ({ issues: new Map(), projects: new Map(), memo: new Map() }));
+    const memo = new Map<string, Promise<unknown>>();
+    const snapshots: ReadAuthorizationSnapshot[] = actors.map(() => ({ issues: new Map(), projects: new Map(), memo }));
     const batchActors: ResponsibleUserActorWithMemo[] = actors.map(actor => ({ ...actor, __responsibleUserSnapshotMemo: undefined }));
     const privacyKey = (actor: AuthorizationActor) => JSON.stringify([actor.type, actor.agentId, actor.userId, actor.onBehalfOfUserId]);
     const privacyActors = [...new Map(batchActors.map(actor => [privacyKey(actor), actor])).values()];
@@ -892,7 +893,8 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA
       return promise;
     }
 
-    const promise = loadResponsibleUserSnapshot(input.companyId, input.userId);
+    const promise = batchRead(`responsible:${input.companyId}:${input.userId}`,
+      () => loadResponsibleUserSnapshot(input.companyId, input.userId));
     actorWithMemo.__responsibleUserSnapshotMemo.set(key, promise);
     void promise.catch(() => {
       if (actorWithMemo.__responsibleUserSnapshotMemo?.get(key) === promise) {

server/src/services/execution-workspace-repositories.ts

@@ -29,6 +29,11 @@ type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 const controllerAcquisitionMessage = "New controller repository acquisition requires a trusted task. Existing verified public checkouts remain available.";
 const lowTrustRepositoryMessage = "Low-trust repository reuse permits anonymous public URLs only; new acquisition is unavailable. Use an authorized configured source for private repositories.";
+const rootReaderRunColumns = {
+  id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
+  responsibleUserId: heartbeatRuns.responsibleUserId, nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
+  contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId', 'taskId', ${heartbeatRuns.contextSnapshot}->'taskId', 'executionPolicy', ${heartbeatRuns.contextSnapshot}->'executionPolicy')`,
+};
 
 function resolveAcquisitionSource(request: PrepareWorkspaceRepository["repository"], available: ProjectRepository[], anonymousOnly: boolean) {
   if (anonymousOnly && request.kind === "catalog") throw forbidden(lowTrustRepositoryMessage);
@@ -180,8 +185,17 @@ async function assertRootProjectAccess(db: Db | DbTransaction, actor: Authorizat
   }
 }
 
+async function assertRootProjectsReadAccessForActors(db: Db | DbTransaction, actors: AuthorizationActor[], companyId: string, projectIds: Set<string>) {
+  if (!actors.length || !projectIds.size) return;
+  const decisions = await authorizationService(db).decideReadsForActors(actors,
+    [...projectIds].map(projectId => ({ type: "project" as const, companyId, projectId })));
+  if (decisions.some(batch => batch.some(decision => !decision.allowed))) {
+    throw forbidden("Workspace source project is no longer accessible");
+  }
+}
+
 async function retainedRootReader(tx: DbTransaction, companyId: string, reader: Pick<typeof executionWorkspaceRepositoryAccess.$inferSelect, "scopeId" | "issueId" | "agentId" | "responsibleUserId">) {
-  const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, reader.scopeId), eq(heartbeatRuns.companyId, companyId)));
+  const [run] = await tx.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, reader.scopeId), eq(heartbeatRuns.companyId, companyId)));
   const [agent] = reader.agentId ? await tx.select({ id: agents.id }).from(agents).where(and(eq(agents.id, reader.agentId), eq(agents.companyId, companyId))) : [];
   const [issue] = reader.issueId ? await tx.select().from(issues).where(and(eq(issues.id, reader.issueId), eq(issues.companyId, companyId))) : [];
   if (!run || !agent || !issue || run.agentId !== reader.agentId || run.responsibleUserId !== reader.responsibleUserId
@@ -260,8 +274,8 @@ async function retainGitMetadataOwners(tx: DbTransaction, input: {
   for (const reader of closure.overlappingEntries.filter(entry => entry.kind === "reader")) {
     const { actor } = await retainedRootReader(tx, input.companyId, reader);
     readerActors.push(actor);
-    await assertRootProjectAccess(tx, actor, input.companyId, projectIds);
   }
+  await assertRootProjectsReadAccessForActors(tx, readerActors, input.companyId, projectIds);
   await assertTaskWorkspacesReadAccessForActors(tx, readerActors, input.companyId, workspaceIds);
   for (const [kind, ids] of [["source", workspaceIds], ["project_source", projectIds]] as const) {
     for (const scopeId of ids) if (!closure.entries.some(entry => entry.kind === kind && entry.scopeId === scopeId)) {
@@ -356,8 +370,8 @@ export async function retainIsolatedWorkspaceSourceAccess(db: Db, input: {
         const { actor, issue } = await retainedRootReader(tx, input.companyId, reader);
         readerActors.push(actor);
         await assertRootProjectTrust(tx, { companyId: input.companyId, actor, issueId: issue.id, projectIds: target.sourceProjectIds });
-        await assertRootProjectAccess(tx, actor, input.companyId, target.sourceProjectIds);
       }
+      await assertRootProjectsReadAccessForActors(tx, readerActors, input.companyId, target.sourceProjectIds);
       await assertTaskWorkspacesReadAccessForActors(tx, readerActors, input.companyId, target.sourceIds);
     } catch (error) { failure ??= error; }
     if (!failure && verifiedCopyProjectIds) {
@@ -385,7 +399,7 @@ async function assertRootProjectTrust(db: Db | DbTransaction, input: {
   if (!restricted.length) return;
   const incompatible = () => forbidden("Physical workspace source trust is incompatible with the task's execution policy");
   const runId = input.actor.runId;
-  const [run] = runId ? await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, input.companyId))) : [];
+  const [run] = runId ? await db.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, input.companyId))) : [];
   const creation = !input.issueId && input.actor.type === "agent" && runId ? input.taskCreation : undefined;
   const runIssueId = run?.nativeIssueId ?? run?.contextSnapshot?.issueId ?? run?.contextSnapshot?.taskId ?? run?.issueId;
   const issueId = input.issueId ?? (creation && typeof runIssueId === "string" ? runIssueId : null);
@@ -471,7 +485,7 @@ export function executionWorkspaceRepositoryService(db: Db) {
     const [project] = issue.projectId ? await reader.select().from(projects).where(and(eq(projects.id, issue.projectId), eq(projects.companyId, issue.companyId))) : [];
     const [workspaceSourceProject] = workspace?.projectId ? await reader.select().from(projects).where(and(eq(projects.id, workspace.projectId), eq(projects.companyId, issue.companyId))) : [];
     const runId = input.runId ?? issue.executionRunId;
-    const [run] = runId ? await reader.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId))) : [];
+    const [run] = runId ? await reader.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId))) : [];
     if (runId && (!run || (input.runId && run.agentId !== agentId) || (run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId ?? run.issueId) !== issue.id)) {
       throw forbidden("Task repository run identity is unavailable");
     }
@@ -545,12 +559,14 @@ export function executionWorkspaceRepositoryService(db: Db) {
     const sourceProjects = sourceIds.size ? await tx.selectDistinct({ projectId: executionWorkspaces.projectId }).from(executionWorkspaces).where(and(
       eq(executionWorkspaces.companyId, input.companyId), inArray(executionWorkspaces.id, [...sourceIds]))) : [];
     const sourceActors: AuthorizationActor[] = [];
+    const projectReaders: AuthorizationActor[] = [];
     for (const reader of readers) {
       const { actor, issue } = await retainedRootReader(tx, input.companyId, reader);
       await assertRootProjectTrust(tx, { companyId: input.companyId, projectIds: sourceProjectIds, actor, issueId: issue.id,
         sourceWorkspaceId: input.reader?.runId === reader.scopeId ? input.sourceWorkspaceId : undefined });
-      await assertRootProjectAccess(tx, actor, input.companyId, sourceProjectIds,
-        input.reader?.runId === reader.scopeId ? { sharedProjectIds, issueId: issue.id } : undefined);
+      if (input.reader?.runId === reader.scopeId) {
+        await assertRootProjectAccess(tx, actor, input.companyId, sourceProjectIds, { sharedProjectIds, issueId: issue.id });
+      } else projectReaders.push(actor);
       if (ownAdmissionOnly) {
         const [ownSource] = await tx.select({ id: executionWorkspaces.id }).from(executionWorkspaces).where(and(
           eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.sourceWorkspaceId!),
@@ -566,6 +582,7 @@ export function executionWorkspaceRepositoryService(db: Db) {
         }
       }
     }
+    await assertRootProjectsReadAccessForActors(tx, projectReaders, input.companyId, sourceProjectIds);
     if (input.actor) {
       if (!input.issueId) throw forbidden("Workspace write authorization requires a task");
       await assertRootProjectTrust(tx, { ...input, projectIds: sourceProjectIds, actor: input.actor });
@@ -676,7 +693,7 @@ export function executionWorkspaceRepositoryService(db: Db) {
       throw forbidden("Task repository publication binding is unavailable");
     }
     await requiresAnonymousAcquisition(tx, { issue, workspace, agentId: input.agentId, runId: input.runId });
-    const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)));
+    const [run] = await tx.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)));
     const { actor } = await retainedRootReader(tx, input.companyId, {
       scopeId: input.runId, issueId: input.issueId, agentId: input.agentId, responsibleUserId: run?.responsibleUserId ?? null,
     });