diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 38c358874..6abe6bfec 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -58,7 +58,8 @@ The daemon validates that the epoch is current and the exact source sequence sti ## 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 authoritative range first. @@ -106,13 +107,14 @@ replica cache. 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 - catch up immediately. Every visibility-driven removal, including app backgrounding, stays - subscribed for a 30-second grace period so brief tab, pane, route, and app switches do not repeatedly - unsubscribe and catch up. Losing window keyboard focus does not make a selected pane invisible. - Disconnecting and disposal clear pending grace because the subscription itself no longer exists. - After grace has expired, revisiting a retained timeline displays its cached state immediately and - authoritative catch-up advances it to the current tail. +- Selective daemons receive every agent visible in any pane plus the most recently viewed hidden + agents, up to five subscribed agents. Visible agents always win: if more than five are visible, + they all remain subscribed and no hidden agent does. Switching and app backgrounding preserve + this connection-scoped hot set, so returning to an agent still covered by it needs no catch-up. + Losing window keyboard focus does not make a selected pane invisible. Disconnecting clears hidden + hot agents; reconnect restores the currently visible set before authoritative catch-up. Revisiting + 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 authoritative catch-up, but the app does not issue selective-subscription RPCs. diff --git a/packages/app/e2e/browser/viewed-agent-timelines.spec.ts b/packages/app/e2e/browser/viewed-agent-timelines.spec.ts index 4a6885766..c597b445e 100644 --- a/packages/app/e2e/browser/viewed-agent-timelines.spec.ts +++ b/packages/app/e2e/browser/viewed-agent-timelines.spec.ts @@ -97,7 +97,7 @@ async function expectAgentConsistentlyIdle(page: Page, title: string): Promise { - 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, }) => { test.setTimeout(150_000); @@ -105,9 +105,9 @@ test.describe("Viewed agent timelines", () => { const scenario = await seedViewedTimelineScenario({ firstAgentModel: "one-minute-stream" }); try { 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 subscriptions.waitForSubscribedAgents([scenario.secondAgentId], { timeout: 45_000 }); + await subscriptions.waitForSubscribedAgents([scenario.firstAgentId, scenario.secondAgentId]); const finish = await scenario.client.waitForFinish(scenario.firstAgentId, 90_000); 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.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("a hidden hot chat stays current", async ({ page }) => { test.setTimeout(60_000); const subscriptions = observeTimelineSubscriptions(page); const scenario = await seedViewedTimelineScenario(); diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 85b52092e..3f327c2f1 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -1,9 +1,6 @@ import { expect, test, vi } from "vitest"; import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan"; -import { - createViewedTimelineSync, - VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS, -} from "./viewed-timeline-sync"; +import { createViewedTimelineSync } from "./viewed-timeline-sync"; interface Deferred { promise: Promise; @@ -129,18 +126,14 @@ class TimelineWorld { return new Promise((resolve) => this.retryWaiters.push(resolve)); } - runUnsubscribeGrace(): void { - const index = this.scheduled.findIndex( - (entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS, + elapse(elapsedMs: number): void { + const due = this.scheduled.filter((entry) => entry.delayMs <= elapsedMs); + this.scheduled.splice( + 0, + this.scheduled.length, + ...this.scheduled.filter((entry) => entry.delayMs > elapsedMs), ); - expect(index).toBeGreaterThanOrEqual(0); - this.scheduled.splice(index, 1)[0].task(); - } - - expectNoPendingUnsubscribe(): void { - expect( - this.scheduled.filter((entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS), - ).toEqual([]); + for (const entry of due) entry.task(); } 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 () => { const world = new TimelineWorld(); 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"]); }); -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(); 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(); - - world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); - world.runUnsubscribeGrace(); + world.sync.replaceVisibleAgentIds("workspace", ["agent-f"]); staleMembership.succeed(); const currentMembership = await world.nextMembership(); currentMembership.succeed(); - const agentB = await world.nextFetch("agent-b"); - agentB.respond({ hasNewer: false }); + const currentCatchUps = await Promise.all( + ["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({ - stale: ["agent-a"], - current: ["agent-b"], + stale: ["agent-a", "agent-b", "agent-c", "agent-d", "agent-e"], + current: ["agent-a", "agent-b", "agent-c", "agent-d", "agent-f"], }); 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 () => { const world = new TimelineWorld(); world.sync.setConnected(true); @@ -347,7 +320,22 @@ test("disconnect cancels paging and reconnect restores membership before fresh c 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(); world.sync.replaceVisibleAgentIds("left-route", ["agent-a"]); 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.expectNoPendingMembership(); - world.expectNoPendingUnsubscribe(); world.sync.replaceVisibleAgentIds("right-route", ["agent-b"]); - world.runUnsubscribeGrace(); - const remaining = await world.nextMembership(); - remaining.succeed(); - expect({ combined: combined.agentIds, remaining: remaining.agentIds }).toEqual({ - combined: ["agent-a", "agent-b"], - remaining: ["agent-b"], - }); + expect(combined.agentIds).toEqual(["agent-a", "agent-b"]); + world.expectNoPendingMembership(); + world.expectNoPendingFetch(); }); 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(); world.sync.setConnected(true); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); - const initial = await world.nextMembership(); - initial.succeed(); - const initialCatchUp = await world.nextFetch("agent-a"); - initialCatchUp.respond({ hasNewer: false }); + const agentAMembership = await world.nextMembership(); + agentAMembership.succeed(); + const agentACatchUp = await world.nextFetch("agent-a"); + agentACatchUp.respond({ hasNewer: false }); - world.sync.setActive(false); - world.expectNoPendingMembership(); - world.runUnsubscribeGrace(); - const background = await world.nextMembership(); - background.succeed(); - world.sync.setActive(true); - 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.replaceVisibleAgentIds("workspace", ["agent-b"]); + const agentBMembership = await world.nextMembership(); + agentBMembership.succeed(); + const agentBCatchUp = await world.nextFetch("agent-b"); + agentBCatchUp.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready")); world.sync.setActive(false); world.expectNoPendingMembership(); + world.elapse(60_000); world.sync.setActive(true); - world.expectNoPendingUnsubscribe(); world.expectNoPendingMembership(); world.expectNoPendingFetch(); }); @@ -527,14 +491,17 @@ test("stale membership retry cannot overwrite a newer effective set", async () = const staleRetry = await world.nextRetry(); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); - world.runUnsubscribeGrace(); const current = await world.nextMembership(); staleRetry(); current.succeed(); - const catchUp = await world.nextFetch("agent-b"); - catchUp.respond({ hasNewer: false }); + const [agentA, agentB] = await Promise.all([ + 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(); }); @@ -558,138 +525,114 @@ test("membership retry cannot run while disconnected", async () => { expect(restored.agentIds).toEqual(["agent-a"]); }); -test("quickly returning to an agent cancels its pending unsubscribe without another catch-up", 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 () => { +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 initialMembership = await world.nextMembership(); initialMembership.succeed(); - const initialCatchUp = await world.nextFetch("agent-a"); - initialCatchUp.respond({ hasNewer: false }); + const agentACatchUp = await world.nextFetch("agent-a"); + agentACatchUp.respond({ hasNewer: false }); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); const expandedMembership = await world.nextMembership(); expandedMembership.succeed(); const agentBCatchUp = await world.nextFetch("agent-b"); agentBCatchUp.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready")); - expect(expandedMembership.agentIds).toEqual(["agent-a", "agent-b"]); - world.runUnsubscribeGrace(); - const settledMembership = await world.nextMembership(); - settledMembership.succeed(); - expect(settledMembership.agentIds).toEqual(["agent-b"]); + world.elapse(60_000); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + + world.expectNoPendingMembership(); + 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(); 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 }); + const agentAMembership = await world.nextMembership(); + agentAMembership.succeed(); + const agentACatchUp = await world.nextFetch("agent-a"); + agentACatchUp.respond({ hasNewer: false }); - world.sync.replaceVisibleAgentIds("workspace", []); - world.sync.setActive(false); - world.expectNoPendingMembership(); - world.runUnsubscribeGrace(); - const unsubscribe = await world.nextMembership(); - unsubscribe.succeed(); + world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); + const agentBMembership = await world.nextMembership(); + agentBMembership.succeed(); + const agentBCatchUp = await world.nextFetch("agent-b"); + agentBCatchUp.respond({ hasNewer: false }); + 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); - - expect(agentAStatuses).toEqual(["pending"]); - world.expectNoPendingUnsubscribe(); world.expectNoPendingMembership(); -}); - -test("disposing cancels pending unsubscribe grace", async () => { - const world = new TimelineWorld(); 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"]); - const membership = await world.nextMembership(); - membership.succeed(); - const catchUp = await world.nextFetch("agent-a"); - catchUp.respond({ hasNewer: false }); - - world.sync.replaceVisibleAgentIds("workspace", []); - world.sync.dispose(); - - world.expectNoPendingUnsubscribe(); - world.expectNoPendingMembership(); + const reopened = await world.nextMembership(); + reopened.succeed(); + const reopenedCatchUp = await world.nextFetch("agent-a"); + reopenedCatchUp.respond({ hasNewer: false }); + expect(reopened.agentIds).toEqual(["agent-a", "agent-b"]); }); 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 () => { const world = new TimelineWorld(); world.sync.setDeliveryMode("legacy"); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index caa4c63d7..8fafac06a 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -39,7 +39,7 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge { } 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"; @@ -99,7 +99,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed // Authoritative fetch owed but not runnable yet: disconnected, unacknowledged, or parked. // Acknowledgement and tail completion are the only drain points. const pendingCatchUps = new Map(); - const lingeringRemovals = new Map void>(); const visibilityCatchUpPending = new Set(); const visibilityCatchUpErrors = new Set(); const listeners = new Set<() => void>(); @@ -114,10 +113,27 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed let reconcileRequested = false; let membershipNeedsRetry = false; let cancelMembershipRetry: (() => void) | null = null; + let recentlyViewedAgentIds: string[] = []; - const visibleAgentIds = () => (active ? normalizeAgentIds([...sources.values()].flat()) : []); - const effectiveAgentIds = () => - normalizeAgentIds([...visibleAgentIds(), ...lingeringRemovals.keys()]); + const visibleAgentIds = () => normalizeAgentIds([...sources.values()].flat()); + + 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 isDesired = (agentId: string) => desired.includes(agentId); @@ -361,32 +377,16 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed void reconcileMembership(); }; - const clearLingeringRemovals = () => { - for (const cancel of lingeringRemovals.values()) cancel(); - lingeringRemovals.clear(); - }; - - const publishVisibleMembership = (allowGrace: boolean) => { + const publishVisibleMembership = () => { const visible = visibleAgentIds(); - for (const agentId of visible) { - lingeringRemovals.get(agentId)?.(); - lingeringRemovals.delete(agentId); + if (!connected || deliveryMode !== "selective") { + const activeVisible = active ? visible : []; + recentlyViewedAgentIds = activeVisible; + commitDesiredMembership(activeVisible); + return; } - - if (allowGrace && connected && deliveryMode === "selective") { - 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()); + if (!active) return; + commitDesiredMembership(selectHotAgentIds(visible)); }; return { @@ -403,19 +403,20 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const normalized = normalizeAgentIds(agentIds); if (normalized.length === 0) sources.delete(sourceId); else sources.set(sourceId, normalized); - publishVisibleMembership(true); + publishVisibleMembership(); }, setActive(nextActive) { if (active === nextActive) return; active = nextActive; - publishVisibleMembership(true); + publishVisibleMembership(); }, setConnected(nextConnected) { if (connected === nextConnected) return; connected = nextConnected; if (!connected) { - clearLingeringRemovals(); - commitDesiredMembership(visibleAgentIds(), { resetCatchUpStatus: true }); + const visible = active ? visibleAgentIds() : []; + recentlyViewedAgentIds = visible; + commitDesiredMembership(visible, { resetCatchUpStatus: true }); cancelMembershipRetry?.(); cancelMembershipRetry = null; acknowledged = []; @@ -434,13 +435,14 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed setDeliveryMode(nextMode) { if (deliveryMode === nextMode) return; deliveryMode = nextMode; - clearLingeringRemovals(); cancelMembershipRetry?.(); cancelMembershipRetry = null; membershipNeedsRetry = false; membershipGeneration += 1; for (const agentId of desired) cancelCatchUp(agentId); - desired = visibleAgentIds(); + const visible = active ? visibleAgentIds() : []; + recentlyViewedAgentIds = visible; + desired = visible; visibilityCatchUpPending.clear(); visibilityCatchUpErrors.clear(); for (const agentId of desired) visibilityCatchUpPending.add(agentId); @@ -467,7 +469,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed }, dispose() { disposed = true; - clearLingeringRemovals(); cancelMembershipRetry?.(); cancelMembershipRetry = null; sources.clear(); @@ -475,6 +476,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed for (const agentId of desired) cancelCatchUp(agentId); desired = []; acknowledged = []; + recentlyViewedAgentIds = []; visibilityCatchUpPending.clear(); visibilityCatchUpErrors.clear(); notifyListeners();