From 6505538fc5caa1fc32ec2c205fd702ab2fc2b2e7 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:18:20 -0600 Subject: [PATCH] fix(realtime): limit a shared live-event read set to 100 events and 250 ms (#20632) 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. > - The server sends the output of a running agent to the browser on the company WebSocket. > - #20624 lets the events that are already in the queue share one set of permission and redaction reads. > - That set could serve a full queue of 1,000 events for one second. > - If access is revoked during that time, the principal still receives the queued output. The security scan on #20624 reported this. > - This pull request limits one set to 100 events and 250 ms. > - The benefit is a small, stated limit on output that a revoked principal can receive, with the same delivery speed. ## Linked Issues or Issue Description Refs #20624 **What happened?** After #20624, one set of permission reads can authorize each event that was in the queue when the set opened. The queue holds at most 1,000 events or 1 MiB. The set stays in use for one second. If a membership, an API key or a task grant is revoked in that time, the server still sends the remaining queued events. **Expected behavior** A revocation stops delivery after a small, fixed quantity of queued output. The documentation states that quantity. **Steps to reproduce** 1. Open a company WebSocket as a user who can read a private task. 2. Publish 1,000 transcript events for a run of that task in one burst. 3. Revoke the task grant while the server delivers the burst. 4. The server delivers the remaining queued events. **Paperclip version or commit** `master` at 90dd864bff. ## What Changed - `live-event-delivery.ts` replaces a shared read set after 100 events or 250 ms. Before, the limit was one second with no event limit. - `live-event-delivery.test.ts` pins the exact limits. One set serves events 2 to 101 and event 102 opens a new set. Access is revoked during that set, and event 102 is not delivered. A set is kept at 249 ms and replaced at 250 ms. - `doc/live-transcript-delivery.md` states the limit and that queued output from before a revocation can still be sent up to that limit. ## Verification - `npx vitest run src/realtime src/__tests__/live-events-run-read.test.ts` in `server/`. 66 tests pass. - The boundary tests fail when either limit changes by one in either direction (99, 101 events; 249, 251 ms). - The 200-event burst test against PostgreSQL issues 22 statements with this limit. Separate reads issued 1,401. - Not run locally: a clean server typecheck. The local worktree reports type errors in `server/src/services/chat-channels/slack/adapter.ts`, which this change does not touch. CI supplies the typecheck. - Not verified: delivery delay on a hosted instance after deployment. ## Risks - More permission reads for large bursts: one set for each 100 queued events. A burst of 1,000 events uses 10 sets, not 1,000. - Output produced before a revocation and already in the queue can still be sent, up to 100 events or 250 ms. Before #20624 the limit was the one event in flight. - 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 - [x] All Paperclip CI gates are green - [x] 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 | 6 ++- .../src/realtime/live-event-delivery.test.ts | 46 ++++++++++++++++++- server/src/realtime/live-event-delivery.ts | 9 ++-- 3 files changed, 54 insertions(+), 7 deletions(-) diff --git a/doc/live-transcript-delivery.md b/doc/live-transcript-delivery.md index d17e4f45da5..98362e8d139 100644 --- a/doc/live-transcript-delivery.md +++ b/doc/live-transcript-delivery.md @@ -21,11 +21,13 @@ 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; +250 ms or 100 events, so a large queue or 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. +event published after it. Events that were already queued when access was revoked +can still be delivered until their set is replaced: at most 100 events or 250 ms +of output produced before the revocation. ## Additive negotiation diff --git a/server/src/realtime/live-event-delivery.test.ts b/server/src/realtime/live-event-delivery.test.ts index d46255525d8..7d3dd8c40b3 100644 --- a/server/src/realtime/live-event-delivery.test.ts +++ b/server/src/realtime/live-event-delivery.test.ts @@ -235,14 +235,14 @@ it("shares one permission read among events queued behind it, never with an even expect(read).toHaveBeenCalledTimes(4); f.delivery.dispose(); }); -it("stops sharing a read after one second and never reuses a failed read", async () => { +it("stops sharing a read after 250 ms 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; + now += 100; return allowed ? value : null; })); f.subscribe(["run-1"]); @@ -261,3 +261,45 @@ it("stops sharing a read after one second and never reuses a failed read", async g.delivery.dispose(); } finally { clock.mockRestore(); } }); +it("serves exactly 100 events from one read set, so the 101st sees a revocation", async () => { + let allowed = true; + const read = vi.fn(async () => allowed); + const sets = new Map(); + const f = fixture(vi.fn(async (value: LiveEvent, reads?: LiveDeliveryReads) => { + sets.set(value.payload.seq as number, reads); + const visible = await sharedDeliveryRead(reads, "permission", read); + if (value.payload.seq === 2) allowed = false; + return visible ? value : null; + })); + f.subscribe(["run-1"]); + for (let seq = 1; seq <= 102; seq++) f.delivery.enqueue(event(seq)); + await flush(); + expect(read).toHaveBeenCalledTimes(3); + expect(sets.get(2)).not.toBe(sets.get(1)); + for (let seq = 3; seq <= 101; seq++) expect(sets.get(seq)).toBe(sets.get(2)); + expect(sets.get(102)).not.toBe(sets.get(101)); + expect(f.frames.filter(frame => frame.type === "heartbeat.run.log").map(frame => frame.payload.seq)) + .toEqual(Array.from({ length: 101 }, (_, index) => index + 1)); + f.delivery.dispose(); +}); +it("keeps a read set at 249 ms and replaces it at 250 ms", async () => { + let now = 1_000_000; + const clock = vi.spyOn(Date, "now").mockImplementation(() => now); + try { + const steps = new Map([[2, 249], [3, 1]]); + const sets = new Map(); + const f = fixture(vi.fn(async (value: LiveEvent, reads?: LiveDeliveryReads) => { + const seq = value.payload.seq as number; + sets.set(seq, reads); + now += steps.get(seq) ?? 0; + return value; + })); + f.subscribe(["run-1"]); + for (let seq = 1; seq <= 4; seq++) f.delivery.enqueue(event(seq)); + await flush(); + expect(sets.get(3)).toBe(sets.get(2)); + expect(sets.get(4)).not.toBe(sets.get(3)); + expect(f.frames.filter(frame => frame.type === "heartbeat.run.log")).toHaveLength(4); + f.delivery.dispose(); + } finally { clock.mockRestore(); } +}); diff --git a/server/src/realtime/live-event-delivery.ts b/server/src/realtime/live-event-delivery.ts index f4aad8b7938..3c49e992bfb 100644 --- a/server/src/realtime/live-event-delivery.ts +++ b/server/src/realtime/live-event-delivery.ts @@ -4,7 +4,8 @@ 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; +const SHARED_READS_MAX_AGE_MS = 250; +const SHARED_READS_MAX_EVENTS = 100; export type LiveDeliveryReads = Map>; export function sharedDeliveryRead(reads: LiveDeliveryReads | undefined, key: string, read: () => Promise): Promise { if (!reads) return read(); @@ -39,11 +40,13 @@ export function createLiveEventDelivery(input: { let reads: LiveDeliveryReads = new Map(); let readsCoverSeq = 0; let readsOpenedAt = 0; + let readsServed = 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; + if (seq > readsCoverSeq || readsServed >= SHARED_READS_MAX_EVENTS || now - readsOpenedAt >= SHARED_READS_MAX_AGE_MS) { + reads = new Map(); readsCoverSeq = enqueuedSeq; readsOpenedAt = now; readsServed = 0; } + readsServed += 1; return reads; } function closeReads() { reads = new Map(); readsCoverSeq = 0; }