fix(guest): pace queued prompts across reconnect and retry room rate refusals (SW-12)

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<string>());
   const socket = useRef<ShareSocket | undefined>(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))

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<typeof setTimeout>),
+};
+
+const decided = (state: GuestPrompt["state"]) =>
+  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 {
+  private connected = false;
+  private stopped = false;
+  private timer?: unknown;
+  private nextAt = 0;
+  private readonly sent = new Set<string>();
+  private readonly retryAt = new Map<string, number>();
+
+  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();
+  }
+}

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({
@@ -211,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();
 });

webviews/test/guest-send-queue.test.ts

@@ -0,0 +1,148 @@
+// 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<number, { at: number; run: () => 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"]);
+});
+
+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();
+});