fix(desktop): fence delayed pulses and order host observations

Signed-off-by: Logan Johnson <loganj@squareup.com>

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();
+  }
+});

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<DesktopObservation[]>("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 };

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",
+    );
+  }
+});

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,
     );
   }
 

desktop/src/shared/api/relayEventPublisher.ts

@@ -19,9 +19,11 @@ export async function publishSessionEvent(
   event: RelayEvent,
   timeoutMessage: string,
   sendErrorMessage: string,
+  assertActive: () => void = () => {},
 ): Promise<RelayEvent> {
   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 ||