Harden pre-hydration timeline reconciliation

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`:

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<StreamViewportHandle>();
-    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<StreamViewportHandle>();
-    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<StreamViewportHandle>();
-    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,

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", () => {

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,

packages/app/src/types/stream.ts

@@ -415,9 +415,20 @@ function preserveReplacementHead(
   currentHead: StreamItem[],
   preserveLiveHead: boolean,
   sendingClientMessageIds: ReadonlySet<string>,
+  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";