diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index ed22654c2e..4a967fe118 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -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 diff --git a/packages/app/e2e/agent-consecutive-turns.spec.ts b/packages/app/e2e/agent-consecutive-turns.spec.ts index 2ef5ccc40d..ff2f3cdef2 100644 --- a/packages/app/e2e/agent-consecutive-turns.spec.ts +++ b/packages/app/e2e/agent-consecutive-turns.spec.ts @@ -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 { 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 { 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 { 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 { 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); diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 37c781a606..5fd5dcb6eb 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -606,7 +606,6 @@ 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/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 3ae4114d2a..f601dce4c0 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -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 | 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: [], }); }); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 2cad88aecc..cde96f6ed1 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -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 diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 369824b1af..6e8efbd82d 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -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[] = [ diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 34e8b9014c..7c151adf05 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -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({ diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 45a712884a..b764890973 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -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, 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, ); diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.ts index b30042b6cc..790e450620 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.ts @@ -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 }; }