diff --git a/DESIGN.md b/DESIGN.md index c7b1b75084..cfbdf11539 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -2616,6 +2616,11 @@ Alert rows of the table above from the `--droplet-*` roles in `src/styles.css`. the message until the transcript proves it. Next-turn delivery and an explicit Send now action claim the same server ID. Connection loss never resumes automatic input. Previously held browser messages retain their explicit Send now/Discard recovery. + Any other follow-up's exact `sent` or `discarded` receipt is kept the same way (ten minutes, 1024, + same browser resume ID and device, local machine only) and replayed with `resumed: true` on its + browser's reattach, or to that browser's already attached successor when the Enter's reply comes + after the reload. It settles only that exact ID's saved copy: an uncertain copy, which ages out as + `dismissed`, stays unconfirmed, and nothing is typed again (#2421). - Not connected, the one word `Reconnecting…` (`Connecting…` before this page's first connection to the PC's socket) is the placeholder while the box is empty and moves into the status content once there is a draft (`composerStatusHint`), on a phone too; it says no more (nothing is queued, and diff --git a/changelog.d/pending-receipt-resume.md b/changelog.d/pending-receipt-resume.md new file mode 100644 index 0000000000..8c5be09373 --- /dev/null +++ b/changelog.d/pending-receipt-resume.md @@ -0,0 +1,5 @@ +### Fixed +- A message that went to the terminal no longer stays "Not confirmed" when the connection dropped + before its receipt arrived: the browser that sent it gets the receipt back when it reconnects or + reloads, within ten minutes. Nothing is sent again, and a message whose delivery was never + confirmed stays unconfirmed. diff --git a/server/core/pending.test.ts b/server/core/pending.test.ts index 5344b247c9..c09acfbd54 100644 --- a/server/core/pending.test.ts +++ b/server/core/pending.test.ts @@ -128,6 +128,50 @@ it.each(["delivery first", "withdrawal first", "ambiguous delivery first", "ambi } }); +it.each(["same browser", "no device", "wrong browser", "wrong device", "read-only", "revoked", "no resume id", "expired", "uncertain", + "settled after reattach", "late wrong browser", "late wrong device", "late read-only", "late revoked"] as const)( + "replays a follow-up's exact receipt only to the browser that sent it (#2421): %s", (name) => { + // "late": the reload attaches before the Enter's reply settles the send, so the receipt is pushed to it + const late = name === "settled after reattach" || name.startsWith("late "); + const scenario = name.replace(/^late /, ""); + let now = Date.now(); + const clock = spyOn(Date, "now").mockImplementation(() => now); + try { + const token = crypto.randomUUID(), frames = new Map(), clients = new Set(), revoked = new Set(); + const client = (deviceId: string | undefined, receiptResumeId: string | undefined, readOnly = false): Client => { + const output: ServerMessage[] = []; + const owner = { data: connectionData({ deviceId, readOnly }), getBufferedAmount: () => 0, + send: (value: string) => { output.push(JSON.parse(value)); return 1; } } as Client; + owner.data.attached.set("p", { keepSize: false, receiptResumeId }); + clients.add(owner); frames.set(owner, output); return owner; + }; + const device = scenario === "no device" ? undefined : "device"; + const first = client(device, scenario === "no resume id" ? undefined : token); + const delivery = createPendingDelivery({ + options: {}, clients, attachments: new Map(), serialize: async (_pane, task) => task(), lastTyped: new Map(), + sight: () => { throw new Error("a receipt replay takes no input lease"); }, + submitText: async () => { throw new Error("a receipt replay never types"); }, + authorizeSocket: (owner) => { if (revoked.has(owner)) throw new Error("revoked"); }, + clientSnapshot: async () => ({ panes: [] }) as unknown as SessionSnapshot, + omo: { sessionOf: () => null }, SUBMIT_DEADLINE_MS: 45_000, SUBMIT_DELAY_MS: 0, TYPED_SETTLE_MS: 0, + }); + const item = delivery.pending.enqueue(first, "p", 1, "follow-up", null, { agent: "codex", terminalId: "p", session: null }); + expect(delivery.pending.claim(item, false)).toBe(true); + // an Enter nobody confirmed leaves an uncertain copy, which only ever ages out as dismissed + if (scenario === "uncertain") delivery.pending.settle(item, { code: "submit_changed", message: "Enter was not confirmed" }, true); + else if (!late) delivery.pending.settle(item); + clients.delete(first); delivery.closeConnection(first); + if (scenario === "uncertain") { now += 60_001; delivery.pending.expire(); } + if (scenario === "expired") now += 10 * 60 * 1000 + 1; + const resumed = client(scenario === "wrong device" ? "other" : device, scenario === "wrong browser" ? crypto.randomUUID() : token, scenario === "read-only"); + if (scenario === "revoked") revoked.add(resumed); + delivery.attached(resumed, "p"); + if (late) delivery.pending.settle(item); + const removed = frames.get(resumed)!.flatMap((frame) => frame.type === "pending-messages" ? frame.removed ?? [] : []); + expect(removed).toEqual(["same browser", "no device", "settled after reattach"].includes(scenario) ? [{ id: item.message.id, outcome: "sent", resumed: true }] : []); + } finally { clock.mockRestore(); } + }); + it.each(["before close", "after close", "overlapping reload", "same live browser", "no device", "read-only", "detach", "wrong browser", "wrong device", "revoked", "expired", "pending cap", "completed cap", "confirmation budget"] as const)( "resumes only exact, authorized Pi presentation receipts: %s", async (when) => { const root = mkdtempSync(join(tmpdir(), "ovis-pi-resume-")); roots.push(root); diff --git a/server/core/pending.ts b/server/core/pending.ts index 33b9c91458..6750e1fd74 100644 --- a/server/core/pending.ts +++ b/server/core/pending.ts @@ -21,7 +21,7 @@ import { closeBackslash, closeMention, type createSubmitText, freeClaudeBox, fre import { type Client, connectionOf, send } from "./connections.ts"; import { ATTACH_HELD_MESSAGE } from "./terminal-process.ts"; import type { AttachmentView, TerminalSighting } from "./terminals.ts"; -import type { ClientMessage, HerdrPane, PendingMessage, SessionSnapshot } from "../../shared/protocol.ts"; +import type { ClientMessage, HerdrPane, PendingMessage, PendingRemoval, SessionSnapshot } from "../../shared/protocol.ts"; import { typingGuards } from "../input-guard.ts"; import { claudeForegroundSession, isClaudeProcess } from "../claude-store.ts"; import { HerdrError, paneProcessInfo, paneRead, paneSendKeys, paneSendText, type ProcessInfo } from "../herdr/client.ts"; @@ -112,7 +112,11 @@ export function createPendingDelivery({ options, clients, attachments, sight, se return typingGuards.pendingLease(sight(client, paneId).facts()); }); const pending = new PendingInputs((owner, paneId, messages, removed) => { - if (isClient(owner)) { send(owner, { type: "pending-messages", pane_id: paneId, messages, ...(removed ? { removed } : {}) }); return; } + if (isClient(owner)) { + send(owner, { type: "pending-messages", pane_id: paneId, messages, ...(removed ? { removed } : {}) }); + keepReceipts(owner, paneId, messages, removed); + return; + } // a first message: every connection attached to the pane sees it, and keeps the receipt once it went if (removed) firstReceipts.set(paneId, [...(firstReceipts.get(paneId) ?? []), ...removed].slice(-FIRST_MESSAGE_RECEIPTS)); for (const client of clients) { @@ -158,6 +162,10 @@ export function createPendingDelivery({ options, clients, attachments, sight, se const PI_RESUME_MS = 10 * 60 * 1000, PI_RESUME_LIMIT = 1024; const livePiOwner = (owner: Client): boolean => clients.has(owner) && !connectionOf(owner).closing; const piCompleted: Array<{ paneId: string; key: string; id: string; at: number }> = []; + /** a follow-up's sent or discarded receipt, kept for its browser's return (#2421): the same bounds and binding as Pi's */ + const completed: Array<{ paneId: string; key: string; id: string; outcome: "sent" | "discarded"; at: number }> = []; + /** the resume key each follow-up's browser had when the bridge took it, by connection and message id */ + const receiptKeys = new WeakMap>(); const resumeKey = (client: Client, paneId: string): string | undefined => { const id = client.data.attached.get(paneId)?.receiptResumeId; return id ? JSON.stringify([client.data.deviceId ?? null, id]) : undefined; @@ -165,6 +173,7 @@ export function createPendingDelivery({ options, clients, attachments, sight, se const prunePi = (): void => { const cutoff = Date.now() - PI_RESUME_MS; while (piCompleted.length && (piCompleted[0]!.at < cutoff || piCompleted.length > PI_RESUME_LIMIT)) piCompleted.shift(); + while (completed.length && (completed[0]!.at < cutoff || completed.length > PI_RESUME_LIMIT)) completed.shift(); let count = 0; for (const [paneId, items] of piTyped) { const kept = items.filter((item) => { @@ -883,13 +892,42 @@ export function createPendingDelivery({ options, clients, attachments, sight, se item.owner = client; rebound = true; } - const removed = piCompleted.filter((item) => item.paneId === paneId && item.key === key) - .map(({ id }) => ({ id, outcome: "consumed" as const })); + const removed = [ + ...piCompleted.filter((item) => item.paneId === paneId && item.key === key).map(({ id }) => ({ id, outcome: "consumed" as const })), + ...completed.filter((item) => item.paneId === paneId && item.key === key).map(({ id, outcome }) => ({ id, outcome, resumed: true as const })), + ]; if (removed.length) send(client, { type: "pending-messages", pane_id: paneId, messages: [], removed }); } catch { /* failed authorization cannot resume even presentation */ } } return rebound; } + /** + * Keeps a follow-up's sent or discarded receipt for its browser's return: a socket that drops before the + * receipt reaches it leaves the browser's copy unconfirmed. Only an exact outcome is kept (never a dismissed + * uncertain copy), bound to the device and browser that sent it, and it is presentation only: nothing is typed. + */ + function keepReceipts(owner: Client, paneId: string, messages: PendingMessage[], removed: readonly PendingRemoval[] | undefined): void { + const key = resumeKey(owner, paneId); + const keys = receiptKeys.get(owner) ?? new Map(); + if (key) for (const message of messages) if (!keys.has(message.id)) keys.set(message.id, key); + for (const item of removed ?? []) { + const kept = keys.get(item.id); + keys.delete(item.id); + if (!kept || (item.outcome !== "sent" && item.outcome !== "discarded")) continue; + completed.push({ paneId, key: kept, id: item.id, outcome: item.outcome, at: Date.now() }); + // a reload can attach its successor before the Enter's reply settles the send + for (const successor of clients) { + if (successor === owner || !livePiOwner(successor) || !successor.data.attached.has(paneId) + || resumeKey(successor, paneId) !== kept || connectionOf(successor).mode === "observe") continue; + try { + authorizeSocket(successor); + send(successor, { type: "pending-messages", pane_id: paneId, messages: [], removed: [{ id: item.id, outcome: item.outcome, resumed: true }] }); + } catch { /* failed authorization cannot resume even presentation */ } + } + } + if (keys.size) receiptKeys.set(owner, keys); else receiptKeys.delete(owner); + prunePi(); + } /** A connection that attached the pane: it sees the pane's first messages and their receipts, and may be the one to type them. */ function attached(client: Client, paneId: string): void { prunePi(); diff --git a/server/pending-input.contract.test.ts b/server/pending-input.contract.test.ts index 8e171f0a2e..089824df99 100644 --- a/server/pending-input.contract.test.ts +++ b/server/pending-input.contract.test.ts @@ -216,6 +216,46 @@ async function setup(label: string, agent = "claude", screen = "› Message\n", return { screenFile, pane, port: server.port, server, socket, bytes, waitBytes, state, showScreen, queue, removed, another, replacementLog }; } +describe("resumable follow-up receipts (#2421)", () => { + it.each(["sent", "discarded"] as const)("replays a Codex follow-up's exact %s receipt to its returning browser without typing again", async (outcome) => { + const f = await setup(`codex-resume-${outcome}`, "codex"); + const token = crypto.randomUUID(), from = f.socket.seen.length; + f.socket.send({ type: "attach", pane_id: f.pane, cols: 100, rows: 30, receipt_resume_id: token }); + await f.socket.wait((frame) => frame.type === "input-ready" && frame.pane_id === f.pane, from); + const row = await f.queue(980, "follow-up"); + if (outcome === "sent") { await f.state("idle"); await f.waitBytes(`${paste(row.text)}\r`); } + else { + f.socket.send({ type: "pending-action", id: 981, pane_id: f.pane, pending_id: row.id, action: "discard" }); + expect((await f.socket.action(981)).ok).toBe(true); + } + await f.removed(row.id, outcome); + const bytes = f.bytes(); + await f.socket.disconnect(); + const resumed = await f.another(false); + await resumed.attach(f.pane, false, token); + const replay = await f.removed(row.id, outcome, resumed); + expect(replay.removed).toContainEqual({ id: row.id, outcome, resumed: true }); + expect(f.bytes()).toBe(bytes); + }, 20_000); + + it("gives no receipt to a browser that attaches with another resume id", async () => { + const f = await setup("codex-resume-other", "codex"); + const from = f.socket.seen.length; + f.socket.send({ type: "attach", pane_id: f.pane, cols: 100, rows: 30, receipt_resume_id: crypto.randomUUID() }); + await f.socket.wait((frame) => frame.type === "input-ready" && frame.pane_id === f.pane, from); + const row = await f.queue(983, "follow-up"); + await f.state("idle"); await f.waitBytes(`${paste(row.text)}\r`); + await f.removed(row.id, "sent"); + await f.socket.disconnect(); + const other = await f.another(false); + await other.attach(f.pane, false, crypto.randomUUID()); + // frames arrive in order: a receipt sent on this attach would come before the pong + other.send({ type: "ping", id: 984 }); + await other.wait((frame) => frame.type === "pong" && frame.id === 984); + expect(other.seen.some((frame) => frame.type === "pending-messages" && frame.removed?.some((item: { id: string }) => item.id === row.id))).toBe(false); + }, 20_000); +}); + describe("Ovis-origin Pi transcript receipts", () => { const piLine = (id: string, parentId: string | null, role: string, text: string, stamp = Date.now()) => JSON.stringify({ type: "message", id, parentId, timestamp: new Date().toISOString(), message: { role, content: [{ type: "text", text }], timestamp: stamp }, diff --git a/shared/protocol.ts b/shared/protocol.ts index 56c85346e9..42f841eb33 100644 --- a/shared/protocol.ts +++ b/shared/protocol.ts @@ -858,6 +858,10 @@ export interface PromptAnswer { /** A connection's authority over the shared ptys: `interact` types and resizes, `observe` only watches. */ export type ClientRole = "interact" | "observe"; +/** A pending message's outcome. `resumed`: replayed after a reconnect, only to the device and browser + * (`receipt_resume_id`) that sent it, for its exact id; it settles a saved copy and never sends input. */ +export interface PendingRemoval { id: string; outcome: "sent" | "discarded" | "consumed" | "dismissed"; resumed?: true } + /** A follow-up accepted by this bridge, scoped to the connection that submitted it. */ export interface PendingMessage { id: string; @@ -1040,7 +1044,7 @@ export type ServerMessage = /** Owner-only receipt and state; only an explicit removed outcome permits deleting a retained client message. * `dismissed` removes an uncertain copy, never proves withdrawal or delivery. Existing clients * remove owned IDs for any explicit outcome; Edit restores only `discarded`. */ - | { type: "pending-messages"; pane_id: string; messages: PendingMessage[]; removed?: Array<{ id: string; outcome: "sent" | "discarded" | "consumed" | "dismissed" }> } + | { type: "pending-messages"; pane_id: string; messages: PendingMessage[]; removed?: PendingRemoval[] } | { type: "secret-result"; id: number; pane_id: string; ok: boolean; code?: string } /** agent-status push for ANY pane, attached or not (server-side status collector) */ | { type: "pane-status"; pane_id: string; agent_status: AgentStatus; finish_id?: number; finish_epoch?: string; /** a pane's running background tasks (OmO's, or a Claude Code pane's subagents and commands), when the frame is about them: it changes no status */ background_tasks?: number; /** every frame says it: absent, the pane does not wait on its turn's background work */ background_wait?: true; /** a pane's plan, when the frame is about it (null: it keeps none now); it changes no status */ plan?: PlanSummary | null } diff --git a/src/lib/machineSession.test.ts b/src/lib/machineSession.test.ts index ca174a9465..bfe3d852fd 100644 --- a/src/lib/machineSession.test.ts +++ b/src/lib/machineSession.test.ts @@ -61,6 +61,69 @@ function withStorage(run: () => void): void { } describe("PC connection ownership", () => { + it.each(["sent", "discarded"] as const)("settles a reloaded copy after the snapshot proves an early resumed %s receipt's lifetime", (outcome) => { + const { session, socket, store } = setup(`reload-receipt-${outcome}`); + const owner = paneStorageId(session.machineId, "w1:p1"); + store.accept(owner, message("saved"), null); + session.connect(); + socket.receive({ type: "features", features: ["pending-input"], connection: "reload", herdr_lifetime: "reload-herdr" }); + expect(session.scope).toBeNull(); + // Open's attach replay can finish while the server still awaits clientSnapshot(). + socket.receive({ type: "pending-messages", pane_id: "w1:p1", messages: [], removed: [{ id: "saved", outcome, resumed: true }] }); + expect(store.read(owner)).toHaveLength(1); + socket.receive(named("reload-herdr")); + expect(store.read(owner)).toEqual([]); + session.close(); + }); + it("settles durable reload state only after the lifetime fence exposes that herdr's storage", () => { + withStorage(() => { + const { session, socket } = setup("durable-reload", pendingMessages); + const owner = paneStorageId(session.machineId, "w1:p1"); + window.localStorage.setItem(`ovis:pane-lifetime:${session.machineId}`, "durable-herdr"); + window.localStorage.setItem(`ovis:pending:${owner}`, JSON.stringify({ version: 1, messages: [message("saved")] })); + session.connect(); + socket.receive({ type: "features", features: ["pending-input"], connection: "reload", herdr_lifetime: "durable-herdr" }); + socket.receive({ type: "pending-messages", pane_id: "w1:p1", messages: [], removed: [{ id: "saved", outcome: "sent", resumed: true }] }); + expect(window.localStorage.getItem(`ovis:pending:${owner}`)).not.toBeNull(); + socket.receive(named("durable-herdr")); + expect(pendingMessages.read(owner)).toEqual([]); + expect(window.localStorage.getItem(`ovis:pending:${owner}`)).toBeNull(); + session.close(); + }); + }); + it.each(["different lifetime", "no lifetime", "disconnected", "ordinary receipt"] as const)("does not settle an early receipt with %s", (scenario) => { + const { session, socket, store } = setup(`reload-reject-${scenario}`); + const owner = paneStorageId(session.machineId, "w1:p1"); + store.accept(owner, message("saved"), null); + session.connect(); + socket.receive({ type: "features", features: ["pending-input"], connection: "reload", herdr_lifetime: "early-herdr" }); + socket.receive({ type: "pending-messages", pane_id: "w1:p1", messages: [], removed: [{ + id: "saved", outcome: "sent", ...(scenario === "ordinary receipt" ? {} : { resumed: true }), + }] }); + if (scenario === "disconnected") { socket.disconnect(); session.connect(); } + socket.receive(scenario === "no lifetime" ? snapshot : named(scenario === "different lifetime" ? "other-herdr" : "early-herdr")); + expect(store.read(owner)).toMatchObject([{ id: "saved", serverOwned: false }]); + // A later matching snapshot cannot resurrect a rejected receipt. + socket.receive(named("early-herdr")); + expect(store.read(owner)).toHaveLength(1); + session.close(); + }); + it("bounds early resumed receipts and keeps only resumed removals from a mixed frame", () => { + const { session, socket, store } = setup("reload-bounded"); + const owner = paneStorageId(session.machineId, "w1:p1"); + store.accept(owner, message("oldest"), null); + store.accept(owner, message("latest"), null); + store.accept(owner, message("ordinary"), null); + session.connect(); + socket.receive({ type: "features", features: ["pending-input"], connection: "reload", herdr_lifetime: "bounded-herdr" }); + for (let i = 0; i < 129; i++) socket.receive({ + type: "pending-messages", pane_id: "w1:p1", messages: [message("live")], + removed: [{ id: i === 0 ? "oldest" : "latest", outcome: "sent", resumed: true }, { id: "ordinary", outcome: "sent" }], + }); + socket.receive(named("bounded-herdr")); + expect(store.read(owner).map(({ id }) => id)).toEqual(["oldest", "ordinary"]); + session.close(); + }); it("knows the PC's id for the open connection from its snapshot, and none once it drops", () => { const { session, socket } = setup(); expect(session.connection).toBeNull(); diff --git a/src/lib/machineSession.ts b/src/lib/machineSession.ts index 2c927ee19c..a75909e51e 100644 --- a/src/lib/machineSession.ts +++ b/src/lib/machineSession.ts @@ -1,4 +1,4 @@ -import type { PendingMessage, ServerMessage } from "../../shared/protocol.ts"; +import type { PendingMessage, PendingRemoval, ServerMessage } from "../../shared/protocol.ts"; import { paneStorageId } from "../../shared/machines.ts"; import { afterSend, afterSettled, greetingMemory, rememberGreeting } from "./greeting.ts"; import { paneLifetimes } from "./paneLifetime.ts"; @@ -24,6 +24,8 @@ export class MachineSession { private currentScope: string | null = null; private currentConnection: string | null = null; private currentEpoch = 0; + private announcedLifetime: string | undefined; + private resumedReceipts: Array<{ paneId: string; lifetime: string; removal: PendingRemoval }> = []; private role: RoleAck | null = null; private readonly handlers = new Set(); private readonly offMessage: () => void; @@ -41,6 +43,7 @@ export class MachineSession { // Installed before view subscribers, and kept even while this PC has no mounted pane. this.offMessage = this.socket.on((message) => { if (this.closed) return; + if (message.type === "features") this.announcedLifetime = message.herdr_lifetime; // a connection that came back to a replaced herdr, or to one it cannot prove, can say so before // the roster does: the earlier herdr's drafts and lenses leave the reused pane IDs before any // pane hears of it @@ -50,6 +53,22 @@ export class MachineSession { const told = message.type === "snapshot" || (message.type === "features" && this.knownHerdr(message.herdr_lifetime)); if (told && this.currentScope === null) this.currentScope = `${this.identity}:${this.currentEpoch}`; if (told && message.connection !== undefined) this.currentConnection = message.connection; + if (message.type === "pending-messages" && this.currentScope === null && this.announcedLifetime !== undefined) { + // Open replays attaches before the first snapshot. Keep only exact resumed receipts, + // tagged with this connection's announced herdr, never its live queue or ordinary removals. + for (const removal of message.removed ?? []) if (removal.resumed === true) { + this.resumedReceipts.push({ paneId: message.pane_id, lifetime: this.announcedLifetime, removal }); + if (this.resumedReceipts.length > 128) this.resumedReceipts.shift(); + } + } + if (message.type === "snapshot" && this.currentScope !== null) { + for (const receipt of this.resumedReceipts) { + if (receipt.lifetime === message.snapshot.herdr_lifetime && this.knownHerdr(receipt.lifetime)) { + this.store.publish(paneStorageId(this.machineId, receipt.paneId), [], [receipt.removal], this.currentScope); + } + } + this.resumedReceipts = []; + } if (message.type === "role-ack") this.role = message; let sent = false; if (message.type === "pending-messages" && this.currentScope !== null) { @@ -81,6 +100,8 @@ export class MachineSession { private disconnected(): void { if (this.currentScope !== null) this.store.suspendScope(this.currentScope); this.currentScope = null; + this.announcedLifetime = undefined; + this.resumedReceipts = []; this.currentConnection = null; this.currentEpoch++; this.role = null; diff --git a/src/lib/pendingMessages.test.ts b/src/lib/pendingMessages.test.ts index 7fc0e45f81..2564ba6dc1 100644 --- a/src/lib/pendingMessages.test.ts +++ b/src/lib/pendingMessages.test.ts @@ -9,6 +9,35 @@ const storage = () => { const message = (id: string, state: PendingMessage["state"] = "queued", text = "check the tests"): PendingMessage => ({ id, request_id: 1, text, state, created_at: "2026-10-06T00:00:00Z" }); describe("pending message ownership", () => { + it("settles a follow-up's exact resumed receipt after reload, never a same-text or live-owned row (#2421)", () => { + const port = storage(), before = new PendingMessageStore(() => port); + before.accept("p", message("delivered", "sending", "same"), "old"); + before.accept("p", message("unproved", "sending", "same"), "old"); + before.suspendScope("old"); + const store = new PendingMessageStore(() => port); + store.accept("p", message("live", "sending", "same"), "other"); + // a plain receipt for a saved copy proves nothing about it + store.publish("p", [], [{ id: "unproved", outcome: "sent" }], "new"); + expect(store.read("p").map((row) => row.id)).toEqual(["delivered", "unproved", "live"]); + store.publish("p", [], [{ id: "delivered", outcome: "sent", resumed: true }, { id: "live", outcome: "sent", resumed: true }], "new"); + expect(store.read("p").map((row) => row.id)).toEqual(["unproved", "live"]); + // an old tab's saved copy and a late acceptance cannot bring the settled id back + port.setItem("ovis:pending:p", JSON.stringify({ messages: [message("delivered", "sending", "same")] })); + store.refresh("p"); + store.accept("p", message("delivered", "sending", "same"), "late"); + // (that tab's save replaced the other saved row; the live connection's row stays) + expect(store.read("p").map((row) => row.id)).toEqual(["live"]); + // another pane's receipt settles nothing here + store.publish("other", [], [{ id: "live", outcome: "sent", resumed: true }], "new"); + expect(store.read("p").map((row) => row.id)).toEqual(["live"]); + }); + it("keeps a resumed receipt that came before another tab saved its copy (#2421)", () => { + const port = storage(), store = new PendingMessageStore(() => port); + store.publish("p", [], [{ id: "early", outcome: "sent", resumed: true }], "new"); + port.setItem("ovis:pending:p", JSON.stringify({ messages: [message("early", "sending"), message("other", "sending")] })); + store.refresh("p"); + expect(store.read("p").map((row) => row.id)).toEqual(["other"]); + }); it("settles only the exact resumed Pi ID after reload, never same-text or unrelated rows", () => { const port = storage(), first = new PendingMessageStore(() => port); const receipt = { ...message("pi:one", "uncertain", "same"), error: { code: "pi_transcript_pending", message: "waiting" } }; diff --git a/src/lib/pendingMessages.ts b/src/lib/pendingMessages.ts index 103364a106..b0c0f5d5f2 100644 --- a/src/lib/pendingMessages.ts +++ b/src/lib/pendingMessages.ts @@ -1,9 +1,9 @@ -import type { PendingMessage } from "../../shared/protocol.ts"; +import type { PendingMessage, PendingRemoval } from "../../shared/protocol.ts"; import { paneLifetimes, paneStateKey, type PaneLifetimes } from "./paneLifetime.ts"; export interface PendingMessageView extends PendingMessage { serverOwned: boolean; resendInFlight?: true; resendGeneration?: number } type StoragePort = Pick; -type Removal = { id: string; outcome: "sent" | "discarded" | "consumed" | "dismissed" }; +type Removal = PendingRemoval; const PREFIX = "ovis:pending:"; const piReceipt = (message: PendingMessage): boolean => message.id.startsWith("pi:") && (message.error?.code === "pi_transcript_pending" || message.error?.code === "transcript_unconfirmed"); @@ -16,7 +16,8 @@ export class PendingMessageStore { private rows = new Map(); private scopes = new Map>(); private removed = new Map>>(); - private settledPi = new Map>(); + /** ids an exact resumed receipt settled (a consumed Pi id, or a resumed sent/discarded one): no saved copy brings them back */ + private settledResumed = new Map>(); private receiptOrder: Array<{ owner: string; id: string; scope: string }> = []; private saved = new Map(); private unsaved = new Set(); @@ -75,7 +76,7 @@ export class PendingMessageStore { try { const raw = this.storage().getItem(PREFIX + owner); if (this.saved.get(owner) === raw) return; - const next = this.restored(raw).filter((message) => !piReceipt(message) || !this.settledPi.get(owner)?.has(message.id)); + const next = this.restored(raw).filter((message) => !this.settledResumed.get(owner)?.has(message.id)); // Another tab cannot alter or discard a live connection's authoritative rows. for (const own of this.read(owner).filter((message) => this.unsaved.has(owner) || message.serverOwned || this.isBusy(owner, message.id))) { const at = next.findIndex((message) => message.id === own.id); @@ -124,7 +125,7 @@ export class PendingMessageStore { } accept(owner: string, message: PendingMessage, scope: string | null): void { const seen = this.removed.get(owner)?.get(message.id); - if (piReceipt(message) && this.settledPi.get(owner)?.has(message.id) || (scope === null ? (seen?.size ?? 0) > 0 : seen?.has(scope))) return; + if (this.settledResumed.get(owner)?.has(message.id) || (scope === null ? (seen?.size ?? 0) > 0 : seen?.has(scope))) return; this.refresh(owner); const proofs = this.scopes.get(owner) ?? new Map(); this.scopes.set(owner, proofs); @@ -151,15 +152,21 @@ export class PendingMessageStore { const unproved = new Set(this.read(owner).filter((message) => message.request_id < 0 && !message.serverOwned && proofs?.get(message.id) === undefined).map((message) => message.id)); // A consumed Pi ID is an exact server-attested receipt, replayed only to the same // browser/device on attach. It can settle a saved non-authoritative copy across - // connection epochs; text, request IDs, and empty snapshots never can. - const resumed = new Set(this.read(owner).filter((message) => piReceipt(message) && !message.serverOwned && proofs?.get(message.id) === undefined).map((message) => message.id)); + // connection epochs; text, request IDs, and empty snapshots never can. A `resumed` sent or discarded + // receipt is the same for any follow-up's saved copy (#2421). + const saved = (message: PendingMessage & { serverOwned: boolean }): boolean => !message.serverOwned && proofs?.get(message.id) === undefined; + const resumed = new Set(this.read(owner).filter((message) => piReceipt(message) && saved(message)).map((message) => message.id)); + const copies = new Set(this.read(owner).filter(saved).map((message) => message.id)); const deleted = new Set(removed.filter((item) => proofs?.get(item.id) === scope || unproved.has(item.id) - || item.outcome === "consumed" && resumed.has(item.id)).map((item) => item.id)); - for (const item of removed) if (item.outcome === "consumed" && deleted.has(item.id) && item.id.startsWith("pi:")) { - const settled = this.settledPi.get(owner) ?? new Set(); + || item.outcome === "consumed" && resumed.has(item.id) || item.resumed === true && copies.has(item.id)).map((item) => item.id)); + // A resumed receipt is remembered even before its copy is saved (another tab may save it later), but + // never for a row a live connection here proves under another scope. + for (const item of removed) if (item.outcome === "consumed" && deleted.has(item.id) && item.id.startsWith("pi:") + || item.resumed === true && (proofs?.get(item.id) === undefined || proofs.get(item.id) === scope)) { + const settled = this.settledResumed.get(owner) ?? new Set(); settled.add(item.id); if (settled.size > 1024) settled.delete(settled.values().next().value!); - this.settledPi.set(owner, settled); + this.settledResumed.set(owner, settled); } const receipts = this.removed.get(owner) ?? new Map>(); this.removed.set(owner, receipts);