From 595208968ff2a56c1de81c1a91ad77248cdb0acf Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Nov 2026 02:45:37 -0800 Subject: [PATCH 1/6] fix(guest): pace chat prompts and retry transient room rate refusals Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- .../src/agent-session/guest/guestSendQueue.ts | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 webviews/src/agent-session/guest/guestSendQueue.ts diff --git a/webviews/src/agent-session/guest/guestSendQueue.ts b/webviews/src/agent-session/guest/guestSendQueue.ts new file mode 100644 index 000000000000..ba65b764b641 --- /dev/null +++ b/webviews/src/agent-session/guest/guestSendQueue.ts @@ -0,0 +1,116 @@ +// One paced outbound path for a guest's fresh messages and replay after reconnect. +// The room accepts two prompt frames/second, burst two; sending a saved backlog at once +// made the third and later messages permanently refused. Keep each client's original id, +// and retry only a temporary rate refusal, so acpmux's deduplication still owns execution. +import type { GuestPromptState } from "./shareSocket"; +import { type GuestPrompt, type GuestPrompts } from "./guestPrompts"; + +export const GUEST_PROMPT_GAP_MS = 600; +export const GUEST_PROMPT_RETRY_MS = 1_200; + +export type GuestSendTimers = { + now: () => number; + set: (run: () => void, ms: number) => unknown; + clear: (timer: unknown) => void; +}; + +const timers: GuestSendTimers = { + now: () => performance.now(), + set: (run, ms) => setTimeout(run, ms), + clear: (timer) => clearTimeout(timer as ReturnType), +}; + +const decided = (state: GuestPrompt["state"]) => + state === "approved" || state === "declined" || state === "removed" || state === "refused"; + +/** One guest tab's outbox. A socket snapshot proves it can send; a close stops its timer. */ +export class GuestSendQueue { + private connected = false; + private stopped = false; + private timer?: unknown; + private nextAt = 0; + private readonly sent = new Set(); + private readonly retryAt = new Map(); + + constructor( + private readonly prompts: GuestPrompts, + private readonly send: (prompt: GuestPrompt) => boolean, + private readonly clock: GuestSendTimers = timers, + ) {} + + /** The current socket is synced. Fold server decisions before retrying unacknowledged messages. */ + snapshot(states: readonly GuestPromptState[]): void { + if (this.stopped) return; + this.disconnected(); + for (const state of states) this.status(state); + this.sent.clear(); + this.connected = true; + this.pump(); + } + + /** The composer recorded another message. Send it through the same budget as replayed ones. */ + queued(): void { + this.pump(); + } + + /** A refusal caused by the room's burst limit is retryable, with the same client id. */ + status(state: GuestPromptState): void { + if (this.stopped) return; + const prompt = this.prompts.list().find((entry) => entry.clientId === state.client_id); + if (!prompt || decided(prompt.state)) return; + if (state.state === "refused" && state.reason === "share.rate_limited") { + // An earlier duplicate's late refusal must never undo a prompt already held by acpmux. + if (prompt.state !== "sending") return; + this.sent.delete(prompt.clientId); + this.retryAt.set(prompt.clientId, this.clock.now() + GUEST_PROMPT_RETRY_MS); + this.cancelTimer(); + this.pump(); + return; + } + this.sent.delete(prompt.clientId); + this.retryAt.delete(prompt.clientId); + this.prompts.update(state); + this.pump(); + } + + /** The socket is down; the next snapshot will replay the same unsent ids. */ + disconnected(): void { + this.connected = false; + this.cancelTimer(); + } + + stop(): void { + this.stopped = true; + this.disconnected(); + this.sent.clear(); + this.retryAt.clear(); + } + + private cancelTimer(): void { + if (this.timer !== undefined) this.clock.clear(this.timer); + this.timer = undefined; + } + + private pump(): void { + if (!this.connected || this.stopped || this.timer !== undefined) return; + const next = this.prompts.unsent().find((prompt) => !this.sent.has(prompt.clientId)); + if (!next) return; + const now = this.clock.now(); + const wait = Math.max(0, this.nextAt - now, (this.retryAt.get(next.clientId) ?? 0) - now); + if (wait > 0) { + this.timer = this.clock.set(() => { + this.timer = undefined; + this.pump(); + }, wait); + return; + } + if (!this.send(next)) { + this.disconnected(); + return; + } + this.sent.add(next.clientId); + this.retryAt.delete(next.clientId); + this.nextAt = now + GUEST_PROMPT_GAP_MS; + this.pump(); + } +} From 697e74abdf57b21f320a6c6874c7a78b84859f3f Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Nov 2026 02:46:12 -0800 Subject: [PATCH 2/6] fix(guest): route reconnect backlog and new sends through one paced outbox Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- .../src/agent-session/guest/GuestPage.tsx | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/webviews/src/agent-session/guest/GuestPage.tsx b/webviews/src/agent-session/guest/GuestPage.tsx index be387d584247..84aca8aa620b 100644 --- a/webviews/src/agent-session/guest/GuestPage.tsx +++ b/webviews/src/agent-session/guest/GuestPage.tsx @@ -13,6 +13,7 @@ import { PresenceStack } from "../acpmux/multiplayer/Faces"; import { displayName, participant, type Participant } from "../acpmux/multiplayer/participants"; import { GuestComposer, GuestMessages } from "./GuestComposer"; import { GuestPrompts } from "./guestPrompts"; +import { GuestSendQueue } from "./guestSendQueue"; import { joinName, type Redeem, type RedeemError } from "./redeem"; import { TypingSignal } from "./typingSignal"; import { @@ -150,6 +151,10 @@ export function LiveChat({ const [hasMore, setHasMore] = useState(false); const [expanded, setExpanded] = useState(() => new Set()); const socket = useRef(undefined); + const sendQueue = useMemo( + () => new GuestSendQueue(prompts, (prompt) => socket.current?.send({ t: "prompt", client_id: prompt.clientId, text: prompt.text }) ?? false), + [prompts], + ); const pageArrived = useRef<(() => void) | undefined>(undefined); const ended = useRef(onEnded); ended.current = onEnded; @@ -166,10 +171,7 @@ export function LiveChat({ setChat(next.chat); setHasMore(next.hasMore); client.feed(next.records, "snapshot"); - prompts.snapshot(next.prompts); - // Messages the room never acknowledged go again, as the same prompt (same client id). - for (const prompt of prompts.unsent()) - room.send({ t: "prompt", client_id: prompt.clientId, text: prompt.text }); + sendQueue.snapshot(next.prompts); typing.reset(); // A page asked for on the old connection never comes; the new history is in place. pageArrived.current?.(); @@ -183,9 +185,10 @@ export function LiveChat({ pageArrived.current = undefined; }, presence: setPeople, - promptState: (state) => prompts.update(state), - connection: setConnection, + promptState: (state) => sendQueue.status(state), + connection: (state) => { setConnection(state); if (state !== "live") sendQueue.disconnected(); }, ended: (reason) => { + sendQueue.stop(); client.dispose(); ended.current(reason); }, @@ -194,11 +197,12 @@ export function LiveChat({ socket.current = room; room.connect(); return () => { + sendQueue.disconnected(); room.stop(); client.dispose(); pageArrived.current?.(); }; - }, [socketUrl, openSocket, prompts, typing]); + }, [socketUrl, openSocket, prompts, typing, sendQueue]); const loadOlder = useCallback( () => @@ -225,9 +229,8 @@ export function LiveChat({ }; const send = (text: string) => { typing.stop(); - const prompt = prompts.add(text); - // Without a live connection it stays `sending` and goes with the next snapshot. - socket.current?.send({ t: "prompt", client_id: prompt.clientId, text }); + prompts.add(text); + sendQueue.queued(); }; const viewing = people .filter((person) => person.viewing !== false && !isMe(person.id, guestId)) From d2fd0830a97fef4fa9207f41cc8a42f78c540dbd Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Nov 2026 02:46:57 -0800 Subject: [PATCH 3/6] test(guest): use fake clock for offline prompt burst and retry identity Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- webviews/test/guest-send-queue.test.ts | 134 +++++++++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 webviews/test/guest-send-queue.test.ts diff --git a/webviews/test/guest-send-queue.test.ts b/webviews/test/guest-send-queue.test.ts new file mode 100644 index 000000000000..8e640b784c6f --- /dev/null +++ b/webviews/test/guest-send-queue.test.ts @@ -0,0 +1,134 @@ +// The guest page's single paced outbox: offline backlog, online bursts, a transient room refusal, +// stable ids through reconnect and no sends after revoke. The fake clock has zero wall-clock sleeps. +import { expect, test } from "bun:test"; +import { GuestPrompts } from "../src/agent-session/guest/guestPrompts"; +import { GUEST_PROMPT_GAP_MS, GUEST_PROMPT_RETRY_MS, GuestSendQueue, type GuestSendTimers } from "../src/agent-session/guest/guestSendQueue"; +import { memoryStorage } from "../src/agent-session/guest/fixtures/liveChat"; + +function fakeClock() { + let now = 0; + let nextId = 0; + const jobs = new Map void }>(); + const timers: GuestSendTimers = { + now: () => now, + set: (run, ms) => { + const id = ++nextId; + jobs.set(id, { at: now + ms, run }); + return id; + }, + clear: (id) => void jobs.delete(id as number), + }; + const advance = (ms: number) => { + const until = now + ms; + for (let steps = 0; steps < 1_000; steps += 1) { + const next = [...jobs].filter(([, job]) => job.at <= until).sort((a, b) => a[1].at - b[1].at)[0]; + if (!next) { now = until; return; } + now = next[1].at; + jobs.delete(next[0]); + next[1].run(); + } + throw new Error("a paced guest outbox scheduled too many timers"); + }; + return { timers, advance, jobs }; +} + +function fixture() { + const clock = fakeClock(); + let id = 0; + const prompts = new GuestPrompts("share", memoryStorage(), (bytes) => bytes.fill(++id), clock.timers.now); + const sent: Array<{ id: string; text: string; at: number }> = []; + const queue = new GuestSendQueue(prompts, (prompt) => { + sent.push({ id: prompt.clientId, text: prompt.text, at: clock.timers.now() }); + return true; + }, clock.timers); + const add = (text: string) => { + const p = prompts.add(text); + queue.queued(); + return p; + }; + return { clock, prompts, sent, queue, add }; +} + +test("five messages queued while offline drain in order, at most one every 600 ms", () => { + const f = fixture(); + const added = Array.from({ length: 5 }, (_, index) => f.add(`offline ${index}`)); + expect(f.sent).toEqual([]); + f.queue.snapshot([]); + expect(f.sent.map((msg) => msg.text)).toEqual(["offline 0"]); + for (let index = 1; index < 5; index += 1) { + f.clock.advance(GUEST_PROMPT_GAP_MS - 1); + expect(f.sent.length).toBe(index); + f.clock.advance(1); + expect(f.sent.length).toBe(index + 1); + } + expect(f.sent.map((msg) => msg.id)).toEqual(added.map((prompt) => prompt.clientId)); + expect(f.sent.map((msg) => msg.at)).toEqual([0, 600, 1200, 1800, 2400]); + f.queue.stop(); +}); + +test("rapid sends while live use the same pacing as a replayed backlog", () => { + const f = fixture(); + f.queue.snapshot([]); + const first = f.add("one"); + f.add("two"); + f.add("three"); + expect(f.sent.map((msg) => msg.text)).toEqual(["one"]); + f.queue.status({ client_id: first.clientId, state: "held" }); + f.clock.advance(GUEST_PROMPT_GAP_MS); + expect(f.sent.map((msg) => msg.text)).toEqual(["one", "two"]); + f.clock.advance(GUEST_PROMPT_GAP_MS); + expect(f.sent.map((msg) => msg.text)).toEqual(["one", "two", "three"]); + f.queue.stop(); +}); + +test("share.rate_limited is retried after a gap with the same client id, then held", () => { + const f = fixture(); + f.queue.snapshot([]); + const prompt = f.add("retry me"); + expect(f.sent.map((msg) => msg.id)).toEqual([prompt.clientId]); + f.queue.status({ client_id: prompt.clientId, state: "refused", reason: "share.rate_limited" }); + expect(f.prompts.unsent().map((each) => each.clientId)).toEqual([prompt.clientId]); + f.clock.advance(GUEST_PROMPT_RETRY_MS - 1); + expect(f.sent.length).toBe(1); + f.clock.advance(1); + expect(f.sent.map((msg) => msg.id)).toEqual([prompt.clientId, prompt.clientId]); + f.queue.status({ client_id: prompt.clientId, state: "held" }); + expect(f.prompts.held()).toEqual([prompt.clientId]); + f.queue.disconnected(); + f.queue.snapshot([{ client_id: prompt.clientId, state: "held" }]); + f.clock.advance(2_000); + expect(f.sent.length).toBe(2); + f.queue.stop(); +}); + +test("a reconnect replays an unanswered message once with the same id", () => { + const f = fixture(); + f.queue.snapshot([]); + const prompt = f.add("sent before drop"); + f.queue.disconnected(); + f.clock.advance(100); + f.queue.snapshot([]); + expect(f.sent.length).toBe(1); + f.clock.advance(GUEST_PROMPT_GAP_MS - 100); + expect(f.sent.map((msg) => msg.id)).toEqual([prompt.clientId, prompt.clientId]); + f.queue.status({ client_id: prompt.clientId, state: "approved" }); + f.queue.disconnected(); + f.queue.snapshot([]); + f.clock.advance(2_000); + expect(f.sent.length).toBe(2); + f.queue.stop(); +}); + +test("permanent refusals remain decided; end cancels unsent messages and timers", () => { + const f = fixture(); + f.queue.snapshot([]); + const first = f.add("accepted"); + f.add("queued until stop"); + f.queue.status({ client_id: first.clientId, state: "refused", reason: "share.too_large" }); + expect(f.prompts.unsent().length).toBe(1); + expect(f.clock.jobs.size).toBe(1); + f.queue.stop(); + expect(f.clock.jobs.size).toBe(0); + f.clock.advance(5_000); + expect(f.sent.map((msg) => msg.text)).toEqual(["accepted"]); +}); From 3f1996156f02ed395ace75d1af609de4f2ae4c33 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Mon, 2 Nov 2026 01:51:32 +0000 Subject: [PATCH 4/6] test(guest): wait for paced queued prompt Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- webviews/test/guest-composer.test.tsx | 3 +++ 1 file changed, 3 insertions(+) diff --git a/webviews/test/guest-composer.test.tsx b/webviews/test/guest-composer.test.tsx index 59ccf386fb48..651deaa01601 100644 --- a/webviews/test/guest-composer.test.tsx +++ b/webviews/test/guest-composer.test.tsx @@ -184,6 +184,9 @@ test("declined and refused messages read as one quiet line, the reason said", as await type(page.host, text); await enter(page.host); } + // The outbox paces rapid submits; observe the second frame instead of reading it + // before the 600 ms queue timer fires. + await fake.until(() => fake.prompts().length === 2); const [skip, bump] = fake.prompts(); await fake.frame({ v: 1, t: "prompt_state", client_id: skip!.client_id, state: "declined" }); await fake.frame({ From c27cfdfe11aed854c7f17c7c78a886e1999ad14e Mon Sep 17 00:00:00 2001 From: Leo Li Date: Mon, 2 Nov 2026 02:02:10 +0000 Subject: [PATCH 5/6] test(guest): await paced reconnect replay Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- webviews/test/guest-composer.test.tsx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/webviews/test/guest-composer.test.tsx b/webviews/test/guest-composer.test.tsx index 651deaa01601..52e22fcce9fd 100644 --- a/webviews/test/guest-composer.test.tsx +++ b/webviews/test/guest-composer.test.tsx @@ -214,6 +214,8 @@ test("a message sent while the room is away goes after the reconnect, with the s // The room heard nothing of it; the next snapshot says nothing about it, so it goes again. await fake.until(() => fake.sockets.length === 2); await fake.join(); + // Reconnect replay shares the outbox pacing; observe the frame after its gap. + await fake.until(() => fake.prompts().length === 1); expect(fake.prompts()).toEqual([sent!]); await page.unmount(); }); From 2bbce664eedb7f3ae5cb08ef41b59dd603e87daa Mon Sep 17 00:00:00 2001 From: Leo Li Date: Mon, 2 Nov 2026 02:23:32 +0000 Subject: [PATCH 6/6] fix(guest): accept delayed state after timeout refusal Agent-Session: https://cmux.com/deeplink/agent?id=aid-b68baba69c1ba3838513 --- webviews/src/agent-session/guest/guestSendQueue.ts | 2 +- webviews/test/guest-send-queue.test.ts | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/webviews/src/agent-session/guest/guestSendQueue.ts b/webviews/src/agent-session/guest/guestSendQueue.ts index ba65b764b641..ad55a6dee7ab 100644 --- a/webviews/src/agent-session/guest/guestSendQueue.ts +++ b/webviews/src/agent-session/guest/guestSendQueue.ts @@ -21,7 +21,7 @@ const timers: GuestSendTimers = { }; const decided = (state: GuestPrompt["state"]) => - state === "approved" || state === "declined" || state === "removed" || state === "refused"; + state === "approved" || state === "declined" || state === "removed"; /** One guest tab's outbox. A socket snapshot proves it can send; a close stops its timer. */ export class GuestSendQueue { diff --git a/webviews/test/guest-send-queue.test.ts b/webviews/test/guest-send-queue.test.ts index 8e640b784c6f..cd86078f2860 100644 --- a/webviews/test/guest-send-queue.test.ts +++ b/webviews/test/guest-send-queue.test.ts @@ -132,3 +132,17 @@ test("permanent refusals remain decided; end cancels unsent messages and timers" f.clock.advance(5_000); expect(f.sent.map((msg) => msg.text)).toEqual(["accepted"]); }); + +test("a timeout refusal can be superseded by a delayed held and owner decision", () => { + const f = fixture(); + f.queue.snapshot([]); + const prompt = f.add("eventually accepted"); + f.queue.status({ client_id: prompt.clientId, state: "refused", reason: "agent_session.timeout" }); + expect(f.prompts.list()[0]?.state).toBe("refused"); + f.queue.status({ client_id: prompt.clientId, state: "held" }); + expect(f.prompts.list()[0]?.state).toBe("held"); + f.queue.status({ client_id: prompt.clientId, state: "approved" }); + expect(f.prompts.list()[0]?.state).toBe("approved"); + expect(f.sent).toHaveLength(1); + f.queue.stop(); +});