From 3f36947e0cf18f58f57f34cefd53b49af54c790c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 02:43:17 -0700 Subject: [PATCH] perf(realtime): batch task and run authorization per event (#17579) Use one fresh authorization batch for each live event task and run binding. Preserve current membership, grant, company, and payload disclosure checks. Build, typecheck, 70 affected local tests, and the complete required CI matrix passed. Greptile scored the exact head 5/5 at medium impact with no unresolved findings. The separate full local suite remains in progress. Co-Authored-By: Paperclip --- doc/DATABASE.md | 6 ++++ .../__tests__/live-events-run-read.test.ts | 31 ++++++++++++++++-- .../src/__tests__/live-events-toast.test.ts | 5 ++- server/src/realtime/live-events-ws.ts | 32 +++++++++---------- 4 files changed, 54 insertions(+), 20 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index a51b7cb6100..dd8126cbcd1 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -123,6 +123,12 @@ launch context does not travel with each policy or parent-report check. JSON types, task-key precedence, and company and agent identity checks are preserved. Each decision still reads current data; this is not an authorization cache. +WebSocket delivery uses one fresh authorization batch for an event's task and +run bindings. When both bindings name the same task, the batch reads its current +authority once. Different task bindings must both pass. The batch is discarded +after that delivery; later events still recheck grants, membership, task +existence, and company boundaries. + Filesystem preparation uses current task/run and selected-configuration checks. Historical root-reader authorization has been retired; no retained-reader batch is loaded at admission or publication. Workspace-record disclosure still uses task diff --git a/server/src/__tests__/live-events-run-read.test.ts b/server/src/__tests__/live-events-run-read.test.ts index 76704f2f63c..b02b0d4cf45 100644 --- a/server/src/__tests__/live-events-run-read.test.ts +++ b/server/src/__tests__/live-events-run-read.test.ts @@ -58,9 +58,19 @@ describe("live-event run read boundaries", () => { }).returning({ id: heartbeatRuns.id }); const [maintenance] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: "company" }).returning({ id: heartbeatRuns.id }); const [foreign] = await db.insert(heartbeatRuns).values({ companyId: foreignCompanyId, agentId: foreignAgent.id, scopeKind: "company" }).returning({ id: heartbeatRuns.id }); + const hiddenIssueId = randomUUID(); + await db.insert(issues).values({ id: hiddenIssueId, companyId, title: "Other private task", visibility: "private", privacyRootIssueId: hiddenIssueId, responsibleUserId: owner }); + const [hiddenRun] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: "issue", issueId: hiddenIssueId }).returning({ id: heartbeatRuns.id }); const reads: string[] = []; + const queries: string[] = []; let failNextRunRead = false; + let failNextIssueRead = false; const observedDb = drizzle(db.$client, { schema, logger: { logQuery(query) { + queries.push(query); + if (query.startsWith("select ") && query.includes('from "issues"') && failNextIssueRead) { + failNextIssueRead = false; + throw new Error("fixture issue lookup unavailable"); + } if (query.startsWith("select ") && query.includes('from "heartbeat_runs"')) { reads.push(query); if (failNextRunRead) { failNextRunRead = false; throw new Error("fixture run lookup unavailable"); } @@ -78,17 +88,32 @@ describe("live-event run read boundaries", () => { socket.on("message", bytes => messages.push(JSON.parse(bytes.toString()))); try { await once(socket, "open"); - async function deliver(runId: string) { + let deliveries = 0; + async function deliver(runId: string, payloadIssueId?: string) { + deliveries += 1; const start = messages.length; const barrier = randomUUID(); - publishLiveEvent({ companyId, type: "heartbeat.run.status", payload: { runId, status: "running" } }); + publishLiveEvent({ companyId, type: "heartbeat.run.status", payload: { + runId, status: "running", ...(payloadIssueId ? { issueId: payloadIssueId } : {}), + } }); publishLiveEvent({ companyId, type: "agent.status", payload: { agentId: barrier, status: "idle" } }); await expect.poll(() => messages.some(message => message.payload.agentId === barrier)).toBe(true); return messages.slice(start).filter(message => message.type === "heartbeat.run.status"); } expect(await deliver(run.id)).toMatchObject([{ payload: { runId: run.id, status: "running" } }]); + const batchStart = queries.length; + expect(await deliver(run.id, issueId)).toMatchObject([{ payload: { runId: run.id, issueId, status: "running" } }]); + expect(queries.length - batchStart).toBeLessThanOrEqual(6); + expect(queries.slice(batchStart).filter(query => query.startsWith("select ") && query.includes('from "issues"')).length).toBeLessThanOrEqual(2); + expect(await deliver(hiddenRun.id, issueId)).toEqual([]); + expect(await deliver(run.id, hiddenIssueId)).toEqual([]); + expect(await deliver(run.id, randomUUID())).toEqual([]); + failNextIssueRead = true; + expect(await deliver(run.id, issueId)).toEqual([]); + expect(await deliver(run.id, issueId)).toHaveLength(1); await db.update(issueAccessGrants).set({ revokedAt: new Date() }).where(eq(issueAccessGrants.id, grant.id)); + expect(await deliver(run.id, issueId)).toEqual([]); expect(await deliver(run.id)).toEqual([]); expect(await deliver(maintenance.id)).toHaveLength(1); expect(await deliver(foreign.id)).toEqual([]); @@ -101,7 +126,7 @@ describe("live-event run read boundaries", () => { await db.delete(issues).where(eq(issues.id, issueId)); expect(await deliver(run.id)).toEqual([]); - expect(reads).toHaveLength(9); + expect(reads).toHaveLength(deliveries); for (const query of reads) { expect(query).toMatch(/^select "company_id", "scope_kind", "issue_id" from "heartbeat_runs"/); expect(query).not.toMatch(/context_snapshot|runner_profile_json|stdout_excerpt|result_json/); diff --git a/server/src/__tests__/live-events-toast.test.ts b/server/src/__tests__/live-events-toast.test.ts index 5c5fbe7bf6c..e82af433d73 100644 --- a/server/src/__tests__/live-events-toast.test.ts +++ b/server/src/__tests__/live-events-toast.test.ts @@ -14,7 +14,10 @@ vi.mock("../realtime/task-toast-eligibility.js", () => ({ taskToastEligible: vi. vi.mock("../services/authorization.js", () => ({ authorizationService: () => ({ decide: async ({ resource }: { resource: { issueId: string } }) => ({ allowed: resource.issueId !== "hidden" }) }), })); -vi.mock("../services/heartbeat-run-privacy.js", () => ({ canActorReadHeartbeatRun: async () => false })); +vi.mock("../services/heartbeat-run-privacy.js", () => ({ + getHeartbeatRunReadDecisions: async (_db: unknown, _access: unknown, _actor: unknown, runs: { issueId?: string }[]) => + runs.map(run => run.issueId !== "hidden"), +})); vi.mock("../services/case-access.js", () => ({ canActorReadCase: async (_db: unknown, _actor: unknown, record: { id: string }) => record.id !== "hidden-case", })); diff --git a/server/src/realtime/live-events-ws.ts b/server/src/realtime/live-events-ws.ts index 6f299bfe672..95159016122 100644 --- a/server/src/realtime/live-events-ws.ts +++ b/server/src/realtime/live-events-ws.ts @@ -5,7 +5,7 @@ import { createRequire } from "node:module"; import type { Duplex } from "node:stream"; import { and, eq, isNull } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { agentApiKeys, companyMemberships, instanceUserRoles, heartbeatRuns, issues, projects } from "@paperclipai/db"; +import { agentApiKeys, companyMemberships, instanceUserRoles, heartbeatRuns, projects } from "@paperclipai/db"; import type { DeploymentMode, LiveEvent } from "@paperclipai/shared"; import type { BetterAuthSessionResult } from "../auth/better-auth.js"; import { logger } from "../middleware/logger.js"; @@ -45,7 +45,7 @@ const { WebSocket, WebSocketServer } = require("ws") as { }; import { authorizationService, type AuthorizationActor } from "../services/authorization.js"; -import { canActorReadHeartbeatRun } from "../services/heartbeat-run-privacy.js"; +import { getHeartbeatRunReadDecisions } from "../services/heartbeat-run-privacy.js"; import { canActorReadCase } from "../services/case-access.js"; interface UpgradeContext { @@ -301,12 +301,22 @@ export function setupLiveEventsWebSocketServer( const payload = event.payload; const issueId = typeof payload.issueId === "string" ? payload.issueId : payload.entityType === "issue" && typeof payload.entityId === "string" ? payload.entityId : null; + const runId = typeof payload.runId === "string" ? payload.runId + : payload.entityType === "heartbeat_run" && typeof payload.entityId === "string" ? payload.entityId : null; + const readBindings: Parameters[3][number][] = []; if (issueId) { - const [issue] = await db.select({ id: issues.id }).from(issues) - .where(and(eq(issues.id, issueId), eq(issues.companyId, event.companyId))).limit(1); - if (!issue || !(await access.decide({ actor: context!.actor, action: "issue:read", - resource: { type: "issue", companyId: event.companyId, issueId } })).allowed) return null; + readBindings.push({ companyId: event.companyId, scopeKind: "issue", issueId }); } + if (runId) { + const [run] = await db.select({ + companyId: heartbeatRuns.companyId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, event.companyId))); + if (!run) return null; + readBindings.push(run); + } + if ((await getHeartbeatRunReadDecisions(db, access, context!.actor, readBindings)).some(allowed => !allowed)) return null; if (["company_skill", "company_skill_source"].includes(String(payload.entityType)) && typeof payload.entityId === "string") { try { const skills = skillAccessService(db, context!.actor); @@ -320,16 +330,6 @@ export function setupLiveEventsWebSocketServer( if (!project || !(await access.decide({ actor: context!.actor, action: "project:read", resource: { type: "project", companyId: event.companyId, projectId: project.id } })).allowed) return null; } - const runId = typeof payload.runId === "string" ? payload.runId - : payload.entityType === "heartbeat_run" && typeof payload.entityId === "string" ? payload.entityId : null; - if (runId) { - const [run] = await db.select({ - companyId: heartbeatRuns.companyId, - scopeKind: heartbeatRuns.scopeKind, - issueId: heartbeatRuns.issueId, - }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, event.companyId))); - if (!run || !(await canActorReadHeartbeatRun(db, access, context!.actor, run))) return null; - } if (event.type.startsWith("heartbeat.run.")) return runId ? event : null; if (event.type === "agent.session.goal.changed") return issueId ? event : null; // Company-wide invalidations carry no task-derived content. Authorized