Fix chat history paging around tool updates (#1247)

* Fix projected timeline paging

* Avoid full timeline fetches for plain pages
This commit is contained in:
Mohamed Boudra
2026-05-31 05:38:29 +00:00
committed by GitHub
parent 0e3eb9cf75
commit 3d96ea6354
18 changed files with 890 additions and 256 deletions
@@ -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 } } : {}),
});
}
@@ -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",
});
});
});
@@ -147,7 +147,7 @@ describe("loadOlderAgentHistory", () => {
direction: "before",
cursor: { epoch: "epoch-1", seq: 10 },
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "canonical",
projection: "projected",
},
},
]);
@@ -10,7 +10,7 @@ export interface LoadOlderAgentHistoryClient {
direction: "before";
cursor: { epoch: string; seq: number };
limit: number;
projection: "canonical";
projection: "projected";
},
) => Promise<unknown>;
}
@@ -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 } }
: {}),
@@ -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",
@@ -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<string, unknown>;
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,
@@ -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;
@@ -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",
});
});
+19 -19
View File
@@ -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;
}