From 44b608b8541c4cd95d1fd2e4df5473e029118b66 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 9 Nov 2026 07:18:19 +0100 Subject: [PATCH] fix(app): stop showing Updating messages for a chat that has not changed Every return to a chat fetched from the daemon just to learn nothing had changed, and the toast covered that fetch. The subscribe request that attaches a chat now carries its cursor, and the daemon answers current or behind in the same round trip, so an unchanged chat needs no fetch and shows nothing. The daemon attaches before it compares, so a commit landing in between is either delivered live or counted as behind. It never loads an agent to answer: an agent it has not loaded answers behind and resumes in the fetch, as before. The delay before an unanswered check shows the toast starts when the request is sent. A cold start's cache read alone can outlast the delay, which would flash the toast for a chat the daemon then reports current. --- docs/timeline-sync.md | 27 +- .../e2e/browser/agent-timeline-resume.spec.ts | 71 ++++- .../browser/agent-timeline-sync-retry.spec.ts | 2 + .../e2e/browser/background-reconnect.spec.ts | 12 +- .../sidebar-workspace-mark-unread.spec.ts | 6 +- .../e2e/support/helpers/plugin-timeline.ts | 8 +- .../e2e/support/helpers/timeline-resume.ts | 112 ++++++- .../app/e2e/support/helpers/workspace-ui.ts | 14 + packages/app/src/contexts/session-context.tsx | 6 +- .../use-agent-screen-state-machine.test.ts | 54 ++++ .../hooks/use-agent-screen-state-machine.ts | 21 +- .../src/stores/workspace-layout-store.test.ts | 2 + packages/app/src/timeline/replica.test.ts | 2 + .../src/timeline/viewed-timeline-sync.test.ts | 291 +++++++++++++++++- .../app/src/timeline/viewed-timeline-sync.ts | 236 +++++++++++++- packages/client/src/connection.test.ts | 25 ++ packages/client/src/daemon-client.ts | 16 +- packages/protocol/src/messages.test.ts | 20 ++ packages/protocol/src/messages.ts | 10 + .../server/src/server/agent/agent-manager.ts | 5 + .../src/server/agent/agent-timeline-store.ts | 5 + .../selective-timeline-delivery.e2e.test.ts | 88 +++++- packages/server/src/server/session.ts | 25 +- .../server/src/server/websocket-server.ts | 2 + 24 files changed, 994 insertions(+), 66 deletions(-) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index e975178da7..111f370d45 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -102,15 +102,26 @@ updated after the checkpoint does not block it. Forking the current context rema ## Resume behavior -Opening, reconnecting, and returning from app background establish the current timeline through -bounded catch-up. Switching between continuously subscribed open chats needs no fetch. -Focus alone does not mutate timeline state; the response is compared with the local -authoritative range first. +Opening, reconnecting, and returning from app background establish whether the chat is current. +Switching between continuously subscribed open chats needs no fetch. Focus alone does not mutate +timeline state; the response is compared with the local authoritative range first. -Cached history remains readable during recovery. The chat shows Reconnecting to host while the host is -offline, then Updating messages until authoritative catch-up completes. 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. +A chat with a cached cursor asks the daemon first. The `agent.timeline.set_subscription.request` +that attaches the chat carries its `{ epoch, seq }`, and the response answers `current` or `behind` +per chat. The daemon attaches before it compares, so an entry committed in between arrives live or +counts as behind. `current` means the agent is loaded, the epoch matches, and the cursor is at the +head; it completes catch-up with no fetch. `behind`, or no entry, runs the bounded catch-up. The +subscribe handler never loads an agent, so an agent the daemon has not loaded answers `behind` and +resumes in that fetch. Hosts without `features.timelineFreshnessCheck` get no cursors and always fetch. + +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 +separate resume workflow. The turn footer in a chat follows its messages, not the agent status. Turn activity comes from the agent directory and messages from the chat's own catch-up, and either can land first. So the timeline diff --git a/packages/app/e2e/browser/agent-timeline-resume.spec.ts b/packages/app/e2e/browser/agent-timeline-resume.spec.ts index 8a093feffc..8d164e954a 100644 --- a/packages/app/e2e/browser/agent-timeline-resume.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-resume.spec.ts @@ -13,10 +13,17 @@ import { import { commitTimelineTurnsWhileDisconnected, disconnectViewedTimeline, - expectOneResumeCheckWithoutTail, + expectResumeAnsweredWithoutFetch, expectResumeOverflowFallsBackToOneTail, + hideTabAndReturn, rememberTimelineRequestCounts, + releaseHeldCatchUp, + restoreUnchangedViewedTimeline, + restoreViewedTimelineHoldingCatchUp, restoreViewedTimelineWithHeldResponse, + waitForResumeAnswer, + reopenAppWatchingForUpdatingToast, + watchForUpdatingToast, } from "../support/helpers/timeline-resume"; test.describe("Agent timeline resume", () => { @@ -42,7 +49,9 @@ test.describe("Agent timeline resume", () => { } }); - test("an unchanged resume leaves a scrolled-up timeline untouched", async ({ page }) => { + test("returning to an unchanged chat shows no update and leaves it untouched", async ({ + page, + }) => { test.setTimeout(120_000); const gate = await installDaemonWebSocketGate(page); const agent = await seedLongMockAgentTimeline({ turns: 30 }); @@ -51,13 +60,65 @@ test.describe("Agent timeline resume", () => { await expectTimelinePromptVisible(page, agent.newestPrompt); await scrollTimelinePromptIntoView(page, agent.initialTailAnchorPrompt); const presentation = await rememberTimelinePresentation(page, agent.initialTailAnchorPrompt); - const requests = rememberTimelineRequestCounts(gate); + + await test.step("back from the background", async () => { + const requests = rememberTimelineRequestCounts(gate); + const expectNoUpdatingToast = await watchForUpdatingToast(page); + + await hideTabAndReturn(page); + await waitForResumeAnswer(gate, requests); + + await expectNoUpdatingToast(); + await expectTimelinePresentationUnchanged(page, presentation); + expectResumeAnsweredWithoutFetch(gate, requests); + }); + + await test.step("back from a dropped connection", async () => { + const requests = rememberTimelineRequestCounts(gate); + const expectNoUpdatingToast = await watchForUpdatingToast(page); + + await disconnectViewedTimeline(page, gate); + await restoreUnchangedViewedTimeline(page, gate); + + await expectNoUpdatingToast(); + await expectTimelinePresentationUnchanged(page, presentation); + expectResumeAnsweredWithoutFetch(gate, requests); + }); + + await test.step("back from reopening the app", async () => { + const requests = rememberTimelineRequestCounts(gate); + + const expectNoUpdatingToast = await reopenAppWatchingForUpdatingToast(page); + await waitForResumeAnswer(gate, requests); + await expectTimelinePromptVisible(page, agent.newestPrompt); + + await expectNoUpdatingToast(); + expectResumeAnsweredWithoutFetch(gate, requests); + }); + } finally { + gate.restore(); + await agent.cleanup(); + } + }); + + test("returning to a changed chat reports the update and shows the new messages", async ({ + page, + }) => { + test.setTimeout(120_000); + const gate = await installDaemonWebSocketGate(page); + const agent = await seedLongMockAgentTimeline({ turns: 3 }); + try { + await openAgentTimeline(page, agent); + await expectTimelinePromptVisible(page, agent.newestPrompt); + const presentation = await rememberTimelinePresentation(page, agent.newestPrompt); await disconnectViewedTimeline(page, gate); - await restoreViewedTimelineWithHeldResponse(page, gate); + const background = await commitTimelineTurnsWhileDisconnected(agent, 1); + await restoreViewedTimelineHoldingCatchUp(page, gate); await expectTimelinePresentationUnchanged(page, presentation); - expectOneResumeCheckWithoutTail(gate, requests); + await releaseHeldCatchUp(page, gate); + await expectTimelinePromptVisible(page, background.newestPrompt); } finally { gate.restore(); await agent.cleanup(); diff --git a/packages/app/e2e/browser/agent-timeline-sync-retry.spec.ts b/packages/app/e2e/browser/agent-timeline-sync-retry.spec.ts index f68efffd02..b95e945056 100644 --- a/packages/app/e2e/browser/agent-timeline-sync-retry.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-sync-retry.spec.ts @@ -6,6 +6,7 @@ import { seedLongMockAgentTimeline, } from "../support/helpers/timeline-pagination"; import { + commitTimelineTurnsWhileDisconnected, disconnectViewedTimeline, restoreViewedTimelineWithHeldResponse, } from "../support/helpers/timeline-resume"; @@ -25,6 +26,7 @@ test.describe("Agent timeline sync retry", () => { gate.failTimelineResponses(agent.agentId); await disconnectViewedTimeline(page, gate); + await commitTimelineTurnsWhileDisconnected(agent, 1); await restoreViewedTimelineWithHeldResponse(page, gate); const callout = page.getByTestId("agent-timeline-sync-error"); diff --git a/packages/app/e2e/browser/background-reconnect.spec.ts b/packages/app/e2e/browser/background-reconnect.spec.ts index e9831c257d..b879354a29 100644 --- a/packages/app/e2e/browser/background-reconnect.spec.ts +++ b/packages/app/e2e/browser/background-reconnect.spec.ts @@ -5,6 +5,7 @@ import { type IsolatedHostDaemon, } from "../support/helpers/isolated-host-daemon"; import { seedSavedSettingsHosts } from "../support/helpers/settings"; +import { setTabVisibility } from "../support/helpers/workspace-ui"; const test = base.extend<{ reconnectHost: IsolatedHostDaemon }>({ reconnectHost: async ({ e2eWorker }, provide) => { @@ -37,17 +38,6 @@ async function refocusConnectedTab(page: Page): Promise { await expect(page.getByText("Online", { exact: true })).toBeVisible(); } -async function setTabVisibility(page: Page, visibility: DocumentVisibilityState): Promise { - await page.evaluate((state) => { - Object.defineProperty(document, "visibilityState", { configurable: true, get: () => state }); - Object.defineProperty(document, "hidden", { - configurable: true, - get: () => state === "hidden", - }); - document.dispatchEvent(new Event("visibilitychange")); - }, visibility); -} - async function openHostOverview(page: Page, daemon: IsolatedHostDaemon): Promise { await page.goto("/"); await seedSavedSettingsHosts(page, [ diff --git a/packages/app/e2e/browser/sidebar-workspace-mark-unread.spec.ts b/packages/app/e2e/browser/sidebar-workspace-mark-unread.spec.ts index a2a695af5f..670f5564b1 100644 --- a/packages/app/e2e/browser/sidebar-workspace-mark-unread.spec.ts +++ b/packages/app/e2e/browser/sidebar-workspace-mark-unread.spec.ts @@ -300,15 +300,11 @@ test("clicking the already focused timeline acknowledges a new completion", asyn async function reconnectWithoutInteracting( page: Page, gate: Awaited>, - agentId: string, ) { await gate.drop(); await gate.waitForBlockedConnection(); - gate.holdTimelineResponses(agentId); gate.restore(); - await gate.waitForHeldTimelineResponse(); await expectReconnectingToastGone(page); - gate.releaseHeldTimelineResponses(); await expect(page.getByRole("alert").filter({ hasText: "Updating messages" })).toHaveCount(0); } @@ -325,7 +321,7 @@ test("reconnecting a focused chat preserves its unread completion until another ); await workspaces.subject.client.waitForFinish(workspaces.subject.agentId, 20_000); await expectStatus(page, workspaces.subject.workspaceId, "attention"); - await reconnectWithoutInteracting(page, gate, workspaces.subject.agentId); + await reconnectWithoutInteracting(page, gate); await expectStatus(page, workspaces.subject.workspaceId, "attention"); await page.getByText("Unread across reconnect.", { exact: true }).click(); await expectStatus(page, workspaces.subject.workspaceId, "done"); diff --git a/packages/app/e2e/support/helpers/plugin-timeline.ts b/packages/app/e2e/support/helpers/plugin-timeline.ts index d321aaa236..923e4fabd7 100644 --- a/packages/app/e2e/support/helpers/plugin-timeline.ts +++ b/packages/app/e2e/support/helpers/plugin-timeline.ts @@ -48,7 +48,13 @@ export function trackPluginChatRecovery(page: Page) { async expectRecoveredWithoutRetransmission() { await this.expectCatalogWithoutRetransmission(); await expect - .poll(() => messages.some((message) => message.type === "fetch_agent_timeline_response")) + .poll(() => + messages.some( + (message) => + message.type === "fetch_agent_timeline_response" || + message.type === "agent.timeline.set_subscription.response", + ), + ) .toBe(true); }, }; diff --git a/packages/app/e2e/support/helpers/timeline-resume.ts b/packages/app/e2e/support/helpers/timeline-resume.ts index d8926013b2..d0ffd4156e 100644 --- a/packages/app/e2e/support/helpers/timeline-resume.ts +++ b/packages/app/e2e/support/helpers/timeline-resume.ts @@ -1,14 +1,25 @@ import { expect, type Page } from "@playwright/test"; import type { installDaemonWebSocketGate } from "./daemon-websocket-gate"; import type { LongTimelineAgent } from "./timeline-pagination"; -import { expectReconnectingToastGone, expectReconnectingToastVisible } from "./workspace-ui"; +import { + expectReconnectingToastGone, + expectReconnectingToastVisible, + setTabVisibility, +} from "./workspace-ui"; type DaemonWebSocketGate = Awaited>; +const SUBSCRIPTION_REQUEST = "agent.timeline.set_subscription.request"; +const SUBSCRIPTION_RESPONSE = "agent.timeline.set_subscription.response"; +const TIMELINE_REQUEST = "fetch_agent_timeline_request"; +const TIMELINE_RESPONSE = "fetch_agent_timeline_response"; + export interface TimelineRequestCounts { agentId?: string; after: number; tail: number; + subscriptions: number; + fetches: number; } export interface BackgroundTimelineTurns { @@ -42,17 +53,89 @@ export function rememberTimelineRequestCounts( agentId, after: gate.getTimelineRequestCount("after", agentId), tail: gate.getTimelineRequestCount("tail", agentId), + subscriptions: gate.getClientRequestCount(SUBSCRIPTION_REQUEST), + fetches: gate.getClientRequestCount(TIMELINE_REQUEST), }; } -export function expectOneResumeCheckWithoutTail( +export function expectResumeAnsweredWithoutFetch( gate: DaemonWebSocketGate, before: TimelineRequestCounts, ): void { - expect(gate.getTimelineRequestCount("after", before.agentId) - before.after).toBe(1); + expect(gate.getTimelineRequestCount("after", before.agentId) - before.after).toBe(0); expect(gate.getTimelineRequestCount("tail", before.agentId) - before.tail).toBe(0); } +/** Resolves once the daemon has told the app where this chat stands, whichever way it asked. */ +export async function waitForResumeAnswer( + gate: DaemonWebSocketGate, + before: TimelineRequestCounts, +): Promise { + await Promise.race([ + gate.waitForServerMessage(SUBSCRIPTION_RESPONSE, before.subscriptions + 1), + gate.waitForServerMessage(TIMELINE_RESPONSE, before.fetches + 1), + ]); +} + +/** Records every appearance of the toast, including one that lasts a single commit. */ +export async function watchForUpdatingToast(page: Page): Promise<() => Promise> { + const watch = await page.evaluateHandle(() => { + let shown = false; + const check = () => { + shown ||= document.querySelector('[data-testid="agent-updating-toast"]') !== null; + }; + const observer = new MutationObserver(check); + observer.observe(document.body, { childList: true, subtree: true }); + check(); + return { + stop() { + observer.disconnect(); + return shown; + }, + }; + }); + return async () => { + await waitForNextPaint(page); + const shown = await watch.evaluate((recorder) => recorder.stop()); + await watch.dispose(); + expect(shown, "the Updating messages toast appeared").toBe(false); + }; +} + +/** Reopens the app on this chat, recording the toast from the first paint of the new page. */ +export async function reopenAppWatchingForUpdatingToast(page: Page): Promise<() => Promise> { + await page.addInitScript(() => { + const recorder = window as Window & { updatingToastShown?: boolean }; + new MutationObserver(() => { + recorder.updatingToastShown ||= + document.querySelector('[data-testid="agent-updating-toast"]') !== null; + }).observe(document, { childList: true, subtree: true }); + }); + await page.reload(); + return async () => { + await waitForNextPaint(page); + const shown = await page.evaluate( + () => (window as Window & { updatingToastShown?: boolean }).updatingToastShown === true, + ); + expect(shown, "the Updating messages toast appeared").toBe(false); + }; +} + +async function waitForNextPaint(page: Page): Promise { + await page.evaluate( + () => + new Promise((resolve) => { + requestAnimationFrame(() => requestAnimationFrame(() => resolve())); + }), + ); +} + +export async function hideTabAndReturn(page: Page): Promise { + await setTabVisibility(page, "hidden"); + await waitForNextPaint(page); + await setTabVisibility(page, "visible"); +} + export function expectResumeOverflowFallsBackToOneTail( gate: DaemonWebSocketGate, before: TimelineRequestCounts, @@ -69,15 +152,34 @@ export async function disconnectViewedTimeline( await expectReconnectingToastVisible(page); } -export async function restoreViewedTimelineWithHeldResponse( +export async function restoreViewedTimelineHoldingCatchUp( page: Page, gate: DaemonWebSocketGate, ): Promise { - gate.holdNextServerMessage("fetch_agent_timeline_response"); + gate.holdNextServerMessage(TIMELINE_RESPONSE); gate.restore(); await gate.waitForHeldServerMessage(); await expectReconnectingToastGone(page); await expect(page.getByRole("alert").filter({ hasText: "Updating messages" })).toBeVisible(); +} + +export async function releaseHeldCatchUp(page: Page, gate: DaemonWebSocketGate): Promise { gate.releaseHeldServerMessage(); await expect(page.getByRole("alert").filter({ hasText: "Updating messages" })).toHaveCount(0); } + +export async function restoreViewedTimelineWithHeldResponse( + page: Page, + gate: DaemonWebSocketGate, +): Promise { + await restoreViewedTimelineHoldingCatchUp(page, gate); + await releaseHeldCatchUp(page, gate); +} + +export async function restoreUnchangedViewedTimeline( + page: Page, + gate: DaemonWebSocketGate, +): Promise { + gate.restore(); + await expectReconnectingToastGone(page); +} diff --git a/packages/app/e2e/support/helpers/workspace-ui.ts b/packages/app/e2e/support/helpers/workspace-ui.ts index fe03ec053c..3e8aa0401c 100644 --- a/packages/app/e2e/support/helpers/workspace-ui.ts +++ b/packages/app/e2e/support/helpers/workspace-ui.ts @@ -91,6 +91,20 @@ export async function expectWorkspaceHeader( }); } +export async function setTabVisibility( + page: Page, + visibility: DocumentVisibilityState, +): Promise { + await page.evaluate((state) => { + Object.defineProperty(document, "visibilityState", { configurable: true, get: () => state }); + Object.defineProperty(document, "hidden", { + configurable: true, + get: () => state === "hidden", + }); + document.dispatchEvent(new Event("visibilitychange")); + }, visibility); +} + export async function expectReconnectingToastVisible( page: Page, options?: { timeout?: number }, diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 0b46f5f2de..a17f7e0a3f 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -386,8 +386,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider }); }; const sync = getHostRuntimeStore().createViewedTimelineOwner(serverId, { - observe: (agentIds) => { - const observation = client.observeTimeline(agentIds); + canCheckFreshness: () => + client.getLastServerInfoMessage()?.features?.timelineFreshnessCheck === true, + observe: (agentIds, cursors) => { + const observation = client.observeTimeline(agentIds, { cursors }); observation.subscribe({ snapshot: () => {}, update: (message) => { 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 68102ed131..1216a6d14e 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 @@ -455,6 +455,60 @@ describe("deriveAgentScreenViewState", () => { expect(ready.sync).toEqual({ status: "idle" }); }); + it("stays silent over cached content while the freshness check is unanswered", () => { + const input: AgentScreenMachineInput = { + ...createBaseInput(), + agent: createAgentWithStatus({ id: "agent-1", status: "idle" }), + visibilityCatchUpStatus: "checking", + hasHydratedHistoryBefore: true, + }; + + const checking = expectReadyState( + deriveAgentScreenViewState({ input, memory: createBaseMemory() }).state, + ); + const behind = expectReadyState( + deriveAgentScreenViewState({ + input: { ...input, visibilityCatchUpStatus: "pending" }, + memory: createBaseMemory(), + }).state, + ); + + expect(checking.sync).toEqual({ status: "catching_up", ui: "silent" }); + expect(behind.sync).toEqual({ status: "catching_up", ui: "status" }); + }); + + it("keeps reconnecting until the daemon answers the check made after a reconnect", () => { + const ready = expectReadyState( + deriveAgentScreenViewState({ + input: { + ...createBaseInput(), + agent: createAgentWithStatus({ id: "agent-1", status: "idle" }), + isConnected: true, + visibilityCatchUpStatus: "reattaching", + hasHydratedHistoryBefore: true, + }, + memory: createBaseMemory(), + }).state, + ); + + expect(ready.sync).toEqual({ status: "reconnecting" }); + }); + + it("keeps the blocking overlay while a chat with nothing cached is checked", () => { + const input: AgentScreenMachineInput = { + ...createBaseInput(), + agent: createAgentWithStatus({ id: "agent-1", status: "idle" }), + visibilityCatchUpStatus: "checking", + hasHydratedHistoryBefore: false, + }; + + const ready = expectReadyState( + deriveAgentScreenViewState({ input, memory: createBaseMemory() }).state, + ); + + expect(ready.sync).toEqual({ status: "catching_up", ui: "overlay" }); + }); + it("keeps optimistic create non-blocking while timeline and authoritative history catch up", () => { const memory = createBaseMemory(); const input: AgentScreenMachineInput = { 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 082d11827e..4718ec43a1 100644 --- a/packages/app/src/hooks/use-agent-screen-state-machine.ts +++ b/packages/app/src/hooks/use-agent-screen-state-machine.ts @@ -157,12 +157,13 @@ function resolveAgentScreenSource(args: { function resolveCatchingUpUi(args: { hasOptimisticCreateContinuity: boolean; + isFreshnessCheckPending: boolean; isVisibilityCatchUpPending: boolean; hasHydratedHistoryBefore: boolean; hadInitialSyncFailure: boolean; }): "overlay" | "status" | "silent" { if (args.hasOptimisticCreateContinuity) return "silent"; - if (args.hasHydratedHistoryBefore) return "status"; + if (args.hasHydratedHistoryBefore) return args.isFreshnessCheckPending ? "silent" : "status"; if (args.isVisibilityCatchUpPending) return "overlay"; if (args.hadInitialSyncFailure) return "silent"; return "overlay"; @@ -180,7 +181,11 @@ function resolveAgentScreenSync(args: { ) { return { status: "idle" }; } - if (!input.isConnected && !input.isArchived) { + // A reconnected chat is not back until the daemon answers whether it is current. + if ( + (!input.isConnected || input.visibilityCatchUpStatus === "reattaching") && + !input.isArchived + ) { return { status: "reconnecting" }; } if (input.missingAgentState.kind === "error") { @@ -189,16 +194,16 @@ function resolveAgentScreenSync(args: { if (input.visibilityCatchUpStatus === "error" || input.visibilityCatchUpStatus === "retrying") { return { status: "sync_error", isRetrying: input.visibilityCatchUpStatus === "retrying" }; } - if ( - input.visibilityCatchUpStatus === "pending" || - input.needsAuthoritativeSync || - input.isHistorySyncing - ) { + const isFreshnessCheckPending = input.visibilityCatchUpStatus === "checking"; + const isVisibilityCatchUpPending = + input.visibilityCatchUpStatus === "pending" || isFreshnessCheckPending; + if (isVisibilityCatchUpPending || input.needsAuthoritativeSync || input.isHistorySyncing) { return { status: "catching_up", ui: resolveCatchingUpUi({ hasOptimisticCreateContinuity: hasOptimisticCreateContinuity(input), - isVisibilityCatchUpPending: input.visibilityCatchUpStatus === "pending", + isFreshnessCheckPending, + isVisibilityCatchUpPending, hasHydratedHistoryBefore: input.hasHydratedHistoryBefore, hadInitialSyncFailure, }), diff --git a/packages/app/src/stores/workspace-layout-store.test.ts b/packages/app/src/stores/workspace-layout-store.test.ts index 75093f4b8c..9542e4b3f6 100644 --- a/packages/app/src/stores/workspace-layout-store.test.ts +++ b/packages/app/src/stores/workspace-layout-store.test.ts @@ -165,6 +165,8 @@ it("feeds restored layout to timeline sync as a release signal, not a subscripti const sync = createViewedTimelineSync({ replaceDemandedAgentIds: () => undefined, prepare: async () => undefined, + canCheckFreshness: () => false, + applyCurrentTimeline: () => undefined, observe: (agentIds) => { observed.push(agentIds); return { ready: Promise.resolve(), release: async () => undefined }; diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index d18bae6ae3..d73206cd8a 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -76,6 +76,7 @@ function createOwner( drainQueuedAgentMessage: () => undefined, applyTurnLiveness, ports: { + canCheckFreshness: () => false, observe, readCursor: () => undefined, fetchPage, @@ -1172,6 +1173,7 @@ describe("create handoff lifetime", () => { drainQueuedAgentMessage: () => undefined, applyTurnLiveness: () => undefined, ports: { + canCheckFreshness: () => false, observe: () => ({ ready: Promise.resolve(), release: async () => undefined }), readCursor: () => undefined, fetchPage: async () => { diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index eeb000aefb..54ebc83d61 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -5,6 +5,7 @@ import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan"; import { consumeForcedTimelineTailReplacement, createViewedTimelineSync, + FRESHNESS_CHECK_QUIET_MS, type TimelineResponsePayload, type ViewedTimelineStatus, type ViewedTimelineSyncPorts, @@ -28,7 +29,8 @@ function deferred(): Deferred { interface MembershipRequest { agentIds: string[]; - succeed(): void; + cursors: Record | undefined; + succeed(timelines?: Record): void; fail(message: string): void; } @@ -126,8 +128,12 @@ test("switching chats promotes the newly visible catch-up while another visible }); class TimelineWorld { - constructor(private readonly observeMembership?: ViewedTimelineSyncPorts["observe"]) {} + constructor( + private readonly observeMembership?: ViewedTimelineSyncPorts["observe"], + private readonly checksFreshness = false, + ) {} readonly errors: string[] = []; + readonly appliedCurrent: string[] = []; readonly releasedMemberships: string[][] = []; readonly cursors = new Map(); readonly cacheRequests: string[] = []; @@ -149,12 +155,17 @@ class TimelineWorld { this.cacheRequestWaiters.shift()?.(agentId); await this.cacheGate?.promise; }, - observe: (agentIds) => { - if (this.observeMembership) return this.observeMembership(agentIds); - const result = deferred(); + canCheckFreshness: () => this.checksFreshness, + applyCurrentTimeline: (agentId) => { + this.appliedCurrent.push(agentId); + }, + observe: (agentIds, cursors) => { + if (this.observeMembership) return this.observeMembership(agentIds, cursors); + const result = deferred<{ timelines?: Record }>(); this.memberships.push({ agentIds, - succeed: () => result.resolve(), + cursors, + succeed: (timelines) => result.resolve({ timelines }), fail: (message) => result.reject(new Error(message)), }); this.releaseMembershipWaiter(); @@ -1200,3 +1211,271 @@ test("disposing a view releases its pending observation before bootstrap complet await Promise.resolve(); world.expectNoPendingFetch(); }); + +function freshnessWorld(): TimelineWorld { + return new TimelineWorld(undefined, true); +} + +test("a chat the daemon reports current finishes catching up without a fetch", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.sync.setConnected(true); + world.show("pane", ["agent-a"]); + + const membership = await world.nextMembership(); + expect(membership.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 42 } }); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + membership.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + expect(world.appliedCurrent).toEqual(["agent-a"]); + expect(world.reportedCurrent).toEqual([{ agentId: "agent-a", status: "ready" }]); + world.expectNoPendingFetch(); + world.sync.dispose(); +}); + +test("a chat the daemon reports behind fetches from its cursor and reports pending", 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": "behind" }); + + const fetch = await world.nextFetch("agent-a"); + expect(fetch.request).toMatchObject({ + direction: "after", + cursor: { epoch: "epoch-a", seq: 42 }, + }); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + expect(world.appliedCurrent).toEqual([]); + fetch.respond({ hasNewer: false }); + world.sync.dispose(); +}); + +test("a chat with no cached cursor is not checked and fetches its tail", async () => { + const world = freshnessWorld(); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + + const membership = await world.nextMembership(); + expect(membership.cursors).toEqual({}); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + membership.succeed(); + + expect((await world.nextFetch("agent-a")).request.direction).toBe("tail"); + world.sync.dispose(); +}); + +test("an answer without an entry for a chat falls back to the fetch", 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({}); + + expect((await world.nextFetch("agent-a")).request.direction).toBe("after"); + world.sync.dispose(); +}); + +test("a chat being opened waits for its cache before the daemon compares its cursor", async () => { + const world = freshnessWorld(); + world.cacheGate = deferred(); + world.sync.setConnected(true); + world.show("pane", ["agent-a"]); + await world.nextCacheRequest(); + + world.expectNoPendingMembership(); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + world.cursors.set("agent-a", { epoch: "cached", endSeq: 9 }); + world.cacheGate.resolve(); + + const membership = await world.nextMembership(); + expect(membership.cursors).toEqual({ "agent-a": { epoch: "cached", seq: 9 } }); + membership.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.expectNoPendingFetch(); + world.sync.dispose(); +}); + +test("an unanswered check starts reporting pending after the delay and resolves on the answer", async () => { + vi.useFakeTimers(); + try { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.sync.setConnected(true); + world.show("pane", ["agent-a"]); + const membership = await world.nextMembership(); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + await vi.advanceTimersByTimeAsync(FRESHNESS_CHECK_QUIET_MS - 1); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + await vi.advanceTimersByTimeAsync(1); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + world.expectNoPendingFetch(); + + membership.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 slow cache read does not use up the delay for the daemon's answer", async () => { + vi.useFakeTimers(); + try { + const world = freshnessWorld(); + world.cacheGate = deferred(); + world.sync.setConnected(true); + world.show("pane", ["agent-a"]); + await world.nextCacheRequest(); + + await vi.advanceTimersByTimeAsync(FRESHNESS_CHECK_QUIET_MS * 4); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + world.cursors.set("agent-a", { epoch: "cached", endSeq: 9 }); + world.cacheGate.resolve(); + const membership = await world.nextMembership(); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + + await vi.advanceTimersByTimeAsync(FRESHNESS_CHECK_QUIET_MS); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + + membership.succeed({ "agent-a": "current" }); + await vi.advanceTimersByTimeAsync(0); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"); + world.sync.dispose(); + } finally { + vi.useRealTimers(); + } +}); + +test("foregrounding a current chat checks over the subscription instead of fetching", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + const first = await world.nextMembership(); + first.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("checking"); + const check = await world.nextMembership(); + expect(check.agentIds).toEqual(["agent-a"]); + expect(check.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 42 } }); + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]])); + + check.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.expectNoPendingFetch(); + world.sync.dispose(); +}); + +test("foregrounding asks only about visible chats; hidden chats stay caught up", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + world.cursors.set("agent-b", { epoch: "epoch-b", endSeq: 2 }); + world.openedEarlier(["agent-b"]); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + (await world.nextMembership()).succeed({ "agent-a": "current", "agent-b": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + + const check = await world.nextMembership(); + expect(check.agentIds).toEqual(["agent-a", "agent-b"]); + expect(check.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 1 } }); + expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"); + check.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.expectNoPendingFetch(); + world.sync.dispose(); +}); + +test("a gap found while the check is outstanding wins over a current answer", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + const membership = await world.nextMembership(); + + world.sync.recoverGap("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.expectNoPendingFetch(); + membership.succeed({ "agent-a": "current" }); + + const fetch = await world.nextFetch("agent-a"); + expect(fetch.request).toMatchObject({ direction: "after", cursor: { seq: 42 } }); + expect(world.appliedCurrent).toEqual([]); + fetch.respond({ hasNewer: false }); + world.sync.dispose(); +}); + +test("a failed check reports the error and retries through membership", 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()).fail("subscribe failed"); + + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error")); + world.sync.retryVisibleAgentTimeline("agent-a"); + const retry = await world.nextMembership(); + expect(retry.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 42 } }); + retry.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.sync.dispose(); +}); + +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")); + + 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(); +}); + +test("without the daemon capability nothing changes: no cursors, no checking, fetch on foreground", async () => { + const world = new TimelineWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 42 }); + world.show("pane", ["agent-a"]); + world.sync.setConnected(true); + + const membership = await world.nextMembership(); + expect(membership.cursors).toBeUndefined(); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + membership.succeed(); + const first = await world.nextFetch("agent-a"); + expect(first.request.direction).toBe("after"); + first.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + + world.sync.setActive(false); + world.sync.setActive(true); + + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + expect((await world.nextFetch("agent-a")).request.direction).toBe("after"); + world.expectNoPendingMembership(); + world.sync.dispose(); +}); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 94d4b685f8..7c2d16e50f 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -339,10 +339,53 @@ function applyAuthoritativeTimelineResponse(input: { return true; } +/** + * Measured on a dev daemon: set_subscription answers in a mean of 1.4 ms, p99 per-window max + * 107 ms, max 227 ms. A check slower than this is slow enough to say so on screen. + */ +export const FRESHNESS_CHECK_QUIET_MS = 500; + +export type TimelineFreshness = "current" | "behind"; + +/** The empty after page the daemon would have returned for a cursor already at the head. */ +function buildCurrentTimelinePage( + agentId: string, + cursor: { epoch: string; endSeq: number }, +): TimelineResponsePayload { + return { + requestId: "", + agentId, + agent: null, + direction: "after", + projection: "projected", + epoch: cursor.epoch, + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: cursor.endSeq, nextSeq: cursor.endSeq + 1 }, + startCursor: null, + endCursor: null, + hasOlder: false, + hasNewer: false, + entries: [], + error: null, + }; +} + export interface ViewedTimelineSyncPorts { prepare(agentId: string): Promise; replaceDemandedAgentIds(agentIds: string[]): void; - observe(agentIds: string[]): { ready: Promise; release(): Promise }; + /** Whether the daemon answers with each cursor's freshness in the subscribe response. */ + canCheckFreshness(): boolean; + observe( + agentIds: string[], + cursors?: Record, + ): { + ready: Promise<{ timelines?: Record } | void>; + release(): Promise; + }; + /** The daemon confirmed the cached timeline is at its head, as a zero-row after page would. */ + applyCurrentTimeline(agentId: string): void; readCursor(agentId: string): { epoch: string; endSeq: number } | undefined; fetchPage( agentId: string, @@ -361,7 +404,17 @@ export interface ViewedTimelineSyncPorts { reportError(error: unknown): void; } -export type ViewedTimelineStatus = "ready" | "pending" | "error" | "retrying"; +/** + * "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. + */ +export type ViewedTimelineStatus = + | "ready" + | "checking" + | "reattaching" + | "pending" + | "error" + | "retrying"; export interface ViewedTimelineUiBridge { replaceVisibleAgentIds(sourceId: string, agentIds: string[]): void; @@ -386,7 +439,12 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge { export type ViewedTimelineOwnerPorts = Omit< ViewedTimelineSyncPorts, - "prepare" | "replaceDemandedAgentIds" | "onCatchUpEnded" | "onCatchUpOwed" | "onCatchUpSettled" + | "prepare" + | "replaceDemandedAgentIds" + | "onCatchUpEnded" + | "onCatchUpOwed" + | "onCatchUpSettled" + | "applyCurrentTimeline" >; export interface ViewedTimelineOwner extends ViewedTimelineSync { @@ -422,6 +480,18 @@ export function createViewedTimelineOwner(input: { prepare: (agentId) => input.replica.prepare(agentId), readCursor: (agentId) => input.replica.readCursor(agentId) ?? input.ports.readCursor(agentId), replaceDemandedAgentIds: input.replaceDemandedAgentIds, + applyCurrentTimeline: (agentId) => { + const cursor = input.replica.readCursor(agentId) ?? input.ports.readCursor(agentId); + if (!cursor) return; + const accepted = applyAuthoritativeTimelineResponse({ + serverId: input.serverId, + payload: buildCurrentTimelinePage(agentId, cursor), + cachedCursor: input.replica.readRange(agentId), + recoverGap: (id, gapCursor) => sync.recoverGap(id, gapCursor), + drainQueuedAgentMessage: input.drainQueuedAgentMessage, + }); + if (accepted) input.replica.timelineUpdated(agentId); + }, onCatchUpEnded: (agentId) => useCreateFlowStore.getState().clearByAgent({ serverId: input.serverId, agentId }), onCatchUpOwed: (agentId) => input.replica.catchUpOwed(agentId), @@ -526,6 +596,12 @@ 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. + const freshnessChecks = new Map< + string, + { quiet: boolean; reattaching: boolean; timer: ReturnType | null } + >(); const listeners = new Set<() => void>(); let active = true; let connected = false; @@ -538,6 +614,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed let acknowledged: string[] = []; let membershipGeneration = 0; let membershipNeedsRetry = false; + let freshnessCheckOwed = false; // 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; @@ -593,6 +670,33 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed ports.onCatchUpOwed(agentId); }; + const startFreshnessCheck = (agentId: string, reattaching = false) => { + if (freshnessChecks.has(agentId)) return; + freshnessChecks.set(agentId, { quiet: true, reattaching, 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. + const startFreshnessCheckDelay = (agentId: string) => { + const check = freshnessChecks.get(agentId); + if (!check || check.timer) return; + check.timer = setTimeout(() => { + check.quiet = false; + notifyListeners(); + }, FRESHNESS_CHECK_QUIET_MS); + }; + + const endFreshnessCheck = (agentId: string) => { + const check = freshnessChecks.get(agentId); + if (!check) return; + if (check.timer) clearTimeout(check.timer); + freshnessChecks.delete(agentId); + }; + + const endAllFreshnessChecks = () => { + for (const agentId of freshnessChecks.keys()) endFreshnessCheck(agentId); + }; + const cancelCatchUp = (agentId: string) => { catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1); catchUps.delete(agentId); @@ -682,6 +786,10 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } = {}, ) => { const { request, supersede = false } = options; + if (freshnessChecks.has(agentId)) { + if (request) pendingCatchUps.set(agentId, request); + return; + } if (!canCatchUp(agentId)) { if (request) pendingCatchUps.set(agentId, request); return; @@ -736,30 +844,96 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } }; + const membershipIsSettled = (requested: string[]) => + !membershipNeedsRetry && + !freshnessCheckOwed && + sameAgentIds(requested, observation?.agentIds ?? acknowledged); + + // A chat with no fetch running and a cursor to compare; the daemon can answer without a fetch. + const collectFreshnessCursors = (requested: string[]) => { + const cursors: Record = {}; + for (const agentId of requested) { + if (catchUps.has(agentId) || pendingCatchUps.has(agentId)) continue; + const cursor = ports.readCursor(agentId); + if (cursor) cursors[agentId] = { epoch: cursor.epoch, seq: cursor.endSeq }; + } + return cursors; + }; + + const completeCurrentCatchUp = (agentId: string) => { + const generation = (catchUpGenerations.get(agentId) ?? 0) + 1; + catchUpGenerations.set(agentId, generation); + catchUps.set(agentId, { generation, status: "complete" }); + ports.applyCurrentTimeline(agentId); + setVisibilityCatchUpReady(agentId); + }; + + const applyFreshnessAnswer = ( + checked: string[], + timelines: Record | undefined, + ) => { + for (const agentId of checked) { + endFreshnessCheck(agentId); + // A gap found during the check still needs its fetch; "current" only speaks for the cursor. + if (timelines?.[agentId] === "current" && !pendingCatchUps.has(agentId)) { + completeCurrentCatchUp(agentId); + } + } + if (checked.length > 0) notifyListeners(); + }; + + const isReconcileSuperseded = (generation: number) => + disposed || !connected || generation !== membershipGeneration; + + const askDaemonAboutFreshness = (checked: string[]) => { + for (const agentId of checked) { + startFreshnessCheck(agentId); + startFreshnessCheckDelay(agentId); + } + for (const agentId of freshnessChecks.keys()) { + if (!checked.includes(agentId)) endFreshnessCheck(agentId); + } + }; + const reconcileMembership = async (): Promise => { if (disposed || !connected) return; const generation = membershipGeneration; const requested = desired; - if (!membershipNeedsRetry && sameAgentIds(requested, observation?.agentIds ?? acknowledged)) - return; + if (membershipIsSettled(requested)) return; + const canCheck = ports.canCheckFreshness(); + if (canCheck) { + // The cursor sent to the daemon must include whatever the cache restores. + await Promise.all(requested.map((agentId) => cacheLoads.get(agentId))); + if (isReconcileSuperseded(generation)) return; + if (membershipIsSettled(requested)) return; + } membershipNeedsRetry = false; + freshnessCheckOwed = false; + const cursors = canCheck ? collectFreshnessCursors(requested) : undefined; + const checked = Object.keys(cursors ?? {}); + askDaemonAboutFreshness(checked); + notifyListeners(); + let timelines: Record | undefined; try { const previous = observation; - const handle = ports.observe(requested); + const handle = ports.observe(requested, cursors); observation = { handle, agentIds: requested }; void previous?.handle.release().catch(ports.reportError); - await handle.ready; + const answer = await handle.ready; + timelines = answer ? answer.timelines : undefined; } catch (error) { - if (disposed || !connected || generation !== membershipGeneration) return; + if (isReconcileSuperseded(generation)) return; membershipNeedsRetry = true; acknowledged = []; + for (const agentId of checked) endFreshnessCheck(agentId); for (const agentId of requested) cancelCatchUp(agentId); setVisibilityCatchUpError(requested, error); ports.reportError(error); return; } - if (disposed || !connected || generation !== membershipGeneration) return; + if (isReconcileSuperseded(generation)) return; acknowledged = requested; + applyFreshnessAnswer(checked, timelines); startAcknowledgedCatchUps(); }; @@ -803,6 +977,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed for (const agentId of desired) { if (!nextDesired.includes(agentId)) { cancelCatchUp(agentId); + endFreshnessCheck(agentId); visibilityCatchUpPending.delete(agentId); visibilityCatchUpErrors.delete(agentId); manualRetries.delete(agentId); @@ -815,6 +990,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed markCatchUpOwed(agentId); visibilityCatchUpErrors.delete(agentId); manualRetries.delete(agentId); + if (connected && ports.canCheckFreshness()) startFreshnessCheck(agentId); ensureCacheLoaded(agentId); } } @@ -826,6 +1002,19 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed void reconcileMembership(); }; + // A render can see the connection return, or a chat become visible, before this owner hears + // about it, so those two states answer as if the check had already started. + const getUnconfirmedStatus = (agentId: string): ViewedTimelineStatus => { + const check = freshnessChecks.get(agentId); + if (check) { + if (!check.quiet) return "pending"; + return check.reattaching ? "reattaching" : "checking"; + } + if (!ports.canCheckFreshness()) return "pending"; + if (!connected) return "reattaching"; + return isDesired(agentId) ? "pending" : "checking"; + }; + const publishVisibleMembership = () => { commitDesiredMembership(normalizeAgentIds([...openedAgentIds, ...visibleAgentIds()])); }; @@ -838,8 +1027,8 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed getAgentTimelineStatus(agentId) { if (manualRetries.has(agentId)) return "retrying"; if (visibilityCatchUpErrors.has(agentId)) return "error"; - if (!isDesired(agentId) || visibilityCatchUpPending.has(agentId)) return "pending"; - return "ready"; + if (isDesired(agentId) && !visibilityCatchUpPending.has(agentId)) return "ready"; + return getUnconfirmedStatus(agentId); }, getAgentTimelineError(agentId) { return visibilityCatchUpErrors.get(agentId) ?? null; @@ -867,11 +1056,27 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed startAcknowledgedCatchUps(); return; } + const canCheck = connected && ports.canCheckFreshness(); for (const agentId of visibleAgentIds()) { if (visibilityCatchUpErrors.has(agentId)) continue; markCatchUpOwed(agentId); + if ( + canCheck && + catchUps.get(agentId)?.status === "complete" && + !pendingCatchUps.has(agentId) && + ports.readCursor(agentId) + ) { + catchUps.delete(agentId); + startFreshnessCheck(agentId); + freshnessCheckOwed = true; + continue; + } startCatchUp(agentId, { supersede: true }); } + if (freshnessCheckOwed) { + membershipGeneration += 1; + void reconcileMembership(); + } notifyListeners(); }, setConnected(nextConnected) { @@ -883,10 +1088,18 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed observation = null; acknowledged = []; membershipGeneration += 1; + freshnessCheckOwed = false; + endAllFreshnessChecks(); for (const agentId of desired) cancelCatchUp(agentId); return; } + if (ports.canCheckFreshness()) { + for (const agentId of desired) { + if (visibilityCatchUpPending.has(agentId)) startFreshnessCheck(agentId, true); + } + } membershipGeneration += 1; + notifyListeners(); void reconcileMembership(); }, recoverGap(agentId, cursor) { @@ -913,6 +1126,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed visibilityCatchUpPending.clear(); visibilityCatchUpErrors.clear(); manualRetries.clear(); + endAllFreshnessChecks(); notifyListeners(); listeners.clear(); }, diff --git a/packages/client/src/connection.test.ts b/packages/client/src/connection.test.ts index a3d12e48f6..ef326bf516 100644 --- a/packages/client/src/connection.test.ts +++ b/packages/client/src/connection.test.ts @@ -11,6 +11,7 @@ function connection( broadcasts?: boolean; browserHost?: { hostKind: string; supportedCommands: string[] }; acknowledgeTimelineReads?: boolean; + timelineFreshnessCheck?: boolean; } = {}, ) { const sent: Array<{ @@ -47,6 +48,7 @@ function connection( workspaceMultiplicity: options.workspaceMultiplicity ?? true, selectiveAgentTimeline: !options.broadcasts, explicitEventSubscriptions: !options.broadcasts, + timelineFreshnessCheck: options.timelineFreshnessCheck === true, }, }, }, @@ -887,6 +889,29 @@ test("old attention stream events reach the current notification interface", asy } }); +test("timeline cursors reach only hosts that answer freshness", async () => { + const cursors = { agent: { epoch: "epoch-1", seq: 9 } }; + const sentCursors = async (timelineFreshnessCheck: boolean) => { + const h = connection({ timelineFreshnessCheck }); + try { + const connected = h.client.connect(); + h.open(); + await connected; + const observation = h.client.observeTimeline(["agent"], { cursors }); + await observation.ready; + await observation.release(); + return h.sent.find( + (frame) => frame.message?.type === "agent.timeline.set_subscription.request", + )?.message; + } finally { + await h.client.close(); + } + }; + + expect(await sentCursors(true)).toMatchObject({ cursors }); + expect(await sentCursors(false)).not.toHaveProperty("cursors"); +}); + test("broadcast-only hosts expose local timeline readiness without sending an unsupported RPC", async () => { const h = connection({ ownedSubscriptions: false, broadcasts: true }); try { diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index fa6a0306cc..a401de273f 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -4167,11 +4167,23 @@ export class DaemonClient { observeTimeline( agentIds: string[], - options?: { signal?: AbortSignal }, + options?: { + signal?: AbortSignal; + /** The caller's position per chat; the response answers whether each is at the head. */ + cursors?: Record; + }, ): OwnedSubscription> { return this.observe( "agent.timeline.set_subscription.response", - { type: "agent.timeline.set_subscription.request", agentIds }, + { + type: "agent.timeline.set_subscription.request", + agentIds, + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + ...(options?.cursors && + this.lastServerInfoMessage?.features?.timelineFreshnessCheck === true + ? { cursors: options.cursors } + : {}), + }, options, ); } diff --git a/packages/protocol/src/messages.test.ts b/packages/protocol/src/messages.test.ts index ebe7d9b7d8..453c7aa1c1 100644 --- a/packages/protocol/src/messages.test.ts +++ b/packages/protocol/src/messages.test.ts @@ -492,6 +492,26 @@ describe("daemon update messages", () => { }); describe("viewed timeline subscription messages", () => { + test("carries a cursor per chat and a freshness answer per chat", () => { + const cursors = { "agent-a": { epoch: "epoch-1", seq: 42 } }; + const timelines = { "agent-a": "current", "agent-b": "behind" }; + + expect( + SessionInboundMessageSchema.parse({ + type: "agent.timeline.set_subscription.request", + agentIds: ["agent-a"], + requestId: "timeline-subscription-2", + cursors, + }), + ).toMatchObject({ cursors }); + expect( + SessionOutboundMessageSchema.parse({ + type: "agent.timeline.set_subscription.response", + payload: { agentIds: ["agent-a"], requestId: "timeline-subscription-2", timelines }, + }), + ).toMatchObject({ payload: { timelines } }); + }); + test("parses a complete viewed-agent set and its acknowledgement", () => { const request = SessionInboundMessageSchema.parse({ type: "agent.timeline.set_subscription.request", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 94d6190f57..4cff2be935 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2087,10 +2087,16 @@ export const ProviderSubagentTimelineRequestMessageSchema = z.object({ limit: z.number().int().nonnegative().optional(), }); +// "current" means the daemon has this agent loaded, the epoch matches, and the cursor is at the head. +// Anything else is "behind", including an agent the daemon has not loaded. +export const TimelineFreshnessSchema = z.enum(["current", "behind"]); + export const SetAgentTimelineSubscriptionRequestMessageSchema = z.object({ type: z.literal("agent.timeline.set_subscription.request"), agentIds: z.array(z.string()), requestId: z.string(), + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + cursors: z.record(z.string(), AgentTimelineCursorSchema).optional(), }); export const AgentForkContextRequestMessageSchema = z.object({ @@ -4076,6 +4082,8 @@ export const ServerInfoStatusPayloadSchema = z forgeProviders: z.boolean().optional(), // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: z.boolean().optional(), + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + timelineFreshnessCheck: z.boolean().optional(), explicitEventSubscriptions: z.boolean().optional(), ownedSubscriptions: z.boolean().optional(), providerOptionControls: z.boolean().optional(), @@ -5249,6 +5257,8 @@ export const SetAgentTimelineSubscriptionResponseMessageSchema = z.object({ subscriptionId: z.string().optional(), agentIds: z.array(z.string()), requestId: z.string(), + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + timelines: z.record(z.string(), TimelineFreshnessSchema).optional(), }), }); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 880709f152..62bb1d9797 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1399,6 +1399,11 @@ export class AgentManager { return this.timelineStore.getRows(id); } + /** The timeline head of a loaded agent, or null when this agent is not loaded. Never loads. */ + getLoadedTimelineHead(id: string): { epoch: string; seq: number } | null { + return this.getAgent(id) ? this.timelineStore.getHead(id) : null; + } + fetchTimeline(id: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { this.requireAgent(id); return this.timelineStore.fetch(id, options); diff --git a/packages/server/src/server/agent/agent-timeline-store.ts b/packages/server/src/server/agent/agent-timeline-store.ts index 4c3b9c5c3a..b07531ae62 100644 --- a/packages/server/src/server/agent/agent-timeline-store.ts +++ b/packages/server/src/server/agent/agent-timeline-store.ts @@ -89,6 +89,11 @@ export class InMemoryAgentTimelineStore { return this.requireState(agentId).epoch; } + getHead(agentId: string): { epoch: string; seq: number } { + const state = this.requireState(agentId); + return { epoch: state.epoch, seq: state.nextSeq - 1 }; + } + fetch(agentId: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { const state = this.requireState(agentId); const direction = options?.direction ?? "tail"; diff --git a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts index f47b30ce85..2f602407ec 100644 --- a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts +++ b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts @@ -184,6 +184,7 @@ async function connect(input: { timelineNotifications?: boolean; pluginTimelineItems?: boolean; workspaceSetupBlocked?: boolean; + ownedSubscriptions?: boolean; }): Promise { let socket!: WebSocket; const client = new DaemonClient({ @@ -194,7 +195,7 @@ async function connect(input: { return socket as unknown as WebSocketLike; }, capabilities: { - [CLIENT_CAPS.ownedSubscriptions]: false, + [CLIENT_CAPS.ownedSubscriptions]: input.ownedSubscriptions ?? false, [CLIENT_CAPS.selectiveAgentTimeline]: input.selective, [CLIENT_CAPS.pluginTimelineItems]: input.pluginTimelineItems ?? false, [CLIENT_CAPS.workspaceSetupBlocked]: input.workspaceSetupBlocked ?? false, @@ -211,6 +212,91 @@ async function connect(input: { return connected; } +async function startDaemonWithMockAgents( + home: { paseoHomeRoot: string; cleanup: boolean } | Record = {}, +): Promise { + await daemon.close(); + daemon = await createTestPaseoDaemon({ + isDev: true, + agentClients: { mock: new MockLoadTestAgentClient() }, + ...home, + }); +} + +async function connectForFreshness(clientId: string): Promise { + return connect({ clientId, selective: true, ownedSubscriptions: true }); +} + +async function createChatAtHead(connected: ConnectedClient, title: string) { + const agent = await connected.client.createAgent({ + provider: "mock", + cwd: "/tmp", + title, + model: "ten-second-stream", + }); + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: "Already on screen", + }); + const page = await connected.client.fetchAgentTimeline(agent.id, { direction: "tail" }); + if (!page.endCursor) throw new Error("Expected a timeline cursor"); + return { agentId: agent.id, head: { epoch: page.endCursor.epoch, seq: page.endCursor.seq } }; +} + +test("subscribing with a cursor answers whether each chat is at its head", async () => { + await startDaemonWithMockAgents(); + const connected = await connectForFreshness("freshness-loaded"); + const atHead = await createChatAtHead(connected, "At head"); + const staleSeq = await createChatAtHead(connected, "Stale seq"); + const otherEpoch = await createChatAtHead(connected, "Other epoch"); + const noCursor = await createChatAtHead(connected, "No cursor"); + + const observation = connected.client.observeTimeline( + [atHead.agentId, staleSeq.agentId, otherEpoch.agentId, noCursor.agentId], + { + cursors: { + [atHead.agentId]: atHead.head, + [staleSeq.agentId]: { ...staleSeq.head, seq: staleSeq.head.seq - 1 }, + [otherEpoch.agentId]: { ...otherEpoch.head, epoch: "another-epoch" }, + }, + }, + ); + + expect((await observation.ready).timelines).toEqual({ + [atHead.agentId]: "current", + [staleSeq.agentId]: "behind", + [otherEpoch.agentId]: "behind", + }); + + connected.clear(); + await daemon.daemon.agentManager.appendTimelineItem(atHead.agentId, { + type: "assistant_message", + text: "Arrives after the answer", + }); + await connected.next(isAgentStream(atHead.agentId), "live row after a current answer"); + await observation.release(); +}); + +test("subscribing with a cursor answers behind for a stored agent and leaves it unloaded", async () => { + const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-freshness-")); + await startDaemonWithMockAgents({ paseoHomeRoot, cleanup: false }); + const before = await connectForFreshness("freshness-before-restart"); + const stored = await createChatAtHead(before, "Stored"); + before.close(); + await before.client.close(); + await startDaemonWithMockAgents({ paseoHomeRoot, cleanup: true }); + expect(daemon.daemon.agentManager.getAgent(stored.agentId)).toBeNull(); + + const connected = await connectForFreshness("freshness-after-restart"); + const observation = connected.client.observeTimeline([stored.agentId], { + cursors: { [stored.agentId]: stored.head }, + }); + + expect((await observation.ready).timelines).toEqual({ [stored.agentId]: "behind" }); + expect(daemon.daemon.agentManager.getAgent(stored.agentId)).toBeNull(); + await observation.release(); +}); + test("notification timeline items are sent only to clients that advertise support", async () => { await daemon.close(); daemon = await createTestPaseoDaemon({ diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 9221aa5e0b..e96b5e6320 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1408,6 +1408,21 @@ export class Session { this.refreshObservationProducers(); } + private compareTimelineCursors( + agentIds: string[], + cursors: Record, + ): Record { + const timelines: Record = {}; + for (const agentId of agentIds) { + const cursor = cursors[agentId]; + if (!cursor) continue; + const head = this.agentManager.getLoadedTimelineHead(agentId); + timelines[agentId] = + head?.epoch === cursor.epoch && head.seq === cursor.seq ? "current" : "behind"; + } + return timelines; + } + private subscribeAgentTimelines(agentIds: string[]): OwnedSubscription { const owner = this.delivery.begin("timelines", undefined, (id) => { this.timelineSubscriptions.delete(id); @@ -3045,11 +3060,19 @@ export class Session { } case "agent.timeline.set_subscription.request": { const agentIds = [...new Set(msg.agentIds)].sort(); + // Attach before comparing: a row appended after the comparison reaches this owner live. const owner = this.subscribeAgentTimelines(agentIds); this.emitForSource( { type: "agent.timeline.set_subscription.response", - payload: { agentIds, requestId: msg.requestId, subscriptionId: owner.responseId }, + payload: { + agentIds, + requestId: msg.requestId, + subscriptionId: owner.responseId, + ...(msg.cursors + ? { timelines: this.compareTimelineCursors(agentIds, msg.cursors) } + : {}), + }, }, source, ); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index e507da044b..fe8e1268ce 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -2196,6 +2196,8 @@ export class VoiceAssistantWebSocketServer { forgeProviders: true, // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: true, + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + timelineFreshnessCheck: true, explicitEventSubscriptions: true, // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. canonicalSubmittedPrompts: true,