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