From 070d6f83fd4129dbe6f4c7ecc0b69f8f73da0994 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 13:37:25 -0600 Subject: [PATCH] perf(realtime): share permission reads among queued live events (#20624) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A task page shows the output of a running agent as it is produced. > - The server sends each output event to the browser on the company WebSocket. > - Before each send, the server checks permissions and redacts secrets. Each event did its own sequential database reads. > - When output arrives faster than those reads complete, the queue grows. The text reaches the browser later and later, and the server then closes the connection. > - This pull request lets the events that are already in the queue share one set of reads. > - The benefit is that live output keeps pace with the run, and a revoked permission still applies to the next event. ## Linked Issues or Issue Description **What happened?** On a hosted instance, the live output of a running task sometimes appeared very slowly. A probe WebSocket on the company event stream received events 0.5 seconds after their creation time at the start. After 30 seconds the delay was 23 seconds. The server delivered about 6 events each second. Then it closed the socket with code 1013. The delivery loop for a connection is serial. For each transcript event it ran a membership read, a scope decision, a run lookup and a task read decision. Native run events added a settings read and a secret registry read. A burst of 200 events issued 1,401 SQL statements. **Expected behavior** Live output reaches the browser at the rate the run produces it. The cost of the permission checks does not grow with the number of events in a burst. A revoked permission stops delivery immediately. **Steps to reproduce** 1. Use a deployment that authenticates users, with a database that has a network round trip. 2. Start several agent runs that produce output at the same time. 3. Open the task page of one run and read the live output. 4. Compare the creation time of each `heartbeat.run.log` or `heartbeat.run.event` message with its arrival time. 5. The delay grows continuously until the server closes the socket with code 1013. **Paperclip version or commit** `master` at 79e28dc0d0. ## What Changed - `live-event-delivery.ts` gives each queued event a sequence number. It passes a shared read set to `authorize`. A set serves only the events that were in the queue when the set opened. The delivery replaces the set after one second and discards it when the queue is empty. - `sharedDeliveryRead` returns the pending read for a key, or starts it. It removes a read that fails, so the next event reads again. - `live-events-ws.ts` uses the shared set for the membership or key check, the transcript scope decision, the run binding and the task read decision. - `live-run-event-redaction.ts` uses the shared set for the username setting and the registered secret values of the run. `run-secret-redaction.ts` adds `redactorForRun`, which reads the values one time and returns a redactor. - `doc/live-transcript-delivery.md` describes the shared reads and their limits. ## Verification - `npx vitest run src/realtime/live-event-delivery.test.ts src/realtime/live-event-delivery.cost.test.ts src/__tests__/live-events-run-read.test.ts src/__tests__/live-events-ws.test.ts src/__tests__/live-events-toast.test.ts` in `server/`. All tests pass. - New integration test against PostgreSQL: 200 transcript events in one burst. All events arrive in order. The burst issues at most 40 statements. Without the shared reads the same test fails with 1,401 statements. After the grant is revoked, the next burst delivers nothing. - New unit tests: 39 events that wait behind one read share the next read. An event that arrives during a read starts a new read. A set is replaced after one second. A failed read is not shared. - The existing tests for revocation, deleted tasks, restricted agent keys and live redaction pass without changes. - Assumption tested: an event never uses a read that began before the event entered the queue. The unit test holds a read open and adds an event to prove it. - Not verified: the delay on a hosted instance after deployment. The measurement above is from before this change. ## Risks - Permission freshness. Before, each event had its own check. Now a check can serve events that were already in the queue, for at most one second. A revocation applies to the first event published after it. Events already in the queue at the time of the check can still be sent. That window existed before between a check and its send. - Secret values. A read set holds decrypted registry values while a burst drains. The delivery discards the set when the queue is empty and when the connection closes. - Plugin run streams use the same redactor and get the same shared reads. - No schema, API or protocol change. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude, model ID `claude-fable-5-2`, in Claude Code with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude --- doc/live-transcript-delivery.md | 26 ++++++++- .../__tests__/live-events-run-read.test.ts | 46 +++++++++++++++ .../src/realtime/live-event-delivery.test.ts | 56 ++++++++++++++++++- server/src/realtime/live-event-delivery.ts | 34 +++++++++-- server/src/realtime/live-events-ws.ts | 40 +++++++------ .../src/services/live-run-event-redaction.ts | 8 ++- server/src/services/run-secret-redaction.ts | 4 ++ 7 files changed, 184 insertions(+), 30 deletions(-) diff --git a/doc/live-transcript-delivery.md b/doc/live-transcript-delivery.md index 5151ab7a59f..d17e4f45da5 100644 --- a/doc/live-transcript-delivery.md +++ b/doc/live-transcript-delivery.md @@ -7,8 +7,25 @@ after access is revoked. A subscription is never an authorization grant. Full ev rows and log chunks require the current `company_scope:read` permission used by HTTP transcript reads, as well as access to the run's task. A restricted agent can still receive authorized lifecycle and lightweight activity notifications. -The check runs for each full delivery on both negotiated and older connections; -changing a trust policy takes effect without reconnecting. +The check runs at delivery on both negotiated and older connections; changing a +trust policy takes effect without reconnecting. + +## Shared reads for queued output + +Output can arrive faster than one permission check completes. Each event then +waited for its own sequential database reads, the queue grew, and the connection +reached its backlog limit. Events that are already queued when a set of reads +opens now share that set: the membership or key check, the transcript scope +decision, the run binding, the task read decision, the username-censoring +setting and the run's registered secret values. + +An event never uses a read that began before the event was queued. An event that +arrives while a read is in flight starts a new set. A set is also replaced after +one second, so slow per-event work cannot extend it. A failed read is not shared; +the next event reads again. The set is discarded when the queue empties, so +decrypted secret values are not retained by an idle connection. An isolated event +therefore costs the same reads as before, and a revocation applies to the first +event published after it. ## Additive negotiation @@ -136,6 +153,11 @@ Both use a disposable server and can use an explicitly supplied disposable `DATABASE_URL` when local embedded Postgres is unavailable. Browser fixtures establish delivery behavior, not native-provider or hosted-upgrade qualification. +`server/src/__tests__/live-events-run-read.test.ts` publishes 200 transcript +events in one burst against PostgreSQL. It requires every event in order, at most +40 statements where separate reads issued 1,401, and no delivery after the grant +is revoked. + `server/src/realtime/live-event-delivery.cost.test.ts` compares identical output with zero interests, one viewer, three browser connections, and simulated socket congestion. Set `PAPERCLIP_LIVE_DELIVERY_COST_REPORT` to retain its JSON report. diff --git a/server/src/__tests__/live-events-run-read.test.ts b/server/src/__tests__/live-events-run-read.test.ts index 2976bd96a01..43e098bc53a 100644 --- a/server/src/__tests__/live-events-run-read.test.ts +++ b/server/src/__tests__/live-events-run-read.test.ts @@ -183,6 +183,52 @@ describe("live-event run read boundaries", () => { } }); + it("delivers a burst of output without one permission read per event, and still stops at a revocation", async () => { + const companyId = randomUUID(), owner = randomUUID(), viewer = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Burst reads", issuePrefix: `BR${companyId.slice(0, 6)}` }); + await db.insert(authUsers).values([owner, viewer].map(id => ({ id, name: "Fixture viewer", email: `${id}@example.test`, createdAt: new Date(), updatedAt: new Date() }))); + await db.insert(companyMemberships).values([owner, viewer].map(principalId => ({ companyId, principalType: "user", principalId, membershipRole: "operator", status: "active" }))); + const [agent] = await db.insert(agents).values({ companyId, name: "Burst agent", adapterType: "process" }).returning(); + await db.insert(issues).values({ id: issueId, companyId, title: "Private task", visibility: "private", privacyRootIssueId: issueId, responsibleUserId: owner }); + const [grant] = await db.insert(issueAccessGrants).values({ issueId, subjectType: "user", subjectId: viewer, source: "explicit" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: "issue", issueId, status: "running" }).returning({ id: heartbeatRuns.id }); + const queries: string[] = []; + const observedDb = drizzle(db.$client, { schema, logger: { logQuery(query) { queries.push(query); } } }) as Db; + const server = createServer(); + const wss = setupLiveEventsWebSocketServer(server, observedDb, { deploymentMode: "authenticated", + resolveCloudActor: async () => ({ userId: viewer, companyIds: [companyId] }) }); + server.listen(0, "127.0.0.1"); await once(server, "listening"); + const socket = new WebSocket(`ws://127.0.0.1:${(server.address() as AddressInfo).port}/api/companies/${companyId}/events/ws`); + const messages: Array<{ type: string; payload: Record }> = []; + socket.on("message", bytes => messages.push(JSON.parse(bytes.toString()))); + const burst = async (count: number) => { + const start = messages.length, queryStart = queries.length, barrier = randomUUID(); + for (let seq = 1; seq <= count; seq++) { + publishLiveEvent({ companyId, type: "heartbeat.run.log", payload: { runId: run.id, seq, chunk: `token-${seq}` } }); + publishLiveEvent({ companyId, type: "heartbeat.run.event", payload: { runId: run.id, seq, + payload: { prpEvent: { eventType: "item.delta", payload: { text: `native-${seq}` } } } } }); + } + publishLiveEvent({ companyId, type: "agent.status", payload: { agentId: barrier, status: "idle" } }); + await expect.poll(() => messages.some(message => message.payload.agentId === barrier), { timeout: 20_000 }).toBe(true); + return { delivered: messages.slice(start).filter(message => message.type.startsWith("heartbeat.run.")), queries: queries.slice(queryStart) }; + }; + try { + await once(socket, "open"); + const first = await burst(100); + expect(first.delivered.filter(message => message.type === "heartbeat.run.log").map(message => message.payload.seq)) + .toEqual(Array.from({ length: 100 }, (_, index) => index + 1)); + expect(first.delivered.filter(message => message.type === "heartbeat.run.event")).toHaveLength(100); + expect(first.queries.length).toBeLessThanOrEqual(40); + await db.update(issueAccessGrants).set({ revokedAt: new Date() }).where(eq(issueAccessGrants.id, grant.id)); + expect((await burst(20)).delivered).toEqual([]); + } finally { + socket.terminate(); + await Promise.all([...wss.clients].map(client => { const closed = new Promise(resolve => client.on("close", resolve)); client.terminate(); return closed; })); + await new Promise((resolve, reject) => (wss as typeof wss & { close(callback: (error?: Error) => void): void }).close(error => error ? reject(error) : resolve())); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + } + }); + it.each([false, true])("keeps transcript permissions current for agent sockets (negotiated=%s)", async negotiated => { const companyId = randomUUID(), viewer = randomUUID(), issueId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Restricted live reads", issuePrefix: `RL${companyId.slice(0, 6)}` }); diff --git a/server/src/realtime/live-event-delivery.test.ts b/server/src/realtime/live-event-delivery.test.ts index 2f8b4c4eac4..d46255525d8 100644 --- a/server/src/realtime/live-event-delivery.test.ts +++ b/server/src/realtime/live-event-delivery.test.ts @@ -1,11 +1,11 @@ import { afterEach, expect, it, vi } from "vitest"; import type { LiveEvent } from "@paperclipai/shared"; -import { createLiveEventDelivery, LIVE_DELIVERY_MAX_BYTES, LIVE_DELIVERY_MAX_ITEMS } from "./live-event-delivery.js"; +import { createLiveEventDelivery, LIVE_DELIVERY_MAX_BYTES, LIVE_DELIVERY_MAX_ITEMS, sharedDeliveryRead, type LiveDeliveryReads } from "./live-event-delivery.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; const event = (seq = 1, runId = "run-1"): LiveEvent => ({ id: seq, companyId: "company", type: "heartbeat.run.log", createdAt: "2026-10-24T00:00:00Z", payload: { runId, agentId: "agent", seq, chunk: "output" } }); -function fixture(authorize = vi.fn(async (value: LiveEvent) => value as LiveEvent | null)) { +function fixture(authorize = vi.fn(async (value: LiveEvent, _reads?: LiveDeliveryReads) => value as LiveEvent | null)) { const frames: any[] = []; const socket = { readyState: 1, bufferedAmount: 0, send: vi.fn((text: string, callback?: (error?: Error) => void) => { frames.push(JSON.parse(text)); callback?.(); @@ -209,3 +209,55 @@ it("cannot restart recovery timers when a timed-out read later rejects", async ( expect(f.authorize).toHaveBeenCalledOnce(); } finally { f.delivery.dispose(); } }); + +const flush = async () => { for (let n = 0; n < 600; n++) await Promise.resolve(); }; +it("shares one permission read among events queued behind it, never with an event that arrived after it began", async () => { + const releases: Array<() => void> = []; + const read = vi.fn(() => new Promise(resolve => { releases.push(() => resolve(true)); })); + const f = fixture(vi.fn(async (value: LiveEvent, reads?: LiveDeliveryReads) => + await sharedDeliveryRead(reads, "permission", read) ? value : null)); + f.subscribe(["run-1"]); + f.delivery.enqueue(event(1)); + for (let seq = 2; seq <= 40; seq++) f.delivery.enqueue(event(seq)); + await flush(); + expect(read).toHaveBeenCalledTimes(1); + releases[0](); await flush(); + expect(read).toHaveBeenCalledTimes(2); + f.delivery.enqueue(event(41)); + releases[1](); await flush(); + expect(read).toHaveBeenCalledTimes(3); + expect(f.frames.filter(frame => frame.type === "heartbeat.run.log")).toHaveLength(40); + releases[2](); await flush(); + expect(f.frames.filter(frame => frame.type === "heartbeat.run.log").map(frame => frame.payload.seq)) + .toEqual(Array.from({ length: 41 }, (_, index) => index + 1)); + expect(f.authorize).toHaveBeenCalledTimes(41); + f.delivery.enqueue(event(42)); await flush(); + expect(read).toHaveBeenCalledTimes(4); + f.delivery.dispose(); +}); +it("stops sharing a read after one second and never reuses a failed read", async () => { + let now = 1_000_000; + const clock = vi.spyOn(Date, "now").mockImplementation(() => now); + try { + const read = vi.fn(async () => true); + const f = fixture(vi.fn(async (value: LiveEvent, reads?: LiveDeliveryReads) => { + const allowed = await sharedDeliveryRead(reads, "permission", read); + now += 400; + return allowed ? value : null; + })); + f.subscribe(["run-1"]); + for (let seq = 1; seq <= 7; seq++) f.delivery.enqueue(event(seq)); + await flush(); + expect(f.frames.filter(frame => frame.type === "heartbeat.run.log")).toHaveLength(7); + expect(read).toHaveBeenCalledTimes(3); + f.delivery.dispose(); + + const failing = vi.fn(async () => true).mockResolvedValueOnce(true).mockRejectedValueOnce(new Error("permission database unavailable")); + const g = fixture(vi.fn(async (value: LiveEvent, reads?: LiveDeliveryReads) => + await sharedDeliveryRead(reads, "permission", failing) ? value : null)); + g.delivery.enqueue(event(1)); g.delivery.enqueue(event(2)); g.delivery.enqueue(event(3)); await flush(); + expect(failing).toHaveBeenCalledTimes(3); + expect(g.frames.map(frame => frame.payload.seq)).toEqual([1, 3]); + g.delivery.dispose(); + } finally { clock.mockRestore(); } +}); diff --git a/server/src/realtime/live-event-delivery.ts b/server/src/realtime/live-event-delivery.ts index 588eb654a35..f4aad8b7938 100644 --- a/server/src/realtime/live-event-delivery.ts +++ b/server/src/realtime/live-event-delivery.ts @@ -4,6 +4,17 @@ import { beginIdleTrackedWork } from "../services/task-admission.js"; export const LIVE_DELIVERY_MAX_ITEMS = 1_000; export const LIVE_DELIVERY_MAX_BYTES = 1024 * 1024; const CONGESTION_DEADLINE_MS = 5_000; +const SHARED_READS_MAX_AGE_MS = 1_000; +export type LiveDeliveryReads = Map>; +export function sharedDeliveryRead(reads: LiveDeliveryReads | undefined, key: string, read: () => Promise): Promise { + if (!reads) return read(); + const existing = reads.get(key); + if (existing) return existing as Promise; + const pending = read(); + reads.set(key, pending); + pending.catch(() => { if (reads.get(key) === pending) reads.delete(key); }); + return pending; +} const transcript = (event: LiveEvent) => event.type === "heartbeat.run.event" || event.type === "heartbeat.run.log"; const runId = (event: LiveEvent) => typeof event.payload.runId === "string" ? event.payload.runId : null; interface Socket { @@ -15,7 +26,7 @@ interface Socket { export function createLiveEventDelivery(input: { socket: Socket; - authorize(event: LiveEvent): Promise; + authorize(event: LiveEvent, reads: LiveDeliveryReads): Promise; decorate(event: LiveEvent): Promise; onError(error: unknown): void; }) { @@ -23,7 +34,19 @@ export function createLiveEventDelivery(input: { let negotiated = false; let revision = 0; let interests = new Set(); - let queued: Array<{ event: LiveEvent; bytes: number; epoch: number }> = []; + let queued: Array<{ event: LiveEvent; bytes: number; epoch: number; seq: number }> = []; + let enqueuedSeq = 0; + let reads: LiveDeliveryReads = new Map(); + let readsCoverSeq = 0; + let readsOpenedAt = 0; + function readsFor(seq: number) { + const now = Date.now(); + if (seq > readsCoverSeq || now - readsOpenedAt >= SHARED_READS_MAX_AGE_MS) { + reads = new Map(); readsCoverSeq = enqueuedSeq; readsOpenedAt = now; + } + return reads; + } + function closeReads() { reads = new Map(); readsCoverSeq = 0; } let retainedBytes = 0; let retainedItems = 0; let epoch = 0; @@ -42,7 +65,7 @@ export function createLiveEventDelivery(input: { function dispose() { disposed = true; epoch += 1; for (const item of queued) { retainedBytes -= item.bytes; retainedItems -= 1; } - queued = []; activityAt.clear(); + queued = []; activityAt.clear(); closeReads(); if (congestionTimer) clearTimeout(congestionTimer); if (pendingReadTimer) clearTimeout(pendingReadTimer); congestionTimer = pendingReadTimer = undefined; @@ -99,7 +122,7 @@ export function createLiveEventDelivery(input: { if (item.epoch !== epoch || !wants(item.event)) continue; pendingReadTimer = setTimeout(closeBacklog, CONGESTION_DEADLINE_MS); pendingReadTimer.unref?.(); - const visible = await input.authorize(item.event); + const visible = await input.authorize(item.event, readsFor(item.seq)); if (!visible || item.epoch !== epoch || !wants(item.event) || !isOpen()) continue; const decorated = transcript(visible) || visible.type === "heartbeat.run.activity" ? visible : await input.decorate(visible); @@ -124,6 +147,7 @@ export function createLiveEventDelivery(input: { } } } finally { + closeReads(); try { draining = false; resync(); } finally { finishIdleWork(); } } @@ -166,7 +190,7 @@ export function createLiveEventDelivery(input: { if (!event) return; const bytes = Buffer.byteLength(JSON.stringify(event)); if (!room(bytes)) { overflow(); return; } - queued.push({ event, bytes, epoch }); retainedBytes += bytes; retainedItems += 1; + queued.push({ event, bytes, epoch, seq: ++enqueuedSeq }); retainedBytes += bytes; retainedItems += 1; void drain(); }, requestResync: () => overflow("controller_change"), diff --git a/server/src/realtime/live-events-ws.ts b/server/src/realtime/live-events-ws.ts index 8cce53eb437..e3e8da18aa9 100644 --- a/server/src/realtime/live-events-ws.ts +++ b/server/src/realtime/live-events-ws.ts @@ -1,4 +1,4 @@ -import { createLiveEventDelivery } from "./live-event-delivery.js"; +import { createLiveEventDelivery, sharedDeliveryRead, type LiveDeliveryReads } from "./live-event-delivery.js"; import { skillAccessService } from "../services/skill-access.js"; import { createHash } from "node:crypto"; import type { IncomingMessage, Server as HttpServer } from "node:http"; @@ -283,7 +283,7 @@ export function setupLiveEventsWebSocketServer( } const access = authorizationService(db); - async function visibleEvent(event: LiveEvent): Promise { + async function visibleEvent(event: LiveEvent, reads: LiveDeliveryReads): Promise { if (event.payload.entityType === "case" && (typeof event.payload.entityId !== "string" || !(await canActorReadCase(db, context!.actor, { id: event.payload.entityId, companyId: event.companyId })))) return null; if (context!.actor.source === "local_implicit") { @@ -294,42 +294,46 @@ export function setupLiveEventsWebSocketServer( } return event; } - if (context!.apiKeyId) { - const active = await db.select({ id: agentApiKeys.id }).from(agentApiKeys) - .where(and(eq(agentApiKeys.id, context!.apiKeyId), isNull(agentApiKeys.revokedAt))).limit(1); - if (!active.length) return null; - } else { + const principalActive = await sharedDeliveryRead(reads, "principal", async () => { + if (context!.apiKeyId) { + const active = await db.select({ id: agentApiKeys.id }).from(agentApiKeys) + .where(and(eq(agentApiKeys.id, context!.apiKeyId), isNull(agentApiKeys.revokedAt))).limit(1); + return active.length > 0; + } const membership = await db.select({ id: companyMemberships.id }).from(companyMemberships) .where(and(eq(companyMemberships.companyId, context!.companyId), eq(companyMemberships.principalType, "user"), eq(companyMemberships.principalId, context!.actorId), eq(companyMemberships.status, "active"))).limit(1); const admin = context!.actor.source !== "cloud_tenant" && await db.select({ id: instanceUserRoles.id }).from(instanceUserRoles) .where(and(eq(instanceUserRoles.userId, context!.actorId), eq(instanceUserRoles.role, "instance_admin"))).limit(1).then(rows => rows.length > 0); - if (!membership.length && !admin) return null; - } + return membership.length > 0 || admin; + }); + if (!principalActive) return null; 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; if (event.type === "heartbeat.run.event" || event.type === "heartbeat.run.log") { - const decision = await access.decide({ actor: context!.actor, action: "company_scope:read", - resource: { type: "company", companyId: event.companyId } }); - if (!decision.allowed) return null; + const allowed = await sharedDeliveryRead(reads, "transcript-scope", async () => (await access.decide({ + actor: context!.actor, action: "company_scope:read", + resource: { type: "company", companyId: event.companyId } })).allowed); + if (!allowed) return null; } const readBindings: Parameters[3][number][] = []; if (issueId) { readBindings.push({ companyId: event.companyId, scopeKind: "issue", issueId }); } if (runId) { - const [run] = await db.select({ + const run = await sharedDeliveryRead(reads, `run:${runId}`, async () => (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))); + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, event.companyId))))[0]); if (!run) return null; readBindings.push(run); } - if ((await getHeartbeatRunReadDecisions(db, access, context!.actor, readBindings)).some(allowed => !allowed)) return null; + if (!(await sharedDeliveryRead(reads, `read:${issueId ?? ""}:${runId ?? ""}`, async () => + (await getHeartbeatRunReadDecisions(db, access, context!.actor, readBindings)).every(Boolean)))) return null; if (["company_skill", "company_skill_source"].includes(String(payload.entityType)) && typeof payload.entityId === "string") { try { const skills = skillAccessService(db, context!.actor); @@ -366,12 +370,12 @@ export function setupLiveEventsWebSocketServer( if (event.type.startsWith("plugin.")) return event; return null; } - const delivery = createLiveEventDelivery({ socket, authorize: async event => { - const visible = await visibleEvent(event); + const delivery = createLiveEventDelivery({ socket, authorize: async (event, reads) => { + const visible = await visibleEvent(event, reads); if (!visible || visible.type !== "heartbeat.run.event") return visible; const payload = visible.payload.payload; if (!payload || typeof payload !== "object" || Array.isArray(payload) || !("prpEvent" in payload)) return visible; - return redactRunEvent(visible); + return redactRunEvent(visible, reads); }, decorate: async visible => { let toastEligible = false; diff --git a/server/src/services/live-run-event-redaction.ts b/server/src/services/live-run-event-redaction.ts index 4818f183fc5..39b29f80a07 100644 --- a/server/src/services/live-run-event-redaction.ts +++ b/server/src/services/live-run-event-redaction.ts @@ -4,16 +4,18 @@ import { redactCurrentUserValue } from "../log-redaction.js"; import { redactEventPayload } from "../redaction.js"; import { instanceSettingsService } from "./instance-settings.js"; import { createRunSecretRedactionRegistry } from "./run-secret-redaction.js"; +import { sharedDeliveryRead, type LiveDeliveryReads } from "../realtime/live-event-delivery.js"; export function createLiveRunEventRedactor(db: Db) { const settings = instanceSettingsService(db); const redactions = createRunSecretRedactionRegistry(db); - return async (event: LiveEvent): Promise => { + return async (event: LiveEvent, reads?: LiveDeliveryReads): Promise => { const runId = event.payload.runId; if (typeof runId !== "string") return null; const redacted = redactCurrentUserValue({ ...event, payload: redactEventPayload(event.payload) ?? {}, - }, { enabled: (await settings.getGeneral()).censorUsernameInLogs }); - return redactions.redactForRun(event.companyId, runId, redacted); + }, { enabled: await sharedDeliveryRead(reads, "censor-usernames", async () => (await settings.getGeneral()).censorUsernameInLogs) }); + const redact = await sharedDeliveryRead(reads, `secrets:${event.companyId}:${runId}`, () => redactions.redactorForRun(event.companyId, runId)); + return redact(redacted); }; } diff --git a/server/src/services/run-secret-redaction.ts b/server/src/services/run-secret-redaction.ts index 725d7342382..ed426fac510 100644 --- a/server/src/services/run-secret-redaction.ts +++ b/server/src/services/run-secret-redaction.ts @@ -162,6 +162,10 @@ export function createRunSecretRedactionRegistry(db: Db) { return redactRegisteredSecretValues(withoutSqlRunError(run), disclosure?.values ?? [], disclosure?.credentialConnectionName ?? null); }); }, + redactorForRun: async (companyId: string, runId: string, viewerUserId: string | null = null) => { + const disclosure = await valuesForRun(companyId, runId, viewerUserId); + return (value: T): T => redactRegisteredSecretValues(withoutSqlRunError(value), disclosure.values, disclosure.credentialConnectionName); + }, redactForRun: async (companyId: string, runId: string, value: T, viewerUserId: string | null = null): Promise => { const disclosure = await valuesForRun(companyId, runId, viewerUserId); return redactRegisteredSecretValues(withoutSqlRunError(value), disclosure.values, disclosure.credentialConnectionName);