diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index bccd4d3ee0..89c32c0bd5 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -455,10 +455,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider reportError: (error) => { console.warn("[Session] viewed timeline synchronization failed", { serverId, error }); }, - schedule: (task, delayMs) => { - const timeout = setTimeout(task, delayMs); - return () => clearTimeout(timeout); - }, }); viewedTimelineSyncRef.current = sync; setViewedTimelineSync(serverId, sync); diff --git a/packages/app/src/hooks/use-agent-screen-state-machine.test.ts b/packages/app/src/hooks/use-agent-screen-state-machine.test.ts index 19caef4f8b..68102ed131 100644 --- a/packages/app/src/hooks/use-agent-screen-state-machine.test.ts +++ b/packages/app/src/hooks/use-agent-screen-state-machine.test.ts @@ -422,6 +422,23 @@ describe("deriveAgentScreenViewState", () => { expect(result.memory.lastReadyAgent).toBeNull(); }); + it.each(["error", "retrying"] as const)("shows archived history sync %s", (status) => { + const result = deriveAgentScreenViewState({ + input: { + ...createBaseInput(), + agent: createAgent("agent-1"), + isArchived: true, + hasHydratedHistoryBefore: true, + visibilityCatchUpStatus: status, + }, + memory: createBaseMemory(), + }); + expect(expectReadyState(result.state).sync).toEqual({ + status: "sync_error", + isRetrying: status === "retrying", + }); + }); + it("renders an archived agent before provider history is initialized", () => { const result = deriveAgentScreenViewState({ input: { diff --git a/packages/app/src/hooks/use-agent-screen-state-machine.ts b/packages/app/src/hooks/use-agent-screen-state-machine.ts index 591fd1abb1..e0cc75ee06 100644 --- a/packages/app/src/hooks/use-agent-screen-state-machine.ts +++ b/packages/app/src/hooks/use-agent-screen-state-machine.ts @@ -175,10 +175,14 @@ function resolveAgentScreenSync(args: { hadInitialSyncFailure: boolean; }): AgentScreenReadySyncState { const { input, hadInitialSyncFailure } = args; - if (input.isArchived) { + if ( + input.isArchived && + input.visibilityCatchUpStatus !== "error" && + input.visibilityCatchUpStatus !== "retrying" + ) { return { status: "idle" }; } - if (!input.isConnected) { + if (!input.isConnected && !input.isArchived) { return { status: "reconnecting" }; } if (input.missingAgentState.kind === "error") { diff --git a/packages/app/src/stores/workspace-layout-store.test.ts b/packages/app/src/stores/workspace-layout-store.test.ts index 1946586e84..8b478b7d27 100644 --- a/packages/app/src/stores/workspace-layout-store.test.ts +++ b/packages/app/src/stores/workspace-layout-store.test.ts @@ -174,7 +174,6 @@ it("feeds restored layout to timeline sync as a release signal, not a subscripti fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), onCatchUpEnded: () => undefined, reportError: () => undefined, - schedule: () => () => undefined, }); // The edge session-context.tsx owns. Pointing it at subscription instead would resume every // agent behind a restored tab on the daemon. diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index 0d825673e9..859ae8e6b7 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -57,7 +57,6 @@ function createOwner(storage: TimelineReplicaStorage): ViewedTimelineOwner { fetchPage: async () => ({ hasNewer: false, endCursor: null }), fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), reportError: () => undefined, - schedule: () => () => undefined, }, }); } @@ -617,7 +616,6 @@ describe("create handoff lifetime", () => { }, fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), reportError: () => undefined, - schedule: () => () => undefined, }, }); diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index a6e7cbeb33..e556da612f 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -90,7 +90,7 @@ test("visible catch-up settles before hidden open chats start fetching", async ( world.sync.dispose(); }); -test("a failed visible catch-up releases hidden chats without bypassing its retry delay", async () => { +test("a failed visible catch-up releases hidden chats without bypassing its error", async () => { const world = new TimelineWorld(); world.openedEarlier(["a-hidden"]); world.sync.replaceVisibleAgentIds("pane", ["z-visible"]); @@ -103,7 +103,7 @@ test("a failed visible catch-up releases hidden chats without bypassing its retr hidden.respond({ hasNewer: false }); await world.nextError(); world.expectNoPendingFetch(); - (await world.nextRetry())(); + world.sync.retryVisibleAgentTimeline("z-visible"); (await world.nextFetch("z-visible")).respond({ hasNewer: false }); world.sync.dispose(); }); @@ -200,20 +200,6 @@ class TimelineWorld { const waiter = this.errorWaiters.shift(); if (waiter) waiter(this.errors.at(-1) ?? ""); }, - schedule: (task, delayMs) => { - const scheduled = { task, delayMs }; - this.scheduled.push(scheduled); - const waiterIndex = this.retryWaiters.findIndex((waiter) => waiter.delayMs === delayMs); - if (waiterIndex >= 0) { - const [waiter] = this.retryWaiters.splice(waiterIndex, 1); - this.scheduled.splice(this.scheduled.indexOf(scheduled), 1); - waiter.resolve(task); - } - return () => { - const index = this.scheduled.indexOf(scheduled); - if (index >= 0) this.scheduled.splice(index, 1); - }; - }, }); show(sourceId: string, agentIds: string[]): void { @@ -235,12 +221,6 @@ class TimelineWorld { }> = []; private readonly errorWaiters: Array<(message: string) => void> = []; private readonly cacheRequestWaiters: Array<(agentId: string) => void> = []; - private readonly scheduled: Array<{ task: () => void; delayMs: number }> = []; - private readonly retryWaiters: Array<{ - delayMs: number; - resolve(retry: () => void): void; - }> = []; - get pendingFetchCount(): number { return this.fetches.length; } @@ -300,22 +280,6 @@ class TimelineWorld { return new Promise((resolve) => this.errorWaiters.push(resolve)); } - nextRetry(delayMs = 1_000): Promise<() => void> { - const index = this.scheduled.findIndex((entry) => entry.delayMs === delayMs); - if (index >= 0) return Promise.resolve(this.scheduled.splice(index, 1)[0].task); - return new Promise((resolve) => this.retryWaiters.push({ delayMs, resolve })); - } - - 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), - ); - for (const entry of due) entry.task(); - } - private releaseMembershipWaiter(): void { const waiter = this.membershipWaiters.shift(); if (!waiter) return; @@ -483,6 +447,28 @@ test("reports a chat current only once its parked follow-up page completes", asy world.sync.dispose(); }); +test("a gap during manual retry is recovered before reporting ready", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + (await world.nextFetch("agent-a")).fail("unavailable"); + await world.nextError(); + world.sync.retryVisibleAgentTimeline("agent-a"); + const retry = await world.nextFetch("agent-a"); + world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 9 }); + retry.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.pendingFetchCount).toBe(1)); + const recovery = await world.nextFetch("agent-a"); + expect(recovery.request).toMatchObject({ + direction: "after", + cursor: { epoch: "epoch-agent-a", seq: 9 }, + }); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying"); + recovery.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); +}); + test("reports a chat current once the latest-tail fallback replaces an overflowing catch-up", async () => { const world = new TimelineWorld(); world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 }); @@ -780,6 +766,21 @@ test("overlapping sources deduplicate membership and retain open hidden chats", world.expectNoPendingFetch(); }); +test("failed timeline fetch does not schedule a retry", async () => { + vi.useFakeTimers(); + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + (await world.nextFetch("agent-a")).fail("unavailable"); + await world.nextError(); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(120_000); + world.expectNoPendingFetch(); + world.expectNoPendingMembership(); + vi.useRealTimers(); +}); + test("a failed catch-up reports once and retries through the explicit retry policy", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); @@ -788,12 +789,12 @@ test("a failed catch-up reports once and retries through the explicit retry poli membership.succeed(); const failed = await world.nextFetch("agent-a"); failed.fail("timeline unavailable"); - const [error, retryCatchUp] = await Promise.all([world.nextError(), world.nextRetry()]); + const error = await world.nextError(); expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); - retryCatchUp(); + world.sync.retryVisibleAgentTimeline("agent-a"); const retry = await world.nextFetch("agent-a"); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying"); retry.respond({ hasNewer: false }); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); @@ -804,35 +805,6 @@ test("a failed catch-up reports once and retries through the explicit retry poli world.expectNoPendingMembership(); }); -test("a failed catch-up retries with exponential backoff", async () => { - const world = new TimelineWorld(); - world.sync.setConnected(true); - world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); - const membership = await world.nextMembership(); - membership.succeed(); - - const first = await world.nextFetch("agent-a"); - first.fail("timeline unavailable"); - const [firstError, retryAfterFirstFailure] = await Promise.all([ - world.nextError(), - world.nextRetry(), - ]); - expect(firstError).toBe("timeline unavailable"); - retryAfterFirstFailure(); - - const second = await world.nextFetch("agent-a"); - second.fail("timeline unavailable"); - await world.nextError(); - - const retryAfterSecondFailure = await world.nextRetry(2_000); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); - - retryAfterSecondFailure(); - const third = await world.nextFetch("agent-a"); - third.respond({ hasNewer: false }); - await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); -}); - test("manual retries can immediately re-attempt a failed catch-up", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); @@ -842,7 +814,7 @@ test("manual retries can immediately re-attempt a failed catch-up", async () => const failed = await world.nextFetch("agent-a"); failed.fail("timeline unavailable"); - await world.nextRetry(); + await world.nextError(); expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); world.sync.retryVisibleAgentTimeline("agent-a"); @@ -853,7 +825,7 @@ test("manual retries can immediately re-attempt a failed catch-up", async () => await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); }); -test("redeclaring unchanged visibility does not bypass catch-up backoff", async () => { +test("redeclaring unchanged visibility does not bypass catch-up error", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); @@ -862,9 +834,11 @@ test("redeclaring unchanged visibility does not bypass catch-up backoff", async const failed = await world.nextFetch("agent-a"); failed.fail("timeline unavailable"); - await world.nextRetry(); + await world.nextError(); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + world.sync.setActive(false); + world.sync.setActive(true); world.expectNoPendingMembership(); world.expectNoPendingFetch(); @@ -880,7 +854,7 @@ test("a manual retry that fails returns to the error state", async () => { const failed = await world.nextFetch("agent-a"); failed.fail("timeline unavailable"); - await world.nextRetry(); + await world.nextError(); world.sync.retryVisibleAgentTimeline("agent-a"); expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying"); @@ -943,20 +917,34 @@ test("repeated recovery for the same running gap reuses the in-flight fetch", as gapPage.respond({ hasNewer: false }); }); -test("membership failure autonomously retries without another visibility declaration", async () => { +test("failed membership does not schedule a retry", async () => { + vi.useFakeTimers(); + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).fail("unavailable"); + await world.nextError(); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(120_000); + world.expectNoPendingFetch(); + world.expectNoPendingMembership(); + vi.useRealTimers(); +}); + +test("membership failure waits for manual retry", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); const failed = await world.nextMembership(); failed.fail("subscription unavailable"); - const [error, retryMembership] = await Promise.all([world.nextError(), world.nextRetry()]); + const error = await world.nextError(); expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); - retryMembership(); + world.sync.retryVisibleAgentTimeline("agent-a"); const retry = await world.nextMembership(); retry.succeed(); const catchUp = await world.nextFetch("agent-a"); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying"); catchUp.respond({ hasNewer: false }); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); @@ -967,44 +955,68 @@ test("membership failure autonomously retries without another visibility declara }); }); -test("membership failures retry with exponential backoff", async () => { +test("retrying membership restores a previously loaded chat", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); - world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); - - const first = await world.nextMembership(); - first.fail("subscription unavailable"); - const [firstError, retryAfterFirstFailure] = await Promise.all([ - world.nextError(), - world.nextRetry(), - ]); - expect(firstError).toBe("subscription unavailable"); - - retryAfterFirstFailure(); - const second = await world.nextMembership(); - second.fail("subscription unavailable again"); - - const retryAfterSecondFailure = await world.nextRetry(2_000); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); - - retryAfterSecondFailure(); - const retry = await world.nextMembership(); - retry.succeed(); - const catchUp = await world.nextFetch("agent-a"); - catchUp.respond({ hasNewer: false }); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + (await world.nextFetch("agent-a")).respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 10 }); + world.show("workspace", ["agent-b"]); + (await world.nextMembership()).fail("subscription unavailable"); + await world.nextError(); + world.sync.retryVisibleAgentTimeline("agent-a"); + (await world.nextMembership()).succeed(); + (await world.nextFetch("agent-b")).respond({ hasNewer: false }); + const resumed = await world.nextFetch("agent-a"); + expect(resumed.request).toMatchObject({ + direction: "after", + cursor: { epoch: "epoch-agent-a", seq: 10 }, + }); + resumed.respond({ hasNewer: false }); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); }); -test("redeclaring unchanged visibility does not bypass membership backoff", async () => { +test("a fetch that finishes after membership failure cannot clear the error", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + const inFlight = await world.nextFetch("agent-a"); + world.show("workspace", ["agent-b"]); + (await world.nextMembership()).fail("subscription unavailable"); + await world.nextError(); + inFlight.respond({ hasNewer: false }); + await Promise.resolve(); + await Promise.resolve(); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"); +}); + +test("a failed membership retry returns to error", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.show("workspace", ["agent-a"]); + (await world.nextMembership()).fail("unavailable"); + await world.nextError(); + world.sync.retryVisibleAgentTimeline("agent-a"); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying"); + (await world.nextMembership()).fail("still unavailable"); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error")); +}); + +test("redeclaring unchanged visibility does not bypass membership error", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); const failed = await world.nextMembership(); failed.fail("subscription unavailable"); - await world.nextRetry(); + await world.nextError(); world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + world.sync.setActive(false); + world.sync.setActive(true); world.expectNoPendingMembership(); world.expectNoPendingFetch(); @@ -1050,7 +1062,6 @@ test("backgrounding preserves subscriptions and returning verifies the visible c world.sync.setActive(false); world.expectNoPendingMembership(); - world.elapse(60_000); world.sync.setActive(true); world.expectNoPendingMembership(); @@ -1067,11 +1078,10 @@ test("stale membership retry cannot overwrite a newer effective set", async () = world.show("workspace", ["agent-a"]); const failed = await world.nextMembership(); failed.fail("subscription unavailable"); - const staleRetry = await world.nextRetry(); + await world.nextError(); world.show("workspace", ["agent-b"]); const current = await world.nextMembership(); - staleRetry(); current.succeed(); const agentB = await world.nextFetch("agent-b"); agentB.respond({ hasNewer: false }); @@ -1088,10 +1098,9 @@ test("membership retry cannot run while disconnected", async () => { world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); const failed = await world.nextMembership(); failed.fail("subscription unavailable"); - const disconnectedRetry = await world.nextRetry(); + await world.nextError(); world.sync.setConnected(false); - disconnectedRetry(); world.expectNoPendingMembership(); world.sync.setConnected(true); const restored = await world.nextMembership(); @@ -1118,7 +1127,6 @@ test("returning to an open hidden agent stays live after inactivity", async () = agentBCatchUp.respond({ hasNewer: false }); await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready")); - world.elapse(60_000); world.show("workspace", ["agent-a"]); world.expectNoPendingMembership(); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 2e14a42093..67defcec04 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -329,7 +329,6 @@ export interface ViewedTimelineSyncPorts { */ onCatchUpEnded(agentId: string): void; reportError(error: unknown): void; - schedule(task: () => void, delayMs: number): () => void; } export type ViewedTimelineStatus = "ready" | "pending" | "error" | "retrying"; @@ -423,26 +422,14 @@ export function createViewedTimelineOwner(input: { }; } -const RETRY_DELAY_MS = 1_000; -const MAX_RETRY_DELAY_MS = 30_000; - type CatchUpStatus = "running" | "complete" | "error"; interface CatchUpState { generation: number; status: CatchUpStatus; request?: ProjectedTimelineForwardFetchPlan; - cancelRetry?: () => void; - retryDelayMs?: number; } -const getNextRetryDelayMs = (previousDelayMs: number | undefined): number => { - if (previousDelayMs == null) { - return RETRY_DELAY_MS; - } - return Math.min(previousDelayMs * 2, MAX_RETRY_DELAY_MS); -}; - function isSameCatchUpRequest( left: ProjectedTimelineForwardFetchPlan | undefined, right: ProjectedTimelineForwardFetchPlan | undefined, @@ -472,9 +459,7 @@ function decideCatchUp(input: { } return "replace"; } - return input.current.status === "running" || input.current.status === "complete" - ? "keep" - : "replace"; + return "keep"; } function normalizeAgentIds(agentIds: string[]): string[] { @@ -494,8 +479,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const pendingCatchUps = new Map(); const visibilityCatchUpPending = new Set(); const visibilityCatchUpErrors = new Map(); - // User-initiated retries only. Background retries stay silent; a retry the user asked for - // owes them a pending state until it settles. + // A user-initiated retry owes a pending state until it settles. const manualRetries = new Set(); const loadedCache = new Set(); const cacheLoads = new Map>(); @@ -511,8 +495,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed let acknowledged: string[] = []; let membershipGeneration = 0; let membershipNeedsRetry = false; - let membershipRetryDelayMs: number | undefined; - let cancelMembershipRetry: (() => void) | null = null; // Chats this session has opened. Visibility is the only way in, so an agent is never // subscribed — and never resumed on the daemon — until the user actually opens it. // Membership then survives hiding, workspace switches, backgrounding and reconnect; @@ -563,7 +545,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const cancelCatchUp = (agentId: string) => { catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1); - catchUps.get(agentId)?.cancelRetry?.(); catchUps.delete(agentId); pendingCatchUps.delete(agentId); }; @@ -607,19 +588,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed setVisibilityCatchUpReady(agentId); } catch (error) { if (catchUps.get(agentId)?.generation === generation) { - const nextRetryDelayMs = getNextRetryDelayMs(catchUps.get(agentId)?.retryDelayMs); - const cancelRetry = ports.schedule(() => { - const current = catchUps.get(agentId); - if (current?.generation !== generation || current.status !== "error") return; - startCatchUp(agentId); - }, nextRetryDelayMs); - catchUps.set(agentId, { - generation, - status: "error", - request, - cancelRetry, - retryDelayMs: nextRetryDelayMs, - }); + catchUps.set(agentId, { generation, status: "error", request }); setVisibilityCatchUpError([agentId], error); ports.reportError(error); } @@ -686,17 +655,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed if (decision === "keep") { return; } - current?.cancelRetry?.(); const generation = (catchUpGenerations.get(agentId) ?? 0) + 1; catchUpGenerations.set(agentId, generation); - const retryDelayMs = - supersede || current?.status !== "error" ? undefined : current.retryDelayMs; - catchUps.set(agentId, { - generation, - status: "running", - request: nextRequest, - retryDelayMs, - }); + catchUps.set(agentId, { generation, status: "running", request: nextRequest }); pendingCatchUps.delete(agentId); void fetchUntilCurrent( agentId, @@ -715,7 +676,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed for (const agentId of ordered) { const pendingCatchUp = pendingCatchUps.get(agentId); const current = catchUps.get(agentId); - // Completion of a different chat must not bypass a failed chat's backoff or + // Completion of a different chat must not bypass a failed chat's error or // supersede an in-flight tail that owns its parked gap recovery. if (current && !(current.status === "complete" && pendingCatchUp)) continue; startCatchUp(agentId, { @@ -741,42 +702,31 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } catch (error) { if (disposed || !connected || generation !== membershipGeneration) return; membershipNeedsRetry = true; + acknowledged = []; + for (const agentId of requested) cancelCatchUp(agentId); setVisibilityCatchUpError(requested, error); - cancelMembershipRetry?.(); - const nextRetryDelayMs = getNextRetryDelayMs(membershipRetryDelayMs); - cancelMembershipRetry = ports.schedule(() => { - cancelMembershipRetry = null; - if (disposed || !connected || membershipGeneration !== generation) return; - void reconcileMembership(); - }, nextRetryDelayMs); - membershipRetryDelayMs = nextRetryDelayMs; ports.reportError(error); return; } if (disposed || !connected || generation !== membershipGeneration) return; - cancelMembershipRetry?.(); - cancelMembershipRetry = null; - membershipRetryDelayMs = undefined; acknowledged = requested; startAcknowledgedCatchUps(); }; const retryVisibleAgentTimeline = (agentId: string) => { - if (!isDesired(agentId) || manualRetries.has(agentId)) return; + if (!connected || !isDesired(agentId) || manualRetries.has(agentId)) return; const catchUp = catchUps.get(agentId); const membershipRetryable = membershipNeedsRetry && connected; if (catchUp?.status !== "error" && !membershipRetryable) return; manualRetries.add(agentId); notifyListeners(); - if (catchUp?.status === "error") { - catchUp.cancelRetry?.(); - startCatchUp(agentId, { request: catchUp.request, supersede: true }); + if (membershipRetryable) { + void reconcileMembership(); return; } - cancelMembershipRetry?.(); - cancelMembershipRetry = null; - membershipRetryDelayMs = undefined; - void reconcileMembership(); + if (catchUp?.status === "error") { + startCatchUp(agentId, { request: catchUp.request, supersede: true }); + } }; const commitDesiredMembership = ( @@ -817,8 +767,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed ensureCacheLoaded(agentId); } } - cancelMembershipRetry?.(); - cancelMembershipRetry = null; desired = nextDesired; ports.replaceDemandedAgentIds(desired); membershipGeneration += 1; @@ -869,6 +817,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed return; } for (const agentId of visibleAgentIds()) { + if (visibilityCatchUpErrors.has(agentId)) continue; visibilityCatchUpPending.add(agentId); visibilityCatchUpErrors.delete(agentId); startCatchUp(agentId, { supersede: true }); @@ -880,9 +829,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed connected = nextConnected; if (!connected) { commitDesiredMembership(desired, { resetCatchUpStatus: true }); - cancelMembershipRetry?.(); - cancelMembershipRetry = null; - membershipRetryDelayMs = undefined; void observation?.handle.release().catch(ports.reportError); observation = null; acknowledged = []; @@ -894,7 +840,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed void reconcileMembership(); }, recoverGap(agentId, cursor) { - if (!isDesired(agentId)) return; + if (!isDesired(agentId) || catchUps.get(agentId)?.status === "error") return; startCatchUp(agentId, { request: planTimelineCatchUpAfter({ epoch: cursor.epoch, seq: cursor.endSeq }), supersede: true, @@ -904,10 +850,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed disposed = true; void observation?.handle.release().catch(ports.reportError); observation = null; - cancelMembershipRetry?.(); - cancelMembershipRetry = null; membershipNeedsRetry = false; - membershipRetryDelayMs = undefined; sources.clear(); membershipGeneration += 1; for (const agentId of desired) cancelCatchUp(agentId);