From 3d96ea63543068ef69687c6a7519099e9c44df37 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 31 May 2026 13:38:29 +0800 Subject: [PATCH] Fix chat history paging around tool updates (#1247) * Fix projected timeline paging * Avoid full timeline fetches for plain pages --- docs/timeline-sync.md | 6 +- packages/app/e2e/helpers/rewind-flow.ts | 4 +- .../rewind/use-rewind-agent-mutation.ts | 2 +- .../hooks/use-agent-initialization.test.ts | 12 +- .../use-load-older-agent-history.test.ts | 2 +- .../src/hooks/use-load-older-agent-history.ts | 2 +- .../screens/workspace/workspace-screen.tsx | 2 +- .../timeline/session-stream-reducers.test.ts | 147 ++++++++++ .../src/timeline/session-stream-reducers.ts | 148 +++++++--- .../app/src/timeline/timeline-fetch-policy.ts | 3 +- .../src/timeline/timeline-sync-plan.test.ts | 12 +- .../app/src/timeline/timeline-sync-plan.ts | 38 +-- ...ts => agent.event-stream.real.e2e.test.ts} | 0 ...gration.test.ts => agent.real.e2e.test.ts} | 0 .../server/agent/timeline-projection.test.ts | 197 ++++++++++--- .../src/server/agent/timeline-projection.ts | 138 +++++++-- .../daemon-e2e/timeline-window.e2e.test.ts | 274 +++++++++++++++++- packages/server/src/server/session.ts | 159 +++------- 18 files changed, 890 insertions(+), 256 deletions(-) rename packages/server/src/server/agent/providers/claude/{agent.event-stream.integration.test.ts => agent.event-stream.real.e2e.test.ts} (100%) rename packages/server/src/server/agent/providers/claude/{agent.integration.test.ts => agent.real.e2e.test.ts} (100%) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 5caab5bef..2d85dfaf2 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -2,8 +2,8 @@ Agent chat delivery has two paths: -1. **Live stream** — `agent_stream` WebSocket messages for immediacy. -2. **Authoritative history** — `fetch_agent_timeline_request` for correctness. +1. **Live stream** — `agent_stream` WebSocket messages for immediacy. These may be delta-shaped lifecycle updates. +2. **Authoritative history** — `fetch_agent_timeline_request` for correctness. This always returns full projected timeline items, never lifecycle deltas. The invariant is: @@ -24,6 +24,8 @@ Heartbeat is used for notification routing. It must not be used as a correctness Large unbounded timeline responses can exceed relay frame limits, so catch-up uses bounded pages. Bounded does not mean partial. +Page limits are projected-item targets. A tool call lifecycle is one projected item even if it spans many source sequence numbers, and assistant/reasoning chunks are merged before counting. The response carries `seqStart`, `seqEnd`, `sourceSeqRanges`, and `collapsed` so clients can advance sequence cursors without rendering delta rows. + When the app fetches `direction: "after"` and the daemon responds with `hasNewer: true`, the app must immediately fetch the next page from `endCursor`. The catch-up is complete only when `hasNewer: false`. The first load of an agent without a local cursor is different: it fetches a bounded latest tail page. Older history remains user-driven by scrolling upward. diff --git a/packages/app/e2e/helpers/rewind-flow.ts b/packages/app/e2e/helpers/rewind-flow.ts index ddfb1c015..57ce79a87 100644 --- a/packages/app/e2e/helpers/rewind-flow.ts +++ b/packages/app/e2e/helpers/rewind-flow.ts @@ -320,12 +320,12 @@ async function fetchTimelineEpoch(handle: AgentHandle): Promise Promise<{ epoch?: string }>; }; const timeline = await client.fetchAgentTimeline(handle.agentId, { direction: "tail", - projection: "canonical", + projection: "projected", limit: 0, }); return timeline.epoch; diff --git a/packages/app/src/components/rewind/use-rewind-agent-mutation.ts b/packages/app/src/components/rewind/use-rewind-agent-mutation.ts index f09112b22..8428a422a 100644 --- a/packages/app/src/components/rewind/use-rewind-agent-mutation.ts +++ b/packages/app/src/components/rewind/use-rewind-agent-mutation.ts @@ -47,7 +47,7 @@ export function useRewindAgentMutation(input: UseRewindAgentMutationInput): { : undefined; await input.client.fetchAgentTimeline(input.agentId, { direction: "tail", - projection: "canonical", + projection: "projected", ...(cursor ? { cursor: { epoch: cursor.epoch, seq: cursor.endSeq } } : {}), }); } diff --git a/packages/app/src/hooks/use-agent-initialization.test.ts b/packages/app/src/hooks/use-agent-initialization.test.ts index 2c7b9f38b..850d5ce00 100644 --- a/packages/app/src/hooks/use-agent-initialization.test.ts +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -34,7 +34,7 @@ afterEach(() => { }); describe("ensureAgentIsInitialized", () => { - it("requests bounded canonical catch-up after the current cursor when authoritative history is loaded", () => { + it("requests bounded projected catch-up after the current cursor when authoritative history is loaded", () => { const client = makeClient(); useSessionStore.getState().initializeSession(serverId, client as never); useSessionStore @@ -56,12 +56,12 @@ describe("ensureAgentIsInitialized", () => { direction: "after", cursor: { epoch: "epoch-1", seq: 42 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after"); }); - it("requests a bounded canonical tail when no authoritative cursor is available", () => { + it("requests a bounded projected tail when no authoritative cursor is available", () => { const client = makeClient(); useSessionStore.getState().initializeSession(serverId, client as never); @@ -75,7 +75,7 @@ describe("ensureAgentIsInitialized", () => { expect(client.fetchAgentTimeline).toHaveBeenCalledWith(agentId, { direction: "tail", limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("tail"); }); @@ -107,7 +107,7 @@ describe("ensureAgentIsInitialized", () => { }); describe("refreshAgent", () => { - it("fetches a bounded canonical tail after refreshing the agent", async () => { + it("fetches a bounded projected tail after refreshing the agent", async () => { const client = makeClient(); useSessionStore.getState().initializeSession(serverId, client as never); @@ -121,7 +121,7 @@ describe("refreshAgent", () => { expect(client.fetchAgentTimeline).toHaveBeenCalledWith(agentId, { direction: "tail", limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); }); 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 index ca86dacf8..28efe17fc 100644 --- a/packages/app/src/hooks/use-load-older-agent-history.test.ts +++ b/packages/app/src/hooks/use-load-older-agent-history.test.ts @@ -147,7 +147,7 @@ describe("loadOlderAgentHistory", () => { direction: "before", cursor: { epoch: "epoch-1", seq: 10 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }, }, ]); diff --git a/packages/app/src/hooks/use-load-older-agent-history.ts b/packages/app/src/hooks/use-load-older-agent-history.ts index 52e3e412e..6474559a0 100644 --- a/packages/app/src/hooks/use-load-older-agent-history.ts +++ b/packages/app/src/hooks/use-load-older-agent-history.ts @@ -10,7 +10,7 @@ export interface LoadOlderAgentHistoryClient { direction: "before"; cursor: { epoch: string; seq: number }; limit: number; - projection: "canonical"; + projection: "projected"; }, ) => Promise; } diff --git a/packages/app/src/screens/workspace/workspace-screen.tsx b/packages/app/src/screens/workspace/workspace-screen.tsx index 18781a497..a32931b61 100644 --- a/packages/app/src/screens/workspace/workspace-screen.tsx +++ b/packages/app/src/screens/workspace/workspace-screen.tsx @@ -2465,7 +2465,7 @@ function WorkspaceScreenContent({ const currentCursor = sessionState?.agentTimelineCursor.get(agentId); await client.fetchAgentTimeline(agentId, { direction: "tail", - projection: "canonical", + projection: "projected", ...(currentCursor ? { cursor: { epoch: currentCursor.epoch, seq: currentCursor.endSeq } } : {}), diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 0f5bda598..6c2ce8d90 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -547,6 +547,153 @@ describe("processTimelineResponse", () => { }); }); + it("hydrates a fetched in-progress tool call as one item and streams the next update on top", () => { + const fetched = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "tail", + payload: { + ...baseTimelineInput.payload, + direction: "tail", + epoch: "epoch-1", + startCursor: { seq: 10 }, + endCursor: { seq: 250 }, + entries: [ + { + ...makeToolCallTimelineEntry(10, "call-1", "running", { + type: "read", + filePath: "/tmp/example.ts", + }), + seqEnd: 250, + sourceSeqRanges: [ + { startSeq: 10, endSeq: 10 }, + { startSeq: 250, endSeq: 250 }, + ], + collapsed: ["tool_lifecycle"], + }, + ], + }, + }); + + expect(getAgentToolCalls(fetched.tail)).toHaveLength(1); + expect(fetched.cursor).toEqual({ epoch: "epoch-1", startSeq: 10, endSeq: 250 }); + + const streamed = processAgentStreamEvent({ + ...baseStreamInput, + currentTail: fetched.tail, + currentHead: fetched.head, + currentCursor: fetched.cursor ?? undefined, + seq: 251, + epoch: "epoch-1", + event: { + type: "timeline", + provider: "claude", + item: { + type: "tool_call", + callId: "call-1", + name: "Read", + status: "completed", + detail: { + type: "read", + filePath: "/tmp/example.ts", + }, + error: null, + }, + } as AgentStreamEventPayload, + }); + + const tools = getAgentToolCalls(streamed.tail); + expect(tools).toHaveLength(1); + expect(tools[0]?.payload.data.status).toBe("completed"); + expect(streamed.cursor).toEqual({ epoch: "epoch-1", startSeq: 10, endSeq: 251 }); + }); + + it("accepts an after-page projected tool update whose item started before the cursor", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 10, + endSeq: 249, + }; + const runningTool = hydrateStreamState( + [ + { + event: makeToolCallTimelineEvent("call-1"), + timestamp: new Date(1010), + }, + ], + { source: "canonical" }, + ); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: runningTool, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + direction: "after", + epoch: "epoch-1", + startCursor: { seq: 250 }, + endCursor: { seq: 250 }, + entries: [ + { + ...makeToolCallTimelineEntry(10, "call-1", "completed", { + type: "read", + filePath: "/tmp/example.ts", + }), + seqEnd: 250, + sourceSeqRanges: [ + { startSeq: 10, endSeq: 10 }, + { startSeq: 250, endSeq: 250 }, + ], + collapsed: ["tool_lifecycle"], + }, + ], + }, + }); + + const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up"); + const tools = getAgentToolCalls(result.tail); + expect(catchUp).toBeUndefined(); + expect(tools).toHaveLength(1); + expect(tools[0]?.payload.data.status).toBe("completed"); + expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 10, endSeq: 250 }); + }); + + it("replaces an active assistant head when after-page returns a full projected assistant item", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }; + const currentHead = [makeAssistantItem("ABC")]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentHead, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + direction: "after", + epoch: "epoch-1", + startCursor: { seq: 4 }, + endCursor: { seq: 5 }, + entries: [ + { + ...makeTimelineEntry(1, "ABCDE"), + seqEnd: 5, + sourceSeqRanges: [{ startSeq: 1, endSeq: 5 }], + collapsed: ["assistant_merge"], + }, + ], + }, + }); + + expect(getAssistantTexts(result.tail)).toEqual([]); + expect(getAssistantTexts(result.head)).toEqual(["ABCDE"]); + expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 5 }); + }); + it("detects gap and emits catch-up side effect", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index d3723532a..d4b1a5c39 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -53,9 +53,16 @@ type SessionTimelineSeqCursor = type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "gap" | "init"; +interface TimelineSeqRange { + startSeq: number; + endSeq: number; +} + interface TimelineResponseEntry { seqStart: number; seqEnd: number; + sourceSeqRanges?: TimelineSeqRange[]; + collapsed?: string[]; provider: string; item: Record; timestamp: string; @@ -96,6 +103,7 @@ export interface ProcessTimelineResponseOutput { interface TimelineUnit { seq: number; seqEnd: number; + sourceSeqRanges: TimelineSeqRange[]; event: AgentStreamEventPayload; timestamp: Date; } @@ -376,44 +384,56 @@ function acceptIncrementalTimelineUnits(args: { currentCursor: TimelineCursor | undefined; }): IncrementalAcceptResult { const { timelineUnits, payload, currentCursor } = args; - const acceptedUnits: TimelineUnit[] = []; - let cursor: TimelineCursor | undefined = currentCursor; - let gapCursor: { epoch: string; endSeq: number } | null = null; + const firstUnit = timelineUnits[0]; + const lastUnit = timelineUnits[timelineUnits.length - 1]; + const responseStartSeq = payload.startCursor?.seq ?? firstUnit?.seq; + const responseEndSeq = payload.endCursor?.seq ?? lastUnit?.seqEnd; - for (const unit of timelineUnits) { - const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({ - cursor: cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null, - epoch: payload.epoch, - seq: unit.seq, - }); - - if (decision === "gap") { - gapCursor = cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null; - break; - } - if (decision === "drop_stale") { - if (cursor && unit.seqEnd > cursor.endSeq) { - gapCursor = { epoch: cursor.epoch, endSeq: cursor.endSeq }; - break; - } - continue; - } - if (decision === "drop_epoch") { - continue; - } - - acceptedUnits.push(unit); - if (decision === "init") { - cursor = { epoch: payload.epoch, startSeq: unit.seq, endSeq: unit.seqEnd }; - continue; - } - if (!cursor) { - continue; - } - cursor = { ...cursor, endSeq: unit.seqEnd }; + if (responseStartSeq === undefined || responseEndSeq === undefined) { + return { acceptedUnits: [], cursor: currentCursor, gapCursor: null }; } - return { acceptedUnits, cursor, gapCursor }; + if (!currentCursor) { + return { + acceptedUnits: timelineUnits, + cursor: { epoch: payload.epoch, startSeq: responseStartSeq, endSeq: responseEndSeq }, + gapCursor: null, + }; + } + + if (currentCursor.epoch !== payload.epoch) { + return { acceptedUnits: [], cursor: currentCursor, gapCursor: null }; + } + + if ( + (!payload.startCursor || !payload.endCursor) && + responseStartSeq <= currentCursor.endSeq && + responseEndSeq > currentCursor.endSeq + ) { + return { + acceptedUnits: [], + cursor: currentCursor, + gapCursor: { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq }, + }; + } + + if (responseEndSeq <= currentCursor.endSeq) { + return { acceptedUnits: [], cursor: currentCursor, gapCursor: null }; + } + + if (responseStartSeq > currentCursor.endSeq + 1) { + return { + acceptedUnits: [], + cursor: currentCursor, + gapCursor: { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq }, + }; + } + + return { + acceptedUnits: timelineUnits, + cursor: { ...currentCursor, endSeq: responseEndSeq }, + gapCursor: null, + }; } function acceptOlderTimelineUnits(args: { @@ -426,16 +446,21 @@ function acceptOlderTimelineUnits(args: { 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 firstUnit = timelineUnits[0]; + const lastUnit = timelineUnits[timelineUnits.length - 1]; + const responseStartSeq = payload.startCursor?.seq ?? firstUnit?.seq; + const responseEndSeq = payload.endCursor?.seq ?? lastUnit?.seqEnd; + if ( + responseStartSeq === undefined || + responseEndSeq === undefined || + responseEndSeq >= currentCursor.startSeq + ) { + return { acceptedUnits: [], cursor: currentCursor, gapCursor: null }; } - const firstAccepted = acceptedUnits[0]; - const startSeq = payload.startCursor?.seq ?? firstAccepted?.seq ?? currentCursor.startSeq; return { - acceptedUnits, - cursor: { ...currentCursor, startSeq }, + acceptedUnits: timelineUnits, + cursor: { ...currentCursor, startSeq: responseStartSeq }, gapCursor: null, }; } @@ -480,6 +505,32 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea ]; } +function replaceLiveAssistantWithProjectedText(params: { + head: StreamItem[]; + event: AgentStreamEventPayload; + timestamp: Date; +}): StreamItem[] | null { + const { head, event, timestamp } = params; + if (event.type !== "timeline" || event.item.type !== "assistant_message") { + return null; + } + const index = head.findLastIndex((item) => item.kind === "assistant_message"); + const current = head[index]; + if (!current || current.kind !== "assistant_message") { + return null; + } + if (!event.item.text.startsWith(current.text)) { + return null; + } + const next = [...head]; + next[index] = { + ...current, + text: event.item.text, + timestamp, + }; + return next; +} + function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -523,6 +574,15 @@ function applyTimelineIncrementalPath(args: { nextTail = mergePrependedCanonicalTail(olderTail, currentTail); } else if (currentHead.length > 0) { for (const { event, timestamp } of acceptedUnits) { + const replacedHead = replaceLiveAssistantWithProjectedText({ + head: nextHead, + event, + timestamp, + }); + if (replacedHead) { + nextHead = replacedHead; + continue; + } const applied = applyStreamEvent({ tail: nextTail, head: nextHead, @@ -597,6 +657,10 @@ export function processTimelineResponse( const timelineUnits = payload.entries.map((entry) => ({ seq: entry.seqStart, seqEnd: entry.seqEnd, + sourceSeqRanges: + entry.sourceSeqRanges && entry.sourceSeqRanges.length > 0 + ? entry.sourceSeqRanges + : [{ startSeq: entry.seqStart, endSeq: entry.seqEnd }], event: { type: "timeline", provider: entry.provider, diff --git a/packages/app/src/timeline/timeline-fetch-policy.ts b/packages/app/src/timeline/timeline-fetch-policy.ts index e249211a8..6fe59d1a0 100644 --- a/packages/app/src/timeline/timeline-fetch-policy.ts +++ b/packages/app/src/timeline/timeline-fetch-policy.ts @@ -1,2 +1,3 @@ -// 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. +// Count is projected timeline items, not delta chunks. Fetch responses never return +// tool lifecycle deltas; `sourceSeqRanges` maps projected items back to source seqs. export const TIMELINE_FETCH_PAGE_SIZE = 100; diff --git a/packages/app/src/timeline/timeline-sync-plan.test.ts b/packages/app/src/timeline/timeline-sync-plan.test.ts index 54b121a1c..913f2501b 100644 --- a/packages/app/src/timeline/timeline-sync-plan.test.ts +++ b/packages/app/src/timeline/timeline-sync-plan.test.ts @@ -18,7 +18,7 @@ describe("timeline sync planning", () => { expect(plan).toEqual({ direction: "tail", limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); @@ -32,7 +32,7 @@ describe("timeline sync planning", () => { direction: "after", cursor: { epoch: "epoch-1", seq: 42 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); @@ -45,7 +45,7 @@ describe("timeline sync planning", () => { direction: "after", cursor: { epoch: "epoch-1", seq: 100 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); @@ -55,7 +55,7 @@ describe("timeline sync planning", () => { expect(plan).toEqual({ direction: "tail", limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); @@ -66,7 +66,7 @@ describe("timeline sync planning", () => { direction: "before", cursor: { epoch: "epoch-1", seq: 25 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); @@ -82,7 +82,7 @@ describe("timeline sync planning", () => { direction: "after", cursor: { epoch: "epoch-1", seq: 200 }, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", }); }); diff --git a/packages/app/src/timeline/timeline-sync-plan.ts b/packages/app/src/timeline/timeline-sync-plan.ts index a9c5083f1..43efec277 100644 --- a/packages/app/src/timeline/timeline-sync-plan.ts +++ b/packages/app/src/timeline/timeline-sync-plan.ts @@ -11,39 +11,39 @@ export interface AgentTimelineCursorRange { endSeq: number; } -export interface CanonicalTimelineTailFetchPlan { +export interface ProjectedTimelineTailFetchPlan { direction: "tail"; limit: number; - projection: "canonical"; + projection: "projected"; } -export interface CanonicalTimelineAfterFetchPlan { +export interface ProjectedTimelineAfterFetchPlan { direction: "after"; cursor: TimelineSyncCursor; limit: number; - projection: "canonical"; + projection: "projected"; } -export interface CanonicalTimelineBeforeFetchPlan { +export interface ProjectedTimelineBeforeFetchPlan { direction: "before"; cursor: TimelineSyncCursor; limit: number; - projection: "canonical"; + projection: "projected"; } -export type CanonicalTimelineFetchPlan = - | CanonicalTimelineTailFetchPlan - | CanonicalTimelineAfterFetchPlan - | CanonicalTimelineBeforeFetchPlan; +export type ProjectedTimelineFetchPlan = + | ProjectedTimelineTailFetchPlan + | ProjectedTimelineAfterFetchPlan + | ProjectedTimelineBeforeFetchPlan; -export type CanonicalTimelineForwardFetchPlan = - | CanonicalTimelineTailFetchPlan - | CanonicalTimelineAfterFetchPlan; +export type ProjectedTimelineForwardFetchPlan = + | ProjectedTimelineTailFetchPlan + | ProjectedTimelineAfterFetchPlan; export function planInitialAgentTimelineSync(input: { cursor: AgentTimelineCursorRange | undefined; hasAuthoritativeHistory: boolean; -}): CanonicalTimelineForwardFetchPlan { +}): ProjectedTimelineForwardFetchPlan { if (input.hasAuthoritativeHistory && input.cursor) { return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq }); } @@ -53,7 +53,7 @@ export function planInitialAgentTimelineSync(input: { export function planResumeTimelineSync(input: { cursor: AgentTimelineCursorRange | undefined; -}): CanonicalTimelineForwardFetchPlan { +}): ProjectedTimelineForwardFetchPlan { if (input.cursor) { return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq }); } @@ -66,7 +66,7 @@ export function planTimelineCatchUpAfter(cursor: TimelineSyncCursor) { direction: "after", cursor, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", } as const; } @@ -74,7 +74,7 @@ export function planTimelineTailFetch() { return { direction: "tail", limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", } as const; } @@ -83,7 +83,7 @@ export function planTimelineOlderFetch(cursor: TimelineSyncCursor) { direction: "before", cursor, limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", + projection: "projected", } as const; } @@ -92,7 +92,7 @@ export function planTimelineCatchUpFollowUp(input: { hasNewer: boolean; endCursor: TimelineSyncCursor | null; error: string | null; -}): CanonicalTimelineAfterFetchPlan | null { +}): ProjectedTimelineAfterFetchPlan | null { if (input.error || input.direction !== "after" || !input.hasNewer || !input.endCursor) { return null; } diff --git a/packages/server/src/server/agent/providers/claude/agent.event-stream.integration.test.ts b/packages/server/src/server/agent/providers/claude/agent.event-stream.real.e2e.test.ts similarity index 100% rename from packages/server/src/server/agent/providers/claude/agent.event-stream.integration.test.ts rename to packages/server/src/server/agent/providers/claude/agent.event-stream.real.e2e.test.ts diff --git a/packages/server/src/server/agent/providers/claude/agent.integration.test.ts b/packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts similarity index 100% rename from packages/server/src/server/agent/providers/claude/agent.integration.test.ts rename to packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts diff --git a/packages/server/src/server/agent/timeline-projection.test.ts b/packages/server/src/server/agent/timeline-projection.test.ts index c7f53abc6..814b7a62e 100644 --- a/packages/server/src/server/agent/timeline-projection.test.ts +++ b/packages/server/src/server/agent/timeline-projection.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "vitest"; import type { AgentTimelineRow } from "./agent-manager.js"; import { projectTimelineRows, + selectProjectedTimelinePage, selectTimelineWindowByProjectedLimit, } from "./timeline-projection.js"; @@ -436,58 +437,178 @@ describe("selectTimelineWindowByProjectedLimit", () => { }); }); - test("can enforce a hard projected limit when tool lifecycle collapsing is disabled", () => { + test("tail limit treats a repeated running tool call as one projected item", () => { const rows: AgentTimelineRow[] = [ - { - seq: 1, - timestamp: "2026-02-13T00:00:00.000Z", + ...Array.from({ length: 6 }, (_, index) => ({ + seq: index + 1, + timestamp: `2026-02-13T00:00:00.00${index}Z`, + item: { type: "assistant_message" as const, text: `old ${index}` }, + })), + ...Array.from({ length: 20 }, (_, index) => ({ + seq: index + 7, + timestamp: `2026-02-13T00:00:01.0${index}Z`, item: { - type: "tool_call", + type: "tool_call" as const, callId: "call_1", name: "shell", - status: "running", + status: "running" as const, error: null, detail: { - type: "unknown", - input: { cmd: "pwd" }, - output: null, + type: "unknown" as const, + input: { cmd: "sleep 10" }, + output: { progress: index }, }, }, - }, - { - seq: 2, - timestamp: "2026-02-13T00:00:00.100Z", - item: { type: "assistant_message", text: "work" }, - }, - { - seq: 3, - timestamp: "2026-02-13T00:00:00.200Z", - item: { - type: "tool_call", - callId: "call_1", - name: "shell", - status: "completed", - error: null, - detail: { - type: "unknown", - input: { cmd: "pwd" }, - output: { stdout: "/tmp" }, - }, - }, - }, + })), ]; const selected = selectTimelineWindowByProjectedLimit({ rows, direction: "tail", - limit: 1, - collapseToolLifecycle: false, + limit: 100, }); - expect(selected.minSeq).toBe(3); - expect(selected.maxSeq).toBe(3); - expect(selected.selectedRows.map((row) => row.seq)).toEqual([3]); - expect(selected.projectedEntries).toHaveLength(1); - expect(selected.projectedEntries[0]?.item.type).toBe("tool_call"); + const tools = selected.projectedEntries.filter((entry) => entry.item.type === "tool_call"); + expect(tools).toHaveLength(1); + expect(tools[0]?.collapsed).toContain("tool_lifecycle"); + expect(selected.projectedEntries).toHaveLength(2); + }); +}); + +describe("selectProjectedTimelinePage", () => { + function toolRow(seq: number, status: "running" | "completed"): AgentTimelineRow { + return { + seq, + timestamp: new Date(1000 + seq).toISOString(), + item: { + type: "tool_call", + callId: "call_1", + name: "shell", + status, + error: null, + detail: { + type: "unknown", + input: { cmd: "sleep 10" }, + output: status === "completed" ? { stdout: "done" } : null, + }, + }, + }; + } + + test("tail page returns full projected items instead of tool lifecycle deltas", () => { + const rows: AgentTimelineRow[] = [ + { seq: 1, timestamp: "2026-02-13T00:00:00.000Z", item: { type: "user_message", text: "go" } }, + ...Array.from({ length: 120 }, (_, index) => toolRow(index + 2, "running")), + ]; + + const page = selectProjectedTimelinePage({ rows, direction: "tail", limit: 100 }); + + expect(page.entries.map((entry) => entry.item.type)).toEqual(["user_message", "tool_call"]); + expect(page.entries[1]?.collapsed).toContain("tool_lifecycle"); + expect(page.entries[1]?.sourceSeqRanges).toEqual([{ startSeq: 2, endSeq: 121 }]); + expect(page.startSeq).toBe(1); + expect(page.endSeq).toBe(121); + expect(page.hasNewer).toBe(false); + }); + + test("after page includes a full projected tool item when only its update is new", () => { + const rows: AgentTimelineRow[] = [ + toolRow(10, "running"), + { + seq: 11, + timestamp: "2026-02-13T00:00:00.011Z", + item: { type: "assistant_message", text: "working" }, + }, + toolRow(250, "completed"), + ]; + + const page = selectProjectedTimelinePage({ + rows, + direction: "after", + cursorSeq: 249, + limit: 100, + }); + + expect(page.entries).toHaveLength(1); + expect(page.entries[0]?.item.type).toBe("tool_call"); + expect(page.entries[0]?.seqStart).toBe(10); + expect(page.entries[0]?.seqEnd).toBe(250); + expect(page.entries[0]?.sourceSeqRanges).toEqual([ + { startSeq: 10, endSeq: 10 }, + { startSeq: 250, endSeq: 250 }, + ]); + expect(page.startSeq).toBe(250); + expect(page.endSeq).toBe(250); + }); + + test("after page cursor advances only through contiguously covered seqs", () => { + const rows: AgentTimelineRow[] = [ + toolRow(1, "running"), + ...Array.from({ length: 498 }, (_, index) => ({ + seq: index + 2, + timestamp: new Date(2000 + index).toISOString(), + item: { type: "user_message" as const, text: `middle ${index + 2}` }, + })), + toolRow(500, "completed"), + ...Array.from({ length: 101 }, (_, index) => ({ + seq: index + 501, + timestamp: new Date(3000 + index).toISOString(), + item: { type: "user_message" as const, text: `later ${index + 501}` }, + })), + ]; + + const page = selectProjectedTimelinePage({ + rows, + direction: "after", + cursorSeq: 0, + limit: 100, + }); + + expect(page.entries[0]?.item.type).toBe("tool_call"); + expect( + page.entries.some((entry) => entry.item.type === "user_message" && entry.seqStart === 101), + ).toBe(false); + expect(page.endSeq).toBe(100); + expect(page.hasNewer).toBe(true); + }); + + test("before page includes a wide tool whose earlier source range is before the cursor", () => { + const rows: AgentTimelineRow[] = [ + toolRow(1, "running"), + ...Array.from({ length: 498 }, (_, index) => ({ + seq: index + 2, + timestamp: new Date(2000 + index).toISOString(), + item: { type: "user_message" as const, text: `middle ${index + 2}` }, + })), + toolRow(500, "completed"), + ]; + + const page = selectProjectedTimelinePage({ + rows, + direction: "before", + cursorSeq: 500, + limit: 100, + }); + + expect(page.entries.some((entry) => entry.item.type === "tool_call")).toBe(true); + expect(page.endSeq).toBeLessThan(500); + expect(page.hasOlder).toBe(true); + }); + + test("tail page includes a wide tool when its completion is the newest seq", () => { + const rows: AgentTimelineRow[] = [ + toolRow(1, "running"), + ...Array.from({ length: 499 }, (_, index) => ({ + seq: index + 2, + timestamp: new Date(2000 + index).toISOString(), + item: { type: "user_message" as const, text: `middle ${index + 2}` }, + })), + toolRow(501, "completed"), + ]; + + const page = selectProjectedTimelinePage({ rows, direction: "tail", limit: 100 }); + + expect(page.entries.some((entry) => entry.item.type === "tool_call")).toBe(true); + expect(page.endSeq).toBe(501); }); }); diff --git a/packages/server/src/server/agent/timeline-projection.ts b/packages/server/src/server/agent/timeline-projection.ts index c86198407..279999e4e 100644 --- a/packages/server/src/server/agent/timeline-projection.ts +++ b/packages/server/src/server/agent/timeline-projection.ts @@ -28,6 +28,14 @@ interface ProjectedWindowSelection { maxSeq: number | null; } +export interface ProjectedTimelinePageSelection { + entries: TimelineProjectionEntry[]; + startSeq: number | null; + endSeq: number | null; + hasOlder: boolean; + hasNewer: boolean; +} + function appendSeqToRanges(ranges: TimelineSeqRange[], seq: number): TimelineSeqRange[] { if (ranges.length === 0) { return [{ startSeq: seq, endSeq: seq }]; @@ -266,15 +274,11 @@ export function selectTimelineWindowByProjectedLimit(input: { rows: readonly AgentTimelineRow[]; direction: TimelineLimitDirection; limit: number; - collapseToolLifecycle?: boolean; }): ProjectedWindowSelection { const { rows, direction } = input; const limit = Math.max(0, Math.floor(input.limit)); - const collapseTools = input.collapseToolLifecycle ?? true; const canonical = makeCanonicalEntries(rows); - const projectedAll = mergeReasoningChunks( - mergeAssistantChunks(collapseTools ? collapseToolLifecycle(canonical) : canonical), - ); + const projectedAll = mergeReasoningChunks(mergeAssistantChunks(collapseToolLifecycle(canonical))); if (projectedAll.length === 0) { return { @@ -320,27 +324,25 @@ export function selectTimelineWindowByProjectedLimit(input: { let { minSeq, maxSeq } = computeWindowBounds(projectedEntries); let expandedEntries = projectedEntries; - if (collapseTools) { - // Expand to include any projected entries that overlap the selected - // canonical range. Tool lifecycle collapse can produce non-monotonic - // seqEnd values, which would otherwise create cursor gaps. - for (let iteration = 0; iteration < projectedAll.length + 1; iteration += 1) { - const overlapping = projectedAll.filter( - (entry) => entry.seqStart <= maxSeq && entry.seqEnd >= minSeq, - ); - const nextBounds = computeWindowBounds(overlapping); - if ( - overlapping.length === expandedEntries.length && - nextBounds.minSeq === minSeq && - nextBounds.maxSeq === maxSeq - ) { - expandedEntries = overlapping; - break; - } + // Expand to include any projected entries that overlap the selected canonical + // range. Tool lifecycle collapse can produce non-monotonic seqEnd values, + // which would otherwise create cursor gaps. + for (let iteration = 0; iteration < projectedAll.length + 1; iteration += 1) { + const overlapping = projectedAll.filter( + (entry) => entry.seqStart <= maxSeq && entry.seqEnd >= minSeq, + ); + const nextBounds = computeWindowBounds(overlapping); + if ( + overlapping.length === expandedEntries.length && + nextBounds.minSeq === minSeq && + nextBounds.maxSeq === maxSeq + ) { expandedEntries = overlapping; - minSeq = nextBounds.minSeq; - maxSeq = nextBounds.maxSeq; + break; } + expandedEntries = overlapping; + minSeq = nextBounds.minSeq; + maxSeq = nextBounds.maxSeq; } const selectedRows = rows.filter((row) => row.seq >= minSeq && row.seq <= maxSeq); @@ -353,6 +355,94 @@ export function selectTimelineWindowByProjectedLimit(input: { }; } +function getTimelineBounds( + rows: readonly AgentTimelineRow[], +): { minSeq: number; maxSeq: number } | null { + const first = rows[0]; + const last = rows[rows.length - 1]; + if (!first || !last) { + return null; + } + return { minSeq: first.seq, maxSeq: last.seq }; +} + +function selectEntriesOverlappingSeqRange(input: { + entries: readonly TimelineProjectionEntry[]; + startSeq: number; + endSeq: number; +}): TimelineProjectionEntry[] { + return input.entries.filter( + (entry) => entry.seqStart <= input.endSeq && entry.seqEnd >= input.startSeq, + ); +} + +export function selectProjectedTimelinePage(input: { + rows: readonly AgentTimelineRow[]; + bounds?: { minSeq: number; maxSeq: number }; + direction: TimelineLimitDirection; + cursorSeq?: number; + limit?: number; +}): ProjectedTimelinePageSelection { + const limit = input.limit === undefined ? 0 : Math.max(0, Math.floor(input.limit)); + const bounds = input.bounds ?? getTimelineBounds(input.rows); + const projectedAll = projectTimelineRows({ rows: input.rows, mode: "projected" }); + if (projectedAll.length === 0 || !bounds) { + return { + entries: [], + startSeq: null, + endSeq: null, + hasOlder: false, + hasNewer: false, + }; + } + + if (input.direction === "tail") { + const selected = selectTimelineWindowByProjectedLimit({ + rows: input.rows, + direction: "tail", + limit, + }); + return { + entries: selected.projectedEntries, + startSeq: selected.minSeq, + endSeq: selected.maxSeq, + hasOlder: selected.minSeq !== null && selected.minSeq > bounds.minSeq, + hasNewer: false, + }; + } + + let startSeq: number; + let endSeq: number; + 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); + } else { + const cursorSeq = input.cursorSeq ?? bounds.maxSeq + 1; + endSeq = Math.min(bounds.maxSeq, cursorSeq - 1); + startSeq = limit === 0 ? bounds.minSeq : Math.max(bounds.minSeq, cursorSeq - limit); + } + + if (startSeq > endSeq) { + return { + entries: [], + startSeq: null, + endSeq: null, + hasOlder: startSeq > bounds.minSeq, + hasNewer: endSeq < bounds.maxSeq, + }; + } + + const entries = selectEntriesOverlappingSeqRange({ entries: projectedAll, startSeq, endSeq }); + return { + entries, + startSeq, + endSeq, + hasOlder: startSeq > bounds.minSeq, + hasNewer: endSeq < bounds.maxSeq, + }; +} + /** * Apply a projected-count limit to a flat AgentTimelineItem[] without seq metadata. * Used by callers that only have items in hand (e.g. MCP tools reading 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 04e1b0369..eb4322dba 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 @@ -1,4 +1,4 @@ -import { describe, test, expect, beforeEach, afterEach } from "vitest"; +import { describe, test, expect, beforeEach, afterEach, vi } from "vitest"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -17,6 +17,7 @@ describe("daemon E2E - timeline window", () => { }); afterEach(async () => { + vi.restoreAllMocks(); await ctx.cleanup(); }, 60_000); @@ -66,6 +67,10 @@ describe("daemon E2E - timeline window", () => { expect((await ctx.client.waitForFinish(agent.id, 5_000)).status).toBe("idle"); const expected = "SECOND"; + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: "next", + }); await ctx.client.sendMessage(agent.id, `Respond with exactly: ${expected}`); expect((await ctx.client.waitForFinish(agent.id, 5_000)).status).toBe("idle"); @@ -86,4 +91,271 @@ describe("daemon E2E - timeline window", () => { rmSync(cwd, { recursive: true, force: true }); } }, 30_000); + + test("timeline fetch returns one projected in-progress tool call instead of lifecycle deltas", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Tool Projection Test", + modeId: "full-access", + }); + + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: "run the tool", + }); + for (let index = 0; index < 120; index += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "tool_call", + callId: "call_1", + name: "shell", + status: "running", + error: null, + detail: { + type: "unknown", + input: { cmd: "sleep 10" }, + output: { progress: index }, + }, + }); + } + + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 100, + projection: "canonical", + }); + + const toolEntries = timeline.entries.filter((entry) => entry.item.type === "tool_call"); + expect(timeline.projection).toBe("projected"); + expect(timeline.entries.map((entry) => entry.item.type)).toEqual([ + "user_message", + "tool_call", + ]); + expect(toolEntries).toHaveLength(1); + expect(toolEntries[0]?.collapsed).toContain("tool_lifecycle"); + expect(toolEntries[0]?.sourceSeqRanges).toEqual([{ startSeq: 2, endSeq: 121 }]); + expect(timeline.endCursor?.seq).toBe(121); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("after fetch returns the full projected tool item for a new lifecycle update", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Tool Catch-up Test", + modeId: "full-access", + }); + + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "tool_call", + callId: "call_1", + name: "shell", + status: "running", + error: null, + detail: { + type: "unknown", + input: { cmd: "sleep 10" }, + output: null, + }, + }); + for (let seq = 2; seq <= 249; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: `background ${seq}`, + }); + } + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "tool_call", + callId: "call_1", + name: "shell", + status: "completed", + error: null, + detail: { + type: "unknown", + input: { cmd: "sleep 10" }, + output: { stdout: "done" }, + }, + }); + + const baseline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + }); + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "after", + cursor: { epoch: baseline.epoch, seq: 249 }, + limit: 100, + projection: "canonical", + }); + + expect(timeline.entries).toHaveLength(1); + expect(timeline.startCursor?.seq).toBe(250); + expect(timeline.endCursor?.seq).toBe(250); + expect(timeline.entries[0]?.seqStart).toBe(1); + expect(timeline.entries[0]?.seqEnd).toBe(250); + expect(timeline.entries[0]?.sourceSeqRanges).toEqual([ + { startSeq: 1, endSeq: 1 }, + { startSeq: 250, endSeq: 250 }, + ]); + expect(timeline.entries[0]?.item.type).toBe("tool_call"); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("reset timeline fetch reports older history when the reset slice starts after window min", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Reset HasOlder Test", + modeId: "full-access", + }); + + for (let seq = 1; seq <= 600; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: `row ${seq}`, + }); + } + + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + cursor: { epoch: "stale-epoch", seq: 600 }, + limit: 200, + }); + + expect(timeline.reset).toBe(true); + expect(timeline.staleCursor).toBe(true); + expect(timeline.startCursor?.seq).toBeGreaterThan(timeline.window.minSeq); + expect(timeline.hasOlder).toBe(true); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("tail fetch does not re-fetch full plain chat history", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Tail Bounded Fetch Test", + modeId: "full-access", + }); + for (let seq = 1; seq <= 600; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: `row ${seq}`, + }); + } + + const fetchSpy = vi.spyOn(ctx.daemon.daemon.agentManager, "fetchTimeline"); + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 100, + }); + + expect(timeline.entries).toHaveLength(100); + expect(timeline.startCursor?.seq).toBe(501); + expect(timeline.endCursor?.seq).toBe(600); + expect(timeline.hasOlder).toBe(true); + expect( + fetchSpy.mock.calls.some( + ([, options]) => options?.direction === "tail" && options.limit === 0, + ), + ).toBe(false); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("after fetch does not re-fetch full plain chat history", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline After Bounded Fetch Test", + modeId: "full-access", + }); + for (let seq = 1; seq <= 600; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: `row ${seq}`, + }); + } + const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, { + direction: "tail", + limit: 1, + }).epoch; + + const fetchSpy = vi.spyOn(ctx.daemon.daemon.agentManager, "fetchTimeline"); + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "after", + cursor: { epoch, seq: 300 }, + limit: 100, + }); + + expect(timeline.entries).toHaveLength(100); + expect(timeline.startCursor?.seq).toBe(301); + expect(timeline.endCursor?.seq).toBe(400); + expect(timeline.hasNewer).toBe(true); + expect( + fetchSpy.mock.calls.some( + ([, options]) => options?.direction === "tail" && options.limit === 0, + ), + ).toBe(false); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("before fetch does not re-fetch full plain chat history", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Timeline Before Bounded Fetch Test", + modeId: "full-access", + }); + for (let seq = 1; seq <= 600; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "user_message", + text: `row ${seq}`, + }); + } + const epoch = ctx.daemon.daemon.agentManager.fetchTimeline(agent.id, { + direction: "tail", + limit: 1, + }).epoch; + + const fetchSpy = vi.spyOn(ctx.daemon.daemon.agentManager, "fetchTimeline"); + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "before", + cursor: { epoch, seq: 501 }, + limit: 100, + }); + + expect(timeline.entries).toHaveLength(100); + expect(timeline.startCursor?.seq).toBe(401); + expect(timeline.endCursor?.seq).toBe(500); + expect(timeline.hasOlder).toBe(true); + expect( + fetchSpy.mock.calls.some( + ([, options]) => options?.direction === "tail" && options.limit === 0, + ), + ).toBe(false); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 172aaebd5..368dfa034 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -99,6 +99,7 @@ import type { AgentManagerEvent, AgentTimelineCursor, AgentTimelineFetchDirection, + AgentTimelineFetchResult, ManagedAgent, } from "./agent/agent-manager.js"; import { createAgentCommand } from "./agent/create-agent/create.js"; @@ -120,8 +121,7 @@ import { emitLiveTimelineItemIfAgentKnown, } from "./agent/timeline-append.js"; import { - projectTimelineRows, - selectTimelineWindowByProjectedLimit, + selectProjectedTimelinePage, type TimelineProjectionMode, } from "./agent/timeline-projection.js"; import { @@ -7415,84 +7415,33 @@ export class Session { }); } - private loadProjectedTimelineWindow(params: { - agentId: string; - direction: AgentTimelineFetchDirection; - cursor: AgentTimelineCursor | undefined; - requestedLimit: number; - timeline: ReturnType; - }): { - timeline: ReturnType; - selectedRows: ReturnType["selectedRows"]; - minSeq: number | null; - maxSeq: number | null; - } { - const { agentId, direction, cursor, requestedLimit } = params; - let timeline = params.timeline; - const projectedLimit = Math.max(1, Math.floor(requestedLimit)); - let fetchLimit = projectedLimit; - let projectedWindow = selectTimelineWindowByProjectedLimit({ - rows: timeline.rows, - direction, - limit: projectedLimit, - collapseToolLifecycle: false, - }); - - while (timeline.hasOlder) { - const needsMoreProjectedEntries = projectedWindow.projectedEntries.length < projectedLimit; - const firstLoadedRow = timeline.rows[0]; - const firstSelectedRow = projectedWindow.selectedRows[0]; - const startsAtLoadedBoundary = - firstLoadedRow != null && - firstSelectedRow != null && - firstSelectedRow.seq === firstLoadedRow.seq; - const boundaryIsAssistantChunk = - startsAtLoadedBoundary && firstLoadedRow.item.type === "assistant_message"; - - if (!needsMoreProjectedEntries && !boundaryIsAssistantChunk) { - break; - } - - const maxRows = Math.max(0, timeline.window.maxSeq - timeline.window.minSeq + 1); - const nextFetchLimit = Math.min(maxRows, fetchLimit * 2); - if (nextFetchLimit <= fetchLimit) { - break; - } - - fetchLimit = nextFetchLimit; - timeline = this.agentManager.fetchTimeline(agentId, { - direction, - cursor, - limit: fetchLimit, - }); - projectedWindow = selectTimelineWindowByProjectedLimit({ - rows: timeline.rows, - direction, - limit: projectedLimit, - collapseToolLifecycle: false, - }); + private shouldUseFullTimelineForProjectedPage(input: { + timeline: AgentTimelineFetchResult; + }): boolean { + const { timeline } = input; + if (timeline.reset || timeline.rows.length === 0 || !timeline.hasOlder) { + return false; } - return { - timeline, - selectedRows: projectedWindow.selectedRows, - minSeq: projectedWindow.minSeq, - maxSeq: projectedWindow.maxSeq, - }; + const firstRow = timeline.rows[0]; + if ( + firstRow?.item.type === "assistant_message" || + firstRow?.item.type === "reasoning" || + firstRow?.item.type === "tool_call" + ) { + return true; + } + + return timeline.rows.some((row) => row.item.type === "tool_call"); } private async handleFetchAgentTimelineRequest( msg: Extract, ): Promise { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); - const projection: TimelineProjectionMode = msg.projection ?? "projected"; + const projection: TimelineProjectionMode = "projected"; const requestedLimit = msg.limit; - const limit = requestedLimit ?? (direction === "after" ? 0 : undefined); - const shouldLimitByProjectedWindow = - projection === "canonical" && - direction === "tail" && - typeof requestedLimit === "number" && - requestedLimit > 0; + const pageLimit = requestedLimit ?? (direction === "after" ? 0 : 200); const cursor: AgentTimelineCursor | undefined = msg.cursor ? { epoch: msg.cursor.epoch, @@ -7508,43 +7457,29 @@ export class Session { }); const agentPayload = await this.buildAgentPayload(snapshot); - let timeline = this.agentManager.fetchTimeline(msg.agentId, { + const controlTimeline = this.agentManager.fetchTimeline(msg.agentId, { direction, cursor, - limit: - shouldLimitByProjectedWindow && typeof requestedLimit === "number" - ? Math.max(1, Math.floor(requestedLimit)) - : limit, + limit: pageLimit, }); - let hasOlder = timeline.hasOlder; - let hasNewer = timeline.hasNewer; - let startCursor: { epoch: string; seq: number } | null = null; - let endCursor: { epoch: string; seq: number } | null = null; - let entries: ReturnType; - - if (shouldLimitByProjectedWindow) { - const projectedResult = this.loadProjectedTimelineWindow({ - agentId: msg.agentId, - direction, - cursor, - requestedLimit, - timeline, - }); - timeline = projectedResult.timeline; - entries = projectTimelineRows({ rows: projectedResult.selectedRows, mode: projection }); - if (projectedResult.minSeq !== null && projectedResult.maxSeq !== null) { - startCursor = { epoch: timeline.epoch, seq: projectedResult.minSeq }; - endCursor = { epoch: timeline.epoch, seq: projectedResult.maxSeq }; - hasOlder = projectedResult.minSeq > timeline.window.minSeq; - hasNewer = false; - } - } else { - const firstRow = timeline.rows[0]; - const lastRow = timeline.rows[timeline.rows.length - 1]; - startCursor = firstRow ? { epoch: timeline.epoch, seq: firstRow.seq } : null; - endCursor = lastRow ? { epoch: timeline.epoch, seq: lastRow.seq } : null; - entries = projectTimelineRows({ rows: timeline.rows, mode: projection }); - } + const timeline = this.shouldUseFullTimelineForProjectedPage({ + timeline: controlTimeline, + }) + ? this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }) + : controlTimeline; + const projectedPage = selectProjectedTimelinePage({ + rows: timeline.rows, + bounds: timeline.window, + direction: controlTimeline.reset ? "tail" : direction, + ...(cursor ? { cursorSeq: cursor.seq } : {}), + limit: pageLimit, + }); + const startCursor = + projectedPage.startSeq !== null + ? { epoch: timeline.epoch, seq: projectedPage.startSeq } + : null; + const endCursor = + projectedPage.endSeq !== null ? { epoch: timeline.epoch, seq: projectedPage.endSeq } : null; this.emit({ type: "fetch_agent_timeline_response", @@ -7555,15 +7490,17 @@ export class Session { direction, projection, epoch: timeline.epoch, - reset: timeline.reset, - staleCursor: timeline.staleCursor, - gap: timeline.gap, + reset: controlTimeline.reset, + staleCursor: controlTimeline.staleCursor, + gap: controlTimeline.gap, window: timeline.window, startCursor, endCursor, - hasOlder, - hasNewer, - entries: entries.map((entry) => ({ + hasOlder: + projectedPage.hasOlder || + (projectedPage.startSeq !== null && projectedPage.startSeq > timeline.window.minSeq), + hasNewer: projectedPage.hasNewer, + entries: projectedPage.entries.map((entry) => ({ provider: snapshot.provider, item: entry.item, timestamp: entry.timestamp,