fix(timeline): preserve optimistic prompts through hydration

Canonical replacement relied on a transport submission record that can settle before a provider echoes the user row. Keep presentation continuity owned by the unreconciled row while destructive history replacement remains governed by active transactions.
2026-10-26 11:12:53 +00:00 · 2026-07-30 23:29:18 +02:00
parent 43c15f707d
commit b98df29b19
9 changed files with 223 additions and 97 deletions
@@ -80,8 +80,10 @@ 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.
+user presentations identify themselves by having client identity without provider identity or a
+timeline position. This prevents a page from duplicating rows it already covers without coupling
+display continuity to the shorter-lived submission registry. Unreconciled local presentations are
+not persisted in the durable replica cache.

 ## Selective and legacy delivery

@@ -117,10 +119,11 @@ footer, and permanently drops attachments because the daemon does not echo them

 A submitted prompt is one `UserMessageItem` row. That row is the authoritative local presentation:
 its stable identity, text, timestamp, images, and attachments do not change when the provider
-acknowledges it. Submission lifecycle is a separate record keyed by agent, not another row shape or
-a property inferred from message identity. The transaction registry holds every unresolved send and
-records RPC acceptance and provider acknowledgement independently. Provider acknowledgement exists
-solely so a later transport error cannot roll back a prompt already observed canonically.
+acknowledges it. Submission lifecycle is a separate record keyed by agent, not another row shape.
+The transaction registry records RPC acceptance and provider acknowledgement independently for
+agent-scoped sends. Provider acknowledgement exists solely so a later transport error cannot roll
+back a prompt already observed canonically; it does not own whether the local presentation remains
+visible.

 The daemon's accepted response already waits for the correlated run start, but its response and the
 directory update reach client state separately. An accepted transaction remains active until the
@@ -140,7 +143,13 @@ arrived first. Generic reducers and consumers do not reimplement message identit
 Ordinary bootstrap, same-epoch reset, and catch-up replacement preserve unmatched locally submitted
 rows because a provider may never echo them. A known epoch change or rewind replaces history and
 drops acknowledged local rows omitted by the new canonical epoch; every transaction not yet
-acknowledged by the provider, and no other local row, crosses that destructive boundary.
+acknowledged by the provider, and no other local row, crosses that destructive boundary. A cold
+reset without an existing epoch is destructive because the client has no continuity anchor.
+
+Tail rows are positioned history, so an unmatched local presentation is appended after the
+canonical replacement rather than ordered by timestamps from different machines. The head is a
+live overlay: cursorless items stay there during continuity replacement until canonical positions
+arrive, while a destructive replacement retains only active submission transactions.

 Canonical replacement owns both timeline lanes. A matching local row keeps its presentation ID and
 payload while taking the canonical row's ordered position. If a live assistant head is the
@@ -1,7 +1,12 @@
 import type { Page } from "@playwright/test";
 import { expect, test as baseTest } from "./fixtures";
 import { expectAgentIdle } from "./helpers/agent-stream";
-import { expectComposerVisible, submitMessage } from "./helpers/composer";
+import {
+  attachImageFromMenu,
+  expectAttachmentPill,
+  expectComposerVisible,
+  submitMessage,
+} from "./helpers/composer";
 import { clickNewChat, gotoWorkspace } from "./helpers/launcher";
 import { seedWorkspace } from "./helpers/seed-client";
 import {
@@ -37,6 +42,7 @@ interface ContentChildFrame extends ElementFrame {
 interface TurnFrame {
  at: number;
  userRow: ElementFrame;
+  attachment: ElementFrame;
  footerRow: ElementFrame;
  spinner: ElementFrame;
  interruptControl: ElementFrame;
@@ -49,6 +55,15 @@ interface TurnFrame {
  };
 }

+const FIRST_PROMPT_IMAGE = {
+  name: "first-prompt.png",
+  mimeType: "image/png",
+  buffer: Buffer.from(
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
+    "base64",
+  ),
+};
+
 declare global {
  interface Window {
    __consecutiveTurnFrames?: { active: boolean; frames: TurnFrame[] };
@@ -124,6 +139,12 @@ async function recordTurnFrames(page: Page, prompt: string): Promise<void> {
        const style = getComputedStyle(candidate);
        if (Number(style.opacity) <= 0) return false;
        if (hasColor(style.backgroundColor)) return true;
+        const hasPaintedBorder = ["Top", "Right", "Bottom", "Left"].some(
+          (side) =>
+            Number.parseFloat(style.getPropertyValue(`border-${side.toLowerCase()}-width`)) > 0 &&
+            hasColor(style.getPropertyValue(`border-${side.toLowerCase()}-color`)),
+        );
+        if (hasPaintedBorder) return true;
        if (candidate.textContent?.trim() && hasColor(style.color)) return true;
        if (candidate instanceof SVGElement) {
          const fill = style.fill || candidate.getAttribute("fill") || "";
@@ -176,6 +197,8 @@ async function recordTurnFrames(page: Page, prompt: string): Promise<void> {
        backgroundColor: getComputedStyle(dots[0]).backgroundColor,
      };
    };
+    const findImageAttachment = (row: Element | undefined) =>
+      row?.querySelector('[role="button"][aria-label="Open image attachment"]');
    const sample = () => {
      const viewport = Array.from(
        document.querySelectorAll('[data-testid="agent-chat-scroll"]'),
@@ -183,6 +206,7 @@ async function recordTurnFrames(page: Page, prompt: string): Promise<void> {
      const promptRow = Array.from(
        viewport?.querySelectorAll('[data-testid="user-message"]') ?? [],
      ).find((candidate) => candidate.textContent?.includes(promptText));
+      const attachment = findImageAttachment(promptRow);
      const footer = Array.from(
        viewport?.querySelectorAll('[data-testid="turn-working-indicator"]') ?? [],
      )[0];
@@ -212,6 +236,7 @@ async function recordTurnFrames(page: Page, prompt: string): Promise<void> {
      state.frames.push({
        at: performance.now(),
        userRow: snapshot(promptRow, viewport),
+        attachment: snapshot(attachment, viewport),
        footerRow: snapshot(footerRow, viewport, spinner.painted),
        spinner,
        interruptControl: snapshot(interrupt),
@@ -308,6 +333,9 @@ function expectRowContinuity(frames: TurnFrame[]): void {
      baseline !== undefined &&
      Math.abs((frame.userRow.rect?.top ?? baseline) - baseline) > 1,
  );
+  const attachmentMissing = frames.findIndex(
+    (frame, index) => index >= first && !hasPaintedLayout(frame.attachment),
+  );
  const violations = [
    ...(first < 0 ? ["submitted row never became visible"] : []),
    ...(disappeared >= 0
@@ -320,8 +348,10 @@ function expectRowContinuity(frames: TurnFrame[]): void {
          `submitted row moved from ${baseline?.toFixed(1)} to ${frames[shifted].userRow.rect?.top.toFixed(1)}`,
        ]
      : []),
+    ...(attachmentMissing >= 0 ? ["submitted image attachment left the painted layout"] : []),
  ];
  let failure = Math.max(0, first);
+  if (attachmentMissing >= 0) failure = attachmentMissing;
  if (shifted >= 0) failure = shifted;
  if (disappeared >= 0) failure = disappeared;
  expect(
@@ -560,6 +590,8 @@ test("keeps the first prompt of a new agent in place through authoritative hydra
    await expectComposerVisible(page);

    const prompt = "Delay synthetic user message by 300ms.";
+    await attachImageFromMenu(page, FIRST_PROMPT_IMAGE);
+    await expectAttachmentPill(page, "composer-image-attachment-pill");
    await recordTurnFrames(page, prompt);
    await submitMessage(page, prompt);

@@ -606,7 +606,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
        hasActiveInitDeferred,
        initRequestDirection: activeInitDeferred?.requestDirection ?? "tail",
        sendingClientMessageIds,
-        hasAuthoritativeBaseline: timeline.status === "synced",
      });

      if (result.error) {
@@ -7,7 +7,7 @@ import {
  selectAgentTimelineState,
  useSessionStore,
 } from "@/stores/session-store";
-import type { StreamItem } from "@/types/stream";
+import { createUserMessage, type StreamItem } from "@/types/stream";
 import { ReplicaCache, type ReplicaCacheStorage } from ".";

 const SERVER_ID = "cached-host";
@@ -223,10 +223,40 @@ describe("ReplicaCache", () => {
      version: number;
      hosts: Array<{ timeline: Record<string, unknown> | null }>;
    };
-    expect(persisted.version).toBe(2);
+    expect(persisted.version).toBe(3);
    expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual(["agentId", "items"]);
  });

+  it("persists reconciled rows without caching unreconciled local presentations", async () => {
+    const storage = new MemoryStorage();
+    const cache = new ReplicaCache(storage);
+    cache.setHosts([SERVER_ID]);
+    seedSession();
+    const unreconciled = createUserMessage({
+      clientMessageId: "client-pending",
+      text: "Pending",
+      timestamp: new Date("2026-07-18T08:01:00.000Z"),
+    });
+    const reconciled = createUserMessage({
+      clientMessageId: "client-sent",
+      messageId: "provider-sent",
+      timelineCursor: { epoch: "epoch-1", seq: 11 },
+      text: "Sent",
+      timestamp: new Date("2026-07-18T08:01:30.000Z"),
+    });
+    useSessionStore
+      .getState()
+      .setAgentStreamTail(SERVER_ID, new Map([["agent-1", [unreconciled, reconciled]]]));
+
+    await cache.flush();
+    useSessionStore.getState().clearSession(SERVER_ID);
+    await cache.restore();
+
+    expect(useSessionStore.getState().sessions[SERVER_ID]?.agentStreamTail.get("agent-1")).toEqual([
+      reconciled,
+    ]);
+  });
+
  it("evicts the least recently written host when the cache exceeds its byte budget", async () => {
    const storage = new MemoryStorage();
    const cache = new ReplicaCache(storage, { maxBytes: 7_000 });
@@ -282,7 +312,7 @@ describe("ReplicaCache", () => {

    expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined();
    expect(JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null")).toEqual({
-      version: 2,
+      version: 3,
      hosts: [],
    });
  });
@@ -18,12 +18,11 @@ import {
  type SessionState,
  type WorkspaceDescriptor,
 } from "@/stores/session-store";
-import type { StreamItem } from "@/types/stream";
+import { isUnreconciledLocalUserMessage, type StreamItem } from "@/types/stream";
 import { normalizeAgentSnapshot } from "@/utils/agent-snapshots";
-import { getSendingClientMessageIds } from "@/composer/submission/model";

 const STORAGE_KEY = "@paseo:replica-cache";
-const CACHE_VERSION = 2;
+const CACHE_VERSION = 3;
 const PERSIST_DELAY_MS = 750;
 const MAX_TIMELINE_ITEMS = 50;
 const MAX_CACHE_BYTES = 1024 * 1024;
@@ -362,11 +361,6 @@ export class ReplicaCache {
            (workspace) => workspace.workspaceDirectory === focusedAgent.cwd,
          ))
        : undefined;
-      const localSubmissionIds = new Set(
-        getSendingClientMessageIds(
-          focusedAgentId ? session.messageSubmissions.get(focusedAgentId) : undefined,
-        ),
-      );
      const timelineState = focusedAgentId
        ? selectAgentTimelineState(session, focusedAgentId)
        : { status: "cold" as const };
@@ -374,11 +368,7 @@ export class ReplicaCache {
        timelineState.status === "cold"
          ? undefined
          : timelineState.items.filter(
-              (item) =>
-                item.kind !== "user_message" ||
-                item.messageId !== undefined ||
-                !item.clientMessageId ||
-                !localSubmissionIds.has(item.clientMessageId),
+              (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item),
            );
      const timeline =
        focusedAgent && items
@@ -199,7 +199,6 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
  hasActiveInitDeferred: false,
  initRequestDirection: "tail",
  sendingClientMessageIds: [],
-  hasAuthoritativeBaseline: true,
 };

 const baseStreamInput: ProcessAgentStreamEventInput = {
@@ -332,7 +331,7 @@ describe("processTimelineResponse", () => {
      currentTail: [],
      currentHead: [liveAssistant, submitted],
      currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
-      sendingClientMessageIds: ["client-new-prompt"],
+      sendingClientMessageIds: [],
      payload: {
        ...baseTimelineInput.payload,
        reset: true,
@@ -412,7 +411,6 @@ describe("processTimelineResponse", () => {
    const result = processTimelineResponse({
      ...baseTimelineInput,
      currentHead: liveCompletion,
-      hasAuthoritativeBaseline: false,
      isInitializing: true,
      hasActiveInitDeferred: true,
      payload: {
@@ -464,7 +462,6 @@ describe("processTimelineResponse", () => {
    const result = processTimelineResponse({
      ...baseTimelineInput,
      currentHead: liveTodo,
-      hasAuthoritativeBaseline: false,
      isInitializing: true,
      hasActiveInitDeferred: true,
      payload: {
@@ -516,7 +513,6 @@ describe("processTimelineResponse", () => {
    const result = processTimelineResponse({
      ...baseTimelineInput,
      currentHead: liveCompaction,
-      hasAuthoritativeBaseline: false,
      isInitializing: true,
      hasActiveInitDeferred: true,
      payload: {
@@ -554,7 +550,6 @@ describe("processTimelineResponse", () => {
      currentTail: [makeAssistantItem("painted replica")],
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -585,7 +580,6 @@ describe("processTimelineResponse", () => {
      ],
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -643,7 +637,6 @@ describe("processTimelineResponse", () => {
      currentHead: liveTool,
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -688,7 +681,6 @@ describe("processTimelineResponse", () => {
      ],
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -734,7 +726,6 @@ describe("processTimelineResponse", () => {
      currentHead: live.head,
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -775,7 +766,6 @@ describe("processTimelineResponse", () => {
      sendingClientMessageIds: [],
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -810,7 +800,6 @@ describe("processTimelineResponse", () => {
      currentHead: live.head,
      isInitializing: true,
      hasActiveInitDeferred: true,
-      hasAuthoritativeBaseline: false,
      payload: {
        ...baseTimelineInput.payload,
        direction: "tail",
@@ -973,31 +962,71 @@ describe("processTimelineResponse", () => {
    expect(result.tail).toEqual([first, second]);
  });

-  it("drops an acknowledged local row omitted by a same-epoch replacement", () => {
-    const acknowledged = createUserMessage({
+  it("keeps an unreconciled local presentation through same-epoch replacement", () => {
+    const image = {
+      id: "local-image",
+      mimeType: "image/png",
+      storageType: "web-indexeddb" as const,
+      storageKey: "local-image",
+      createdAt: 1000,
+    };
+    const attachment = {
+      type: "text" as const,
+      mimeType: "text/plain" as const,
+      text: "attached context",
+      title: "context.txt",
+    };
+    const local = createUserMessage({
      clientMessageId: "client-local-only",
      text: "provider may not echo this",
      timestamp: new Date(1000),
+      images: [image],
+      attachments: [attachment],
    });

    const result = processTimelineResponse({
      ...baseTimelineInput,
-      currentTail: [acknowledged],
+      currentTail: [local],
      currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
      sendingClientMessageIds: [],
      payload: {
        ...baseTimelineInput.payload,
        reset: true,
        epoch: "epoch-1",
-        entries: [],
+        startCursor: { seq: 2 },
+        endCursor: { seq: 2 },
+        entries: [makeTimelineEntry(2, "another client's response")],
      },
    });

-    expect(result.tail).toEqual([]);
+    expect(result.tail).toEqual([
+      expect.objectContaining({
+        kind: "assistant_message",
+        text: "another client's response",
+      }),
+      local,
+    ]);
  });

-  it("drops an acknowledged local row omitted by a known epoch change", () => {
-    const acknowledged = createUserMessage({
+  it("keeps an unreconciled local presentation during non-reset bootstrap", () => {
+    const local = makeSubmittedUserMessage("first prompt", "client-first-prompt");
+
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      currentTail: [local],
+      isInitializing: true,
+      hasActiveInitDeferred: true,
+      payload: {
+        ...baseTimelineInput.payload,
+        direction: "tail",
+      },
+    });
+
+    expect(result.tail).toEqual([local]);
+  });
+
+  it("drops an unreconciled local row omitted by a known epoch change", () => {
+    const local = createUserMessage({
      clientMessageId: "client-prior-epoch",
      text: "prior prompt",
      timestamp: new Date(1000),
@@ -1005,7 +1034,7 @@ describe("processTimelineResponse", () => {

    const result = processTimelineResponse({
      ...baseTimelineInput,
-      currentTail: [acknowledged],
+      currentTail: [local],
      currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
      sendingClientMessageIds: [],
      payload: {
@@ -1019,6 +1048,44 @@ describe("processTimelineResponse", () => {
    expect(result.tail).toEqual([]);
  });

+  it("drops an unreconciled head row omitted by a known epoch change", () => {
+    const local = makeSubmittedUserMessage("prior prompt", "client-prior-epoch");
+
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      currentHead: [local],
+      currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
+      sendingClientMessageIds: [],
+      payload: {
+        ...baseTimelineInput.payload,
+        reset: true,
+        epoch: "epoch-2",
+        entries: [],
+      },
+    });
+
+    expect(result.head).toEqual([]);
+  });
+
+  it("keeps an active head submission across a known epoch change", () => {
+    const local = makeSubmittedUserMessage("pending prompt", "client-pending");
+
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      currentHead: [local],
+      currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
+      sendingClientMessageIds: ["client-pending"],
+      payload: {
+        ...baseTimelineInput.payload,
+        reset: true,
+        epoch: "epoch-2",
+        entries: [],
+      },
+    });
+
+    expect(result.head).toEqual([local]);
+  });
+
  it("keeps an unmatched submission after the canonical replacement range", () => {
    const unmatched = makeSubmittedUserMessage("first submission", "client-first");
    const acknowledged: StreamItem[] = [
@@ -89,7 +89,6 @@ export interface ProcessTimelineResponseInput {
  hasActiveInitDeferred: boolean;
  initRequestDirection: InitRequestDirection;
  sendingClientMessageIds: readonly string[];
-  hasAuthoritativeBaseline: boolean;
 }

 export interface ProcessTimelineResponseOutput {
@@ -212,7 +211,7 @@ function applyTimelineReplacePath(args: {
  currentTail: StreamItem[];
  currentHead: StreamItem[];
  sendingClientMessageIds: readonly string[];
-  preserveLiveHead: boolean;
+  preserveContinuity: boolean;
  toHydratedEvents: (
    units: TimelineUnit[],
  ) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>;
@@ -224,7 +223,7 @@ function applyTimelineReplacePath(args: {
    currentTail,
    currentHead,
    sendingClientMessageIds,
-    preserveLiveHead,
+    preserveContinuity,
    toHydratedEvents,
  } = args;
  const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" });
@@ -233,7 +232,7 @@ function applyTimelineReplacePath(args: {
    previousTail: currentTail,
    previousHead: currentHead,
    sendingClientMessageIds,
-    preserveLiveHead,
+    preserveContinuity,
    canonicalCoverage: {
      epoch: payload.epoch,
      endSeq: payload.endCursor?.seq ?? null,
@@ -823,7 +822,6 @@ export function processTimelineResponse(
    hasActiveInitDeferred,
    initRequestDirection,
    sendingClientMessageIds,
-    hasAuthoritativeBaseline,
  } = input;

  // ------------------------------------------------------------------
@@ -895,8 +893,7 @@ export function processTimelineResponse(
        currentTail,
        currentHead,
        sendingClientMessageIds,
-        preserveLiveHead:
-          currentCursor?.epoch === payload.epoch || (!payload.reset && !hasAuthoritativeBaseline),
+        preserveContinuity: currentCursor?.epoch === payload.epoch || !payload.reset,
        toHydratedEvents,
      })
    : applyTimelineIncrementalPath({
@@ -126,6 +126,14 @@ export function createUserMessage(input: UserMessageInput): UserMessageItem {
  };
 }

+export function isUnreconciledLocalUserMessage(message: UserMessageItem): boolean {
+  return (
+    message.clientMessageId !== undefined &&
+    message.messageId === undefined &&
+    message.timelineCursor === undefined
+  );
+}
+
 export function appendSubmittedUserMessage(input: {
  tail: StreamItem[];
  head: StreamItem[];
@@ -396,7 +404,7 @@ export interface CanonicalStreamReplacementInput {
  previousTail: StreamItem[];
  previousHead: StreamItem[];
  sendingClientMessageIds: readonly string[];
-  preserveLiveHead: boolean;
+  preserveContinuity: boolean;
  canonicalCoverage: { epoch: string; endSeq: number | null };
 }

@@ -504,12 +512,12 @@ function reconcileReplacementHeadAgainstTail(
 function preserveReplacementHead(
  tail: StreamItem[],
  currentHead: StreamItem[],
-  preserveLiveHead: boolean,
+  preserveContinuity: boolean,
  sendingClientMessageIds: ReadonlySet<string>,
  canonicalCoverage: CanonicalStreamReplacementInput["canonicalCoverage"],
 ): CanonicalStreamReplacementResult {
  const canonicalTailAssistant = tail.at(-1);
-  const retainedHead = preserveLiveHead
+  const retainedHead = preserveContinuity
    ? currentHead.filter(
        (item) =>
          !item.timelineCursor ||
@@ -609,10 +617,7 @@ export function replaceWithCanonicalStream(
    if (tailResult.matched) {
      unmatchedTailMessages = removeUserMessageAt(unmatchedTailMessages, tailResult.index);
      nextTail.push(tailResult.message);
-      if (
-        tailResult.message.clientMessageId &&
-        sendingClientMessageIds.has(tailResult.message.clientMessageId)
-      ) {
+      if (tailResult.message.clientMessageId) {
        acknowledgedClientMessageIds.add(tailResult.message.clientMessageId);
      }
      continue;
@@ -625,10 +630,7 @@ export function replaceWithCanonicalStream(
        ...headResult.items.slice(headResult.index + 1),
      ];
      nextTail.push(headResult.message);
-      if (
-        headResult.message.clientMessageId &&
-        sendingClientMessageIds.has(headResult.message.clientMessageId)
-      ) {
+      if (headResult.message.clientMessageId) {
        acknowledgedClientMessageIds.add(headResult.message.clientMessageId);
      }
      continue;
@@ -639,10 +641,13 @@ export function replaceWithCanonicalStream(

  const retainedTailMessages: UserMessageItem[] = [];
  for (const local of unmatchedTailMessages) {
-    if (local.clientMessageId && sendingClientMessageIds.has(local.clientMessageId)) {
+    const preserveLocal = input.preserveContinuity
+      ? isUnreconciledLocalUserMessage(local)
+      : local.clientMessageId !== undefined && sendingClientMessageIds.has(local.clientMessageId);
+    if (preserveLocal) {
      nextTail.push(local);
    } else if (
-      input.preserveLiveHead &&
+      input.preserveContinuity &&
      local.timelineCursor &&
      isAfterCanonicalCoverage(local.timelineCursor, input.canonicalCoverage)
    ) {
@@ -651,17 +656,10 @@ export function replaceWithCanonicalStream(
  }
  nextHead = [...retainedTailMessages, ...nextHead];

-  nextHead = nextHead.filter((item) => {
-    if (item.kind !== "user_message" || !item.clientMessageId) return true;
-    if (sendingClientMessageIds.has(item.clientMessageId)) return true;
-    if (!input.preserveLiveHead || !item.timelineCursor) return false;
-    return isAfterCanonicalCoverage(item.timelineCursor, input.canonicalCoverage);
-  });
-
  const replacement = preserveReplacementHead(
    nextTail,
    nextHead,
-    input.preserveLiveHead,
+    input.preserveContinuity,
    sendingClientMessageIds,
    input.canonicalCoverage,
  );
@@ -662,7 +662,32 @@ export class MockLoadTestAgentSession implements AgentSession {
      turnStarted: false,
    };
    this.activeTurn = turn;
+    const largePayload = parseLargeAgentStreamPayloadPrompt(prompt);
+    const stress = parseAgentStreamStressPrompt(prompt);
+    const questionPrompt = parseMockQuestionPrompt(prompt);
+    const structuredBranchName = parseStructuredBranchNamePrompt(prompt);
+    const settledAssistantImageMarkdown = parseSettledAssistantImageMarkdown(prompt);
+    const scheduleTurn = () => {
+      if (shouldEmitTurnFailure(prompt)) {
+        this.scheduleFailedTurn(turn);
+      } else if (structuredBranchName) {
+        this.scheduleSettledAssistantTurn(turn, JSON.stringify(structuredBranchName));
+      } else if (settledAssistantImageMarkdown) {
+        this.scheduleSettledAssistantTurn(turn, settledAssistantImageMarkdown);
+      } else if (shouldEmitPlanApprovalPrompt(prompt)) {
+        this.schedulePlanApprovalTurn(turn);
+      } else if (questionPrompt) {
+        this.scheduleQuestionPromptTurn(turn, questionPrompt);
+      } else if (largePayload) {
+        this.scheduleLargePayloadTurn(turn, largePayload);
+      } else if (stress) {
+        this.scheduleStressTurn(turn, stress);
+      } else {
+        this.schedule(turn, 0);
+      }
+    };
    const userMessageId = randomUUID();
+    const userMessageDelayMs = parseUserMessageDelayMs(prompt);
    const userMessageTimer = setTimeout(() => {
      if (this.activeTurn?.turnId !== turnId) {
        return;
@@ -678,31 +703,10 @@ export class MockLoadTestAgentSession implements AgentSession {
          ...(options?.clientMessageId ? { clientMessageId: options.clientMessageId } : {}),
        },
      });
-    }, parseUserMessageDelayMs(prompt));
+      if (userMessageDelayMs > 0) scheduleTurn();
+    }, userMessageDelayMs);
    userMessageTimer.unref?.();
-
-    const largePayload = parseLargeAgentStreamPayloadPrompt(prompt);
-    const stress = parseAgentStreamStressPrompt(prompt);
-    const questionPrompt = parseMockQuestionPrompt(prompt);
-    const structuredBranchName = parseStructuredBranchNamePrompt(prompt);
-    const settledAssistantImageMarkdown = parseSettledAssistantImageMarkdown(prompt);
-    if (shouldEmitTurnFailure(prompt)) {
-      this.scheduleFailedTurn(turn);
-    } else if (structuredBranchName) {
-      this.scheduleSettledAssistantTurn(turn, JSON.stringify(structuredBranchName));
-    } else if (settledAssistantImageMarkdown) {
-      this.scheduleSettledAssistantTurn(turn, settledAssistantImageMarkdown);
-    } else if (shouldEmitPlanApprovalPrompt(prompt)) {
-      this.schedulePlanApprovalTurn(turn);
-    } else if (questionPrompt) {
-      this.scheduleQuestionPromptTurn(turn, questionPrompt);
-    } else if (largePayload) {
-      this.scheduleLargePayloadTurn(turn, largePayload);
-    } else if (stress) {
-      this.scheduleStressTurn(turn, stress);
-    } else {
-      this.schedule(turn, 0);
-    }
+    if (userMessageDelayMs === 0) scheduleTurn();
    return { turnId };
  }