diff --git a/packages/app/src/components/agent-stream-view.test.tsx b/packages/app/src/components/agent-stream-view.test.tsx index 8f2596be0..82fceef6e 100644 --- a/packages/app/src/components/agent-stream-view.test.tsx +++ b/packages/app/src/components/agent-stream-view.test.tsx @@ -19,8 +19,12 @@ const mockSessionState = vi.hoisted(() => ({ client: null, agentStreamHead: new Map(), workspaces: new Map(), + agentTimelineCursor: new Map(), + agentTimelineHasOlder: new Map(), + agentTimelineOlderFetchInFlight: new Map(), }, }, + setAgentTimelineOlderFetchInFlight: () => {}, })); vi.mock("react-native-unistyles", () => ({ diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 0ef3f7c8d..53f579894 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -56,6 +56,8 @@ import type { import type { AgentScreenAgent } from "@/hooks/use-agent-screen-state-machine"; import { useSessionStore } from "@/stores/session-store"; import { useFileExplorerActions } from "@/hooks/use-file-explorer-actions"; +import { useLoadOlderAgentHistory } from "@/hooks/use-load-older-agent-history"; +import type { ToastApi } from "@/components/toast-host"; import type { DaemonClient } from "@server/client/daemon-client"; import { ToolCallDetailsContent } from "./tool-call-details"; import { QuestionFormCard } from "./question-form-card"; @@ -145,6 +147,7 @@ export interface AgentStreamViewProps { pendingPermissions: Map; routeBottomAnchorRequest?: BottomAnchorRouteRequest | null; isAuthoritativeHistoryReady?: boolean; + toast?: ToastApi | null; onOpenWorkspaceFile?: (input: { filePath: string }) => void; } @@ -158,6 +161,7 @@ const AgentStreamViewComponent = forwardRef(null); + const historyStartReadyRef = useRef(false); const historyRows = useMemo(() => { if (segments.historyVirtualized.length === 0) { @@ -155,6 +162,13 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat scrollOffsetYRef.current = 0; clearNativeViewportSettling(); setIsNativeViewportSettling(false); + historyStartReadyRef.current = false; + const frame = requestAnimationFrame(() => { + historyStartReadyRef.current = true; + }); + return () => { + cancelAnimationFrame(frame); + }; }, [agentId, clearNativeViewportSettling]); useEffect(() => { @@ -227,6 +241,18 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat }); onNearBottomChange(nearBottom); + const distanceFromOldestEdge = + streamViewportMetricsRef.current.contentHeight - + streamViewportMetricsRef.current.viewportHeight - + contentOffset.y; + if ( + historyStartReadyRef.current && + hasOlderHistory && + distanceFromOldestEdge <= HISTORY_START_THRESHOLD_PX + ) { + onNearHistoryStart(); + } + if (programmaticScrollEventBudgetRef.current > 0 && contentOffset.y <= 8) { programmaticScrollEventBudgetRef.current -= 1; } else { @@ -307,6 +333,17 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat ); }, [boundary, listEmptyComponent, renderLiveAuxiliary, renderLiveHeadRow, segments.liveHead]); + const historyFooterContent = useMemo(() => { + if (!isLoadingOlderHistory) { + return null; + } + return ( + + + + ); + }, [isLoadingOlderHistory]); + return ( { routeBottomAnchorRequest: null, isAuthoritativeHistoryReady: true, onNearBottomChange: vi.fn(), + onNearHistoryStart: vi.fn(), + isLoadingOlderHistory: false, + hasOlderHistory: false, scrollEnabled: true, listStyle: null, baseListContentContainerStyle: null, @@ -145,4 +148,63 @@ describe("createWebStreamStrategy", () => { expect(rowRenderCount.mock.calls.length).toBeGreaterThan(0); expect(rowRenderCount.mock.calls.length).toBeLessThanOrEqual(historyVirtualized.length); }); + + it("fires near-history-start when the user scrolls near the top", async () => { + const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); + const viewportRef = React.createRef(); + const onNearHistoryStart = vi.fn(); + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + + act(() => { + root?.render( + <> + {strategy.render({ + agentId: "agent", + segments: { + historyVirtualized: [], + historyMounted: [userMessage(1), userMessage(2)], + liveHead: [], + }, + boundary: { + hasVirtualizedHistory: false, + hasMountedHistory: true, + hasLiveHead: false, + historyToHeadGap: 0, + }, + renderers: createRenderers(vi.fn()), + listEmptyComponent: null, + viewportRef, + routeBottomAnchorRequest: null, + isAuthoritativeHistoryReady: true, + onNearBottomChange: vi.fn(), + onNearHistoryStart, + isLoadingOlderHistory: false, + hasOlderHistory: true, + scrollEnabled: true, + listStyle: null, + baseListContentContainerStyle: null, + forwardListContentContainerStyle: null, + })} + , + ); + }); + + await act(async () => { + await new Promise((resolve) => requestAnimationFrame(resolve)); + }); + + const scrollContainer = container.querySelector('[data-testid="agent-chat-scroll"]'); + expect(scrollContainer).toBeInstanceOf(HTMLElement); + Object.defineProperty(scrollContainer, "clientHeight", { configurable: true, value: 400 }); + Object.defineProperty(scrollContainer, "scrollHeight", { configurable: true, value: 1200 }); + Object.defineProperty(scrollContainer, "scrollTop", { configurable: true, value: 64 }); + + act(() => { + scrollContainer?.dispatchEvent(new Event("scroll")); + }); + + expect(onNearHistoryStart).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/app/src/components/stream-strategy-web.tsx b/packages/app/src/components/stream-strategy-web.tsx index 2b5fd0d91..660e4ff3d 100644 --- a/packages/app/src/components/stream-strategy-web.tsx +++ b/packages/app/src/components/stream-strategy-web.tsx @@ -8,6 +8,7 @@ import React, { useRef, useState, } from "react"; +import { ActivityIndicator } from "react-native"; import { measureElement as measureVirtualElement, useVirtualizer } from "@tanstack/react-virtual"; import { estimateStreamItemHeight } from "./agent-stream-web-virtualization"; import type { StreamRenderInput, StreamStrategy, StreamViewportHandle } from "./stream-strategy"; @@ -23,8 +24,18 @@ const WEB_BOTTOM_SETTLE_TIMEOUT_MS = 200; const USER_SCROLL_DELTA_EPSILON = 1; const AUTO_SCROLL_BOTTOM_THRESHOLD_PX = 64; const AUTO_SCROLL_RESUME_THRESHOLD_PX = 1; +const HISTORY_START_THRESHOLD_PX = 96; import { useWebElementScrollbar } from "./use-web-scrollbar"; +const historyStartSlotStyle: CSSProperties = { + display: "flex", + alignItems: "center", + justifyContent: "center", + minHeight: 32, + paddingTop: 4, + paddingBottom: 8, +}; + function isScrollContainerNearBottom( scrollContainer: Pick, thresholdPx = AUTO_SCROLL_BOTTOM_THRESHOLD_PX, @@ -91,6 +102,9 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool routeBottomAnchorRequest, isAuthoritativeHistoryReady, onNearBottomChange, + onNearHistoryStart, + isLoadingOlderHistory, + hasOlderHistory, scrollEnabled, isMobileBreakpoint, } = props; @@ -115,6 +129,7 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool const pendingAutoScrollFrameRef = useRef(null); const pendingAutoScrollTimeoutRef = useRef(null); const pendingVirtualRowMeasureFramesRef = useRef(new Map()); + const historyStartReadyRef = useRef(false); const showDesktopWebScrollbar = !isMobileBreakpoint; const scrollbarOverlay = useWebElementScrollbar(scrollContainerRef, { enabled: showDesktopWebScrollbar, @@ -280,7 +295,24 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool lastKnownScrollTopRef.current = currentScrollTop; updateScrollMetrics(); - }, [cancelPendingStickToBottom, updateScrollMetrics]); + if ( + historyStartReadyRef.current && + hasOlderHistory && + currentScrollTop <= HISTORY_START_THRESHOLD_PX + ) { + onNearHistoryStart(); + } + }, [cancelPendingStickToBottom, hasOlderHistory, onNearHistoryStart, updateScrollMetrics]); + + useEffect(() => { + const frame = window.requestAnimationFrame(() => { + historyStartReadyRef.current = true; + }); + return () => { + window.cancelAnimationFrame(frame); + historyStartReadyRef.current = false; + }; + }, [props.agentId]); useLayoutEffect(() => { if (!isActivationReady) { @@ -501,6 +533,16 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool const liveAuxiliary = useMemo(() => { return renderLiveAuxiliary(); }, [renderLiveAuxiliary]); + const historyStartSlot = useMemo(() => { + if (!isLoadingOlderHistory) { + return null; + } + return ( +
+ +
+ ); + }, [isLoadingOlderHistory]); const shouldRenderEmpty = !boundary.hasMountedHistory && !boundary.hasVirtualizedHistory && @@ -516,6 +558,7 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool style={scrollContainerStyle} >
+ {historyStartSlot} {shouldUseVirtualizer ? (
{virtualRows.map((virtualRow) => { diff --git a/packages/app/src/components/stream-strategy.ts b/packages/app/src/components/stream-strategy.ts index 6d7cd681b..67fb78da7 100644 --- a/packages/app/src/components/stream-strategy.ts +++ b/packages/app/src/components/stream-strategy.ts @@ -60,6 +60,9 @@ export interface StreamRenderInput { routeBottomAnchorRequest: BottomAnchorRouteRequest | null; isAuthoritativeHistoryReady: boolean; onNearBottomChange: (value: boolean) => void; + onNearHistoryStart: () => void; + isLoadingOlderHistory: boolean; + hasOlderHistory: boolean; scrollEnabled: boolean; listStyle: StyleProp; baseListContentContainerStyle: StyleProp; diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index ecaf39ccb..fa14527c1 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -12,7 +12,8 @@ import { processTimelineResponse, type ProcessTimelineResponseOutput, type TimelineReducerSideEffect, -} from "@/contexts/session-stream-reducers"; +} from "@/timeline/session-stream-reducers"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; import type { AgentAttachment, AgentStreamEventPayload, @@ -451,6 +452,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const setAgentStreamState = useSessionStore((state) => state.setAgentStreamState); const clearAgentStreamHead = useSessionStore((state) => state.clearAgentStreamHead); const setAgentTimelineCursor = useSessionStore((state) => state.setAgentTimelineCursor); + const setAgentTimelineHasOlder = useSessionStore((state) => state.setAgentTimelineHasOlder); const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents); const bumpHistorySyncGeneration = useSessionStore((state) => state.bumpHistorySyncGeneration); const markAgentHistorySynchronized = useSessionStore( @@ -748,17 +750,15 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider if (isNative) { const session = useSessionStore.getState().sessions[serverId]; const agentId = session?.focusedAgentId; - const cursor = agentId ? session?.agentTimelineCursor.get(agentId) : undefined; - if (agentId && cursor) { + if (agentId) { void client .fetchAgentTimeline(agentId, { - direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, - limit: 0, + direction: "tail", + limit: TIMELINE_FETCH_PAGE_SIZE, projection: "canonical", }) .catch((error) => { - console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error); + console.warn("[Session] failed to fetch tail timeline on resume", agentId, error); }); } } @@ -1024,7 +1024,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider .fetchAgentTimeline(agentId, { direction: "after", cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, - limit: 0, + limit: TIMELINE_FETCH_PAGE_SIZE, projection: "canonical", }) .catch((error) => { @@ -1055,6 +1055,15 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const currentTail = session?.agentStreamTail.get(agentId) ?? []; const currentHead = session?.agentStreamHead.get(agentId) ?? []; + setAgentTimelineHasOlder(serverId, (prev) => { + if (prev.get(agentId) === payload.hasOlder) { + return prev; + } + const next = new Map(prev); + next.set(agentId, payload.hasOlder); + return next; + }); + if (payload.agent) { const normalized = normalizeAgentSnapshot(payload.agent, serverId); applyAuthoritativeAgentSnapshot({ @@ -1127,6 +1136,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setAgentStreamHead, setAgentStreamTail, setAgentTimelineCursor, + setAgentTimelineHasOlder, setInitializingAgents, ], ); diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts deleted file mode 100644 index b25cb485d..000000000 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts +++ /dev/null @@ -1,47 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { deriveInitialTimelineRequest } from "./session-timeline-bootstrap-policy"; - -describe("deriveInitialTimelineRequest", () => { - it("requests tail history when authoritative history has not been applied", () => { - expect( - deriveInitialTimelineRequest({ - cursor: { epoch: "epoch-1", seq: 42 }, - hasAuthoritativeHistory: false, - initialTimelineLimit: 100, - }), - ).toEqual({ - direction: "tail", - limit: 100, - projection: "canonical", - }); - }); - - it("requests tail history when no cursor is available", () => { - expect( - deriveInitialTimelineRequest({ - cursor: null, - hasAuthoritativeHistory: true, - initialTimelineLimit: 100, - }), - ).toEqual({ - direction: "tail", - limit: 100, - projection: "canonical", - }); - }); - - it("requests only entries after the known cursor when authoritative history exists", () => { - expect( - deriveInitialTimelineRequest({ - cursor: { epoch: "epoch-1", seq: 42 }, - hasAuthoritativeHistory: true, - initialTimelineLimit: 100, - }), - ).toEqual({ - direction: "after", - cursor: { epoch: "epoch-1", seq: 42 }, - limit: 0, - projection: "canonical", - }); - }); -}); diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts deleted file mode 100644 index ac26e9a37..000000000 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts +++ /dev/null @@ -1,34 +0,0 @@ -type InitialTimelineCursor = { - epoch: string; - seq: number; -} | null; - -export function deriveInitialTimelineRequest({ - cursor, - hasAuthoritativeHistory, - initialTimelineLimit, -}: { - cursor: InitialTimelineCursor; - hasAuthoritativeHistory: boolean; - initialTimelineLimit: number; -}): { - direction: "tail" | "after"; - cursor?: { epoch: string; seq: number }; - limit: number; - projection: "canonical"; -} { - if (!hasAuthoritativeHistory || !cursor) { - return { - direction: "tail", - limit: initialTimelineLimit, - projection: "canonical", - }; - } - - return { - direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.seq }, - limit: 0, - projection: "canonical", - }; -} diff --git a/packages/app/src/hooks/use-agent-initialization.test.ts b/packages/app/src/hooks/use-agent-initialization.test.ts index dd32a15d6..9051ea9d2 100644 --- a/packages/app/src/hooks/use-agent-initialization.test.ts +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -1,71 +1,71 @@ -import { describe, expect, it } from "vitest"; -import { __private__ } from "./use-agent-initialization"; +// @vitest-environment jsdom -describe("useAgentInitialization timeline request policy", () => { - it("uses canonical tail bootstrap when history has not synced yet", () => { - expect( - __private__.deriveInitialTimelineRequest({ - cursor: { - epoch: "epoch-1", - seq: 42, - }, - hasAuthoritativeHistory: false, - initialTimelineLimit: 200, - }), - ).toEqual({ +import { act, renderHook } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { useSessionStore } from "@/stores/session-store"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { getInitDeferred, getInitKey, resolveInitDeferred } from "@/utils/agent-initialization"; +import { useAgentInitialization } from "./use-agent-initialization"; + +const serverId = "server-1"; +const agentId = "agent-1"; + +function makeClient() { + return { + fetchAgentTimeline: vi.fn().mockResolvedValue(undefined), + refreshAgent: vi.fn().mockResolvedValue(undefined), + }; +} + +afterEach(() => { + resolveInitDeferred(getInitKey(serverId, agentId)); + useSessionStore.setState({ sessions: {}, agentLastActivity: new Map() }); + vi.restoreAllMocks(); +}); + +describe("useAgentInitialization", () => { + it("always requests a canonical tail bootstrap even with an authoritative cursor", () => { + const client = makeClient(); + useSessionStore.getState().initializeSession(serverId, client as never); + useSessionStore + .getState() + .setAgentTimelineCursor( + serverId, + new Map([[agentId, { epoch: "epoch-1", startSeq: 1, endSeq: 42 }]]), + ); + useSessionStore.getState().setAgentAuthoritativeHistoryApplied(serverId, agentId, true); + + const { result } = renderHook(() => + useAgentInitialization({ serverId, client: client as never }), + ); + + act(() => { + void result.current.ensureAgentIsInitialized(agentId); + }); + + expect(client.fetchAgentTimeline).toHaveBeenCalledWith(agentId, { direction: "tail", - limit: 200, + limit: TIMELINE_FETCH_PAGE_SIZE, projection: "canonical", }); + expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("tail"); }); - it("uses canonical tail bootstrap when cursor is missing", () => { - expect( - __private__.deriveInitialTimelineRequest({ - cursor: null, - hasAuthoritativeHistory: true, - initialTimelineLimit: 200, - }), - ).toEqual({ + it("refresh fetches a bounded canonical tail after refreshing the agent", async () => { + const client = makeClient(); + const { result } = renderHook(() => + useAgentInitialization({ serverId, client: client as never }), + ); + + await act(async () => { + await result.current.refreshAgent(agentId); + }); + + expect(client.refreshAgent).toHaveBeenCalledWith(agentId); + expect(client.fetchAgentTimeline).toHaveBeenCalledWith(agentId, { direction: "tail", - limit: 200, + limit: TIMELINE_FETCH_PAGE_SIZE, projection: "canonical", }); }); - - it("uses canonical catch-up after the current cursor once history is synced", () => { - expect( - __private__.deriveInitialTimelineRequest({ - cursor: { - epoch: "epoch-1", - seq: 42, - }, - hasAuthoritativeHistory: true, - initialTimelineLimit: 200, - }), - ).toEqual({ - direction: "after", - cursor: { epoch: "epoch-1", seq: 42 }, - limit: 0, - projection: "canonical", - }); - }); - - it("supports unbounded tail bootstrap policy", () => { - expect( - __private__.deriveInitialTimelineRequest({ - cursor: null, - hasAuthoritativeHistory: false, - initialTimelineLimit: 0, - }), - ).toEqual({ - direction: "tail", - limit: 0, - projection: "canonical", - }); - }); - - it("does not expose an RPC-success init fallback", () => { - expect("shouldResolveInitFromRpcSuccess" in __private__).toBe(false); - }); }); diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index dd1f9aafc..b53e95e23 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -8,21 +8,9 @@ import { getInitKey, rejectInitDeferred, } from "@/utils/agent-initialization"; -import { deriveInitialTimelineRequest } from "@/contexts/session-timeline-bootstrap-policy"; -import { isWeb } from "@/constants/platform"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; const INIT_TIMEOUT_MS = 5 * 60_000; -const NATIVE_INITIAL_TIMELINE_LIMIT = 200; -const UNBOUNDED_TIMELINE_LIMIT = 0; - -function resolveInitialTimelineLimit(): number { - return isWeb ? UNBOUNDED_TIMELINE_LIMIT : NATIVE_INITIAL_TIMELINE_LIMIT; -} - -export const __private__ = { - deriveInitialTimelineRequest, - resolveInitialTimelineLimit, -}; export function useAgentInitialization({ serverId, @@ -54,19 +42,13 @@ export function useAgentInitialization({ return existing.promise; } - const session = useSessionStore.getState().sessions[serverId]; - const cursor = session?.agentTimelineCursor.get(agentId); - const initialTimelineLimit = resolveInitialTimelineLimit(); - const hasAuthoritativeHistory = - session?.agentAuthoritativeHistoryApplied.get(agentId) === true; - const timelineRequest = deriveInitialTimelineRequest({ - cursor: cursor ? { epoch: cursor.epoch, seq: cursor.endSeq } : null, - hasAuthoritativeHistory, - initialTimelineLimit, - }); - const initRequestDirection = timelineRequest.direction === "after" ? "after" : "tail"; + const timelineRequest = { + direction: "tail" as const, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical" as const, + }; - const deferred = createInitDeferred(key, initRequestDirection); + const deferred = createInitDeferred(key, "tail"); const timeoutId = setTimeout(() => { setAgentInitializing(agentId, false); rejectInitDeferred( @@ -103,10 +85,9 @@ export function useAgentInitialization({ try { await client.refreshAgent(agentId); - const initialTimelineLimit = resolveInitialTimelineLimit(); await client.fetchAgentTimeline(agentId, { direction: "tail", - limit: initialTimelineLimit, + limit: TIMELINE_FETCH_PAGE_SIZE, projection: "canonical", }); } catch (error) { diff --git a/packages/app/src/hooks/use-load-older-agent-history.test.ts b/packages/app/src/hooks/use-load-older-agent-history.test.ts new file mode 100644 index 000000000..140166e4b --- /dev/null +++ b/packages/app/src/hooks/use-load-older-agent-history.test.ts @@ -0,0 +1,143 @@ +// @vitest-environment jsdom + +import { act, renderHook, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { useSessionStore } from "@/stores/session-store"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { useLoadOlderAgentHistory } from "./use-load-older-agent-history"; + +const serverId = "server-1"; +const agentId = "agent-1"; + +function makeClient(fetchAgentTimeline = vi.fn().mockResolvedValue(undefined)) { + return { fetchAgentTimeline }; +} + +function initialize(input?: { + cursor?: { epoch: string; startSeq: number; endSeq: number }; + hasOlder?: boolean; + inFlight?: boolean; + fetchAgentTimeline?: ReturnType; +}) { + const client = makeClient(input?.fetchAgentTimeline); + useSessionStore.getState().initializeSession(serverId, client as never); + if (input?.cursor) { + useSessionStore.getState().setAgentTimelineCursor(serverId, new Map([[agentId, input.cursor]])); + } + if (input?.hasOlder !== undefined) { + useSessionStore + .getState() + .setAgentTimelineHasOlder(serverId, new Map([[agentId, input.hasOlder]])); + } + if (input?.inFlight !== undefined) { + useSessionStore + .getState() + .setAgentTimelineOlderFetchInFlight(serverId, new Map([[agentId, input.inFlight]])); + } + return client; +} + +afterEach(() => { + useSessionStore.setState({ sessions: {}, agentLastActivity: new Map() }); + vi.restoreAllMocks(); +}); + +describe("useLoadOlderAgentHistory", () => { + it("no-ops without a cursor", () => { + const client = initialize({ hasOlder: true }); + const { result } = renderHook(() => useLoadOlderAgentHistory({ serverId, agentId })); + + act(() => { + result.current.loadOlder(); + }); + + expect(client.fetchAgentTimeline).not.toHaveBeenCalled(); + }); + + it("no-ops when the daemon says no older history exists", () => { + const client = initialize({ + cursor: { epoch: "epoch-1", startSeq: 10, endSeq: 20 }, + hasOlder: false, + }); + const { result } = renderHook(() => useLoadOlderAgentHistory({ serverId, agentId })); + + act(() => { + result.current.loadOlder(); + }); + + expect(client.fetchAgentTimeline).not.toHaveBeenCalled(); + }); + + it("dedupes concurrent older-page requests", () => { + const fetchAgentTimeline = vi.fn( + () => + new Promise((resolve) => { + setTimeout(resolve, 20); + }), + ); + const client = initialize({ + cursor: { epoch: "epoch-1", startSeq: 10, endSeq: 20 }, + hasOlder: true, + fetchAgentTimeline, + }); + const { result } = renderHook(() => useLoadOlderAgentHistory({ serverId, agentId })); + + act(() => { + result.current.loadOlder(); + result.current.loadOlder(); + }); + + expect(client.fetchAgentTimeline).toHaveBeenCalledTimes(1); + }); + + it("requests the page before the current start cursor and clears in-flight on success", async () => { + const client = initialize({ + cursor: { epoch: "epoch-1", startSeq: 10, endSeq: 20 }, + hasOlder: true, + }); + const { result } = renderHook(() => useLoadOlderAgentHistory({ serverId, agentId })); + + act(() => { + result.current.loadOlder(); + }); + + expect(client.fetchAgentTimeline).toHaveBeenCalledWith(agentId, { + direction: "before", + cursor: { epoch: "epoch-1", seq: 10 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + await waitFor(() => { + expect( + useSessionStore.getState().sessions[serverId]?.agentTimelineOlderFetchInFlight.get(agentId), + ).toBe(false); + }); + }); + + it("shows a panel toast and clears in-flight on failure", async () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + const error = new Error("network"); + const client = initialize({ + cursor: { epoch: "epoch-1", startSeq: 10, endSeq: 20 }, + hasOlder: true, + fetchAgentTimeline: vi.fn().mockRejectedValue(error), + }); + const toast = { show: vi.fn(), copied: vi.fn(), error: vi.fn() }; + const { result } = renderHook(() => useLoadOlderAgentHistory({ serverId, agentId, toast })); + + act(() => { + result.current.loadOlder(); + }); + + expect(client.fetchAgentTimeline).toHaveBeenCalledTimes(1); + await waitFor(() => { + expect(toast.show).toHaveBeenCalledWith("Couldn't load older history", { + durationMs: 2200, + testID: "agent-load-older-history-toast", + }); + }); + expect( + useSessionStore.getState().sessions[serverId]?.agentTimelineOlderFetchInFlight.get(agentId), + ).toBe(false); + }); +}); diff --git a/packages/app/src/hooks/use-load-older-agent-history.ts b/packages/app/src/hooks/use-load-older-agent-history.ts new file mode 100644 index 000000000..ba8a3691f --- /dev/null +++ b/packages/app/src/hooks/use-load-older-agent-history.ts @@ -0,0 +1,76 @@ +import { useCallback } from "react"; +import type { ToastApi } from "@/components/toast-host"; +import { useSessionStore } from "@/stores/session-store"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; + +export function useLoadOlderAgentHistory({ + serverId, + agentId, + toast, +}: { + serverId: string; + agentId: string; + toast?: ToastApi | null; +}) { + const client = useSessionStore((state) => state.sessions[serverId]?.client ?? null); + const cursor = useSessionStore((state) => + state.sessions[serverId]?.agentTimelineCursor.get(agentId), + ); + const hasOlder = + useSessionStore((state) => state.sessions[serverId]?.agentTimelineHasOlder.get(agentId)) === + true; + const isLoadingOlder = + useSessionStore((state) => + state.sessions[serverId]?.agentTimelineOlderFetchInFlight.get(agentId), + ) === true; + const setOlderFetchInFlight = useSessionStore( + (state) => state.setAgentTimelineOlderFetchInFlight, + ); + + const setInFlight = useCallback( + (value: boolean) => { + setOlderFetchInFlight(serverId, (prev) => { + if (prev.get(agentId) === value) { + return prev; + } + const next = new Map(prev); + next.set(agentId, value); + return next; + }); + }, + [agentId, serverId, setOlderFetchInFlight], + ); + + const loadOlder = useCallback(() => { + const latestSession = useSessionStore.getState().sessions[serverId]; + const latestIsLoading = latestSession?.agentTimelineOlderFetchInFlight.get(agentId) === true; + if (!client || !cursor || !hasOlder || isLoadingOlder || latestIsLoading) { + return; + } + + setInFlight(true); + void client + .fetchAgentTimeline(agentId, { + direction: "before", + cursor: { epoch: cursor.epoch, seq: cursor.startSeq }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }) + .catch((error) => { + console.warn("[Timeline] failed to load older agent history", agentId, error); + toast?.show("Couldn't load older history", { + durationMs: 2200, + testID: "agent-load-older-history-toast", + }); + }) + .finally(() => { + setInFlight(false); + }); + }, [agentId, client, cursor, hasOlder, isLoadingOlder, serverId, setInFlight, toast]); + + return { + isLoadingOlder, + hasOlder, + loadOlder, + }; +} diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index a84884515..9ab7a2925 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -961,6 +961,7 @@ function ChatAgentContent({ canFinalizePendingCreate={canFinalizePendingCreate} routeBottomAnchorRequest={routeBottomAnchorRequest} hasAppliedAuthoritativeHistory={hasAppliedAuthoritativeHistory} + toast={panelToast.api} onOpenWorkspaceFile={onOpenWorkspaceFile} /> @@ -1018,6 +1019,7 @@ function AgentStreamSection({ canFinalizePendingCreate, routeBottomAnchorRequest, hasAppliedAuthoritativeHistory, + toast, onOpenWorkspaceFile, }: { streamViewRef: React.RefObject; @@ -1029,6 +1031,7 @@ function AgentStreamSection({ canFinalizePendingCreate: boolean; routeBottomAnchorRequest: RouteBottomAnchorRequest; hasAppliedAuthoritativeHistory: boolean; + toast: ReturnType["api"]; onOpenWorkspaceFile?: (input: { filePath: string }) => void; }) { const streamItemsRaw = useSessionStore((state) => @@ -1163,6 +1166,7 @@ function AgentStreamSection({ pendingPermissions={pendingPermissions} routeBottomAnchorRequest={routeBottomAnchorRequest} isAuthoritativeHistoryReady={hasAppliedAuthoritativeHistory} + toast={toast} onOpenWorkspaceFile={onOpenWorkspaceFile} /> ); diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 4d88cb1a3..ef4329fe7 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -268,6 +268,8 @@ export interface SessionState { agentStreamTail: Map; agentStreamHead: Map; agentTimelineCursor: Map; + agentTimelineHasOlder: Map; + agentTimelineOlderFetchInFlight: Map; historySyncGeneration: number; agentHistorySyncGeneration: Map; agentAuthoritativeHistoryApplied: Map; @@ -351,6 +353,14 @@ interface SessionStoreActions { | Map | ((prev: Map) => Map), ) => void; + setAgentTimelineHasOlder: ( + serverId: string, + state: Map | ((prev: Map) => Map), + ) => void; + setAgentTimelineOlderFetchInFlight: ( + serverId: string, + state: Map | ((prev: Map) => Map), + ) => void; bumpHistorySyncGeneration: (serverId: string) => void; markAgentHistorySynchronized: (serverId: string, agentId: string) => void; setAgentAuthoritativeHistoryApplied: ( @@ -443,6 +453,8 @@ function createInitialSessionState(serverId: string, client: DaemonClient): Sess agentStreamTail: new Map(), agentStreamHead: new Map(), agentTimelineCursor: new Map(), + agentTimelineHasOlder: new Map(), + agentTimelineOlderFetchInFlight: new Map(), historySyncGeneration: 0, agentHistorySyncGeneration: new Map(), agentAuthoritativeHistoryApplied: new Map(), @@ -861,6 +873,48 @@ export const useSessionStore = create()( }); }, + setAgentTimelineHasOlder: (serverId, state) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session) { + return prev; + } + const nextState = + typeof state === "function" ? state(session.agentTimelineHasOlder) : state; + if (session.agentTimelineHasOlder === nextState) { + return prev; + } + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTimelineHasOlder: nextState }, + }, + }; + }); + }, + + setAgentTimelineOlderFetchInFlight: (serverId, state) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session) { + return prev; + } + const nextState = + typeof state === "function" ? state(session.agentTimelineOlderFetchInFlight) : state; + if (session.agentTimelineOlderFetchInFlight === nextState) { + return prev; + } + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTimelineOlderFetchInFlight: nextState }, + }, + }; + }); + }, + bumpHistorySyncGeneration: (serverId) => { set((prev) => { const session = prev.sessions[serverId]; diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts similarity index 90% rename from packages/app/src/contexts/session-stream-reducers.test.ts rename to packages/app/src/timeline/session-stream-reducers.test.ts index ee4ffff4b..f19ff6277 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -16,10 +16,15 @@ import { // Test helpers // --------------------------------------------------------------------------- -function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") { +function makeTimelineEntry( + seq: number, + text: string, + type: string = "assistant_message", + seqEnd = seq, +) { return { seqStart: seq, - seqEnd: seq, + seqEnd, provider: "claude", item: { type, text }, timestamp: new Date(1000 + seq).toISOString(), @@ -84,6 +89,14 @@ function getAssistantTexts(items: StreamItem[]): string[] { .map((item) => item.text); } +function getUserTexts(items: StreamItem[]): string[] { + return items + .filter((item): item is Extract => { + return item.kind === "user_message"; + }) + .map((item) => item.text); +} + const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", @@ -94,6 +107,8 @@ const baseTimelineInput: ProcessTimelineResponseInput = { endCursor: null, entries: [], error: null, + hasNewer: false, + hasOlder: false, }, currentTail: [], currentHead: [], @@ -356,6 +371,117 @@ describe("processTimelineResponse", () => { expect(result.cursorChanged).toBe(false); }); + it("prepends older before-cursor entries and only expands the start cursor", () => { + const currentTail: StreamItem[] = [ + { + kind: "user_message", + id: "current-3", + text: "current-3", + timestamp: new Date(3000), + }, + ]; + const currentHead = [makeAssistantItem("live head")]; + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 3, + endSeq: 5, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentHead, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + direction: "before", + epoch: "epoch-1", + startCursor: { seq: 1 }, + endCursor: { seq: 2 }, + entries: [ + makeTimelineEntry(1, "older-1", "user_message"), + makeTimelineEntry(2, "older-2", "user_message"), + ], + }, + }); + + expect(getUserTexts(result.tail)).toEqual(["older-1", "older-2", "current-3"]); + expect(result.head).toBe(currentHead); + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }); + }); + + it("leaves the cursor alone when a before page makes no progress", () => { + const currentTail: StreamItem[] = [ + { + kind: "user_message", + id: "current-3", + text: "current-3", + timestamp: new Date(3000), + }, + ]; + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 3, + endSeq: 5, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + direction: "before", + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 4 }, + entries: [ + makeTimelineEntry(3, "overlap-3", "user_message"), + makeTimelineEntry(4, "overlap-4", "user_message"), + ], + }, + }); + + expect(result.tail).toBe(currentTail); + expect(result.cursorChanged).toBe(false); + expect(result.cursor).toBe(existingCursor); + }); + + it("merges assistant chunks across the older-page prepend boundary", () => { + const currentTail = [makeAssistantItem("newer chunk", "assistant-newer")]; + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 3, + endSeq: 5, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + direction: "before", + epoch: "epoch-1", + startCursor: { seq: 1 }, + endCursor: { seq: 2 }, + entries: [makeTimelineEntry(1, "older chunk ")], + }, + }); + + expect(getAssistantTexts(result.tail)).toEqual(["older chunk newer chunk"]); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }); + }); + it("requests canonical catch-up when a projected entry overlaps unseen seqs", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", @@ -946,6 +1072,8 @@ describe("processAgentStreamEvents", () => { endCursor: { seq: 186 }, entries: [makeTimelineEntry(186, seq186Text)], error: null, + hasNewer: false, + hasOlder: false, }, }); diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts similarity index 92% rename from packages/app/src/contexts/session-stream-reducers.ts rename to packages/app/src/timeline/session-stream-reducers.ts index 7459192e2..4b932ebfc 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -65,6 +65,8 @@ export interface ProcessTimelineResponseInput { endCursor: { seq: number } | null; entries: TimelineResponseEntry[]; error: string | null; + hasNewer: boolean; + hasOlder: boolean; }; currentTail: StreamItem[]; currentHead: StreamItem[]; @@ -310,6 +312,55 @@ function acceptIncrementalTimelineUnits(args: { return { acceptedUnits, cursor, gapCursor }; } +function acceptOlderTimelineUnits(args: { + timelineUnits: TimelineUnit[]; + payload: ProcessTimelineResponseInput["payload"]; + currentCursor: TimelineCursor | undefined; +}): IncrementalAcceptResult { + const { timelineUnits, payload, currentCursor } = args; + if (!currentCursor || currentCursor.epoch !== payload.epoch) { + return { acceptedUnits: [], cursor: currentCursor, gapCursor: null }; + } + + const acceptedUnits = timelineUnits.filter((unit) => unit.seqEnd < currentCursor.startSeq); + if (acceptedUnits.length === 0) { + return { acceptedUnits, cursor: currentCursor, gapCursor: null }; + } + + const firstAccepted = acceptedUnits[0]; + const startSeq = payload.startCursor?.seq ?? firstAccepted?.seq ?? currentCursor.startSeq; + return { + acceptedUnits, + cursor: { ...currentCursor, startSeq }, + gapCursor: null, + }; +} + +function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: StreamItem[]) { + if (olderTail.length === 0) { + return currentTail; + } + if (currentTail.length === 0) { + return olderTail; + } + + const olderLast = olderTail.at(-1); + const currentFirst = currentTail[0]; + if (olderLast?.kind !== "assistant_message" || currentFirst?.kind !== "assistant_message") { + return [...olderTail, ...currentTail]; + } + + return [ + ...olderTail.slice(0, -1), + { + ...olderLast, + text: `${olderLast.text}${currentFirst.text}`, + timestamp: currentFirst.timestamp, + }, + ...currentTail.slice(1), + ]; +} + function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -328,14 +379,27 @@ function applyTimelineIncrementalPath(args: { return { tail: nextTail, head: nextHead, cursor: nextCursor, cursorChanged, sideEffects }; } - const { acceptedUnits, cursor, gapCursor } = acceptIncrementalTimelineUnits({ - timelineUnits, - payload, - currentCursor, - }); + const { acceptedUnits, cursor, gapCursor } = + payload.direction === "before" + ? acceptOlderTimelineUnits({ + timelineUnits, + payload, + currentCursor, + }) + : acceptIncrementalTimelineUnits({ + timelineUnits, + payload, + currentCursor, + }); if (acceptedUnits.length > 0) { - if (currentHead.length > 0) { + if (payload.direction === "before") { + const olderTail = hydrateStreamState( + acceptedUnits.map(({ event, timestamp }) => ({ event, timestamp })), + { source: "canonical" }, + ); + nextTail = mergePrependedCanonicalTail(olderTail, currentTail); + } else if (currentHead.length > 0) { for (const { event, timestamp } of acceptedUnits) { const applied = applyStreamEvent({ tail: nextTail, diff --git a/packages/app/src/timeline/timeline-fetch-policy.ts b/packages/app/src/timeline/timeline-fetch-policy.ts new file mode 100644 index 000000000..e249211a8 --- /dev/null +++ b/packages/app/src/timeline/timeline-fetch-policy.ts @@ -0,0 +1,2 @@ +// Count is projected timeline items, not delta chunks. The daemon's `selectTimelineWindowByProjectedLimit` interprets this against canonical entries: `assistant_merge`, `reasoning_merge`, and `tool_lifecycle`. Do not confuse this with raw stream deltas. +export const TIMELINE_FETCH_PAGE_SIZE = 100;