diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index e975178da7..0a0a7187c4 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -102,30 +102,45 @@ 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. The response carries the agent row for each `current` +chat, because the timeline fetch is also what refreshes that row and the replica cache stores only +part of it. `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 -owner keeps an unsettled turn per chat: the turn its messages were captured during. It is set when a -catch-up becomes owed while the turn is open, or when cached messages are painted that were saved -while one was, and it is saved with those messages in the replica. It carries the turn's start time. -While it is set and the agent reads idle, the chat keeps showing that running turn, with its loader -and elapsed timer unchanged, and there is no separate catching-up footer. The commit that brings the -chat current clears it, in the same store write as the rows, so the final messages and the completed -footer land together. A failed catch-up clears it too, and the footer takes the true status. Losing -the connection is not a failed catch-up: none can run or fail until the host returns, so the running -footer stays beside "Reconnecting to host" and the catch-up on reconnect resolves it. Releasing it -offline would put a completed footer under messages known to be behind. Surfaces outside the timeline (the sidebar, tabs, the composer's stop and send controls) read the true status -and never see this state. Live delivery needs no hold because `applyStreamEvent` commits queued rows -before it applies turn liveness. +agent directory and messages from the chat's own catch-up, and either can land first. So the +timeline owner keeps an unsettled turn per chat: the turn its messages were captured during. It is +set when a check or catch-up becomes owed while the turn is open, or when cached messages are +painted that were saved while one was, and it is saved with those messages in the replica. It +carries the turn's start time. While it is set and the agent reads idle, the chat keeps showing that +running turn, with its loader and elapsed timer unchanged, and there is no separate catching-up +footer. The commit that brings the chat current clears it, in the same store write as the rows, so +the final messages and the completed footer land together. A `current` answer is that commit with no +rows. A check in flight keeps the hold, and `behind` keeps it until the fetched rows land. A failed +catch-up clears it too, and the footer takes the true status. Losing the connection is not a failed +catch-up: none can run or fail until the host returns, so the running footer stays beside +"Reconnecting to host" and the check on reconnect resolves it. Releasing it offline would put a +completed footer under messages known to be behind. Surfaces outside the timeline (the sidebar, +tabs, the composer's stop and send controls) read the true status and never see this state. Live +delivery needs no hold because `applyStreamEvent` commits queued rows before it applies turn +liveness. The draft-create handoff has the same lifetime: the viewed-timeline owner releases it when the sync stops owing that chat a catch-up, not when the first authoritative page lands. Releasing it at the 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/plugin-timeline.spec.ts b/packages/app/e2e/browser/plugin-timeline.spec.ts index d75653298e..aacd7c187b 100644 --- a/packages/app/e2e/browser/plugin-timeline.spec.ts +++ b/packages/app/e2e/browser/plugin-timeline.spec.ts @@ -24,11 +24,12 @@ for (const width of [1100, 390]) { await expectWholeCompletedCard(page); // Local state is checked during growth; the viewport remounts on the history handoff. const recovery = trackPluginChatRecovery(page); - gate.holdTimelineResponses(agent.agentId); + // A chat with a cached cursor answers current and never fetches, so hold the subscribe itself. + gate.holdNextClientRequest("agent.timeline.set_subscription.request"); await page.reload({ waitUntil: "domcontentloaded" }); - await gate.waitForHeldTimelineResponse(); + await gate.waitForHeldClientRequest(); await recovery.expectCatalogWithoutRetransmission(); - gate.releaseHeldTimelineResponses(); + gate.releaseHeldClientRequest(); await expectWholeCompletedCard(page); await recovery.expectRecoveredWithoutRetransmission(); recovery.reset(); 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/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index a91f335048..ba13bb8a9f 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -726,13 +726,11 @@ function ChatAgentContent({ (listener: () => void) => viewedTimelineSync?.subscribe(listener) ?? (() => {}), [viewedTimelineSync], ); - const readTimelineStatus = useCallback( - () => - !agentId || !viewedTimelineSync - ? ("ready" as const) - : viewedTimelineSync.getAgentTimelineStatus(agentId), - [agentId, viewedTimelineSync], - ); + const readTimelineStatus = useCallback((): ViewedTimelineStatus => { + if (!agentId) return "ready"; + // Until the owner is published nothing has confirmed this chat, so its check has not begun. + return viewedTimelineSync?.getAgentTimelineStatus(agentId) ?? "checking"; + }, [agentId, viewedTimelineSync]); const timelineStatus = useSyncExternalStore( subscribeToVisibilityCatchUp, readTimelineStatus, diff --git a/packages/app/src/runtime/directory-sync/index.ts b/packages/app/src/runtime/directory-sync/index.ts index 1ffb4fae5f..0f539a2b1d 100644 --- a/packages/app/src/runtime/directory-sync/index.ts +++ b/packages/app/src/runtime/directory-sync/index.ts @@ -4,6 +4,7 @@ import type { FetchAgentsEntry, FetchAgentsOptions, } from "@getpaseo/client/internal/daemon-client"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once"; import { normalizeProjectDescriptor, @@ -482,6 +483,26 @@ export class DirectorySync { return page; } + /** + * Captures these chats' lifecycle before a subscribe. The returned function keeps the agent rows + * its answer carries, which is what a timeline fetch would have delivered for each of them. + */ + captureSubscribedAgents( + agentIds: string[], + ): (agents: Record) => void { + const tokens = new Map(agentIds.map((id) => [id, this.agents.captureTimeline(id)])); + return (agents) => { + let accepted = false; + for (const [agentId, agent] of Object.entries(agents)) { + const token = tokens.get(agentId); + if (token && this.agents.submitTimelineAgent(token, agent)) accepted = true; + } + if (!accepted) return; + this.revision += 1; + this.persistCheckpoint(); + }; + } + async refreshAgents( input: RefreshAgentDirectoryInput = {}, ): Promise { diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index d6fe75ab12..15a86b7bb4 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -2880,7 +2880,20 @@ export class HostRuntimeStore { drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(serverId, agentId), applyTurnLiveness: (agentId, transitions) => directory.applyAgentTurnLiveness(agentId, transitions), - ports, + ports: { + ...ports, + observe: (agentIds, cursors) => { + const keepAgents = directory.captureSubscribedAgents(agentIds); + const observation = ports.observe(agentIds, cursors); + return { + ready: observation.ready.then((answer) => { + if (answer?.agents) keepAgents(answer.agents); + return answer; + }), + release: () => observation.release(), + }; + }, + }, }); } 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..13046701ee 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -63,6 +63,7 @@ function createOwner( ready: Promise.resolve(), release: async () => undefined, }), + checksFreshness = false, ): ViewedTimelineOwner { const replica = createTimelineReplica({ serverId: SERVER_ID, @@ -76,8 +77,16 @@ function createOwner( drainQueuedAgentMessage: () => undefined, applyTurnLiveness, ports: { + canCheckFreshness: () => checksFreshness, observe, - readCursor: () => undefined, + readCursor: () => { + if (!checksFreshness) return undefined; + const session = useSessionStore.getState().sessions[SERVER_ID]; + const timeline = selectAgentTimelineState(session, AGENT_ID); + return timeline.status === "synced" && timeline.range + ? { epoch: timeline.range.epoch, endSeq: timeline.range.endSeq } + : undefined; + }, fetchPage, fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), reportError: () => undefined, @@ -894,6 +903,155 @@ describe("the turn footer changes together with its messages", () => { owner.dispose(); }); + type Answer = "current" | "behind"; + + // A chat whose subscribes wait for the test to answer them, and whose fetches are recorded. + function createCheckedChat(cachedReply: string | null) { + useSessionStore.getState().initializeSession(SERVER_ID, null); + useSessionStore.getState().applyAgentTimelineResponseState(SERVER_ID, AGENT_ID, { + items: cachedReply ? [item("reply-1", cachedReply, 4)] : [], + head: [], + range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 }, + older: "none", + newer: false, + synchronized: true, + acknowledgedClientMessageIds: [], + }); + const unanswered: ((answer: Answer, beforeAnswer?: () => void) => void)[] = []; + let fetches = 0; + let finishFetch = () => {}; + const owner = createOwner( + { readTimeline: async () => undefined, commitTimeline: () => undefined }, + undefined, + () => { + fetches += 1; + return new Promise((resolve) => { + finishFetch = () => resolve({ hasNewer: false, endCursor: null }); + }); + }, + () => ({ + // The host runtime keeps the answer's agent rows in this same continuation. + ready: new Promise((resolve) => { + unanswered.push((answer, beforeAnswer) => { + beforeAnswer?.(); + resolve({ timelines: { [AGENT_ID]: answer } }); + }); + }), + release: async () => undefined, + }), + true, + ); + const answer = async (value: Answer, beforeAnswer?: () => void) => { + await expect.poll(() => unanswered.length).toBeGreaterThan(0); + unanswered.shift()?.(value, beforeAnswer); + }; + const open = async () => { + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + owner.setConnected(true); + await answer("current"); + await expect.poll(() => owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + }; + return { owner, answer, open, fetchCount: () => fetches, finishFetch: () => finishFetch() }; + } + + function leaveAndReturn(owner: ViewedTimelineOwner, way: "reconnect" | "foreground"): void { + if (way === "reconnect") { + owner.setConnected(false); + owner.setConnected(true); + return; + } + owner.setActive(false); + owner.setActive(true); + } + + it.each([ + ["reconnect", "reattaching"], + ["foreground", "checking"], + ] as const)( + "freshness check on %s: the held turn stays through the check and current settles it", + async (way, status) => { + const chat = createCheckedChat("Done."); + agentIsRunning(); + await chat.open(); + + leaveAndReturn(chat.owner, way); + idleSnapshotArrives(); + expect(chat.owner.getAgentTimelineStatus(AGENT_ID)).toBe(status); + expect(footer()).toBe("running"); + + const recorder = recordCommits(); + await chat.answer("current"); + await expect.poll(() => chat.owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + recorder.stop(); + + expect(footer()).toBe("settled"); + expect(replies()).toEqual(["Done."]); + expect(chat.fetchCount()).toBe(0); + expect(recorder.commits.map((commit) => commit.footer).slice(0, 2)).toEqual([ + "running", + "settled", + ]); + chat.owner.dispose(); + }, + ); + + it("freshness check: an idle row carried by the current answer settles the footer in the same step", async () => { + const chat = createCheckedChat("Done."); + agentIsRunning(); + await chat.open(); + leaveAndReturn(chat.owner, "reconnect"); + + let footerAfterRow: Footer | null = null; + await chat.answer("current", () => { + idleSnapshotArrives(); + footerAfterRow = footer(); + }); + await expect.poll(() => chat.owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + + expect(footerAfterRow).toBe("running"); + expect(footer()).toBe("settled"); + chat.owner.dispose(); + }); + + it("freshness check: an agent still running keeps its footer after current", async () => { + const chat = createCheckedChat("Half"); + agentIsRunning(); + await chat.open(); + leaveAndReturn(chat.owner, "reconnect"); + + await chat.answer("current"); + await expect.poll(() => chat.owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + + expect(footer()).toBe("running"); + idleSnapshotArrives(); + expect(footer()).toBe("settled"); + chat.owner.dispose(); + }); + + it("freshness check: behind holds the turn until the fetched rows land with the completed footer", async () => { + const chat = createCheckedChat("Half"); + agentIsRunning(); + await chat.open(); + leaveAndReturn(chat.owner, "reconnect"); + idleSnapshotArrives(); + + await chat.answer("behind"); + await expect.poll(() => chat.fetchCount()).toBe(1); + expect(footer()).toBe("running"); + + const recorder = recordCommits(); + chat.owner.applyTimelineResponse(catchUpPage({ replyThroughSeq: 8, reply: "Done." })); + recorder.stop(); + + expect(footer()).toBe("settled"); + expect(replies()).toEqual(["HalfDone."]); + expect(recorder.commits.at(-2)).toMatchObject({ footer: "running", replies: ["Half"] }); + expect(recorder.commits.at(-1)).toMatchObject({ footer: "settled", replies: ["HalfDone."] }); + chat.finishFetch(); + await expect.poll(() => chat.owner.getAgentTimelineStatus(AGENT_ID)).toBe("ready"); + chat.owner.dispose(); + }); + it("catch-up failure gives the footer the true status", async () => { let failNext = false; const owner = createChat("Half", async () => { @@ -1172,6 +1330,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..021cea2bf8 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(); @@ -162,6 +173,7 @@ class TimelineWorld { ready: result.promise, release: async () => { this.releasedMemberships.push(agentIds); + result.reject(new Error("Subscription released")); }, }; }, @@ -1200,3 +1212,325 @@ 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 } }); + + check.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + world.expectNoPendingFetch(); + 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 }); + 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(); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(world.releasedMemberships).toEqual([]); + + check.succeed({ "agent-a": "current" }); + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a", "agent-b"]])); + world.sync.dispose(); +}); + +test("losing the connection during a check releases both the check and what it replaced", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + 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(); + world.sync.setConnected(false); + + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"], ["agent-a"]])); + world.sync.dispose(); +}); + +test("a replacement the daemon refuses still releases the subscription it replaced", async () => { + const world = freshnessWorld(); + world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 }); + 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); + const check = await world.nextMembership(); + check.fail("refused"); + + await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]])); + 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..e2a623c4d0 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -12,7 +12,7 @@ import { type AgentTimelineState, } from "@/stores/session-store"; import { useCreateFlowStore } from "@/stores/create-flow-store"; -import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; +import type { AgentSnapshotPayload, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { getSendingClientMessageIds } from "@/composer/submission/model"; import { getInitDeferred, @@ -339,10 +339,56 @@ 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; + agents?: 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 +407,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 +442,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 +483,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 +599,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 +617,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 +673,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 +789,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 +847,98 @@ 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; + // A subscription routes events only once the daemon accepts it, so the old one delivers until then. + const answer = await handle.ready.finally(() => { + void previous?.handle.release().catch(ports.reportError); + }); + 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 +982,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 +995,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 +1007,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 +1032,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 +1061,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 +1093,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 +1131,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..72a3249fa7 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,11 @@ 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(), + // The agent row for each chat answered "current", which gets no timeline fetch to carry it. + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + agents: z.record(z.string(), AgentSnapshotPayloadSchema).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..7b096256cc 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,97 @@ 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" }, + }, + }, + ); + + const answer = await observation.ready; + expect(answer.timelines).toEqual({ + [atHead.agentId]: "current", + [staleSeq.agentId]: "behind", + [otherEpoch.agentId]: "behind", + }); + // Only a current chat skips the fetch that would have delivered its agent row. + expect(Object.keys(answer.agents ?? {})).toEqual([atHead.agentId]); + expect(answer.agents?.[atHead.agentId]).toMatchObject({ id: atHead.agentId, title: "At head" }); + + 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 }, + }); + + const answer = await observation.ready; + expect(answer.timelines).toEqual({ [stored.agentId]: "behind" }); + expect(answer.agents).toEqual({}); + 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..13f8db7397 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1408,6 +1408,65 @@ 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 async answerTimelineFreshness( + agentIds: string[], + cursors: Record, + ): Promise<{ + timelines: Record; + agents: Record; + }> { + const snapshots = await Promise.all( + agentIds.flatMap((agentId) => { + const agent = cursors[agentId] ? this.agentManager.getAgent(agentId) : null; + return agent ? [this.buildAgentPayload(agent)] : []; + }), + ); + // Compared after the await, so a row appended while reading the snapshots answers behind. + const timelines = this.compareTimelineCursors(agentIds, cursors); + const agents: Record = {}; + for (const snapshot of snapshots) { + if (timelines[snapshot.id] === "current") agents[snapshot.id] = snapshot; + } + return { timelines, agents }; + } + + private async handleAgentTimelineSubscription( + msg: Extract, + source?: object, + ): Promise { + 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); + const freshness = msg.cursors ? await this.answerTimelineFreshness(agentIds, msg.cursors) : {}; + this.emitForSource( + { + type: "agent.timeline.set_subscription.response", + payload: { + agentIds, + requestId: msg.requestId, + subscriptionId: owner.responseId, + ...freshness, + }, + }, + source, + ); + } + private subscribeAgentTimelines(agentIds: string[]): OwnedSubscription { const owner = this.delivery.begin("timelines", undefined, (id) => { this.timelineSubscriptions.delete(id); @@ -3043,18 +3102,8 @@ export class Session { throw error; }); } - case "agent.timeline.set_subscription.request": { - const agentIds = [...new Set(msg.agentIds)].sort(); - const owner = this.subscribeAgentTimelines(agentIds); - this.emitForSource( - { - type: "agent.timeline.set_subscription.response", - payload: { agentIds, requestId: msg.requestId, subscriptionId: owner.responseId }, - }, - source, - ); - return undefined; - } + case "agent.timeline.set_subscription.request": + return this.handleAgentTimelineSubscription(msg, source); case "agent.fork_context.request": return this.handleAgentForkContextRequest(msg); default: 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,