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 });