feat(app): keep recently viewed chats live (#2842)

Replace time-based subscription expiry with a bounded hot set so switching among a small working set and backgrounding the app do not force stale timeline catch-up. Visible chats always retain priority, while disconnect and legacy delivery preserve their existing authoritative boundaries.
This commit is contained in:
Mohamed Boudra
2026-08-03 23:18:28 +02:00
committed by GitHub
parent f25e68b8cc
commit b4a4507fb2
4 changed files with 227 additions and 279 deletions
+10 -8
View File
@@ -58,7 +58,8 @@ The daemon validates that the epoch is current and the exact source sequence sti
## Resume behavior ## Resume behavior
Opening, reconnecting, or revisiting after selective-delivery grace fetches the latest tail page. Opening, reconnecting, or revisiting after a selective-delivery coverage gap fetches the latest tail
page.
Focus alone does not mutate timeline state; the tail response is compared with the local Focus alone does not mutate timeline state; the tail response is compared with the local
authoritative range first. authoritative range first.
@@ -106,13 +107,14 @@ replica cache.
The app chooses one delivery policy from `server_info.features.selectiveAgentTimeline`: The app chooses one delivery policy from `server_info.features.selectiveAgentTimeline`:
- Selective daemons receive the union of agents visible in every pane. Additions subscribe and - Selective daemons receive every agent visible in any pane plus the most recently viewed hidden
catch up immediately. Every visibility-driven removal, including app backgrounding, stays agents, up to five subscribed agents. Visible agents always win: if more than five are visible,
subscribed for a 30-second grace period so brief tab, pane, route, and app switches do not repeatedly they all remain subscribed and no hidden agent does. Switching and app backgrounding preserve
unsubscribe and catch up. Losing window keyboard focus does not make a selected pane invisible. this connection-scoped hot set, so returning to an agent still covered by it needs no catch-up.
Disconnecting and disposal clear pending grace because the subscription itself no longer exists. Losing window keyboard focus does not make a selected pane invisible. Disconnecting clears hidden
After grace has expired, revisiting a retained timeline displays its cached state immediately and hot agents; reconnect restores the currently visible set before authoritative catch-up. Revisiting
authoritative catch-up advances it to the current tail. an evicted retained timeline displays its cached state immediately while authoritative catch-up
advances it to the current tail.
- Legacy daemons keep globally streaming agent timelines. Visibility still triggers the existing - Legacy daemons keep globally streaming agent timelines. Visibility still triggers the existing
authoritative catch-up, but the app does not issue selective-subscription RPCs. authoritative catch-up, but the app does not issue selective-subscription RPCs.
@@ -97,7 +97,7 @@ async function expectAgentConsistentlyIdle(page: Page, title: string): Promise<v
} }
test.describe("Viewed agent timelines", () => { test.describe("Viewed agent timelines", () => {
test("a turn that finishes while unsubscribed reopens with consistently idle chrome", async ({ test("a turn that finishes while hidden reopens with consistently idle chrome", async ({
page, page,
}) => { }) => {
test.setTimeout(150_000); test.setTimeout(150_000);
@@ -105,9 +105,9 @@ test.describe("Viewed agent timelines", () => {
const scenario = await seedViewedTimelineScenario({ firstAgentModel: "one-minute-stream" }); const scenario = await seedViewedTimelineScenario({ firstAgentModel: "one-minute-stream" });
try { try {
await openAgent(page, scenario, scenario.firstAgentId); await openAgent(page, scenario, scenario.firstAgentId);
await startVisibleTurn(page, scenario, "Finish after this chat becomes unsubscribed."); await startVisibleTurn(page, scenario, "Finish after this chat becomes hidden.");
await selectAgent(page, "Second viewed chat"); await selectAgent(page, "Second viewed chat");
await subscriptions.waitForSubscribedAgents([scenario.secondAgentId], { timeout: 45_000 }); await subscriptions.waitForSubscribedAgents([scenario.firstAgentId, scenario.secondAgentId]);
const finish = await scenario.client.waitForFinish(scenario.firstAgentId, 90_000); const finish = await scenario.client.waitForFinish(scenario.firstAgentId, 90_000);
expect(finish.status).toBe("idle"); expect(finish.status).toBe("idle");
@@ -118,33 +118,7 @@ test.describe("Viewed agent timelines", () => {
} }
}); });
test("an unsubscribed hidden chat catches up when shown", async ({ page }) => { test("a hidden hot chat stays current", async ({ page }) => {
test.setTimeout(90_000);
const subscriptions = observeTimelineSubscriptions(page);
const scenario = await seedViewedTimelineScenario();
try {
await openAgent(page, scenario, scenario.firstAgentId);
await selectAgent(page, "Second viewed chat");
await subscriptions.waitForSubscribedAgents([scenario.secondAgentId], { timeout: 45_000 });
await commitMessage(
scenario,
scenario.firstAgentId,
"Committed after the first chat unsubscribed.",
);
await expect(
page.getByText("Committed after the first chat unsubscribed.", { exact: true }),
).toHaveCount(0);
await selectAgent(page, "First viewed chat");
await expect(
page.getByText("Committed after the first chat unsubscribed.", { exact: true }),
).toBeVisible();
await expect(page.getByText("(end of synthetic stream)", { exact: true })).toBeVisible();
} finally {
await scenario.cleanup();
}
});
test("a hidden retained chat stays current during unsubscribe grace", async ({ page }) => {
test.setTimeout(60_000); test.setTimeout(60_000);
const subscriptions = observeTimelineSubscriptions(page); const subscriptions = observeTimelineSubscriptions(page);
const scenario = await seedViewedTimelineScenario(); const scenario = await seedViewedTimelineScenario();
@@ -1,9 +1,6 @@
import { expect, test, vi } from "vitest"; import { expect, test, vi } from "vitest";
import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan"; import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan";
import { import { createViewedTimelineSync } from "./viewed-timeline-sync";
createViewedTimelineSync,
VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS,
} from "./viewed-timeline-sync";
interface Deferred<T> { interface Deferred<T> {
promise: Promise<T>; promise: Promise<T>;
@@ -129,18 +126,14 @@ class TimelineWorld {
return new Promise((resolve) => this.retryWaiters.push(resolve)); return new Promise((resolve) => this.retryWaiters.push(resolve));
} }
runUnsubscribeGrace(): void { elapse(elapsedMs: number): void {
const index = this.scheduled.findIndex( const due = this.scheduled.filter((entry) => entry.delayMs <= elapsedMs);
(entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS, this.scheduled.splice(
0,
this.scheduled.length,
...this.scheduled.filter((entry) => entry.delayMs > elapsedMs),
); );
expect(index).toBeGreaterThanOrEqual(0); for (const entry of due) entry.task();
this.scheduled.splice(index, 1)[0].task();
}
expectNoPendingUnsubscribe(): void {
expect(
this.scheduled.filter((entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS),
).toEqual([]);
} }
private releaseMembershipWaiter(): void { private releaseMembershipWaiter(): void {
@@ -161,10 +154,6 @@ class TimelineWorld {
} }
} }
test("keeps hidden agents subscribed for thirty seconds", () => {
expect(VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS).toBe(30_000);
});
test("uses a tail fetch when an agent becomes visible", async () => { test("uses a tail fetch when an agent becomes visible", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
@@ -279,51 +268,35 @@ test("all acknowledged agents begin catch-up independently", async () => {
expect(membership.agentIds).toEqual(["agent-a", "agent-b"]); expect(membership.agentIds).toEqual(["agent-a", "agent-b"]);
}); });
test("membership changes during acknowledgement never catch up the stale set", async () => { test("an eviction during acknowledgement never catches up the stale hot set", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); world.sync.replaceVisibleAgentIds("workspace", [
"agent-a",
"agent-b",
"agent-c",
"agent-d",
"agent-e",
]);
const staleMembership = await world.nextMembership(); const staleMembership = await world.nextMembership();
world.sync.replaceVisibleAgentIds("workspace", ["agent-f"]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.runUnsubscribeGrace();
staleMembership.succeed(); staleMembership.succeed();
const currentMembership = await world.nextMembership(); const currentMembership = await world.nextMembership();
currentMembership.succeed(); currentMembership.succeed();
const agentB = await world.nextFetch("agent-b"); const currentCatchUps = await Promise.all(
agentB.respond({ hasNewer: false }); ["agent-a", "agent-b", "agent-c", "agent-d", "agent-f"].map((agentId) =>
world.nextFetch(agentId),
),
);
for (const catchUp of currentCatchUps) catchUp.respond({ hasNewer: false });
expect({ stale: staleMembership.agentIds, current: currentMembership.agentIds }).toEqual({ expect({ stale: staleMembership.agentIds, current: currentMembership.agentIds }).toEqual({
stale: ["agent-a"], stale: ["agent-a", "agent-b", "agent-c", "agent-d", "agent-e"],
current: ["agent-b"], current: ["agent-a", "agent-b", "agent-c", "agent-d", "agent-f"],
}); });
world.expectNoPendingFetch(); world.expectNoPendingFetch();
}); });
test("removing one agent during paging cancels only that agent", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a", "agent-b"]);
const initialMembership = await world.nextMembership();
initialMembership.succeed();
const [agentA, agentB] = await Promise.all([
world.nextFetch("agent-a"),
world.nextFetch("agent-b"),
]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.runUnsubscribeGrace();
const replacement = await world.nextMembership();
agentA.respond({ hasNewer: true, seq: 4 });
agentB.respond({ hasNewer: true, seq: 7 });
const agentBNext = await world.nextFetch("agent-b");
replacement.succeed();
agentBNext.respond({ hasNewer: false });
expect(replacement.agentIds).toEqual(["agent-b"]);
world.expectNoPendingFetch();
});
test("disconnect cancels paging and reconnect restores membership before fresh catch-up", async () => { test("disconnect cancels paging and reconnect restores membership before fresh catch-up", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
@@ -347,7 +320,22 @@ test("disconnect cancels paging and reconnect restores membership before fresh c
world.expectNoPendingFetch(); world.expectNoPendingFetch();
}); });
test("overlapping sources deduplicate membership and source removal preserves remaining views", async () => { test("navigation while disconnected reconnects only the currently visible agent", async () => {
const world = new TimelineWorld();
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.sync.setConnected(true);
const membership = await world.nextMembership();
membership.succeed();
const catchUp = await world.nextFetch("agent-b");
catchUp.respond({ hasNewer: false });
expect(membership.agentIds).toEqual(["agent-b"]);
world.expectNoPendingFetch();
});
test("overlapping sources deduplicate membership and retain hidden hot agents", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.replaceVisibleAgentIds("left-route", ["agent-a"]); world.sync.replaceVisibleAgentIds("left-route", ["agent-a"]);
world.sync.replaceVisibleAgentIds("right-route", ["agent-a", "agent-b"]); world.sync.replaceVisibleAgentIds("right-route", ["agent-a", "agent-b"]);
@@ -363,16 +351,11 @@ test("overlapping sources deduplicate membership and source removal preserves re
world.sync.replaceVisibleAgentIds("left-route", []); world.sync.replaceVisibleAgentIds("left-route", []);
world.expectNoPendingMembership(); world.expectNoPendingMembership();
world.expectNoPendingUnsubscribe();
world.sync.replaceVisibleAgentIds("right-route", ["agent-b"]); world.sync.replaceVisibleAgentIds("right-route", ["agent-b"]);
world.runUnsubscribeGrace();
const remaining = await world.nextMembership();
remaining.succeed();
expect({ combined: combined.agentIds, remaining: remaining.agentIds }).toEqual({ expect(combined.agentIds).toEqual(["agent-a", "agent-b"]);
combined: ["agent-a", "agent-b"], world.expectNoPendingMembership();
remaining: ["agent-b"], world.expectNoPendingFetch();
});
}); });
test("a failed catch-up reports once and retries through the explicit retry policy", async () => { test("a failed catch-up reports once and retries through the explicit retry policy", async () => {
@@ -474,46 +457,27 @@ test("membership failure autonomously retries without another visibility declara
}); });
}); });
test("background waits for grace before unsubscribing and catches up on return", async () => { test("backgrounding preserves the hot membership without catch-up on return", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const initial = await world.nextMembership(); const agentAMembership = await world.nextMembership();
initial.succeed(); agentAMembership.succeed();
const initialCatchUp = await world.nextFetch("agent-a"); const agentACatchUp = await world.nextFetch("agent-a");
initialCatchUp.respond({ hasNewer: false }); agentACatchUp.respond({ hasNewer: false });
world.sync.setActive(false); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.expectNoPendingMembership(); const agentBMembership = await world.nextMembership();
world.runUnsubscribeGrace(); agentBMembership.succeed();
const background = await world.nextMembership(); const agentBCatchUp = await world.nextFetch("agent-b");
background.succeed(); agentBCatchUp.respond({ hasNewer: false });
world.sync.setActive(true); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
const foreground = await world.nextMembership();
foreground.succeed();
const resumedCatchUp = await world.nextFetch("agent-a");
resumedCatchUp.respond({ hasNewer: false });
expect({ background: background.agentIds, foreground: foreground.agentIds }).toEqual({
background: [],
foreground: ["agent-a"],
});
});
test("foregrounding within grace preserves the live membership", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const catchUp = await world.nextFetch("agent-a");
catchUp.respond({ hasNewer: false });
world.sync.setActive(false); world.sync.setActive(false);
world.expectNoPendingMembership(); world.expectNoPendingMembership();
world.elapse(60_000);
world.sync.setActive(true); world.sync.setActive(true);
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership(); world.expectNoPendingMembership();
world.expectNoPendingFetch(); world.expectNoPendingFetch();
}); });
@@ -527,14 +491,17 @@ test("stale membership retry cannot overwrite a newer effective set", async () =
const staleRetry = await world.nextRetry(); const staleRetry = await world.nextRetry();
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.runUnsubscribeGrace();
const current = await world.nextMembership(); const current = await world.nextMembership();
staleRetry(); staleRetry();
current.succeed(); current.succeed();
const catchUp = await world.nextFetch("agent-b"); const [agentA, agentB] = await Promise.all([
catchUp.respond({ hasNewer: false }); world.nextFetch("agent-a"),
world.nextFetch("agent-b"),
]);
agentA.respond({ hasNewer: false });
agentB.respond({ hasNewer: false });
expect(current.agentIds).toEqual(["agent-b"]); expect(current.agentIds).toEqual(["agent-a", "agent-b"]);
world.expectNoPendingMembership(); world.expectNoPendingMembership();
}); });
@@ -558,138 +525,114 @@ test("membership retry cannot run while disconnected", async () => {
expect(restored.agentIds).toEqual(["agent-a"]); expect(restored.agentIds).toEqual(["agent-a"]);
}); });
test("quickly returning to an agent cancels its pending unsubscribe without another catch-up", async () => { test("returning to a hot hidden agent stays live after inactivity", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const initialCatchUp = await world.nextFetch("agent-a");
initialCatchUp.respond({ hasNewer: false });
await vi.waitFor(() => {
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready");
});
world.sync.replaceVisibleAgentIds("workspace", []);
world.expectNoPendingMembership();
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready");
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership();
world.expectNoPendingFetch();
});
test("unsubscribe grace expiry removes the agent exactly once", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const catchUp = await world.nextFetch("agent-a");
catchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
const readinessChanges: string[] = [];
world.sync.subscribe(() => {
readinessChanges.push(world.sync.getAgentTimelineStatus("agent-a"));
});
world.sync.replaceVisibleAgentIds("workspace", []);
world.runUnsubscribeGrace();
const unsubscribe = await world.nextMembership();
unsubscribe.succeed();
expect(unsubscribe.agentIds).toEqual([]);
expect(readinessChanges.at(-1)).toBe("pending");
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership();
});
test("a new visible agent subscribes immediately while the previous agent lingers", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const initialMembership = await world.nextMembership(); const initialMembership = await world.nextMembership();
initialMembership.succeed(); initialMembership.succeed();
const initialCatchUp = await world.nextFetch("agent-a"); const agentACatchUp = await world.nextFetch("agent-a");
initialCatchUp.respond({ hasNewer: false }); agentACatchUp.respond({ hasNewer: false });
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
const expandedMembership = await world.nextMembership(); const expandedMembership = await world.nextMembership();
expandedMembership.succeed(); expandedMembership.succeed();
const agentBCatchUp = await world.nextFetch("agent-b"); const agentBCatchUp = await world.nextFetch("agent-b");
agentBCatchUp.respond({ hasNewer: false }); agentBCatchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
expect(expandedMembership.agentIds).toEqual(["agent-a", "agent-b"]); world.elapse(60_000);
world.runUnsubscribeGrace(); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const settledMembership = await world.nextMembership();
settledMembership.succeed(); world.expectNoPendingMembership();
expect(settledMembership.agentIds).toEqual(["agent-b"]); world.expectNoPendingFetch();
}); });
test("backgrounding preserves an existing unsubscribe grace period", async () => { test("the hot set evicts the least-recent hidden agent and promotes revisited agents", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
for (const agentId of ["agent-a", "agent-b", "agent-c", "agent-d", "agent-e"]) {
world.sync.replaceVisibleAgentIds("workspace", [agentId]);
const membership = await world.nextMembership();
membership.succeed();
const catchUp = await world.nextFetch(agentId);
catchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus(agentId)).toBe("ready"));
}
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.expectNoPendingMembership();
world.expectNoPendingFetch();
world.sync.replaceVisibleAgentIds("workspace", ["agent-f"]);
const eviction = await world.nextMembership();
eviction.succeed();
const agentFCatchUp = await world.nextFetch("agent-f");
agentFCatchUp.respond({ hasNewer: false });
expect(eviction.agentIds).toEqual(["agent-b", "agent-c", "agent-d", "agent-e", "agent-f"]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.expectNoPendingMembership();
world.expectNoPendingFetch();
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const restoreEvicted = await world.nextMembership();
restoreEvicted.succeed();
const agentACatchUp = await world.nextFetch("agent-a");
agentACatchUp.respond({ hasNewer: false });
expect(restoreEvicted.agentIds).toEqual(["agent-a", "agent-b", "agent-d", "agent-e", "agent-f"]);
});
test("visible agents are never evicted when they exceed the hot-set limit", async () => {
const world = new TimelineWorld();
const visible = ["agent-a", "agent-b", "agent-c", "agent-d", "agent-e", "agent-f"];
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", visible);
const membership = await world.nextMembership();
membership.succeed();
const catchUps = await Promise.all(visible.map((agentId) => world.nextFetch(agentId)));
for (const catchUp of catchUps) catchUp.respond({ hasNewer: false });
expect(membership.agentIds).toEqual(visible);
world.sync.replaceVisibleAgentIds("workspace", visible.slice(0, 5));
const hiddenEviction = await world.nextMembership();
hiddenEviction.succeed();
expect(hiddenEviction.agentIds).toEqual(visible.slice(0, 5));
});
test("disconnect clears hidden hot agents before reconnecting the visible set", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership(); const agentAMembership = await world.nextMembership();
membership.succeed(); agentAMembership.succeed();
const catchUp = await world.nextFetch("agent-a"); const agentACatchUp = await world.nextFetch("agent-a");
catchUp.respond({ hasNewer: false }); agentACatchUp.respond({ hasNewer: false });
world.sync.replaceVisibleAgentIds("workspace", []); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
world.sync.setActive(false); const agentBMembership = await world.nextMembership();
world.expectNoPendingMembership(); agentBMembership.succeed();
world.runUnsubscribeGrace(); const agentBCatchUp = await world.nextFetch("agent-b");
const unsubscribe = await world.nextMembership(); agentBCatchUp.respond({ hasNewer: false });
unsubscribe.succeed(); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
expect(unsubscribe.agentIds).toEqual([]);
world.expectNoPendingMembership();
});
test("disconnecting cancels pending unsubscribe grace without publishing on the closed socket", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a", "agent-b"]);
const membership = await world.nextMembership();
membership.succeed();
const [agentA, agentB] = await Promise.all([
world.nextFetch("agent-a"),
world.nextFetch("agent-b"),
]);
agentA.respond({ hasNewer: false });
agentB.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const agentAStatuses: string[] = [];
world.sync.subscribe(() => {
agentAStatuses.push(world.sync.getAgentTimelineStatus("agent-a"));
});
world.sync.setConnected(false); world.sync.setConnected(false);
expect(agentAStatuses).toEqual(["pending"]);
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership(); world.expectNoPendingMembership();
});
test("disposing cancels pending unsubscribe grace", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true); world.sync.setConnected(true);
const restored = await world.nextMembership();
restored.succeed();
const restoredCatchUp = await world.nextFetch("agent-b");
restoredCatchUp.respond({ hasNewer: false });
expect(restored.agentIds).toEqual(["agent-b"]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership(); const reopened = await world.nextMembership();
membership.succeed(); reopened.succeed();
const catchUp = await world.nextFetch("agent-a"); const reopenedCatchUp = await world.nextFetch("agent-a");
catchUp.respond({ hasNewer: false }); reopenedCatchUp.respond({ hasNewer: false });
expect(reopened.agentIds).toEqual(["agent-a", "agent-b"]);
world.sync.replaceVisibleAgentIds("workspace", []);
world.sync.dispose();
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership();
}); });
test("legacy delivery skips subscription RPCs while retaining visibility catch-up and gap recovery", async () => { test("legacy delivery skips subscription RPCs while retaining visibility catch-up and gap recovery", async () => {
@@ -715,6 +658,33 @@ test("legacy delivery skips subscription RPCs while retaining visibility catch-u
}); });
}); });
test("legacy delivery catches up after returning to a view or foreground", async () => {
const world = new TimelineWorld();
world.sync.setDeliveryMode("legacy");
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const firstAgentA = await world.nextFetch("agent-a");
firstAgentA.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]);
const agentB = await world.nextFetch("agent-b");
agentB.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const secondAgentA = await world.nextFetch("agent-a");
secondAgentA.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.setActive(false);
world.sync.setActive(true);
const foregroundAgentA = await world.nextFetch("agent-a");
foregroundAgentA.respond({ hasNewer: false });
world.expectNoPendingMembership();
});
test("switching from legacy to selective delivery publishes membership and catches up once", async () => { test("switching from legacy to selective delivery publishes membership and catches up once", async () => {
const world = new TimelineWorld(); const world = new TimelineWorld();
world.sync.setDeliveryMode("legacy"); world.sync.setDeliveryMode("legacy");
@@ -39,7 +39,7 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge {
} }
const RETRY_DELAY_MS = 1_000; const RETRY_DELAY_MS = 1_000;
export const VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS = 30_000; const VIEWED_TIMELINE_HOT_AGENT_LIMIT = 5;
type CatchUpStatus = "running" | "complete" | "error"; type CatchUpStatus = "running" | "complete" | "error";
@@ -99,7 +99,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
// Authoritative fetch owed but not runnable yet: disconnected, unacknowledged, or parked. // Authoritative fetch owed but not runnable yet: disconnected, unacknowledged, or parked.
// Acknowledgement and tail completion are the only drain points. // Acknowledgement and tail completion are the only drain points.
const pendingCatchUps = new Map<string, ProjectedTimelineForwardFetchPlan>(); const pendingCatchUps = new Map<string, ProjectedTimelineForwardFetchPlan>();
const lingeringRemovals = new Map<string, () => void>();
const visibilityCatchUpPending = new Set<string>(); const visibilityCatchUpPending = new Set<string>();
const visibilityCatchUpErrors = new Set<string>(); const visibilityCatchUpErrors = new Set<string>();
const listeners = new Set<() => void>(); const listeners = new Set<() => void>();
@@ -114,10 +113,27 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
let reconcileRequested = false; let reconcileRequested = false;
let membershipNeedsRetry = false; let membershipNeedsRetry = false;
let cancelMembershipRetry: (() => void) | null = null; let cancelMembershipRetry: (() => void) | null = null;
let recentlyViewedAgentIds: string[] = [];
const visibleAgentIds = () => (active ? normalizeAgentIds([...sources.values()].flat()) : []); const visibleAgentIds = () => normalizeAgentIds([...sources.values()].flat());
const effectiveAgentIds = () =>
normalizeAgentIds([...visibleAgentIds(), ...lingeringRemovals.keys()]); const selectHotAgentIds = (visible: string[]) => {
const visibleSet = new Set(visible);
recentlyViewedAgentIds = [
...visible,
...recentlyViewedAgentIds.filter((agentId) => !visibleSet.has(agentId)),
];
const hiddenBudget = Math.max(0, VIEWED_TIMELINE_HOT_AGENT_LIMIT - visible.length);
const desiredAgentIds = normalizeAgentIds([
...visible,
...recentlyViewedAgentIds
.filter((agentId) => !visibleSet.has(agentId))
.slice(0, hiddenBudget),
]);
const desiredSet = new Set(desiredAgentIds);
recentlyViewedAgentIds = recentlyViewedAgentIds.filter((agentId) => desiredSet.has(agentId));
return desiredAgentIds;
};
const isAcknowledged = (agentId: string) => acknowledged.includes(agentId); const isAcknowledged = (agentId: string) => acknowledged.includes(agentId);
const isDesired = (agentId: string) => desired.includes(agentId); const isDesired = (agentId: string) => desired.includes(agentId);
@@ -361,32 +377,16 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
void reconcileMembership(); void reconcileMembership();
}; };
const clearLingeringRemovals = () => { const publishVisibleMembership = () => {
for (const cancel of lingeringRemovals.values()) cancel();
lingeringRemovals.clear();
};
const publishVisibleMembership = (allowGrace: boolean) => {
const visible = visibleAgentIds(); const visible = visibleAgentIds();
for (const agentId of visible) { if (!connected || deliveryMode !== "selective") {
lingeringRemovals.get(agentId)?.(); const activeVisible = active ? visible : [];
lingeringRemovals.delete(agentId); recentlyViewedAgentIds = activeVisible;
commitDesiredMembership(activeVisible);
return;
} }
if (!active) return;
if (allowGrace && connected && deliveryMode === "selective") { commitDesiredMembership(selectHotAgentIds(visible));
for (const agentId of desired) {
if (visible.includes(agentId) || lingeringRemovals.has(agentId)) continue;
const cancel = ports.schedule(() => {
lingeringRemovals.delete(agentId);
commitDesiredMembership(effectiveAgentIds());
}, VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS);
lingeringRemovals.set(agentId, cancel);
}
} else {
clearLingeringRemovals();
}
commitDesiredMembership(effectiveAgentIds());
}; };
return { return {
@@ -403,19 +403,20 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const normalized = normalizeAgentIds(agentIds); const normalized = normalizeAgentIds(agentIds);
if (normalized.length === 0) sources.delete(sourceId); if (normalized.length === 0) sources.delete(sourceId);
else sources.set(sourceId, normalized); else sources.set(sourceId, normalized);
publishVisibleMembership(true); publishVisibleMembership();
}, },
setActive(nextActive) { setActive(nextActive) {
if (active === nextActive) return; if (active === nextActive) return;
active = nextActive; active = nextActive;
publishVisibleMembership(true); publishVisibleMembership();
}, },
setConnected(nextConnected) { setConnected(nextConnected) {
if (connected === nextConnected) return; if (connected === nextConnected) return;
connected = nextConnected; connected = nextConnected;
if (!connected) { if (!connected) {
clearLingeringRemovals(); const visible = active ? visibleAgentIds() : [];
commitDesiredMembership(visibleAgentIds(), { resetCatchUpStatus: true }); recentlyViewedAgentIds = visible;
commitDesiredMembership(visible, { resetCatchUpStatus: true });
cancelMembershipRetry?.(); cancelMembershipRetry?.();
cancelMembershipRetry = null; cancelMembershipRetry = null;
acknowledged = []; acknowledged = [];
@@ -434,13 +435,14 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
setDeliveryMode(nextMode) { setDeliveryMode(nextMode) {
if (deliveryMode === nextMode) return; if (deliveryMode === nextMode) return;
deliveryMode = nextMode; deliveryMode = nextMode;
clearLingeringRemovals();
cancelMembershipRetry?.(); cancelMembershipRetry?.();
cancelMembershipRetry = null; cancelMembershipRetry = null;
membershipNeedsRetry = false; membershipNeedsRetry = false;
membershipGeneration += 1; membershipGeneration += 1;
for (const agentId of desired) cancelCatchUp(agentId); for (const agentId of desired) cancelCatchUp(agentId);
desired = visibleAgentIds(); const visible = active ? visibleAgentIds() : [];
recentlyViewedAgentIds = visible;
desired = visible;
visibilityCatchUpPending.clear(); visibilityCatchUpPending.clear();
visibilityCatchUpErrors.clear(); visibilityCatchUpErrors.clear();
for (const agentId of desired) visibilityCatchUpPending.add(agentId); for (const agentId of desired) visibilityCatchUpPending.add(agentId);
@@ -467,7 +469,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
}, },
dispose() { dispose() {
disposed = true; disposed = true;
clearLingeringRemovals();
cancelMembershipRetry?.(); cancelMembershipRetry?.();
cancelMembershipRetry = null; cancelMembershipRetry = null;
sources.clear(); sources.clear();
@@ -475,6 +476,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
for (const agentId of desired) cancelCatchUp(agentId); for (const agentId of desired) cancelCatchUp(agentId);
desired = []; desired = [];
acknowledged = []; acknowledged = [];
recentlyViewedAgentIds = [];
visibilityCatchUpPending.clear(); visibilityCatchUpPending.clear();
visibilityCatchUpErrors.clear(); visibilityCatchUpErrors.clear();
notifyListeners(); notifyListeners();