From 05c2869e2ce8b9da1e9cfd7ff0b0380822331c64 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 27 Jul 2026 21:57:10 +0200 Subject: [PATCH] Integrate timeline hydration with submission authority --- .../src/agent-stream/strategy-web.test.tsx | 9 ++ packages/app/src/contexts/session-context.tsx | 7 +- packages/app/src/stores/session-store.test.ts | 2 + .../timeline/session-stream-reducers.test.ts | 128 ++++++++++++++++++ .../src/timeline/session-stream-reducers.ts | 63 ++------- packages/app/src/types/stream.test.ts | 6 +- packages/app/src/types/stream.ts | 54 +++++++- 7 files changed, 213 insertions(+), 56 deletions(-) diff --git a/packages/app/src/agent-stream/strategy-web.test.tsx b/packages/app/src/agent-stream/strategy-web.test.tsx index 0204db41eb..9e86b3602a 100644 --- a/packages/app/src/agent-stream/strategy-web.test.tsx +++ b/packages/app/src/agent-stream/strategy-web.test.tsx @@ -5,11 +5,16 @@ 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: () => ({ @@ -23,6 +28,10 @@ vi.hoisted(() => { removeListener: () => {}, }), }); + Object.defineProperty(window.screen, "orientation", { + configurable: true, + value: { type: "landscape-primary" }, + }); }); function userMessage(index: number): StreamItem { diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index f92322c17c..de18f551f5 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -575,8 +575,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const isInitializing = session?.initializingAgents.get(agentId) === true; const activeInitDeferred = getInitDeferred(initKey); const hasActiveInitDeferred = Boolean(activeInitDeferred); - const currentCursor = session?.agentTimelineCursor.get(agentId); - const currentTail = session?.agentStreamTail.get(agentId) ?? []; + const timeline = selectAgentTimelineState(session, agentId); + const currentCursor = + timeline.status === "synced" ? (timeline.range ?? undefined) : undefined; + const currentTail = timeline.status === "cold" ? [] : timeline.items; const currentHead = session?.agentStreamHead.get(agentId) ?? []; const sendingClientMessageIds = getSendingClientMessageIds( session?.messageSubmissions.get(agentId), @@ -592,6 +594,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider hasActiveInitDeferred, initRequestDirection: activeInitDeferred?.requestDirection ?? "tail", sendingClientMessageIds, + hasAuthoritativeBaseline: timeline.status === "synced", }); if (result.error) { diff --git a/packages/app/src/stores/session-store.test.ts b/packages/app/src/stores/session-store.test.ts index cd386660fe..369097d267 100644 --- a/packages/app/src/stores/session-store.test.ts +++ b/packages/app/src/stores/session-store.test.ts @@ -73,6 +73,7 @@ describe("agent timeline state", () => { range: { epoch: "epoch-1", startSeq: 51, endSeq: 100 }, older: "available", synchronized: true, + acknowledgedClientMessageIds: [], }); expect( @@ -93,6 +94,7 @@ describe("agent timeline state", () => { range: null, older: "none", synchronized: true, + acknowledgedClientMessageIds: [], }); expect( diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 10eed0188c..d2bf026e11 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -173,6 +173,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = { hasActiveInitDeferred: false, initRequestDirection: "tail", sendingClientMessageIds: [], + hasAuthoritativeBaseline: true, }; const baseStreamInput: ProcessAgentStreamEventInput = { @@ -362,6 +363,111 @@ describe("processTimelineResponse", () => { expect(result.head).toEqual([liveThought, liveAssistant]); }); + it("replaces a painted replica with an empty authoritative timeline", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [makeAssistantItem("painted replica")], + isInitializing: true, + hasActiveInitDeferred: true, + hasAuthoritativeBaseline: false, + payload: { + ...baseTimelineInput.payload, + direction: "tail", + startCursor: null, + endCursor: null, + entries: [], + hasOlder: false, + }, + }); + + expect(result.tail).toEqual([]); + expect(result.cursor).toBeNull(); + }); + + it("keeps a newer live assistant continuation when bootstrap ends on the same message", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [makeAssistantItem("painted replica")], + currentHead: [ + { + kind: "assistant_message", + id: "assistant-live", + messageId: "assistant-live", + text: " continuation", + timestamp: new Date(2001), + timelineCursor: { epoch: "epoch-1", seq: 101 }, + }, + ], + isInitializing: true, + hasActiveInitDeferred: true, + hasAuthoritativeBaseline: false, + payload: { + ...baseTimelineInput.payload, + direction: "tail", + startCursor: { seq: 61 }, + endCursor: { seq: 100 }, + entries: [ + { + ...makeTimelineEntry(100, "canonical prefix"), + item: { + type: "assistant_message", + text: "canonical prefix", + messageId: "assistant-live", + }, + }, + ], + }, + }); + + const assistants = [...result.tail, ...result.head].filter( + (item) => item.kind === "assistant_message", + ); + expect(assistants).toHaveLength(1); + expect(assistants[0]?.text).toBe("canonical prefix continuation"); + expect(assistants[0]?.timelineCursor).toEqual({ epoch: "epoch-1", seq: 101 }); + }); + + it("keeps a canonical live user row newer than the bootstrap page", () => { + const live = processAgentStreamEvent({ + ...baseStreamInput, + event: { + type: "timeline", + provider: "claude", + item: { + type: "user_message", + text: "remote live prompt", + clientMessageId: "remote-client-message", + }, + }, + seq: 101, + epoch: "epoch-1", + currentTail: [makeAssistantItem("painted replica")], + hasAuthoritativeBaseline: false, + }); + expect(live.head[0]).toMatchObject({ + kind: "user_message", + timelineCursor: { epoch: "epoch-1", seq: 101 }, + }); + + 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, "canonical answer")], + }, + }); + + expect(getUserTexts(result.head)).toEqual(["remote live prompt"]); + }); + it("uses the timeline entry timestamp as canonical", () => { const result = processTimelineResponse({ ...baseTimelineInput, @@ -2417,6 +2523,28 @@ describe("processAgentStreamEvent", () => { endSeq: 1, }); }); + + it("renders a live row over painted items without creating an authoritative cursor", () => { + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("live before bootstrap", "user_message"), + seq: 51, + epoch: "epoch-1", + currentTail: [makeAssistantItem("painted replica")], + currentCursor: undefined, + hasAuthoritativeBaseline: false, + }); + + expect(getAssistantTexts(result.tail)).toEqual(["painted replica"]); + expect(getUserTexts(result.head)).toEqual(["live before bootstrap"]); + expect(result.head[0]).toMatchObject({ + kind: "user_message", + timelineCursor: { epoch: "epoch-1", seq: 51 }, + }); + expect(result.cursorChanged).toBe(false); + expect(result.cursor).toBeNull(); + expect(result.sideEffects).toEqual([]); + }); }); describe("processAgentStreamEvents", () => { diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 57b75107ca..252afe98be 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -89,6 +89,7 @@ export interface ProcessTimelineResponseInput { hasActiveInitDeferred: boolean; initRequestDirection: InitRequestDirection; sendingClientMessageIds: readonly string[]; + hasAuthoritativeBaseline: boolean; } export interface ProcessTimelineResponseOutput { @@ -233,6 +234,10 @@ function applyTimelineReplacePath(args: { previousHead: currentHead, sendingClientMessageIds, preserveLiveHead, + canonicalCoverage: { + epoch: payload.epoch, + endSeq: payload.endCursor?.seq ?? null, + }, }); const cursor: TimelineCursor | null = payload.startCursor && payload.endCursor @@ -574,43 +579,6 @@ function reconcileOverlappingProjectedReasoning(params: { : { tail: params.tail, head: params.head, reconciled: false }; } -function reconcileProjectedUserOverlay(params: { - tail: StreamItem[]; - head: StreamItem[]; - unit: TimelineUnit; -}): { tail: StreamItem[]; head: StreamItem[]; reconciled: boolean } { - const { event } = params.unit; - if (event.type !== "timeline" || event.item.type !== "user_message") { - return { tail: params.tail, head: params.head, reconciled: false }; - } - const userItem = event.item; - const overlayIndex = params.head.findIndex( - (item) => - item.kind === "user_message" && - matchesLocalUserMessageIdentity( - { - messageId: userItem.messageId, - clientMessageId: userItem.clientMessageId, - text: userItem.text, - }, - item, - ), - ); - const overlay = params.head[overlayIndex]; - if (overlayIndex < 0 || !overlay || overlay.kind !== "user_message") { - return { tail: params.tail, head: params.head, reconciled: false }; - } - const canonical = reduceStreamUpdate([], event, params.unit.timestamp, { - source: "canonical", - })[0]; - if (!canonical || canonical.kind !== "user_message") { - return { tail: params.tail, head: params.head, reconciled: false }; - } - const head = [...params.head]; - head[overlayIndex] = mergeCanonicalUserWithLocalPresentation(canonical, overlay); - return { tail: params.tail, head, reconciled: true }; -} - function reconcileOverlappingProjectedStreamItems(params: { tail: StreamItem[]; head: StreamItem[]; @@ -624,16 +592,13 @@ function reconcileOverlappingProjectedStreamItems(params: { if (params.currentEndSeq === undefined) return { tail, head, reconciledUnits }; for (const unit of params.units) { - let reconciled = reconcileProjectedUserOverlay({ tail, head, unit }); - if (!reconciled.reconciled) { - reconciled = reconcileOverlappingProjectedAssistant({ - tail, - head, - unit, - epoch: params.epoch, - currentEndSeq: params.currentEndSeq, - }); - } + let reconciled = reconcileOverlappingProjectedAssistant({ + tail, + head, + unit, + epoch: params.epoch, + currentEndSeq: params.currentEndSeq, + }); if (!reconciled.reconciled) { reconciled = reconcileOverlappingProjectedReasoning({ tail, @@ -859,6 +824,7 @@ export function processTimelineResponse( hasActiveInitDeferred, initRequestDirection, sendingClientMessageIds, + hasAuthoritativeBaseline, } = input; // ------------------------------------------------------------------ @@ -930,7 +896,8 @@ export function processTimelineResponse( currentTail, currentHead, sendingClientMessageIds, - preserveLiveHead: currentCursor?.epoch === payload.epoch, + preserveLiveHead: + currentCursor?.epoch === payload.epoch || (!payload.reset && !hasAuthoritativeBaseline), toHydratedEvents, }) : applyTimelineIncrementalPath({ diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 00d73b7f65..e3de58f420 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -1350,7 +1350,10 @@ describe("turn lifecycle events", () => { }, }, new Date("2025-01-01T15:03:11Z"), - { source: "canonical" }, + { + source: "canonical", + timelineCursor: { epoch: "epoch-1", seq: 42 }, + }, ); const userMessages = state.filter((item) => item.kind === "user_message"); @@ -1361,6 +1364,7 @@ describe("turn lifecycle events", () => { assert.strictEqual(userMessage.messageId, "provider-owned-canonical"); assert.strictEqual(userMessage.text, "Analyze this"); assert.strictEqual(userMessage.timestamp.getTime(), submittedTimestamp.getTime()); + assert.deepStrictEqual(userMessage.timelineCursor, { epoch: "epoch-1", seq: 42 }); assert.deepStrictEqual(userMessage.images, [image]); assert.deepStrictEqual(userMessage.attachments, [attachment]); }); diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 8b66ab80bd..a46a6e31f6 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -88,6 +88,7 @@ export interface UserMessageItem { id: string; clientMessageId?: string; messageId?: string; + timelineCursor?: TimelinePosition; text: string; timestamp: Date; images?: UserMessageImageAttachment[]; @@ -98,6 +99,7 @@ export interface UserMessageInput { id?: string; clientMessageId?: string; messageId?: string; + timelineCursor?: TimelinePosition; text: string; timestamp: Date; images?: UserMessageImageAttachment[]; @@ -114,6 +116,7 @@ export function createUserMessage(input: UserMessageInput): UserMessageItem { id, ...(input.clientMessageId ? { clientMessageId: input.clientMessageId } : {}), ...(input.messageId ? { messageId: input.messageId } : {}), + ...(input.timelineCursor ? { timelineCursor: input.timelineCursor } : {}), text: input.text, timestamp: input.timestamp, ...(input.images && input.images.length > 0 ? { images: input.images } : {}), @@ -236,11 +239,13 @@ function produceUserMessage( ...presentation, clientMessageId: incoming.clientMessageId ?? existing.clientMessageId, messageId: incoming.messageId ?? existing.messageId, + timelineCursor: incoming.timelineCursor ?? existing.timelineCursor, }); if ( existing.id === merged.id && existing.clientMessageId === merged.clientMessageId && existing.messageId === merged.messageId && + existing.timelineCursor === merged.timelineCursor && existing.text === merged.text && existing.timestamp === merged.timestamp && existing.images === merged.images && @@ -392,6 +397,7 @@ export interface CanonicalStreamReplacementInput { previousHead: StreamItem[]; sendingClientMessageIds: readonly string[]; preserveLiveHead: boolean; + canonicalCoverage: { epoch: string; endSeq: number | null }; } export interface CanonicalStreamReplacementResult { @@ -434,12 +440,33 @@ function preserveReplacementHead( if ( liveAssistant.kind !== "assistant_message" || !tailAssistant || - tailAssistant.kind !== "assistant_message" || - !liveAssistant.text.startsWith(tailAssistant.text) + tailAssistant.kind !== "assistant_message" ) { return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] }; } + const isNewerContinuation = + liveAssistant.messageId !== undefined && + liveAssistant.messageId === tailAssistant.messageId && + liveAssistant.timelineCursor !== undefined && + tailAssistant.timelineCursor !== undefined && + liveAssistant.timelineCursor.epoch === tailAssistant.timelineCursor.epoch && + liveAssistant.timelineCursor.seq > tailAssistant.timelineCursor.seq; + if (isNewerContinuation) { + const text = liveAssistant.text.startsWith(tailAssistant.text) + ? liveAssistant.text + : `${tailAssistant.text}${liveAssistant.text}`; + const head = [ + ...unreconciledHead.slice(0, liveAssistantIndex), + { ...liveAssistant, text }, + ...unreconciledHead.slice(liveAssistantIndex + 1), + ]; + return { tail: tail.slice(0, -1), head, acknowledgedClientMessageIds: [] }; + } + if (!liveAssistant.text.startsWith(tailAssistant.text)) { + return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] }; + } + const head = [ ...unreconciledHead.slice(0, liveAssistantIndex), { ...liveAssistant, text: tailAssistant.text }, @@ -507,7 +534,13 @@ export function replaceWithCanonicalStream( nextHead = nextHead.filter((item) => { if (item.kind !== "user_message" || !item.clientMessageId) return true; - return sendingClientMessageIds.has(item.clientMessageId); + if (sendingClientMessageIds.has(item.clientMessageId)) return true; + if (!input.preserveLiveHead || !item.timelineCursor) return false; + return ( + item.timelineCursor.epoch === input.canonicalCoverage.epoch && + (input.canonicalCoverage.endSeq === null || + item.timelineCursor.seq > input.canonicalCoverage.endSeq) + ); }); const replacement = preserveReplacementHead( @@ -675,6 +708,7 @@ function appendUserMessage( _source: StreamUpdateSource, messageId?: string, clientMessageId?: string, + timelineCursor?: TimelinePosition, ): StreamItem[] { const { chunk, hasContent } = normalizeChunk(text); if (!hasContent) { @@ -686,6 +720,7 @@ function appendUserMessage( id: messageId ?? createUniqueTimelineId(state, "user", chunkSeed, timestamp), clientMessageId, messageId, + timelineCursor, text: chunk, timestamp, }); @@ -1148,6 +1183,7 @@ function reduceTimelineEvent( source, item.messageId, item.clientMessageId, + timelineCursor, ), ); case "assistant_message": @@ -1510,8 +1546,9 @@ function applyCanonicalUserMessageEvent(params: { head: StreamItem[]; event: AgentStreamEventPayload; timestamp: Date; + timelineCursor?: TimelinePosition; }): ApplyStreamEventResult | null { - const { tail, head, event, timestamp } = params; + const { tail, head, event, timestamp, timelineCursor } = params; if (event.type !== "timeline" || event.item.type !== "user_message") return null; const normalized = normalizeChunk(event.item.text); @@ -1523,6 +1560,7 @@ function applyCanonicalUserMessageEvent(params: { createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp), messageId: event.item.messageId, clientMessageId: event.item.clientMessageId, + timelineCursor, text: normalized.chunk, timestamp, }); @@ -1559,7 +1597,13 @@ export function applyStreamEvent(params: { timelineCursor?: TimelinePosition; }): ApplyStreamEventResult { const { tail, head, event, timestamp } = params; - const canonicalUserResult = applyCanonicalUserMessageEvent({ tail, head, event, timestamp }); + const canonicalUserResult = applyCanonicalUserMessageEvent({ + tail, + head, + event, + timestamp, + timelineCursor: params.timelineCursor, + }); if (canonicalUserResult) return canonicalUserResult; const source = params.source ?? "live"; let nextTail = tail;