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 <noreply@paperclip.ing>

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

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/);

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",
 }));

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<typeof getHeartbeatRunReadDecisions>[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