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
2026-08-02 12:21:47 +02:00
parent 0a106dd35f
commit 26bdb1d171
9 changed files with 407 additions and 44 deletions
@@ -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<void>;
 }
@@ -61,6 +62,62 @@ function getPayload(message: Record<string, unknown>): Record<string, unknown> |
    : null;
 }

+interface ObservedTimelineEntry {
+  seqStart: number;
+  seqEnd: number;
+  item: { type: string };
+}
+
+function readTimelineEntries(payload: Record<string, unknown>): 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<ObservedTimelineEntry>;
+    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<string, unknown>,
+  ownedEntries: Map<string, string>,
+): 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<string, unknown>,
+  entryKeys: Set<string>,
+): 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<DaemonHydrationGate> {
  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<OlderTimelinePagesGate> {
  let requestCount = 0;
  let responseCount = 0;
-  let repeatedAssistantEntryCount = 0;
-  const assistantEntryKeys = new Set<string>();
+  let repeatedEntryCount = 0;
+  const entryKeys = new Set<string>();
+  const ownedEntries = new Map<string, string>();
  const releasedPages = new Set<number>();
  const delayedForwards = new Map<number, Array<() => void>>();
  const requestWaiters = new Map<number, Array<() => 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();
@@ -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<number> {
@@ -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<ChildProcess> => {
-    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 });
+      }
    },
  };
 }
@@ -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. */
@@ -56,7 +56,9 @@ interface OlderHistoryLoadingOperation {
 interface OlderHistoryPages {
  expectRequestedPages(count: number): Promise<void>;
  expectSettledWithRequestedPages(count: number): Promise<void>;
-  expectNoRepeatedAssistantEntries(): void;
+  expectNoRepeatedEntries(): void;
+  expectRepeatedEntries(): void;
+  expectOwnedTextRendered(text: string): Promise<void>;
  releasePage(pageNumber: number): void;
 }

@@ -161,8 +163,12 @@ export async function expectTimelinePromptPositionPreserved(
    .toBeLessThanOrEqual(2);
 }

-export async function openAgentTimeline(page: Page, agent: MockAgentWorkspace): Promise<void> {
-  await page.goto(buildAgentRoute(agent.workspaceId, agent.agentId));
+export async function openAgentTimeline(
+  page: Page,
+  agent: MockAgentWorkspace,
+  serverId?: string,
+): Promise<void> {
+  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<OlderHistoryPages> {
-  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);