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