diff --git a/desktop/src/features/agents/desktopObservations.test.mjs b/desktop/src/features/agents/desktopObservations.test.mjs index af9418365f..b3166ba25d 100644 --- a/desktop/src/features/agents/desktopObservations.test.mjs +++ b/desktop/src/features/agents/desktopObservations.test.mjs @@ -101,3 +101,25 @@ test("bounded/invalid reads are not silently authoritative, clocks and staleness ]) assert.ok(desktopFreshness(heard, now).startsWith(expected)); }); + +test("history/live replacement races select newest signed time and lower ID, not arrival", async () => { + const f = fixture(); + const events = [ + { id: "old", created_at: 10 }, + { id: "b", created_at: 20 }, + { id: "a", created_at: 20 }, + ]; + const ipc = f.ipc; + f.ipc = async (command, args) => { + if (command === "read_desktop_observations") + assert.deepEqual( + args.events.map((e) => e.id), + ["a", "b", "old"], + ); + return ipc(command, args); + }; + for (const batch of [events, [...events].reverse()]) { + f.relay.fetchEvents = async () => batch; + await f.refresh(); + } +}); diff --git a/desktop/src/features/agents/desktopObservations.ts b/desktop/src/features/agents/desktopObservations.ts index 5de30fe813..d548fa4243 100644 --- a/desktop/src/features/agents/desktopObservations.ts +++ b/desktop/src/features/agents/desktopObservations.ts @@ -35,6 +35,7 @@ export async function refreshDesktopObservations( event, "Desktop pulse timed out", "Desktop pulse failed", + check, ), ); } catch { @@ -48,7 +49,13 @@ export async function refreshDesktopObservations( const rows = await wait( ipc("read_desktop_observations", { ...scope, - events, + // History is chronological and may include a live replacement before EOSE. + // First matching host wins in the view: newest signed time, then lower ID. + events: [...events].sort( + (a, b) => + b.created_at - a.created_at || + (a.id < b.id ? -1 : a.id > b.id ? 1 : 0), + ), }), ); return { rows, warning, partial: events.length === 100 }; diff --git a/desktop/src/shared/api/relayClientPublishRejection.test.mjs b/desktop/src/shared/api/relayClientPublishRejection.test.mjs index 7875b8679a..f323623ef6 100644 --- a/desktop/src/shared/api/relayClientPublishRejection.test.mjs +++ b/desktop/src/shared/api/relayClientPublishRejection.test.mjs @@ -293,3 +293,52 @@ test("a community switch after send failure cannot retry through its replacement ); assert.equal(eventFrames().length, 0); }); + +test("Desktop pulse cancellation fences rate-limit and reconnect continuations", async () => { + const { refreshDesktopObservations } = await import( + "../../features/agents/desktopObservations.ts" + ); + for (const boundary of ["rate-limit", "reconnect"]) { + reset(); + const client = connectedClient(); + const scope = { owner: "owner", community: "wss://a.example" }; + let active = true; + const reconnect = deferred(); + let reconnecting = false; + client.ensureConnected = async () => { + reconnecting = true; + await reconnect.promise; + client.wsId = 8; + return client.connectionGeneration; + }; + if (boundary === "rate-limit") activateRateLimit(4); + else + sendTransport = async () => { + throw Error("socket failed"); + }; + const result = refreshDesktopObservations( + scope, + () => active, + async (command) => { + assert.equal(command, "prepare_desktop_observation"); + return { event: { id: "pulse", kind: 30181 } }; + }, + client, + ); + const rejected = assert.rejects(result, /scope changed/); + if (boundary === "reconnect") await flushUntil(() => reconnecting); + else await Promise.resolve(); + active = false; + const generation = client.connectionGeneration; + resetRateLimitGate(); + reconnect.resolve(); + await rejected; + assert.equal(eventFrames().length, 0, boundary); + assert.equal(client.pendingEvents.size, 0); + assert.equal( + client.connectionGeneration, + generation, + "cancellation is not socket failure", + ); + } +}); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 93384a4074..3839d01c7e 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -725,6 +725,7 @@ export class RelayClient { event: RelayEvent, timeoutMessage: string, sendErrorMessage: string, + assertActive?: () => void, ) { return publishSessionEvent( { @@ -742,6 +743,7 @@ export class RelayClient { event, timeoutMessage, sendErrorMessage, + assertActive, ); } diff --git a/desktop/src/shared/api/relayEventPublisher.ts b/desktop/src/shared/api/relayEventPublisher.ts index ff71992670..6f6b6b7c82 100644 --- a/desktop/src/shared/api/relayEventPublisher.ts +++ b/desktop/src/shared/api/relayEventPublisher.ts @@ -19,9 +19,11 @@ export async function publishSessionEvent( event: RelayEvent, timeoutMessage: string, sendErrorMessage: string, + assertActive: () => void = () => {}, ): Promise { const publishOwnership = session.ownership(); await waitForRateLimit(); + assertActive(); if (publishOwnership !== session.ownership()) { throw new Error("Relay disconnected for community switch."); } @@ -48,6 +50,14 @@ export async function publishSessionEvent( return; } + try { + assertActive(); + } catch (error) { + window.clearTimeout(timeout); + session.pendingEvents.delete(event.id); + reject(error); + return; + } // Expected socket recovery must not reject the operation being retried. session.pendingEvents.delete(event.id); const sendError = session.recoverSocketFailure(error, sendErrorMessage); @@ -55,7 +65,9 @@ export async function publishSessionEvent( let retryGeneration: number | null = null; try { - retryGeneration = await session.reconnect(); + const reconnected = await session.reconnect(); + assertActive(); + retryGeneration = reconnected; if ( publishOwnership !== session.ownership() || session.generation() !== retryGeneration ||