From ef19db587e176fe9215837c3cd2dd5e70a66fbdd Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 9 Nov 2026 10:51:22 +0100 Subject: [PATCH] fix(app): keep hidden chats live while a subscription is replaced The old timeline subscription was released in the same tick the new one was requested. The client deletes a route on release and adds the new route only when the daemon responds, so events sent in between were dropped. Visible chats recover through the cursor check; hidden chats skip it and stayed behind until something else fetched them. Release the old subscription when the new one settles. Accepting the new route and releasing the old one happen without a socket frame in between, so each event reaches the client once. --- .../src/timeline/viewed-timeline-sync.test.ts | 57 ++++++++++++++++++- .../app/src/timeline/viewed-timeline-sync.ts | 6 +- 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 54ebc83d61..021cea2bf8 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -173,6 +173,7 @@ class TimelineWorld { ready: result.promise, release: async () => { this.releasedMemberships.push(agentIds); + result.reject(new Error("Subscription released")); }, }; }, @@ -1371,7 +1372,6 @@ test("foregrounding a current chat checks over the subscription instead of fetch const check = await world.nextMembership(); expect(check.agentIds).toEqual(["agent-a"]); expect(check.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 42 } }); - await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]])); check.succeed({ "agent-a": "current" }); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); @@ -1379,6 +1379,61 @@ test("foregrounding a current chat checks over the subscription instead of fetch world.sync.dispose(); }); +test("the old subscription keeps delivering until the daemon accepts its replacement", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + world.cursors.set("agent-b", { epoch: "epoch-b", endSeq: 2 }); + world.openedEarlier(["agent-b"]); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current", "agent-b": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + const check = await world.nextMembership(); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(world.releasedMemberships).toEqual([]); + + check.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a", "agent-b"]])); + world.sync.dispose(); +}); + +test("losing the connection during a check releases both the check and what it replaced", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + await world.nextMembership(); + world.sync.setConnected(false); + + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"], ["agent-a"]])); + world.sync.dispose(); +}); + +test("a replacement the daemon refuses still releases the subscription it replaced", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + const check = await world.nextMembership(); + check.fail("refused"); + + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]])); + world.sync.dispose(); +}); + test("foregrounding asks only about visible chats; hidden chats stay caught up", async () => { const world = freshnessWorld(); world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index eb79d87a69..e2a623c4d0 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -921,8 +921,10 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const previous = observation; const handle = ports.observe(requested, cursors); observation = { handle, agentIds: requested }; - void previous?.handle.release().catch(ports.reportError); - const answer = await handle.ready; + // A subscription routes events only once the daemon accepts it, so the old one delivers until then. + const answer = await handle.ready.finally(() => { + void previous?.handle.release().catch(ports.reportError); + }); timelines = answer ? answer.timelines : undefined; } catch (error) { if (isReconcileSuperseded(generation)) return;