From 07d988488e66de777443484847dc6d790d680042 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 20 Jul 2026 17:24:45 +0200 Subject: [PATCH] fix(sync): show complete workspace and agent history (#2263) Cached focused state was leaking into the sidebar before authoritative workspace hydration, while catch-up limits counted raw events instead of projected entries. Keep the focused cache fast, gate directory presentation on hydration, and preserve canonical cursors while paging projected history. --- .../src/hooks/use-sidebar-workspaces-list.ts | 11 +-- packages/app/src/runtime/host-runtime.ts | 51 ++++++------ .../src/stores/session-store-hooks/index.ts | 9 +++ .../session-store-hooks/selectors.test.ts | 70 +++++++++++++++++ .../stores/session-store-hooks/selectors.ts | 7 ++ .../server/agent/timeline-projection.test.ts | 55 +++++++++++++ .../src/server/agent/timeline-projection.ts | 78 ++++++++++++++++++- .../daemon-e2e/timeline-window.e2e.test.ts | 69 ++++++++++++++++ packages/server/src/server/session.ts | 24 ++++-- 9 files changed, 331 insertions(+), 43 deletions(-) diff --git a/packages/app/src/hooks/use-sidebar-workspaces-list.ts b/packages/app/src/hooks/use-sidebar-workspaces-list.ts index e620e2acd..92ae9d367 100644 --- a/packages/app/src/hooks/use-sidebar-workspaces-list.ts +++ b/packages/app/src/hooks/use-sidebar-workspaces-list.ts @@ -1,7 +1,6 @@ import { useCallback, useEffect, useMemo } from "react"; -import { shallow } from "zustand/shallow"; -import { useStoreWithEqualityFn } from "zustand/traditional"; import { useSessionStore, type WorkspaceDescriptor } from "@/stores/session-store"; +import { useHydratedWorkspaceServerIds } from "@/stores/session-store-hooks"; import { useHostProjects } from "@/projects/host-projects"; import { fetchAllWorkspaceDescriptors } from "@/projects/workspace-fetching"; import { getHostRuntimeStore, useHostRegistryLoaded, useHosts } from "@/runtime/host-runtime"; @@ -89,13 +88,9 @@ export function useSidebarWorkspacesList(options?: { const persistedProjectOrder = useSidebarOrderStore((state) => state.projectOrder ?? EMPTY_ORDER); - const hydratedServerIds = useStoreWithEqualityFn( - useSessionStore, - (state) => serverIds.filter((id) => state.sessions[id]?.hasHydratedWorkspaces ?? false), - shallow, - ); + const hydratedServerIds = useHydratedWorkspaceServerIds(serverIds); - const hostProjects = useHostProjects(serverIds); + const hostProjects = useHostProjects(hydratedServerIds); const sidebarModel = useMemo( () => diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 59fdf9c09..af76ba4fa 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1350,7 +1350,7 @@ export class HostRuntimeStore { private hostRegistryStatus: HostRegistryStatus = "loading"; private deps: HostRuntimeControllerDeps; private lastConnectionStatusByServer = new Map(); - private agentDirectoryBootstrapInFlight = new Map>(); + private directoryBootstrapInFlight = new Map>(); private queuedAgentDrainInFlight = new Set(); private directorySyncByServer = new Map(); private configuredOverrideBootstrapInFlight: Promise | null = null; @@ -1586,17 +1586,14 @@ export class HostRuntimeStore { controller.adoptReconciledServerId(newServerId); rekeyMap(this.lastConnectionStatusByServer, oldServerId, newServerId); - rekeyMap(this.agentDirectoryBootstrapInFlight, oldServerId, newServerId); + rekeyMap(this.directoryBootstrapInFlight, oldServerId, newServerId); this.replicaCache.reconcileServerId(oldServerId, newServerId); this.directorySyncByServer.get(oldServerId)?.dispose(); this.directorySyncByServer.delete(oldServerId); const directory = new DirectorySync(newServerId, { drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId), markAgentLoading: () => controller.markAgentDirectorySyncLoading(), - markAgentReady: () => { - this.agentDirectoryBootstrapInFlight.delete(newServerId); - controller.markAgentDirectorySyncReady(); - }, + markAgentReady: () => controller.markAgentDirectorySyncReady(), markAgentError: (error) => controller.markAgentDirectorySyncError(error), }); this.directorySyncByServer.set(newServerId, directory); @@ -1899,7 +1896,7 @@ export class HostRuntimeStore { } this.controllers.delete(serverId); this.lastConnectionStatusByServer.delete(serverId); - this.agentDirectoryBootstrapInFlight.delete(serverId); + this.directoryBootstrapInFlight.delete(serverId); this.directorySyncByServer.get(serverId)?.dispose(); this.directorySyncByServer.delete(serverId); void controller.stop(); @@ -1930,10 +1927,7 @@ export class HostRuntimeStore { drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(host.serverId, agentId), markAgentLoading: () => controller.markAgentDirectorySyncLoading(), - markAgentReady: () => { - this.agentDirectoryBootstrapInFlight.delete(host.serverId); - controller.markAgentDirectorySyncReady(); - }, + markAgentReady: () => controller.markAgentDirectorySyncReady(), markAgentError: (error) => controller.markAgentDirectorySyncError(error), }), ); @@ -1942,7 +1936,7 @@ export class HostRuntimeStore { controller.getSnapshot().connectionStatus, ); controller.subscribe(() => { - this.maybeAutoBootstrapAgentDirectory(host.serverId); + this.maybeAutoBootstrapDirectories(host.serverId); this.emit(host.serverId); }); void controller @@ -1961,11 +1955,11 @@ export class HostRuntimeStore { } } - private maybeAutoBootstrapAgentDirectory(serverId: string): void { + private maybeAutoBootstrapDirectories(serverId: string): void { const controller = this.controllers.get(serverId); if (!controller) { this.lastConnectionStatusByServer.delete(serverId); - this.agentDirectoryBootstrapInFlight.delete(serverId); + this.directoryBootstrapInFlight.delete(serverId); return; } const snapshot = controller.getSnapshot(); @@ -2000,7 +1994,7 @@ export class HostRuntimeStore { if (!didTransitionOnline && snapshot.hasEverLoadedAgentDirectory) { return; } - if (this.agentDirectoryBootstrapInFlight.has(serverId) && !directorySourceChanged) { + if (this.directoryBootstrapInFlight.has(serverId) && !directorySourceChanged) { return; } @@ -2011,25 +2005,28 @@ export class HostRuntimeStore { serverId, subscribe: { subscriptionId: `app:${serverId}` }, page: { limit: DEFAULT_AGENT_DIRECTORY_PAGE_LIMIT }, + }).catch((error) => { + console.error("[HostRuntime] agent directory bootstrap failed", { + serverId, + error: toErrorMessage(error), + }); }), - this.refreshWorkspaceDirectory({ serverId, subscribe: true }), - ]), + this.refreshWorkspaceDirectory({ serverId, subscribe: true }).catch((error) => { + console.error("[HostRuntime] workspace directory bootstrap failed", { + serverId, + error: toErrorMessage(error), + }); + }), + ]).then(() => undefined), ) - .then(() => undefined) - .catch((error) => { - console.error("[HostRuntime] agent directory bootstrap failed", { - serverId, - error: toErrorMessage(error), - }); - }) .finally(() => { - const inFlight = this.agentDirectoryBootstrapInFlight.get(serverId); + const inFlight = this.directoryBootstrapInFlight.get(serverId); if (inFlight === bootstrap) { - this.agentDirectoryBootstrapInFlight.delete(serverId); + this.directoryBootstrapInFlight.delete(serverId); } }); - this.agentDirectoryBootstrapInFlight.set(serverId, bootstrap); + this.directoryBootstrapInFlight.set(serverId, bootstrap); } drainQueuedAgentMessage(serverId: string, agentId: string): void { diff --git a/packages/app/src/stores/session-store-hooks/index.ts b/packages/app/src/stores/session-store-hooks/index.ts index 1c77c1005..f2021a490 100644 --- a/packages/app/src/stores/session-store-hooks/index.ts +++ b/packages/app/src/stores/session-store-hooks/index.ts @@ -4,6 +4,7 @@ import { useSidebarOrderStore } from "@/stores/sidebar-order-store"; import { composeWorkspaceStructure, selectHasHydratedWorkspaces, + selectHydratedWorkspaceServerIds, selectHasWorkspaces, selectProjectOrder, selectRecommendedProjectPaths, @@ -70,6 +71,14 @@ export function useHasHydratedWorkspaces(serverId: string | null): boolean { ); } +export function useHydratedWorkspaceServerIds(serverIds: string[]): string[] { + return useStoreWithEqualityFn( + useSessionStore, + (state) => selectHydratedWorkspaceServerIds(state, serverIds), + workspaceEqualityFns.deep, + ); +} + export function useWorkspaceDirectory( serverId: string | null, workspaceId: string | null, diff --git a/packages/app/src/stores/session-store-hooks/selectors.test.ts b/packages/app/src/stores/session-store-hooks/selectors.test.ts index 583b9b4fa..dde20788a 100644 --- a/packages/app/src/stores/session-store-hooks/selectors.test.ts +++ b/packages/app/src/stores/session-store-hooks/selectors.test.ts @@ -4,6 +4,7 @@ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import { composeWorkspaceStructure, selectHasWorkspaces, + selectHydratedWorkspaceServerIds, selectProjectOrder, selectRecommendedProjectPaths, selectWorkspace, @@ -101,6 +102,75 @@ afterEach(() => { useSessionStore.getState().clearSession(SERVER_ID); }); +describe("workspace replica authority", () => { + it("keeps a cached workspace addressable without publishing it as an authoritative directory", () => { + const cachedWorkspace = createWorkspace({ id: "cached-workspace" }); + initializeWorkspaces([cachedWorkspace]); + + const cachedServerIds = selectHydratedWorkspaceServerIds(useSessionStore.getState(), [ + SERVER_ID, + ]); + + expect(selectWorkspace(useSessionStore.getState(), SERVER_ID, cachedWorkspace.id)).toBe( + cachedWorkspace, + ); + expect(selectWorkspaceStructureProjects(useSessionStore.getState(), cachedServerIds)).toEqual( + [], + ); + + const authoritativeWorkspace = createWorkspace({ + id: "authoritative-workspace", + projectId: "authoritative-project", + }); + useSessionStore + .getState() + .setWorkspaces(SERVER_ID, new Map([[authoritativeWorkspace.id, authoritativeWorkspace]])); + useSessionStore.getState().setHasHydratedWorkspaces(SERVER_ID, true); + const hydratedServerIds = selectHydratedWorkspaceServerIds(useSessionStore.getState(), [ + SERVER_ID, + ]); + + expect( + selectWorkspaceStructureProjects(useSessionStore.getState(), hydratedServerIds).map( + (project) => project.workspaceKeys, + ), + ).toEqual([[`${SERVER_ID}:${authoritativeWorkspace.id}`]]); + }); + + it("publishes each host to the workspace directory independently", () => { + const loadingServerId = "loading-server"; + const hydratedServerId = "hydrated-server"; + const loadingWorkspace = createWorkspace({ id: "loading-workspace" }); + const hydratedWorkspace = createWorkspace({ + id: "hydrated-workspace", + projectId: "hydrated-project", + }); + const state = { + sessions: { + [loadingServerId]: { + hasHydratedWorkspaces: false, + workspaces: new Map([[loadingWorkspace.id, loadingWorkspace]]), + }, + [hydratedServerId]: { + hasHydratedWorkspaces: true, + workspaces: new Map([[hydratedWorkspace.id, hydratedWorkspace]]), + }, + }, + }; + + const directoryServerIds = selectHydratedWorkspaceServerIds(state, [ + loadingServerId, + hydratedServerId, + ]); + const directoryProjects = selectWorkspaceStructureProjects(state, directoryServerIds); + + expect(directoryServerIds).toEqual([hydratedServerId]); + expect(directoryProjects.map((project) => project.workspaceKeys)).toEqual([ + [`${hydratedServerId}:${hydratedWorkspace.id}`], + ]); + }); +}); + describe("selectWorkspace", () => { it("resolves a descriptor when the route id matches workspace identity but not the map key", () => { const workspace = createWorkspace({ id: "workspace-a" }); diff --git a/packages/app/src/stores/session-store-hooks/selectors.ts b/packages/app/src/stores/session-store-hooks/selectors.ts index f38154d3d..d0377cd58 100644 --- a/packages/app/src/stores/session-store-hooks/selectors.ts +++ b/packages/app/src/stores/session-store-hooks/selectors.ts @@ -131,6 +131,13 @@ export function selectHasHydratedWorkspaces( return serverId ? (state.sessions[serverId]?.hasHydratedWorkspaces ?? false) : false; } +export function selectHydratedWorkspaceServerIds( + state: SessionsSnapshot, + serverIds: readonly string[], +): string[] { + return serverIds.filter((serverId) => state.sessions[serverId]?.hasHydratedWorkspaces === true); +} + export function selectWorkspaceStructureProjects( state: SessionsSnapshot, serverIds: readonly string[], diff --git a/packages/server/src/server/agent/timeline-projection.test.ts b/packages/server/src/server/agent/timeline-projection.test.ts index 814b7a62e..8bfbcef83 100644 --- a/packages/server/src/server/agent/timeline-projection.test.ts +++ b/packages/server/src/server/agent/timeline-projection.test.ts @@ -572,6 +572,61 @@ describe("selectProjectedTimelinePage", () => { expect(page.hasNewer).toBe(true); }); + test("after limit counts projected entries while preserving their canonical sequence coverage", () => { + const rows: AgentTimelineRow[] = [ + ...Array.from({ length: 200 }, (_, index) => ({ + seq: index + 1, + timestamp: new Date(1000 + index).toISOString(), + item: { type: "assistant_message" as const, text: "x" }, + })), + ...Array.from({ length: 150 }, (_, index) => ({ + seq: index + 201, + timestamp: new Date(2000 + index).toISOString(), + item: { type: "user_message" as const, text: `message ${index + 1}` }, + })), + ]; + + const page = selectProjectedTimelinePage({ + rows, + direction: "after", + cursorSeq: 0, + limit: 100, + }); + + expect({ + entryCount: page.entries.length, + firstEntry: page.entries[0], + lastEntry: page.entries.at(-1), + startSeq: page.startSeq, + endSeq: page.endSeq, + hasNewer: page.hasNewer, + }).toEqual({ + entryCount: 100, + firstEntry: { + item: { + type: "assistant_message", + text: "x".repeat(200), + }, + timestamp: new Date(1199).toISOString(), + seqStart: 1, + seqEnd: 200, + sourceSeqRanges: [{ startSeq: 1, endSeq: 200 }], + collapsed: ["assistant_merge"], + }, + lastEntry: { + item: { type: "user_message", text: "message 99" }, + timestamp: new Date(2098).toISOString(), + seqStart: 299, + seqEnd: 299, + sourceSeqRanges: [{ startSeq: 299, endSeq: 299 }], + collapsed: [], + }, + startSeq: 1, + endSeq: 299, + hasNewer: true, + }); + }); + test("before page includes a wide tool whose earlier source range is before the cursor", () => { const rows: AgentTimelineRow[] = [ toolRow(1, "running"), diff --git a/packages/server/src/server/agent/timeline-projection.ts b/packages/server/src/server/agent/timeline-projection.ts index bf9344677..e6bd1d161 100644 --- a/packages/server/src/server/agent/timeline-projection.ts +++ b/packages/server/src/server/agent/timeline-projection.ts @@ -376,6 +376,69 @@ function selectEntriesOverlappingSeqRange(input: { ); } +function firstSourceSeqInRange( + entry: TimelineProjectionEntry, + startSeq: number, + endSeq: number, +): number | null { + for (const range of entry.sourceSeqRanges) { + const firstSeq = Math.max(range.startSeq, startSeq); + if (firstSeq <= Math.min(range.endSeq, endSeq)) return firstSeq; + } + return null; +} + +interface ProjectedEntryCandidate { + entry: TimelineProjectionEntry; + index: number; + firstSourceSeq: number; +} + +function selectProjectedEntriesAfter(input: { + entries: readonly TimelineProjectionEntry[]; + rows: readonly AgentTimelineRow[]; + startSeq: number; + maxSeq: number; + limit: number; +}): { entries: TimelineProjectionEntry[]; endSeq: number | null } { + const eligible = input.entries + .map((entry, index) => ({ + entry, + index, + firstSourceSeq: firstSourceSeqInRange(entry, input.startSeq, input.maxSeq), + })) + .filter((candidate): candidate is ProjectedEntryCandidate => candidate.firstSourceSeq !== null) + .sort((left, right) => left.firstSourceSeq - right.firstSourceSeq || left.index - right.index); + const selected = input.limit === 0 ? eligible : eligible.slice(0, input.limit); + const selectedEntries = selected + .sort((left, right) => left.index - right.index) + .map((candidate) => candidate.entry); + if (selectedEntries.length === 0) return { entries: [], endSeq: null }; + + const selectedRanges = selectedEntries + .flatMap((entry) => entry.sourceSeqRanges) + .sort((left, right) => left.startSeq - right.startSeq || left.endSeq - right.endSeq); + // Wide projected entries can include future, discontiguous source ranges. The + // page cursor advances only through source rows covered without a gap. + let endSeq = input.startSeq - 1; + let rangeIndex = 0; + for (const row of input.rows) { + if (row.seq < input.startSeq) continue; + if (row.seq > input.maxSeq || row.seq !== endSeq + 1) break; + while (selectedRanges[rangeIndex] && selectedRanges[rangeIndex].endSeq < row.seq) { + rangeIndex += 1; + } + const range = selectedRanges[rangeIndex]; + if (!range || row.seq < range.startSeq || row.seq > range.endSeq) break; + endSeq = row.seq; + } + + return { + entries: selectedEntries, + endSeq: endSeq >= input.startSeq ? endSeq : null, + }; +} + export function selectProjectedTimelinePage(input: { rows: readonly AgentTimelineRow[]; bounds?: { minSeq: number; maxSeq: number }; @@ -446,7 +509,20 @@ export function selectProjectedTimelinePage(input: { if (input.direction === "after") { const cursorSeq = input.cursorSeq ?? bounds.minSeq - 1; startSeq = Math.max(bounds.minSeq, cursorSeq + 1); - endSeq = limit === 0 ? bounds.maxSeq : Math.min(bounds.maxSeq, cursorSeq + limit); + const selected = selectProjectedEntriesAfter({ + entries: projectedAll, + rows: input.rows, + startSeq, + maxSeq: bounds.maxSeq, + limit, + }); + return { + entries: selected.entries, + startSeq: selected.endSeq === null ? null : startSeq, + endSeq: selected.endSeq, + hasOlder: startSeq > bounds.minSeq, + hasNewer: selected.endSeq !== null && selected.endSeq < bounds.maxSeq, + }; } else { const cursorSeq = input.cursorSeq ?? bounds.maxSeq + 1; endSeq = Math.min(bounds.maxSeq, cursorSeq - 1); diff --git a/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts b/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts index 0c3e0da0d..1e809314b 100644 --- a/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts @@ -362,6 +362,75 @@ describe("daemon E2E - timeline window", () => { } }); + test("projected after limit returns 100 projected entries with their canonical coverage", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Projected Page Limit Test", + modeId: "full-access", + }); + for (let seq = 1; seq <= 200; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: "x", + }); + } + for (let seq = 201; seq <= 350; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: `message ${seq - 200}`, + }); + } + const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, { + direction: "tail", + limit: 1, + }).epoch; + + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "after", + cursor: { epoch, seq: 0 }, + limit: 100, + projection: "projected", + }); + + expect({ + entryCount: timeline.entries.length, + firstEntry: timeline.entries[0], + lastEntry: timeline.entries.at(-1), + startCursor: timeline.startCursor, + endCursor: timeline.endCursor, + hasNewer: timeline.hasNewer, + }).toEqual({ + entryCount: 100, + firstEntry: { + provider: "codex", + item: { type: "assistant_message", text: "x".repeat(200) }, + timestamp: expect.any(String), + seqStart: 1, + seqEnd: 200, + sourceSeqRanges: [{ startSeq: 1, endSeq: 200 }], + collapsed: ["assistant_merge"], + }, + lastEntry: { + provider: "codex", + item: { type: "user_message", text: "message 99" }, + timestamp: expect.any(String), + seqStart: 299, + seqEnd: 299, + sourceSeqRanges: [{ startSeq: 299, endSeq: 299 }], + collapsed: [], + }, + startCursor: { epoch, seq: 1 }, + endCursor: { epoch, seq: 299 }, + hasNewer: true, + }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + test("projected empty after fetch preserves older history availability", async () => { const cwd = tmpCwd(); try { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index b974559cb..1b9e57b5a 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -5686,22 +5686,31 @@ export class Session { private shouldUseFullTimelineForProjectedPage(input: { timeline: AgentTimelineFetchResult; + pageLimit: number; }): boolean { const { timeline } = input; - if (timeline.reset || timeline.rows.length === 0 || !timeline.hasOlder) { - return false; - } + if (timeline.rows.length === 0) return false; + + if (timeline.rows.some((row) => row.item.type === "tool_call")) return true; const firstRow = timeline.rows[0]; if ( - firstRow?.item.type === "assistant_message" || - firstRow?.item.type === "reasoning" || - firstRow?.item.type === "tool_call" + timeline.hasOlder && + (firstRow?.item.type === "assistant_message" || firstRow?.item.type === "reasoning") ) { return true; } - return timeline.rows.some((row) => row.item.type === "tool_call"); + const lastRow = timeline.rows.at(-1); + if ( + timeline.hasNewer && + (lastRow?.item.type === "assistant_message" || lastRow?.item.type === "reasoning") + ) { + return true; + } + + if (!timeline.hasNewer || input.pageLimit === 0) return false; + return projectTimelineRows({ rows: timeline.rows, mode: "projected" }).length < input.pageLimit; } private selectCanonicalTimelineProjection(input: { @@ -5727,6 +5736,7 @@ export class Session { }): AgentTimelineProjectionSelection { const timeline = this.shouldUseFullTimelineForProjectedPage({ timeline: input.controlTimeline, + pageLimit: input.pageLimit, }) ? this.agentManager.fetchTimeline(input.agentId, { direction: "tail", limit: 0 }) : input.controlTimeline;