diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 0a0a7187c4..245829d925 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -117,11 +117,13 @@ agent, so an agent the daemon has not loaded answers `behind` and resumes in tha Cached history remains readable during recovery. The chat shows Reconnecting to host while the host is offline and until a reconnected chat's check answers, so an unchanged chat never shows a second -toast. A check on cached content shows nothing. Updating messages appears for a chat that is behind, -until the fetch completes, and for a check still unanswered `FRESHNESS_CHECK_QUIET_MS` after it was -sent. That delay measures the daemon, so it starts at the request: on a cold start the cache read -alone can take longer. Socket connectivity alone cannot certify that the displayed conversation is -current. The timeline owner publishes freshness; the view renders it without a toast timer or a +toast. Returning from background on a socket that stayed connected shows nothing until the check +answers: the chat stayed subscribed, so a slow answer is no reason to doubt it, and a socket that +died without closing fails the foreground liveness probe and becomes a reconnect. Updating messages +appears for a chat that is behind, until the fetch completes, and for an opened chat whose check is +still unanswered `FRESHNESS_CHECK_QUIET_MS` after it was sent. That delay measures the daemon, so it +starts at the request: on a cold start the cache read alone can take longer. Socket connectivity +alone cannot certify that the displayed conversation is current. The timeline owner publishes freshness; the view renders it without a toast timer or a separate resume workflow. The turn footer in a chat follows its messages, not the agent status. Turn activity comes from the diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 021cea2bf8..d61581cc73 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -1379,6 +1379,54 @@ test("foregrounding a current chat checks over the subscription instead of fetch world.sync.dispose(); }); +test("a slow answer to the foreground check stays silent while the socket stayed connected", async () => { + vi.useFakeTimers(); + try { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current" }); + await vi.advanceTimersByTimeAsync(0); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"); + + world.sync.setActive(false); + world.sync.setActive(true); + const check = await world.nextMembership(); + + await vi.advanceTimersByTimeAsync(FRESHNESS_CHECK_QUIET_MS * 4); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + check.succeed({ "agent-a": "current" }); + await vi.advanceTimersByTimeAsync(0); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"); + world.expectNoPendingFetch(); + world.sync.dispose(); + } finally { + vi.useRealTimers(); + } +}); + +test("a foreground check answered behind reports pending until the fetch lands", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + (await world.nextMembership()).succeed({ "agent-a": "behind" }); + + const fetch = await world.nextFetch("agent-a"); + expect(fetch.request).toMatchObject({ direction: "after", cursor: { seq: 42 } }); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + fetch.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.sync.dispose(); +}); + test("the old subscription keeps delivering until the daemon accepts its replacement", async () => { const world = freshnessWorld(); world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); @@ -1493,22 +1541,33 @@ test("a failed check reports the error and retries through membership", async () }); test("reconnecting reports reattaching instead of pending until the check answers", async () => { - const world = freshnessWorld(); - world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); - world.show("pane", ["agent-a"]); - world.sync.setConnected(true); - (await world.nextMembership()).succeed({ "agent-a": "current" }); - await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + vi.useFakeTimers(); + try { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current" }); + await vi.advanceTimersByTimeAsync(0); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"); - world.sync.setConnected(false); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("reattaching"); - world.sync.setConnected(true); - expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("reattaching"); + world.sync.setConnected(false); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("reattaching"); + world.sync.setConnected(true); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("reattaching"); - (await world.nextMembership()).succeed({ "agent-a": "current" }); - await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); - world.expectNoPendingFetch(); - world.sync.dispose(); + const check = await world.nextMembership(); + await vi.advanceTimersByTimeAsync(FRESHNESS_CHECK_QUIET_MS * 4); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("reattaching"); + + check.succeed({ "agent-a": "current" }); + await vi.advanceTimersByTimeAsync(0); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"); + world.expectNoPendingFetch(); + world.sync.dispose(); + } finally { + vi.useRealTimers(); + } }); test("without the daemon capability nothing changes: no cursors, no checking, fetch on foreground", async () => { diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index e2a623c4d0..06c3fff0d4 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -407,6 +407,12 @@ export interface ViewedTimelineSyncPorts { reportError(error: unknown): void; } +/** + * Why a freshness check was asked: opening a chat, the socket coming back, or the app + * returning to the foreground on a socket that stayed connected. + */ +type FreshnessCheckReason = "open" | "reconnect" | "resume"; + /** * "checking": the daemon has not said whether this chat is current; nothing to report yet. * "reattaching": the same question right after a reconnect, so the chat is still not back. @@ -599,11 +605,15 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const manualRetries = new Set(); const loadedCache = new Set(); const cacheLoads = new Map>(); - // Chats whose freshness answer is outstanding. No catch-up starts for them until it arrives; - // `quiet` holds the status short of "pending" until FRESHNESS_CHECK_QUIET_MS elapses. + // Chats whose freshness answer is outstanding. No catch-up starts for them until it arrives. + // An "open" check stays `quiet`, short of "pending", until FRESHNESS_CHECK_QUIET_MS elapses. const freshnessChecks = new Map< string, - { quiet: boolean; reattaching: boolean; timer: ReturnType | null } + { + reason: FreshnessCheckReason; + quiet: boolean; + timer: ReturnType | null; + } >(); const listeners = new Set<() => void>(); let active = true; @@ -673,16 +683,19 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed ports.onCatchUpOwed(agentId); }; - const startFreshnessCheck = (agentId: string, reattaching = false) => { + const startFreshnessCheck = (agentId: string, reason: FreshnessCheckReason = "open") => { if (freshnessChecks.has(agentId)) return; - freshnessChecks.set(agentId, { quiet: true, reattaching, timer: null }); + freshnessChecks.set(agentId, { reason, quiet: true, timer: null }); }; // The delay measures the daemon's answer, so it starts when the question is sent. On a cold - // start the cache read that precedes the question can outlast it on its own. + // start the cache read that precedes the question can outlast it on its own. Only opening a + // chat escalates. A reconnect already shows Reconnecting to host until the answer. A resume + // kept its socket subscribed, so a slow answer is no reason to doubt the chat; a socket that + // died quietly is caught by the foreground liveness probe and becomes a reconnect. const startFreshnessCheckDelay = (agentId: string) => { const check = freshnessChecks.get(agentId); - if (!check || check.timer) return; + if (!check || check.timer || check.reason !== "open") return; check.timer = setTimeout(() => { check.quiet = false; notifyListeners(); @@ -1012,8 +1025,8 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const getUnconfirmedStatus = (agentId: string): ViewedTimelineStatus => { const check = freshnessChecks.get(agentId); if (check) { - if (!check.quiet) return "pending"; - return check.reattaching ? "reattaching" : "checking"; + if (check.reason === "reconnect") return "reattaching"; + return check.quiet ? "checking" : "pending"; } if (!ports.canCheckFreshness()) return "pending"; if (!connected) return "reattaching"; @@ -1072,7 +1085,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed ports.readCursor(agentId) ) { catchUps.delete(agentId); - startFreshnessCheck(agentId); + startFreshnessCheck(agentId, "resume"); freshnessCheckOwed = true; continue; } @@ -1100,7 +1113,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } if (ports.canCheckFreshness()) { for (const agentId of desired) { - if (visibilityCatchUpPending.has(agentId)) startFreshnessCheck(agentId, true); + if (visibilityCatchUpPending.has(agentId)) startFreshnessCheck(agentId, "reconnect"); } } membershipGeneration += 1;