From 26bdb1d1713e8f884950e560bdf067980f175213 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 2 Aug 2026 12:21:47 +0200 Subject: [PATCH] Keep timeline history duplicate-free across daemon upgrades (#2789) * fix(app): keep timeline pagination duplicate-free Older daemons page projected history by canonical sequence ranges, so one projected block can appear in adjacent responses. Accept each block on the page containing its projection anchor while still advancing across empty projected pages.\n\nProjected siblings can share a provider message ID, so render identity also includes the timeline cursor instead of collapsing distinct blocks onto one layout row. * fix(e2e): preserve isolated daemon launch contract Keep the source-daemon spawn path recognizable to the repository launch-contract check while selecting the built supervisor only for published package fixtures. * fix(app): preserve stable timeline item identity --- ...ent-timeline-pagination-old-daemon.spec.ts | 86 +++++++++++++ .../browser/agent-timeline-pagination.spec.ts | 2 +- .../support/helpers/agent-timeline-gate.ts | 103 +++++++++++---- .../support/helpers/isolated-host-daemon.ts | 51 +++++++- .../app/e2e/support/helpers/mock-agent.ts | 8 +- .../support/helpers/timeline-pagination.ts | 32 ++++- .../timeline/session-stream-reducers.test.ts | 120 ++++++++++++++++++ .../src/timeline/session-stream-reducers.ts | 47 ++++++- packages/app/src/types/stream.ts | 2 +- 9 files changed, 407 insertions(+), 44 deletions(-) create mode 100644 packages/app/e2e/browser/agent-timeline-pagination-old-daemon.spec.ts diff --git a/packages/app/e2e/browser/agent-timeline-pagination-old-daemon.spec.ts b/packages/app/e2e/browser/agent-timeline-pagination-old-daemon.spec.ts new file mode 100644 index 000000000..16e2c7eab --- /dev/null +++ b/packages/app/e2e/browser/agent-timeline-pagination-old-daemon.spec.ts @@ -0,0 +1,86 @@ +import { randomUUID } from "node:crypto"; +import { metroTest as test } from "../support/fixtures"; +import { buildCreateAgentPreferences, buildSeededHost } from "../support/helpers/daemon-registry"; +import { startIsolatedHostDaemon } from "../support/helpers/isolated-host-daemon"; +import type { MockAgentWorkspace } from "../support/helpers/mock-agent"; +import { seedWorkspace } from "../support/helpers/seed-client"; +import { + expectTimelinePromptVisible, + holdOlderHistoryPages, + openAgentTimeline, + scrollThroughOlderHistoryPages, +} from "../support/helpers/timeline-pagination"; + +test("does not repeat an assistant block when the current app paginates a published 0.2.5 daemon", async ({ + page, +}) => { + test.setTimeout(120_000); + const serverId = `srv_old_pagination_${randomUUID().replaceAll("-", "").slice(0, 12)}`; + const daemon = await startIsolatedHostDaemon(serverId, { publishedVersion: "0.2.5" }); + const workspace = await seedWorkspace({ + repoPrefix: "timeline-old-daemon-pagination-", + port: daemon.port, + }); + const createdAgent = await workspace.client.createAgent({ + provider: "mock", + cwd: workspace.repoPath, + workspaceId: workspace.workspaceId, + title: "Published daemon pagination regression", + modeId: "load-test", + model: "ten-second-stream", + }); + const agent: MockAgentWorkspace = { + agentId: createdAgent.id, + workspaceId: workspace.workspaceId, + cwd: workspace.repoPath, + client: workspace.client, + cleanup: workspace.cleanup, + }; + + try { + for (let index = 0; index < 40; index += 1) { + await agent.client.sendAgentMessage( + agent.agentId, + `timeline-pagination-older-turn-${index}: emit 1 coalesced agent stream updates`, + ); + await agent.client.waitForFinish(agent.agentId, 15_000); + } + await agent.client.sendAgentMessage(agent.agentId, "build a realistic long mock timeline"); + await agent.client.waitForFinish(agent.agentId, 20_000); + for (let index = 0; index < 20; index += 1) { + await agent.client.sendAgentMessage( + agent.agentId, + `timeline-pagination-turn-${index}: emit 1 coalesced agent stream updates`, + ); + await agent.client.waitForFinish(agent.agentId, 15_000); + } + + const host = buildSeededHost({ + serverId, + endpoint: `127.0.0.1:${daemon.port}`, + nowIso: new Date().toISOString(), + }); + await page.addInitScript( + ({ seededHost, preferences }) => { + localStorage.setItem("@paseo:e2e", "1"); + localStorage.setItem("@paseo:daemon-registry", JSON.stringify([seededHost])); + localStorage.setItem("@paseo:create-agent-preferences", JSON.stringify(preferences)); + }, + { seededHost: host, preferences: buildCreateAgentPreferences(serverId) }, + ); + + const history = await holdOlderHistoryPages(page, agent, daemon.port); + await openAgentTimeline(page, agent, serverId); + await expectTimelinePromptVisible( + page, + "timeline-pagination-turn-19: emit 1 coalesced agent stream updates", + ); + await scrollThroughOlderHistoryPages(page, 3, history); + + history.expectRepeatedEntries(); + await history.expectOwnedTextRendered("Now I have a clearer picture."); + } finally { + await agent.cleanup(); + await daemon.close(); + } +}); diff --git a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts index 86ca233f5..72e7491de 100644 --- a/packages/app/e2e/browser/agent-timeline-pagination.spec.ts +++ b/packages/app/e2e/browser/agent-timeline-pagination.spec.ts @@ -97,7 +97,7 @@ test.describe("Agent timeline pagination", () => { await expectTimelinePromptVisible(page, agent.newestPrompt); await scrollThroughOlderHistoryPages(page, 3, history); - history.expectNoRepeatedAssistantEntries(); + history.expectNoRepeatedEntries(); } finally { await agent.cleanup(); } diff --git a/packages/app/e2e/support/helpers/agent-timeline-gate.ts b/packages/app/e2e/support/helpers/agent-timeline-gate.ts index 3c7f7865a..332c5d241 100644 --- a/packages/app/e2e/support/helpers/agent-timeline-gate.ts +++ b/packages/app/e2e/support/helpers/agent-timeline-gate.ts @@ -1,5 +1,5 @@ import type { Page } from "@playwright/test"; -import { daemonWsRoutePattern } from "./daemon-port"; +import { daemonWsRoutePattern, wsRoutePatternForPort } from "./daemon-port"; type WebSocketMessage = string | Buffer; @@ -17,7 +17,8 @@ export interface AgentTimelineResponseGate { export interface OlderTimelinePagesGate { getRequestCount(): number; - getRepeatedAssistantEntryCount(): number; + getRepeatedEntryCount(): number; + getOwnedEntryCountContaining(text: string): number; releasePage(pageNumber: number): void; waitForRequestCount(count: number): Promise; } @@ -61,6 +62,62 @@ function getPayload(message: Record): Record | : null; } +interface ObservedTimelineEntry { + seqStart: number; + seqEnd: number; + item: { type: string }; +} + +function readTimelineEntries(payload: Record): ObservedTimelineEntry[] { + if (!Array.isArray(payload.entries)) return []; + return payload.entries.filter((entry): entry is ObservedTimelineEntry => { + if (!entry || typeof entry !== "object") return false; + const candidate = entry as Partial; + return ( + typeof candidate.seqStart === "number" && + typeof candidate.seqEnd === "number" && + typeof candidate.item?.type === "string" + ); + }); +} + +function readCursorSeq(cursor: unknown): number | undefined { + if (!cursor || typeof cursor !== "object") return undefined; + const seq = (cursor as { seq?: unknown }).seq; + return typeof seq === "number" ? seq : undefined; +} + +function recordOwnedTimelineEntries( + payload: Record, + ownedEntries: Map, +): void { + const pageStartSeq = readCursorSeq(payload.startCursor); + const pageEndSeq = readCursorSeq(payload.endCursor); + for (const entry of readTimelineEntries(payload)) { + const belongsToPage = + payload.direction === "tail" || + (pageStartSeq !== undefined && + pageEndSeq !== undefined && + entry.seqStart >= pageStartSeq && + entry.seqStart <= pageEndSeq); + if (!belongsToPage) continue; + ownedEntries.set(`${entry.seqStart}:${entry.seqEnd}`, JSON.stringify(entry.item)); + } +} + +function recordRepeatedTimelineEntries( + payload: Record, + entryKeys: Set, +): number { + let repeats = 0; + for (const entry of readTimelineEntries(payload)) { + const key = `${entry.seqStart}:${entry.seqEnd}:${entry.item.type}`; + if (entryKeys.has(key)) repeats += 1; + else entryKeys.add(key); + } + return repeats; +} + export async function holdDaemonHydration(page: Page): Promise { let released = false; const delayedForwards: Array<() => void> = []; @@ -176,11 +233,13 @@ export async function delayAgentOlderTimelineResponse( export async function holdAgentOlderTimelinePages( page: Page, agentId: string, + daemonPort?: number, ): Promise { let requestCount = 0; let responseCount = 0; - let repeatedAssistantEntryCount = 0; - const assistantEntryKeys = new Set(); + let repeatedEntryCount = 0; + const entryKeys = new Set(); + const ownedEntries = new Map(); const releasedPages = new Set(); const delayedForwards = new Map void>>(); const requestWaiters = new Map void>>(); @@ -193,7 +252,9 @@ export async function holdAgentOlderTimelinePages( } }; - await page.routeWebSocket(daemonWsRoutePattern(), (ws) => { + const routePattern = + daemonPort === undefined ? daemonWsRoutePattern() : wsRoutePatternForPort(String(daemonPort)); + await page.routeWebSocket(routePattern, (ws) => { const server = ws.connectToServer(); ws.onMessage((message) => { const sessionMessage = getSessionMessage(message); @@ -210,32 +271,20 @@ export async function holdAgentOlderTimelinePages( server.onMessage((message) => { const sessionMessage = getSessionMessage(message); const payload = sessionMessage ? getPayload(sessionMessage) : null; + if ( + sessionMessage?.type === "fetch_agent_timeline_response" && + payload?.agentId === agentId && + (payload.direction === "tail" || payload.direction === "before") + ) { + recordOwnedTimelineEntries(payload, ownedEntries); + } if ( sessionMessage?.type === "fetch_agent_timeline_response" && payload?.agentId === agentId && payload.direction === "before" ) { responseCount += 1; - const entries = Array.isArray(payload.entries) ? payload.entries : []; - for (const entry of entries) { - if (!entry || typeof entry !== "object") continue; - const projectedEntry = entry as { - seqStart?: unknown; - seqEnd?: unknown; - item?: { type?: unknown }; - }; - if ( - typeof projectedEntry.seqStart !== "number" || - typeof projectedEntry.seqEnd !== "number" || - typeof projectedEntry.item?.type !== "string" - ) { - continue; - } - if (projectedEntry.item.type !== "assistant_message") continue; - const key = `${projectedEntry.seqStart}:${projectedEntry.seqEnd}:${projectedEntry.item.type}`; - if (assistantEntryKeys.has(key)) repeatedAssistantEntryCount += 1; - else assistantEntryKeys.add(key); - } + repeatedEntryCount += recordRepeatedTimelineEntries(payload, entryKeys); const pageNumber = responseCount; if (releasedPages.has(pageNumber)) { ws.send(message); @@ -252,7 +301,9 @@ export async function holdAgentOlderTimelinePages( return { getRequestCount: () => requestCount, - getRepeatedAssistantEntryCount: () => repeatedAssistantEntryCount, + getRepeatedEntryCount: () => repeatedEntryCount, + getOwnedEntryCountContaining: (value) => + [...ownedEntries.values()].filter((text) => text.includes(value)).length, releasePage(pageNumber) { releasedPages.add(pageNumber); for (const forward of delayedForwards.get(pageNumber) ?? []) forward(); diff --git a/packages/app/e2e/support/helpers/isolated-host-daemon.ts b/packages/app/e2e/support/helpers/isolated-host-daemon.ts index c6dcd2d01..062b444d9 100644 --- a/packages/app/e2e/support/helpers/isolated-host-daemon.ts +++ b/packages/app/e2e/support/helpers/isolated-host-daemon.ts @@ -1,5 +1,11 @@ import { once } from "node:events"; -import { spawn, execSync, type ChildProcess } from "node:child_process"; +import { + spawn, + execFileSync, + execSync, + type ChildProcess, + type SpawnOptions, +} from "node:child_process"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import net from "node:net"; import { tmpdir } from "node:os"; @@ -23,6 +29,7 @@ export interface IsolatedHostDaemonOptions { }; paseoHome?: string; preserveHome?: boolean; + publishedVersion?: string; } async function getAvailablePort(): Promise { @@ -100,6 +107,31 @@ export async function startIsolatedHostDaemon( const paseoHome = options.paseoHome ?? (await mkdtemp(path.join(tmpdir(), "paseo-e2e-secondary-host-"))); + let publishedPackageRoot: string | null = null; + if (options.publishedVersion) { + publishedPackageRoot = await mkdtemp(path.join(tmpdir(), "paseo-e2e-published-server-")); + await writeFile( + path.join(publishedPackageRoot, "package.json"), + `${JSON.stringify({ private: true })}\n`, + ); + const npmCommand = process.platform === "win32" ? "npm.cmd" : "npm"; + try { + execFileSync( + npmCommand, + [ + "install", + "--no-audit", + "--no-fund", + "--no-package-lock", + `@getpaseo/server@${options.publishedVersion}`, + ], + { cwd: publishedPackageRoot, stdio: "ignore" }, + ); + } catch (error) { + await rm(publishedPackageRoot, { recursive: true, force: true }); + throw error; + } + } if (options.mutableRelay) { const endpoint = options.mutableRelay.endpoint ?? @@ -120,10 +152,12 @@ export async function startIsolatedHostDaemon( })}\n`, ); } - const serverDir = path.resolve(__dirname, "../../../../server"); + const serverDir = publishedPackageRoot + ? path.join(publishedPackageRoot, "node_modules", "@getpaseo", "server") + : path.resolve(__dirname, "../../../../server"); const tsxBin = execSync("which tsx").toString().trim(); const spawnDaemon = async (): Promise => { - const child = spawn(tsxBin, ["scripts/supervisor-entrypoint.ts", "--dev"], { + const spawnOptions: SpawnOptions = { cwd: serverDir, env: withDisabledE2ESpeechEnv({ ...process.env, @@ -138,7 +172,10 @@ export async function startIsolatedHostDaemon( }), stdio: ["ignore", "ignore", "pipe"], detached: false, - }); + }; + const child = publishedPackageRoot + ? spawn(process.execPath, ["dist/scripts/supervisor-entrypoint.js"], spawnOptions) + : spawn(tsxBin, ["scripts/supervisor-entrypoint.ts", "--dev"], spawnOptions); let stderr = ""; child.stderr?.on("data", (chunk: Buffer) => { @@ -165,6 +202,9 @@ export async function startIsolatedHostDaemon( if (!options.preserveHome) { await rm(paseoHome, { recursive: true, force: true }); } + if (publishedPackageRoot) { + await rm(publishedPackageRoot, { recursive: true, force: true }); + } throw error; } let closed = false; @@ -186,6 +226,9 @@ export async function startIsolatedHostDaemon( if (!options.preserveHome) { await rm(paseoHome, { recursive: true, force: true }); } + if (publishedPackageRoot) { + await rm(publishedPackageRoot, { recursive: true, force: true }); + } }, }; } diff --git a/packages/app/e2e/support/helpers/mock-agent.ts b/packages/app/e2e/support/helpers/mock-agent.ts index 8999cab85..6320ce3c8 100644 --- a/packages/app/e2e/support/helpers/mock-agent.ts +++ b/packages/app/e2e/support/helpers/mock-agent.ts @@ -70,8 +70,12 @@ export async function seedRunningMockAgentWorkspace( } } -export function buildAgentRoute(workspaceId: string, agentId: string): string { - return buildHostAgentDetailRoute(getServerId(), agentId, workspaceId); +export function buildAgentRoute( + workspaceId: string, + agentId: string, + serverId = getServerId(), +): string { + return buildHostAgentDetailRoute(serverId, agentId, workspaceId); } /** Boots the app directly at the agent's workspace route and waits for the open intent to settle. */ diff --git a/packages/app/e2e/support/helpers/timeline-pagination.ts b/packages/app/e2e/support/helpers/timeline-pagination.ts index 2d4320de5..8f1d301e1 100644 --- a/packages/app/e2e/support/helpers/timeline-pagination.ts +++ b/packages/app/e2e/support/helpers/timeline-pagination.ts @@ -56,7 +56,9 @@ interface OlderHistoryLoadingOperation { interface OlderHistoryPages { expectRequestedPages(count: number): Promise; expectSettledWithRequestedPages(count: number): Promise; - expectNoRepeatedAssistantEntries(): void; + expectNoRepeatedEntries(): void; + expectRepeatedEntries(): void; + expectOwnedTextRendered(text: string): Promise; releasePage(pageNumber: number): void; } @@ -161,8 +163,12 @@ export async function expectTimelinePromptPositionPreserved( .toBeLessThanOrEqual(2); } -export async function openAgentTimeline(page: Page, agent: MockAgentWorkspace): Promise { - await page.goto(buildAgentRoute(agent.workspaceId, agent.agentId)); +export async function openAgentTimeline( + page: Page, + agent: MockAgentWorkspace, + serverId?: string, +): Promise { + await page.goto(buildAgentRoute(agent.workspaceId, agent.agentId, serverId)); await page.waitForURL( (url) => url.pathname.includes("/workspace/") && !url.searchParams.has("open"), { timeout: 60_000 }, @@ -262,8 +268,13 @@ export async function holdNextOlderTimelinePage( export async function holdOlderHistoryPages( page: Page, agent: MockAgentWorkspace, + daemonPort?: number, ): Promise { - const gate: OlderTimelinePagesGate = await holdAgentOlderTimelinePages(page, agent.agentId); + const gate: OlderTimelinePagesGate = await holdAgentOlderTimelinePages( + page, + agent.agentId, + daemonPort, + ); return { async expectRequestedPages(count) { await gate.waitForRequestCount(count); @@ -273,8 +284,17 @@ export async function holdOlderHistoryPages( await waitForTimelineGeometryToSettle(page); expect(gate.getRequestCount()).toBe(count); }, - expectNoRepeatedAssistantEntries() { - expect(gate.getRepeatedAssistantEntryCount()).toBe(0); + expectNoRepeatedEntries() { + expect(gate.getRepeatedEntryCount()).toBe(0); + }, + expectRepeatedEntries() { + expect(gate.getRepeatedEntryCount()).toBeGreaterThan(0); + }, + async expectOwnedTextRendered(text) { + const expectedCount = gate.getOwnedEntryCountContaining(text); + expect(expectedCount).toBeGreaterThan(0); + const timeline = page.locator('[data-testid="agent-chat-scroll"]:visible').first(); + await expect(timeline.getByText(text, { exact: false })).toHaveCount(expectedCount); }, releasePage(pageNumber) { gate.releasePage(pageNumber); diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index dc0505f28..7d7a2aae2 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -224,6 +224,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", direction: "after", + projection: "projected", reset: false, epoch: "epoch-1", window: { minSeq: 1, maxSeq: 0, nextSeq: 1 }, @@ -2540,6 +2541,123 @@ describe("processTimelineResponse", () => { expect(result.older).toBe("none"); }); + it("keeps only projected entries anchored in an older response page", () => { + const currentTail: StreamItem[] = [ + { + kind: "user_message", + id: "current-81", + text: "current-81", + timestamp: new Date(81000), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentCursor: { epoch: "epoch-1", startSeq: 81, endSeq: 100 }, + payload: { + ...baseTimelineInput.payload, + direction: "before", + projection: "projected", + epoch: "epoch-1", + startCursor: { seq: 41 }, + endCursor: { seq: 80 }, + hasOlder: true, + entries: [ + makeTimelineEntry(1, "wide assistant", "assistant_message", 80), + makeTimelineEntry(50, "owned by this page", "user_message"), + ], + }, + }); + + expect(getAssistantTexts(result.tail)).toEqual([]); + expect(getUserTexts(result.tail)).toEqual(["owned by this page", "current-81"]); + expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 41, endSeq: 100 }); + expect(result.older).toBe("available"); + }); + + it("advances through an older projected page with no anchored entries", () => { + const currentTail: StreamItem[] = [ + { + kind: "user_message", + id: "current-81", + text: "current-81", + timestamp: new Date(81000), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentCursor: { epoch: "epoch-1", startSeq: 81, endSeq: 100 }, + payload: { + ...baseTimelineInput.payload, + direction: "before", + projection: "projected", + epoch: "epoch-1", + startCursor: { seq: 41 }, + endCursor: { seq: 80 }, + hasOlder: true, + entries: [makeTimelineEntry(1, "wide assistant", "assistant_message", 80)], + }, + }); + + expect(result.tail).toBe(currentTail); + expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 41, endSeq: 100 }); + expect(result.cursorChanged).toBe(true); + expect(result.older).toBe("available"); + }); + + it("gives prepended assistant blocks unique ids across independently hydrated pages", () => { + const currentTail: StreamItem[] = [ + { + kind: "user_message", + id: "current-81", + text: "current-81", + timestamp: new Date(81000), + }, + { + kind: "assistant_message", + id: "shared-provider-message", + messageId: "shared-provider-message", + text: "newer assistant block", + timestamp: new Date(82000), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentCursor: { epoch: "epoch-1", startSeq: 81, endSeq: 100 }, + payload: { + ...baseTimelineInput.payload, + direction: "before", + projection: "projected", + epoch: "epoch-1", + startCursor: { seq: 41 }, + endCursor: { seq: 80 }, + hasOlder: true, + entries: [ + { + ...makeTimelineEntry(50, "older assistant block"), + item: { + type: "assistant_message", + text: "older assistant block", + messageId: "shared-provider-message", + }, + }, + ], + }, + }); + + const assistants = result.tail.filter((item) => item.kind === "assistant_message"); + expect(assistants.map((item) => item.messageId)).toEqual([ + "shared-provider-message", + "shared-provider-message", + ]); + expect(new Set(assistants.map((item) => item.id)).size).toBe(2); + }); + it("drops a stale before page anchored before a resume-tail replacement", () => { const currentTail: StreamItem[] = [ { @@ -3616,6 +3734,7 @@ describe("processAgentStreamEvents", () => { payload: { agentId: "agent-1", direction: "tail", + projection: "projected", reset: false, epoch: "epoch-1", window: { minSeq: 186, maxSeq: 186, nextSeq: 187 }, @@ -3660,6 +3779,7 @@ describe("processAgentStreamEvents", () => { payload: { agentId: "agent-1", direction: "tail", + projection: "projected", reset: false, epoch: "epoch-1", window: { minSeq: 10, maxSeq: 10, nextSeq: 11 }, diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 282552d52..ec256f54f 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -74,6 +74,7 @@ export interface ProcessTimelineResponseInput { payload: { agentId: string; direction: TimelineDirection; + projection: "projected" | "canonical"; reset: boolean; epoch: string; window: { minSeq: number; maxSeq: number; nextSeq: number }; @@ -786,6 +787,27 @@ function deriveCanonicalAcknowledgements(params: { return [...acknowledged]; } +function selectEntriesOwnedByTimelinePage( + payload: ProcessTimelineResponseInput["payload"], +): TimelineResponseEntry[] { + // COMPAT(projectedBeforePageOwnership): added in v0.2.6, remove after 2027-02-02 + // once the supported daemon floor paginates projected before pages. + if ( + payload.direction !== "before" || + payload.projection !== "projected" || + !payload.startCursor || + !payload.endCursor + ) { + return payload.entries; + } + + const pageStartSeq = payload.startCursor.seq; + const pageEndSeq = payload.endCursor.seq; + return payload.entries.filter( + (entry) => entry.seqStart >= pageStartSeq && entry.seqStart <= pageEndSeq, + ); +} + function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -802,7 +824,7 @@ function applyTimelineIncrementalPath(args: { const sideEffects: TimelineReducerSideEffect[] = []; let acknowledgedClientMessageIds: string[] = []; - if (timelineUnits.length === 0) { + if (timelineUnits.length === 0 && payload.direction !== "before") { return { tail: nextTail, head: nextHead, @@ -827,16 +849,33 @@ function applyTimelineIncrementalPath(args: { currentCursor, }); + if ( + payload.direction === "before" && + cursor && + currentCursor && + cursor.startSeq !== currentCursor.startSeq + ) { + older = payload.hasOlder ? "available" : "none"; + } + if (acceptedUnits.length > 0) { if (payload.direction === "before") { - older = payload.hasOlder ? "available" : "none"; const olderTail = hydrateStreamState( acceptedUnits.map(({ event, timestamp, seqEnd }) => ({ event, timestamp, timelineCursor: { epoch: payload.epoch, seq: seqEnd }, })), - { source: "canonical" }, + { + source: "canonical", + reservedItemIds: new Set( + currentTail.flatMap((item) => + item.kind === "assistant_message" && item.blockGroupId + ? [item.id, item.blockGroupId] + : [item.id], + ), + ), + }, ); nextTail = mergePrependedCanonicalTail(olderTail, currentTail); } else { @@ -915,7 +954,7 @@ export function processTimelineResponse( // ------------------------------------------------------------------ // Convert entries to timeline units // ------------------------------------------------------------------ - const timelineUnits = payload.entries.map((entry) => ({ + const timelineUnits = selectEntriesOwnedByTimelinePage(payload).map((entry) => ({ seq: entry.seqStart, seqEnd: entry.seqEnd, sourceSeqRanges: diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 1c02d1b31..50fcfbd02 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -1423,7 +1423,7 @@ export function hydrateStreamState( timestamp: Date; timelineCursor?: TimelinePosition; }>, - options?: { source?: StreamUpdateSource }, + options?: { source?: StreamUpdateSource; reservedItemIds?: ReadonlySet }, ): StreamItem[] { const hydrated = events.reduce((state, { event, timestamp, timelineCursor }) => { return reduceStreamUpdate(state, event, timestamp, { ...options, timelineCursor });