jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Keep focused agent timelines live and catch up instantly (#2196)

Browse Source 
* fix(app): keep focused agent timelines live

Window focus was conflated with app visibility, so switching OS windows could remove the selected timeline subscription. Separate those signals, grace every visibility-driven removal, and cover retained history until authoritative catch-up completes.

* fix(app): keep timeline catch-up failures non-blocking

* fix(app): surface timeline catch-up failures

* fix(app): preserve file preview refocus

* fix(app): preserve optimistic catch-up flow
This commit is contained in:
Mohamed Boudra
2026-07-18 16:30:59 +02:00
committed by GitHub
parent fcaa84f0e4
commit 7917532716
26 changed files with 527 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/timeline/viewed-timeline-sync.test.ts
+72 -8
View File
@@ -1,4 +1,4 @@
import { expect, test } from "vitest";
import { expect, test, vi } from "vitest";
import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan";
import {
createViewedTimelineSync,
@@ -191,6 +191,7 @@ test("unchanged visible-set publication does not cancel paged catch-up", async (
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending");
const membership = await world.nextMembership();
membership.succeed();
const firstPage = await world.nextFetch("agent-a");
@@ -198,8 +199,13 @@ test("unchanged visible-set publication does not cancel paged catch-up", async (
world.sync.replaceVisibleAgentIds("workspace", ["agent-a", "agent-a"]);
firstPage.respond({ hasNewer: true, seq: 5 });
const secondPage = await world.nextFetch("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending");
secondPage.respond({ hasNewer: false });
await vi.waitFor(() => {
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready");
});
expect(secondPage.request).toEqual({
direction: "after",
cursor: { epoch: "epoch-agent-a", seq: 5 },
@@ -280,6 +286,7 @@ test("disconnect cancels paging and reconnect restores membership before fresh c
const stalePage = await world.nextFetch("agent-a");
world.sync.setConnected(false);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending");
stalePage.respond({ hasNewer: true, seq: 8 });
world.sync.setConnected(true);
const restoredMembership = await world.nextMembership();
@@ -287,6 +294,8 @@ test("disconnect cancels paging and reconnect restores membership before fresh c
const restoredPage = await world.nextFetch("agent-a");
restoredPage.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
expect(restoredMembership.agentIds).toEqual(["agent-a"]);
world.expectNoPendingFetch();
});
@@ -328,10 +337,13 @@ test("a failed catch-up reports once and retries through the explicit retry poli
const failed = await world.nextFetch("agent-a");
failed.fail("timeline unavailable");
const [error, retryCatchUp] = await Promise.all([world.nextError(), world.nextRetry()]);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryCatchUp();
const retry = await world.nextFetch("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retry.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
expect({ error, retryDirection: retry.request.direction }).toEqual({
error: "timeline unavailable",
@@ -396,12 +408,15 @@ test("membership failure autonomously retries without another visibility declara
const failed = await world.nextMembership();
failed.fail("subscription unavailable");
const [error, retryMembership] = await Promise.all([world.nextError(), world.nextRetry()]);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryMembership();
const retry = await world.nextMembership();
retry.succeed();
const catchUp = await world.nextFetch("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
catchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
expect({ error, failed: failed.agentIds, retry: retry.agentIds }).toEqual({
error: "subscription unavailable",
@@ -410,7 +425,7 @@ test("membership failure autonomously retries without another visibility declara
});
});
test("background sends an empty set and foreground restores visible membership", async () => {
test("background waits for grace before unsubscribing and catches up on return", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
@@ -420,6 +435,8 @@ test("background sends an empty set and foreground restores visible membership",
initialCatchUp.respond({ hasNewer: false });
world.sync.setActive(false);
world.expectNoPendingMembership();
world.runUnsubscribeGrace();
const background = await world.nextMembership();
background.succeed();
world.sync.setActive(true);
@@ -434,6 +451,24 @@ test("background sends an empty set and foreground restores visible membership",
});
});
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.expectNoPendingMembership();
world.sync.setActive(true);
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership();
world.expectNoPendingFetch();
});
test("stale membership retry cannot overwrite a newer effective set", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -483,10 +518,16 @@ test("quickly returning to an agent cancels its pending unsubscribe without anot
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();
@@ -500,6 +541,12 @@ test("unsubscribe grace expiry removes the agent exactly once", async () => {
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();
@@ -507,6 +554,7 @@ test("unsubscribe grace expiry removes the agent exactly once", async () => {
unsubscribe.succeed();
expect(unsubscribe.agentIds).toEqual([]);
expect(readinessChanges.at(-1)).toBe("pending");
world.expectNoPendingUnsubscribe();
world.expectNoPendingMembership();
});
@@ -533,7 +581,7 @@ test("a new visible agent subscribes immediately while the previous agent linger
expect(settledMembership.agentIds).toEqual(["agent-b"]);
});
test("backgrounding clears pending unsubscribe grace and membership immediately", async () => {
test("backgrounding preserves an existing unsubscribe grace period", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
@@ -544,25 +592,37 @@ test("backgrounding clears pending unsubscribe grace and membership immediately"
world.sync.replaceVisibleAgentIds("workspace", []);
world.sync.setActive(false);
world.expectNoPendingMembership();
world.runUnsubscribeGrace();
const unsubscribe = await world.nextMembership();
unsubscribe.succeed();
expect(unsubscribe.agentIds).toEqual([]);
world.expectNoPendingUnsubscribe();
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"]);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a", "agent-b"]);
const membership = await world.nextMembership();
membership.succeed();
const catchUp = await world.nextFetch("agent-a");
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 });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.replaceVisibleAgentIds("workspace", []);
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();
});
@@ -612,13 +672,17 @@ test("switching from legacy to selective delivery publishes membership and catch
world.sync.setConnected(true);
const legacyCatchUp = await world.nextFetch("agent-a");
legacyCatchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.setDeliveryMode("selective");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending");
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"));
expect(membership.agentIds).toEqual(["agent-a"]);
world.expectNoPendingMembership();
world.expectNoPendingFetch();
packages/app/src/timeline/viewed-timeline-sync.ts
+76 -5
View File
@@ -25,9 +25,12 @@ interface ViewedTimelineSyncPorts {
}
export type TimelineDeliveryMode = "legacy" | "selective";
export type ViewedTimelineStatus = "ready" | "pending" | "error";
export interface ViewedTimelineUiBridge {
replaceVisibleAgentIds(sourceId: string, agentIds: string[]): void;
subscribe(listener: () => void): () => void;
getAgentTimelineStatus(agentId: string): ViewedTimelineStatus;
}
export interface ViewedTimelineSync extends ViewedTimelineUiBridge {
@@ -88,6 +91,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const catchUpGenerations = new Map<string, number>();
const pendingGaps = new Map<string, ProjectedTimelineForwardFetchPlan>();
const lingeringRemovals = new Map<string, () => void>();
const visibilityCatchUpPending = new Set<string>();
const visibilityCatchUpErrors = new Set<string>();
const listeners = new Set<() => void>();
let active = true;
let connected = false;
let deliveryMode = ports.initialDeliveryMode;
@@ -107,6 +113,26 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const isAcknowledged = (agentId: string) => acknowledged.includes(agentId);
const isDesired = (agentId: string) => desired.includes(agentId);
const notifyListeners = () => {
for (const listener of listeners) listener();
};
const setVisibilityCatchUpReady = (agentId: string) => {
const wasPending = visibilityCatchUpPending.delete(agentId);
const hadError = visibilityCatchUpErrors.delete(agentId);
if (wasPending || hadError) notifyListeners();
};
const setVisibilityCatchUpError = (agentIds: string[]) => {
let changed = false;
for (const agentId of agentIds) {
if (!visibilityCatchUpPending.delete(agentId)) continue;
visibilityCatchUpErrors.add(agentId);
changed = true;
}
if (changed) notifyListeners();
};
const cancelCatchUp = (agentId: string) => {
catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1);
catchUps.get(agentId)?.cancelRetry?.();
@@ -148,6 +174,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
throw new Error(`Timeline page for ${agentId} hasNewer without an end cursor`);
}
catchUps.set(agentId, { generation, status: "complete" });
setVisibilityCatchUpReady(agentId);
} catch (error) {
if (catchUps.get(agentId)?.generation === generation) {
const cancelRetry = ports.schedule(() => {
@@ -156,6 +183,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
startCatchUp(agentId);
}, RETRY_DELAY_MS);
catchUps.set(agentId, { generation, status: "error", cancelRetry });
setVisibilityCatchUpError([agentId]);
ports.reportError(error);
}
}
@@ -208,6 +236,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
await ports.setSubscription(requested);
} catch (error) {
membershipNeedsRetry = true;
setVisibilityCatchUpError(requested);
cancelMembershipRetry?.();
cancelMembershipRetry = ports.schedule(() => {
cancelMembershipRetry = null;
@@ -268,20 +297,45 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
}
};
const commitDesiredMembership = (nextDesired: string[]) => {
const commitDesiredMembership = (
nextDesired: string[],
options: { resetCatchUpStatus?: boolean } = {},
) => {
let statusChanged = false;
if (options.resetCatchUpStatus) {
for (const agentId of nextDesired) {
if (!visibilityCatchUpPending.has(agentId)) {
visibilityCatchUpPending.add(agentId);
statusChanged = true;
}
if (visibilityCatchUpErrors.delete(agentId)) statusChanged = true;
}
}
if (sameAgentIds(nextDesired, desired)) {
if (statusChanged) notifyListeners();
if (deliveryMode === "selective" && membershipNeedsRetry) void reconcileMembership();
retryFailedCatchUps();
return;
}
for (const agentId of desired) {
if (!nextDesired.includes(agentId)) cancelCatchUp(agentId);
if (!nextDesired.includes(agentId)) {
cancelCatchUp(agentId);
visibilityCatchUpPending.delete(agentId);
visibilityCatchUpErrors.delete(agentId);
}
}
for (const agentId of nextDesired) {
if (!desired.includes(agentId)) {
visibilityCatchUpPending.add(agentId);
visibilityCatchUpErrors.delete(agentId);
}
}
cancelMembershipRetry?.();
cancelMembershipRetry = null;
desired = nextDesired;
membershipGeneration += 1;
notifyListeners();
if (deliveryMode === "legacy") {
acknowledged = connected ? desired : [];
if (connected) startAcknowledgedCatchUps();
@@ -302,7 +356,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
lingeringRemovals.delete(agentId);
}
if (allowGrace && connected && active && deliveryMode === "selective") {
if (allowGrace && connected && deliveryMode === "selective") {
for (const agentId of desired) {
if (visible.includes(agentId) || lingeringRemovals.has(agentId)) continue;
const cancel = ports.schedule(() => {
@@ -319,6 +373,15 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
};
return {
subscribe(listener) {
listeners.add(listener);
return () => listeners.delete(listener);
},
getAgentTimelineStatus(agentId) {
if (visibilityCatchUpErrors.has(agentId)) return "error";
if (!isDesired(agentId) || visibilityCatchUpPending.has(agentId)) return "pending";
return "ready";
},
replaceVisibleAgentIds(sourceId, agentIds) {
const normalized = normalizeAgentIds(agentIds);
if (normalized.length === 0) sources.delete(sourceId);
@@ -328,14 +391,14 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
setActive(nextActive) {
if (active === nextActive) return;
active = nextActive;
publishVisibleMembership(false);
publishVisibleMembership(true);
},
setConnected(nextConnected) {
if (connected === nextConnected) return;
connected = nextConnected;
if (!connected) {
clearLingeringRemovals();
commitDesiredMembership(visibleAgentIds());
commitDesiredMembership(visibleAgentIds(), { resetCatchUpStatus: true });
cancelMembershipRetry?.();
cancelMembershipRetry = null;
acknowledged = [];
@@ -361,7 +424,11 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
membershipGeneration += 1;
for (const agentId of desired) cancelCatchUp(agentId);
desired = visibleAgentIds();
visibilityCatchUpPending.clear();
visibilityCatchUpErrors.clear();
for (const agentId of desired) visibilityCatchUpPending.add(agentId);
acknowledged = deliveryMode === "legacy" && connected ? desired : [];
notifyListeners();
if (deliveryMode === "selective" && connected) void reconcileMembership();
else if (connected) startAcknowledgedCatchUps();
},
@@ -382,6 +449,10 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
for (const agentId of desired) cancelCatchUp(agentId);
desired = [];
acknowledged = [];
visibilityCatchUpPending.clear();
visibilityCatchUpErrors.clear();
notifyListeners();
listeners.clear();
},
};
}