diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 7c84824618..ed22654c2e 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -78,6 +78,11 @@ items, range, and older-history availability. Live rows received between cache p response stay in the separate live head, do not advance a cursor or trigger gap recovery, and are reconciled with the authoritative tail and subsequent catch-up. +Every daemon-derived live item carries its timeline epoch and sequence position. Bootstrap +replacement keeps only positioned rows newer than the page it installs, while unresolved local +submissions remain governed by the submission registry. This prevents a page from duplicating rows +it already covers without making the display replica authoritative. + ## Selective and legacy delivery The app chooses one delivery policy from `server_info.features.selectiveAgentTimeline`: diff --git a/packages/app/src/agent-stream/strategy-web.test.tsx b/packages/app/src/agent-stream/strategy-web.test.tsx index 9e86b3602a..c9035fc68a 100644 --- a/packages/app/src/agent-stream/strategy-web.test.tsx +++ b/packages/app/src/agent-stream/strategy-web.test.tsx @@ -5,16 +5,11 @@ import React from "react"; import { act } from "react"; import { createRoot, type Root } from "react-dom/client"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import "@/styles/unistyles"; import type { StreamItem } from "@/types/stream"; import type { StreamRenderInput, StreamSegmentRenderers, StreamViewportHandle } from "./strategy"; import { createWebStreamStrategy } from "./strategy-web"; vi.hoisted(() => { - Object.defineProperty(globalThis, "__DEV__", { - configurable: true, - value: false, - }); Object.defineProperty(window, "matchMedia", { configurable: true, value: () => ({ @@ -28,10 +23,6 @@ vi.hoisted(() => { removeListener: () => {}, }), }); - Object.defineProperty(window.screen, "orientation", { - configurable: true, - value: { type: "landscape-primary" }, - }); }); function userMessage(index: number): StreamItem { @@ -379,223 +370,6 @@ describe("createWebStreamStrategy", () => { expect(onNearHistoryStart).not.toHaveBeenCalled(); }); - it("retries when a history callback does not start a request", async () => { - const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); - const viewportRef = React.createRef(); - const onNearHistoryStart = vi.fn(async () => undefined); - const renderInput: StreamRenderInput = { - agentId: "agent", - segments: { - historyVirtualized: [], - historyMounted: [userMessage(1), userMessage(2)], - liveHead: [], - }, - boundary: { - hasVirtualizedHistory: false, - hasMountedHistory: true, - hasLiveHead: false, - }, - renderers: createRenderers(vi.fn()), - listEmptyComponent: null, - viewportRef, - routeBottomAnchorRequest: null, - isAuthoritativeHistoryReady: true, - onNearBottomChange: vi.fn(), - onNearHistoryStart: onNearHistoryStart as unknown as StreamRenderInput["onNearHistoryStart"], - isLoadingOlderHistory: false, - hasOlderHistory: true, - olderHistoryProgressKey: "epoch-1:20", - scrollEnabled: true, - listStyle: null, - baseListContentContainerStyle: null, - forwardListContentContainerStyle: null, - }; - container = document.createElement("div"); - document.body.appendChild(container); - root = createRoot(container); - - act(() => { - root?.render(strategy.render(renderInput)); - }); - const scrollContainer = container.querySelector('[data-testid="agent-chat-scroll"]'); - if (!(scrollContainer instanceof HTMLElement)) { - throw new Error("Expected agent chat scroll container"); - } - Object.defineProperty(scrollContainer, "clientHeight", { configurable: true, value: 1000 }); - Object.defineProperty(scrollContainer, "scrollHeight", { configurable: true, value: 500 }); - Object.defineProperty(scrollContainer, "scrollTop", { configurable: true, value: 0 }); - await act(async () => { - await new Promise((resolve) => requestAnimationFrame(resolve)); - await Promise.resolve(); - }); - expect(onNearHistoryStart).toHaveBeenCalledTimes(1); - - await act(async () => { - scrollContainer.dispatchEvent(new WheelEvent("wheel", { deltaY: -120 })); - await Promise.resolve(); - }); - expect(onNearHistoryStart).toHaveBeenCalledTimes(2); - }); - - it("loads older history when the current page projects no anchor row", async () => { - const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); - const viewportRef = React.createRef(); - const onNearHistoryStart = vi.fn().mockReturnValue(true); - const renderInput: StreamRenderInput = { - agentId: "agent", - segments: { - historyVirtualized: [], - historyMounted: [], - liveHead: [], - }, - boundary: { - hasVirtualizedHistory: false, - hasMountedHistory: false, - hasLiveHead: false, - }, - renderers: createRenderers(vi.fn()), - listEmptyComponent: null, - viewportRef, - routeBottomAnchorRequest: null, - isAuthoritativeHistoryReady: true, - onNearBottomChange: vi.fn(), - onNearHistoryStart, - isLoadingOlderHistory: false, - hasOlderHistory: true, - olderHistoryProgressKey: "epoch-1:20", - scrollEnabled: true, - listStyle: null, - baseListContentContainerStyle: null, - forwardListContentContainerStyle: null, - }; - container = document.createElement("div"); - document.body.appendChild(container); - root = createRoot(container); - - act(() => { - root?.render(strategy.render(renderInput)); - }); - const scrollContainer = container.querySelector('[data-testid="agent-chat-scroll"]'); - if (!(scrollContainer instanceof HTMLElement)) { - throw new Error("Expected agent chat scroll container"); - } - Object.defineProperty(scrollContainer, "clientHeight", { configurable: true, value: 1000 }); - Object.defineProperty(scrollContainer, "scrollHeight", { configurable: true, value: 500 }); - Object.defineProperty(scrollContainer, "scrollTop", { configurable: true, value: 0 }); - - await act(async () => { - await new Promise((resolve) => requestAnimationFrame(resolve)); - await Promise.resolve(); - }); - - expect(onNearHistoryStart).toHaveBeenCalledTimes(1); - }); - - it("anchors prepended history independently from live content growth below it", async () => { - let anchorTop = 0; - let measuredScrollTop = 0; - vi.spyOn(HTMLElement.prototype, "getBoundingClientRect").mockImplementation( - function (this: HTMLElement) { - const top = this.dataset.historyRowId === "message-79" ? anchorTop - measuredScrollTop : 0; - return { - x: 0, - y: top, - top, - right: 0, - bottom: top + 24, - left: 0, - width: 0, - height: 24, - toJSON: () => ({}), - }; - }, - ); - const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); - const viewportRef = React.createRef(); - const onNearHistoryStart = vi.fn().mockReturnValue(true); - const initialHistory = Array.from({ length: 20 }, (_, index) => userMessage(index + 60)); - const renderInput: StreamRenderInput = { - agentId: "agent", - segments: { historyVirtualized: [], historyMounted: initialHistory, liveHead: [] }, - boundary: { - hasVirtualizedHistory: false, - hasMountedHistory: true, - hasLiveHead: false, - }, - renderers: createRenderers(vi.fn()), - listEmptyComponent: null, - viewportRef, - routeBottomAnchorRequest: null, - isAuthoritativeHistoryReady: true, - onNearBottomChange: vi.fn(), - onNearHistoryStart, - isLoadingOlderHistory: false, - hasOlderHistory: true, - olderHistoryProgressKey: "epoch-1:60", - scrollEnabled: true, - listStyle: null, - baseListContentContainerStyle: null, - forwardListContentContainerStyle: null, - }; - container = document.createElement("div"); - document.body.appendChild(container); - root = createRoot(container); - act(() => { - root?.render(strategy.render(renderInput)); - }); - const scrollContainer = container.querySelector('[data-testid="agent-chat-scroll"]'); - if (!(scrollContainer instanceof HTMLElement)) { - throw new Error("Expected agent chat scroll container"); - } - let scrollHeight = 1000; - let scrollTop = 0; - Object.defineProperties(scrollContainer, { - clientHeight: { configurable: true, value: 400 }, - scrollHeight: { configurable: true, get: () => scrollHeight }, - scrollTop: { - configurable: true, - get: () => scrollTop, - set: (value: number) => { - scrollTop = value; - measuredScrollTop = value; - }, - }, - }); - await act(async () => { - await new Promise((resolve) => requestAnimationFrame(resolve)); - }); - act(() => { - scrollContainer.dispatchEvent(new WheelEvent("wheel", { deltaY: -120 })); - scrollContainer.dispatchEvent(new Event("scroll")); - }); - expect(onNearHistoryStart).toHaveBeenCalledTimes(1); - - anchorTop = 600; - scrollHeight = 1700; - const prependedHistory = Array.from({ length: 40 }, (_, index) => userMessage(index + 40)); - act(() => { - root?.render( - strategy.render({ - ...renderInput, - segments: { - historyVirtualized: [], - historyMounted: [...prependedHistory, ...initialHistory], - liveHead: [userMessage(80)], - }, - boundary: { - hasVirtualizedHistory: false, - hasMountedHistory: true, - hasLiveHead: true, - }, - isLoadingOlderHistory: true, - olderHistoryProgressKey: "epoch-1:40", - }), - ); - }); - - expect(scrollTop).toBe(600); - }); - it("keeps initial route entry anchored when delayed route readiness arrives before user scroll", async () => { const scrollTo = vi.fn(function ( this: HTMLElement, diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index d2bf026e11..1fad0845ad 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -468,6 +468,37 @@ describe("processTimelineResponse", () => { expect(getUserTexts(result.head)).toEqual(["remote live prompt"]); }); + it("does not duplicate a live row already covered by the bootstrap page", () => { + const live = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("thinking", "reasoning"), + seq: 100, + epoch: "epoch-1", + currentTail: [makeAssistantItem("painted replica")], + hasAuthoritativeBaseline: false, + }); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + isInitializing: true, + hasActiveInitDeferred: true, + hasAuthoritativeBaseline: false, + payload: { + ...baseTimelineInput.payload, + direction: "tail", + startCursor: { seq: 61 }, + endCursor: { seq: 100 }, + entries: [makeTimelineEntry(100, "thinking", "reasoning")], + }, + }); + + expect([...result.tail, ...result.head].filter((item) => item.kind === "thought")).toHaveLength( + 1, + ); + }); + it("uses the timeline entry timestamp as canonical", () => { const result = processTimelineResponse({ ...baseTimelineInput, @@ -2545,6 +2576,39 @@ describe("processAgentStreamEvent", () => { expect(result.cursor).toBeNull(); expect(result.sideEffects).toEqual([]); }); + + it("reconciles a pre-bootstrap echo with its submitted row", () => { + const submitted = makeSubmittedUserMessage("submitted before bootstrap", "client-message-1"); + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: { + type: "timeline", + provider: "claude", + item: { + type: "user_message", + text: "canonical presentation", + clientMessageId: "client-message-1", + messageId: "provider-message-1", + }, + }, + seq: 51, + epoch: "epoch-1", + currentTail: [makeAssistantItem("painted replica"), submitted], + currentCursor: undefined, + hasAuthoritativeBaseline: false, + }); + + const users = [...result.tail, ...result.head].filter((item) => item.kind === "user_message"); + expect(users).toEqual([ + expect.objectContaining({ + id: "client-message-1", + clientMessageId: "client-message-1", + messageId: "provider-message-1", + text: "submitted before bootstrap", + timelineCursor: { epoch: "epoch-1", seq: 51 }, + }), + ]); + }); }); describe("processAgentStreamEvents", () => { diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 252afe98be..8792cfffce 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -1132,20 +1132,32 @@ export function processAgentStreamEvent( changedHead: false, }; } else if (!hasAuthoritativeBaseline) { - const overlay = applyStreamEvent({ - tail: currentHead, - head: [], - event, - timestamp, - source: "live", - timelineCursor, - }); - streamResult = { - tail: currentTail, - head: [...overlay.tail, ...overlay.head], - changedTail: false, - changedHead: overlay.changedTail || overlay.changedHead, - }; + if (event.type === "timeline" && event.item.type === "user_message") { + streamResult = applyStreamEvent({ + tail: currentTail, + head: currentHead, + event, + timestamp, + source: "live", + timelineCursor, + unmatchedUserMessageInsert: "head", + }); + } else { + const overlay = applyStreamEvent({ + tail: currentHead, + head: [], + event, + timestamp, + source: "live", + timelineCursor, + }); + streamResult = { + tail: currentTail, + head: [...overlay.tail, ...overlay.head], + changedTail: false, + changedHead: overlay.changedTail || overlay.changedHead, + }; + } } else { streamResult = applyStreamEvent({ tail: sequencing.resetLiveTimeline ? [] : currentTail, diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index a46a6e31f6..5f47bdd707 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -415,9 +415,20 @@ function preserveReplacementHead( currentHead: StreamItem[], preserveLiveHead: boolean, sendingClientMessageIds: ReadonlySet, + canonicalCoverage: CanonicalStreamReplacementInput["canonicalCoverage"], ): CanonicalStreamReplacementResult { + const canonicalTailAssistant = tail.at(-1); const retainedHead = preserveLiveHead - ? currentHead + ? currentHead.filter( + (item) => + !item.timelineCursor || + (item.timelineCursor.epoch === canonicalCoverage.epoch && + (canonicalCoverage.endSeq === null || + item.timelineCursor.seq > canonicalCoverage.endSeq)) || + (item.kind === "assistant_message" && + canonicalTailAssistant?.kind === "assistant_message" && + item.text.startsWith(canonicalTailAssistant.text)), + ) : currentHead.filter( (item) => item.kind === "user_message" && @@ -548,6 +559,7 @@ export function replaceWithCanonicalStream( nextHead, input.preserveLiveHead, sendingClientMessageIds, + input.canonicalCoverage, ); return { ...replacement, @@ -576,6 +588,7 @@ export type ThoughtStatus = "loading" | "ready"; export interface ThoughtItem { kind: "thought"; id: string; + timelineCursor?: TimelinePosition; text: string; timestamp: Date; status: ThoughtStatus; @@ -610,6 +623,7 @@ export type ToolCallPayload = export interface ToolCallItem { kind: "tool_call"; id: string; + timelineCursor?: TimelinePosition; timestamp: Date; payload: ToolCallPayload; } @@ -627,6 +641,7 @@ type ActivityLogType = "system" | "info" | "success" | "error"; export interface ActivityLogItem { kind: "activity_log"; id: string; + timelineCursor?: TimelinePosition; timestamp: Date; activityType: ActivityLogType; message: string; @@ -636,6 +651,7 @@ export interface ActivityLogItem { export interface CompactionItem { kind: "compaction"; id: string; + timelineCursor?: TimelinePosition; timestamp: Date; status: "loading" | "completed"; trigger?: "auto" | "manual"; @@ -650,6 +666,7 @@ export interface TodoEntry { export interface TodoListItem { kind: "todo_list"; id: string; + timelineCursor?: TimelinePosition; timestamp: Date; provider: AgentProvider; items: TodoEntry[]; @@ -791,7 +808,12 @@ function appendAssistantMessage( return [...state, item]; } -function appendThought(state: StreamItem[], text: string, timestamp: Date): StreamItem[] { +function appendThought( + state: StreamItem[], + text: string, + timestamp: Date, + timelineCursor?: TimelinePosition, +): StreamItem[] { const { chunk, hasContent } = normalizeChunk(text); if (!chunk) { return state; @@ -801,6 +823,7 @@ function appendThought(state: StreamItem[], text: string, timestamp: Date): Stre if (last && last.kind === "thought") { const updated: ThoughtItem = { ...last, + ...(timelineCursor ? { timelineCursor } : {}), text: `${last.text}${chunk}`, timestamp, status: "loading", @@ -816,6 +839,7 @@ function appendThought(state: StreamItem[], text: string, timestamp: Date): Stre const item: ThoughtItem = { kind: "thought", id: createUniqueTimelineId(state, "thought", idSeed, timestamp), + ...(timelineCursor ? { timelineCursor } : {}), text: chunk, timestamp, status: "loading", @@ -951,6 +975,7 @@ export function mergeAgentToolCallItem( existing: AgentToolCallItem, data: AgentToolCallData, timestamp: Date, + timelineCursor?: TimelinePosition, ): AgentToolCallItem { const mergedStatus = mergeAgentToolCallStatus(existing.payload.data.status, data.status); const mergedError = @@ -962,6 +987,7 @@ export function mergeAgentToolCallItem( return { ...existing, + ...(timelineCursor ? { timelineCursor } : {}), timestamp, payload: { source: "agent", @@ -981,6 +1007,7 @@ function appendAgentToolCall( state: StreamItem[], data: AgentToolCallData, timestamp: Date, + timelineCursor?: TimelinePosition, ): StreamItem[] { const existingIndex = findExistingAgentToolCallIndex(state, data.callId); @@ -989,7 +1016,7 @@ function appendAgentToolCall( if (!existing || !isAgentToolCallItem(existing)) { return state; } - const merged = mergeAgentToolCallItem(existing, data, timestamp); + const merged = mergeAgentToolCallItem(existing, data, timestamp, timelineCursor); if ( merged.payload.data.provider === existing.payload.data.provider && @@ -998,7 +1025,8 @@ function appendAgentToolCall( merged.payload.data.status === existing.payload.data.status && merged.payload.data.error === existing.payload.data.error && merged.payload.data.detail === existing.payload.data.detail && - merged.payload.data.metadata === existing.payload.data.metadata + merged.payload.data.metadata === existing.payload.data.metadata && + merged.timelineCursor === existing.timelineCursor ) { return state; } @@ -1011,6 +1039,7 @@ function appendAgentToolCall( const item: ToolCallItem = { kind: "tool_call", id: `agent_tool_${data.callId}`, + ...(timelineCursor ? { timelineCursor } : {}), timestamp, payload: { source: "agent", @@ -1039,6 +1068,7 @@ function appendTodoList( provider: AgentProvider, items: TodoEntry[], timestamp: Date, + timelineCursor?: TimelinePosition, ): StreamItem[] { const normalizedItems = items.map((item) => ({ text: item.text, @@ -1050,6 +1080,7 @@ function appendTodoList( const next = [...state]; const updated: TodoListItem = { ...lastItem, + ...(timelineCursor ? { timelineCursor } : {}), items: normalizedItems, timestamp, }; @@ -1063,6 +1094,7 @@ function appendTodoList( const entry: TodoListItem = { kind: "todo_list", id: entryId, + ...(timelineCursor ? { timelineCursor } : {}), timestamp, provider, items: normalizedItems, @@ -1079,6 +1111,7 @@ function reduceTimelineToolCall( { type: "tool_call" } >, timestamp: Date, + timelineCursor?: TimelinePosition, ): StreamItem[] { const normalizedToolName = item.name .trim() @@ -1101,6 +1134,7 @@ function reduceTimelineToolCall( event.provider, tasks.map((entry) => ({ text: entry.text, completed: entry.completed })), timestamp, + timelineCursor, ); } @@ -1111,6 +1145,7 @@ function reduceTimelineToolCall( event.provider, tasks.map((entry) => ({ text: entry.text, completed: entry.completed })), timestamp, + timelineCursor, ); } @@ -1126,6 +1161,7 @@ function reduceTimelineToolCall( metadata: item.metadata, }, timestamp, + timelineCursor, ); } @@ -1136,6 +1172,7 @@ function reduceTimelineCompaction( { type: "compaction" } >, timestamp: Date, + timelineCursor?: TimelinePosition, ): StreamItem[] { if (item.status === "completed") { const loadingIdx = state.findIndex((s) => s.kind === "compaction" && s.status === "loading"); @@ -1143,6 +1180,7 @@ function reduceTimelineCompaction( if (loadingIdx >= 0 && existing && existing.kind === "compaction") { const updated: CompactionItem = { ...existing, + ...(timelineCursor ? { timelineCursor } : {}), status: "completed", trigger: item.trigger ?? existing.trigger, preTokens: item.preTokens ?? existing.preTokens, @@ -1156,6 +1194,7 @@ function reduceTimelineCompaction( const compaction: CompactionItem = { kind: "compaction", id: createTimelineId("compaction", item.status, timestamp), + ...(timelineCursor ? { timelineCursor } : {}), timestamp, status: item.status, trigger: item.trigger, @@ -1199,9 +1238,11 @@ function reduceTimelineEvent( ), ); case "reasoning": - return appendThought(state, item.text, timestamp); + return appendThought(state, item.text, timestamp, timelineCursor); case "tool_call": - return finalizeActiveThoughts(reduceTimelineToolCall(state, event, item, timestamp)); + return finalizeActiveThoughts( + reduceTimelineToolCall(state, event, item, timestamp, timelineCursor), + ); case "todo": { if (event.provider === "claude") { return finalizeActiveThoughts(state); @@ -1210,12 +1251,15 @@ function reduceTimelineEvent( text: todo.text, completed: todo.completed, })); - return finalizeActiveThoughts(appendTodoList(state, event.provider, items, timestamp)); + return finalizeActiveThoughts( + appendTodoList(state, event.provider, items, timestamp, timelineCursor), + ); } case "error": { const activity: ActivityLogItem = { kind: "activity_log", id: createTimelineId("error", item.message ?? "", timestamp), + ...(timelineCursor ? { timelineCursor } : {}), timestamp, activityType: "error", message: item.message ?? "Unknown error", @@ -1223,7 +1267,9 @@ function reduceTimelineEvent( return finalizeActiveThoughts(appendActivityLog(state, activity)); } case "compaction": - return finalizeActiveThoughts(reduceTimelineCompaction(state, item, timestamp)); + return finalizeActiveThoughts( + reduceTimelineCompaction(state, item, timestamp, timelineCursor), + ); default: return state; } @@ -1547,8 +1593,9 @@ function applyCanonicalUserMessageEvent(params: { event: AgentStreamEventPayload; timestamp: Date; timelineCursor?: TimelinePosition; + unmatchedInsert?: "tail" | "head"; }): ApplyStreamEventResult | null { - const { tail, head, event, timestamp, timelineCursor } = params; + const { tail, head, event, timestamp, timelineCursor, unmatchedInsert = "tail" } = params; if (event.type !== "timeline" || event.item.type !== "user_message") return null; const normalized = normalizeChunk(event.item.text); @@ -1564,6 +1611,25 @@ function applyCanonicalUserMessageEvent(params: { text: normalized.chunk, timestamp, }); + if (unmatchedInsert === "head") { + const reconciled = upsertUserMessageAcrossStream({ + tail, + head, + message: canonical, + insert: normalized.hasContent ? "head" : "none", + presentation: "existing", + }); + return { + tail: reconciled.tail, + head: reconciled.head, + changedTail: reconciled.changedTail, + changedHead: reconciled.changedHead, + acknowledgedClientMessageIds: + reconciled.location?.matched && reconciled.location.message.clientMessageId + ? [reconciled.location.message.clientMessageId] + : [], + }; + } const reconciled = placeCanonicalUserMessageAtTail(flushedTail, canonical, normalized.hasContent); return { tail: reconciled.items, @@ -1595,6 +1661,7 @@ export function applyStreamEvent(params: { timestamp: Date; source?: StreamUpdateSource; timelineCursor?: TimelinePosition; + unmatchedUserMessageInsert?: "tail" | "head"; }): ApplyStreamEventResult { const { tail, head, event, timestamp } = params; const canonicalUserResult = applyCanonicalUserMessageEvent({ @@ -1603,6 +1670,7 @@ export function applyStreamEvent(params: { event, timestamp, timelineCursor: params.timelineCursor, + unmatchedInsert: params.unmatchedUserMessageInsert, }); if (canonicalUserResult) return canonicalUserResult; const source = params.source ?? "live";