fix(omp): keep transcript replay responsive

Index transcript branches in bounded asynchronous chunks and retain payloads only for the active chain, so large histories do not block unrelated daemon requests.

packages/server/src/server/agent/providers/omp/agent.test.ts

@@ -1133,6 +1133,17 @@ describe("OMP agent client and session", () => {
         messageId: "assistant-history",
       },
     ]);
+    await expect(
+      omp.runPrompt("continue after resume", "resumed follow-up complete"),
+    ).resolves.toMatchObject({
+      finalText: "resumed follow-up complete",
+    });
+    expect(omp.timeline()).toContainEqual(
+      expect.objectContaining({
+        type: "assistant_message",
+        text: "resumed follow-up complete",
+      }),
+    );
     expect(omp.usageSession()).toMatchObject({ provider: "omp", sessionKey: expect.any(String) });
     await omp.close();
     expect(omp.usageSession()).toBeNull();

packages/server/src/server/agent/providers/omp/history-mapper.test.ts

@@ -1,4 +1,5 @@
 import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { describe, expect, test } from "vitest";
@@ -8,7 +9,7 @@ import { streamOmpCoreHistory, type OmpCapturedUserMessageEntry } from "./messag
 import type { OmpAgentMessage } from "./rpc-types.js";
 import { FakeOmp } from "./test-utils/fake-omp.js";
 import { OMP_HISTORY_MAPPER_HOOKS } from "./history-hooks.js";
-import { readOmpHistoryTodoState, streamOmpHistory } from "./history.js";
+import { readActiveOmpEntryChain, readOmpHistoryTodoState, streamOmpHistory } from "./history.js";
 
 async function collectHistory(
   messages: OmpAgentMessage[],
@@ -1245,3 +1246,53 @@ describe("OMP history mapper", () => {
     );
   });
 });
+
+test("indexes out-of-order branches, duplicate IDs and cycles before selecting history", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "omp-history-branches-"));
+  const sessionFile = join(directory, "session.jsonl");
+  const root = { id: "root", parentId: null };
+  const earlyLeaf = {
+    id: "early",
+    parentId: "root",
+    message: { role: "user", content: "你好🌍".repeat(15000) },
+  };
+  const branch = { id: "branch", parentId: "root" };
+  const active = { id: "active", parentId: "branch" };
+  const duplicate = {
+    id: "branch",
+    parentId: "root",
+    message: { role: "user", content: "replacement" },
+  };
+  try {
+    await writeFile(
+      sessionFile,
+      [earlyLeaf, active, root, branch, duplicate]
+        .map((entry) => JSON.stringify(entry))
+        .join("\n") + '\n{"id":',
+    );
+    // The final record is a branch ancestor, not the active leaf. Latest duplicate wins
+    // when following parents; the invalid tail contributes neither a leaf nor a parent.
+    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual([root, duplicate, active]);
+    await expect(readActiveOmpEntryChain(sessionFile, "early")).resolves.toEqual([root, earlyLeaf]);
+    await expect(readActiveOmpEntryChain(sessionFile, "missing")).resolves.toEqual([
+      root,
+      duplicate,
+      active,
+    ]);
+    const cycle = [
+      { id: "a", parentId: "b" },
+      { id: "b", parentId: "a" },
+    ];
+    await writeFile(sessionFile, cycle.map((entry) => JSON.stringify(entry)).join("\n"));
+    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual(cycle);
+    await writeFile(sessionFile, "null\n{}\ninvalid\n");
+    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual([]);
+    await expect(readActiveOmpEntryChain(directory)).rejects.toMatchObject({ code: "EISDIR" });
+    await expect(readActiveOmpEntryChain(join(directory, "missing"))).rejects.toMatchObject({
+      code: "ENOENT",
+    });
+    await expect(readOmpHistoryTodoState(join(directory, "missing"))).resolves.toBeNull();
+  } finally {
+    await rm(directory, { recursive: true, force: true });
+  }
+});

packages/server/src/server/agent/providers/omp/history.e2e.test.ts

@@ -0,0 +1,89 @@
+import { mkdtemp, open, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { performance } from "node:perf_hooks";
+import { setTimeout as delay } from "node:timers/promises";
+import { expect, test } from "vitest";
+import { createTestPaseoDaemon } from "../../../test-utils/paseo-daemon.js";
+import { DaemonClient } from "../../../test-utils/daemon-client.js";
+import { readActiveOmpEntryChain } from "./history.js";
+
+test("replays a 300 MiB active chain without blocking the event loop", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "omp-history-budget-"));
+  const sessionFile = join(directory, "session.jsonl");
+  try {
+    const daemon = await createTestPaseoDaemon();
+    const client = new DaemonClient({
+      url: `ws://127.0.0.1:${daemon.port}/ws`,
+      appVersion: "0.13.2",
+    });
+    try {
+      await client.connect();
+      await client.fetchAgents({ subscribe: {} });
+      await client.getDaemonStatus();
+      const file = await open(sessionFile, "w");
+      const text = "x".repeat(16 * 1024);
+      let bytes = 0;
+      let count = 0;
+      try {
+        while (bytes < 300 * 1024 * 1024) {
+          const line =
+            JSON.stringify({
+              type: "message",
+              id: `row-${count}`,
+              parentId: count ? `row-${count - 1}` : null,
+              message: { role: "user", content: text },
+            }) + "\n";
+          bytes += Buffer.byteLength(line);
+          await file.write(line);
+          count++;
+        }
+      } finally {
+        await file.close();
+      }
+      let last = performance.now();
+      let longestGap = 0;
+      const probe = setInterval(() => {
+        const now = performance.now();
+        longestGap = Math.max(longestGap, now - last);
+        last = now;
+      }, 5);
+      try {
+        const replayState = { finished: false };
+        let repliesDuringReplay = 0;
+        const replay = readActiveOmpEntryChain(sessionFile).finally(() => {
+          replayState.finished = true;
+        });
+        // A real WebSocket RPC served by the isolated daemon on the same event loop.
+        const requests = (async () => {
+          while (!replayState.finished) {
+            await client.getDaemonStatus();
+            if (!replayState.finished) repliesDuringReplay++;
+            await delay(10);
+          }
+        })();
+        const [entries] = await Promise.all([replay, requests]);
+        await delay(10);
+        console.log(JSON.stringify({ bytes, count, longestGap, repliesDuringReplay }));
+        expect(repliesDuringReplay).toBeGreaterThan(0);
+        expect(entries.length).toBe(count);
+        for (let index = 0; index < entries.length; index++) {
+          expect(entries[index]).toEqual({
+            type: "message",
+            id: `row-${index}`,
+            parentId: index ? `row-${index - 1}` : null,
+            message: { role: "user", content: text },
+          });
+        }
+        expect(longestGap).toBeLessThan(50);
+      } finally {
+        clearInterval(probe);
+      }
+    } finally {
+      await client.close();
+      await daemon.close();
+    }
+  } finally {
+    await rm(directory, { recursive: true, force: true });
+  }
+}, 30000);

packages/server/src/server/agent/providers/omp/history.ts

@@ -1,4 +1,5 @@
-import { readFile } from "node:fs/promises";
+import { open, type FileHandle } from "node:fs/promises";
+import { setImmediate } from "node:timers/promises";
 import { basename, extname, join } from "node:path";
 import type { AgentProvider, AgentStreamEvent, AgentTimelineItem } from "../../agent-sdk-types.js";
 import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
@@ -96,7 +97,9 @@ export async function* streamOmpHistory(input: {
   const messages: OmpAgentMessage[] = [];
   const replayEntries: OmpReplayEntry[] = [];
   const userEntries: OmpCapturedUserMessageEntry[] = [];
+  let prepared = 0;
   for (const entry of entries) {
+    if (++prepared % 32 === 0) await setImmediate();
     if (entry.type === "compaction") {
       replayEntries.push({ entry, item: mapCompactionEntry(entry) });
       continue;
@@ -115,7 +118,9 @@ export async function* streamOmpHistory(input: {
     OMP_HISTORY_MAPPER_HOOKS,
     input.bridgedTools,
   );
+  let replayed = 0;
   for (const replayEntry of replayEntries) {
+    if (++replayed % 32 === 0) await setImmediate();
     const timestamp = normalizeProviderReplayTimestamp(replayEntry.entry.timestamp);
     const events: AgentStreamEvent[] =
       "item" in replayEntry
@@ -329,34 +334,139 @@ function stripExtension(filePath: string): string {
   return extension ? filePath.slice(0, -extension.length) : filePath;
 }
 
-export async function readActiveOmpEntryChain(
-  sessionFile: string,
-  activeEntryId?: string,
-): Promise<OmpSessionEntry[]> {
-  const content = await readFile(sessionFile, "utf8");
-  const entries = content.split("\n").flatMap((line) => {
-    if (!line.trim()) return [];
-    try {
-      const value = JSON.parse(line) as OmpSessionEntry;
-      return value && typeof value === "object" && typeof value.id === "string" ? [value] : [];
-    } catch {
-      return [];
+interface OmpEntryLocation {
+  id: string;
+  parentId?: string | null;
+  offset: number;
+}
+
+// Yield after each bounded IO chunk, including chunks containing only tiny or invalid records.
+// Keep bytes until a complete line so multi-byte UTF-8 characters survive chunk boundaries.
+async function* readOmpLines(
+  file: FileHandle,
+  size: number,
+): AsyncGenerator<{
+  line: Buffer;
+  offset: number;
+}> {
+  const chunkSize = 64 * 1024;
+  let fragments: Buffer[] = [];
+  let lineOffset = 0;
+  for (let position = 0; position < size; ) {
+    const chunk = Buffer.allocUnsafe(Math.min(chunkSize, size - position));
+    const { bytesRead } = await file.read(chunk, 0, chunk.length, position);
+    if (!bytesRead) break;
+    let start = 0;
+    for (let index = 0; index < bytesRead; index++) {
+      if (chunk[index] !== 10) continue;
+      const tail = chunk.subarray(start, index);
+      const line = fragments.length ? Buffer.concat([...fragments, tail]) : tail;
+      yield { line, offset: lineOffset };
+      fragments = [];
+      start = index + 1;
+      lineOffset = position + start;
     }
-  });
-  if (entries.length === 0) return [];
-  const byId = new Map(entries.map((entry) => [entry.id!, entry]));
-  const parentIds = new Set(entries.flatMap((entry) => (entry.parentId ? [entry.parentId] : [])));
-  const leaves = entries.filter((entry) => !parentIds.has(entry.id!));
-  let current: OmpSessionEntry | undefined =
-    (activeEntryId ? byId.get(activeEntryId) : undefined) ?? leaves.at(-1) ?? entries.at(-1);
-  const chain: OmpSessionEntry[] = [];
+    if (start < bytesRead) fragments.push(chunk.subarray(start, bytesRead));
+    position += bytesRead;
+    await setImmediate();
+  }
+  if (fragments.length) yield { line: Buffer.concat(fragments), offset: lineOffset };
+}
+
+function parseOmpEntry(line: Buffer): OmpSessionEntry | null {
+  try {
+    const value: unknown = JSON.parse(line.toString("utf8"));
+    return value && typeof value === "object" && typeof Reflect.get(value, "id") === "string"
+      ? (value as OmpSessionEntry)
+      : null;
+  } catch {
+    // OMP can leave a partial last record; malformed records have always been ignored.
+    return null;
+  }
+}
+
+async function selectOmpChain(
+  locations: readonly OmpEntryLocation[],
+  byId: ReadonlyMap<string, OmpEntryLocation>,
+  parentIds: ReadonlySet<string>,
+  activeEntryId?: string,
+): Promise<OmpEntryLocation[]> {
+  let current = activeEntryId ? byId.get(activeEntryId) : undefined;
+  if (!current) {
+    for (let index = locations.length - 1; index >= 0; index--) {
+      const location = locations[index]!;
+      if (!parentIds.has(location.id)) {
+        current = location;
+        break;
+      }
+      if (index % 256 === 0) await setImmediate();
+    }
+    current ??= locations.at(-1);
+  }
+  const chain: OmpEntryLocation[] = [];
   const seen = new Set<string>();
   while (current?.id && !seen.has(current.id)) {
     chain.push(current);
     seen.add(current.id);
     current = current.parentId ? byId.get(current.parentId) : undefined;
+    if (chain.length % 256 === 0) await setImmediate();
+  }
+  return chain;
+}
+
+export async function readActiveOmpEntryChain(
+  sessionFile: string,
+  activeEntryId?: string,
+): Promise<OmpSessionEntry[]> {
+  const file = await open(sessionFile, "r");
+  try {
+    const stat = await file.stat();
+    if (stat.isDirectory()) {
+      throw Object.assign(
+        new Error(`EISDIR: illegal operation on a directory, read '${sessionFile}'`),
+        {
+          code: "EISDIR",
+        },
+      );
+    }
+    // One descriptor and a fixed byte boundary exclude appends made while replay is yielding.
+    // Retain graph metadata only: abandoned branches must not retain their message payloads.
+    const locations: OmpEntryLocation[] = [];
+    const byId = new Map<string, OmpEntryLocation>();
+    const parentIds = new Set<string>();
+    for await (const { line, offset } of readOmpLines(file, stat.size)) {
+      const entry = parseOmpEntry(line);
+      if (!entry) continue;
+      const location = { id: entry.id!, parentId: entry.parentId, offset };
+      locations.push(location);
+      byId.set(location.id, location);
+      if (location.parentId) parentIds.add(location.parentId);
+    }
+    const chain = await selectOmpChain(locations, byId, parentIds, activeEntryId);
+    if (chain.length === 0) return [];
+    const selected = new Map<number, number>();
+    for (let index = 0; index < chain.length; index++) {
+      selected.set(chain[index]!.offset, chain.length - index - 1);
+      if (index % 256 === 0) await setImmediate();
+    }
+    const entries: OmpSessionEntry[] = [];
+    for await (const { line, offset } of readOmpLines(file, stat.size)) {
+      const index = selected.get(offset);
+      if (index === undefined) continue;
+      const entry = parseOmpEntry(line);
+      // Truncation in place invalidates the indexed snapshot; never publish a partial chain.
+      if (!entry || entry.id !== chain[chain.length - index - 1]!.id) {
+        throw new Error(`OMP transcript changed during replay: ${sessionFile}`);
+      }
+      entries[index] = entry;
+      selected.delete(offset);
+      if (selected.size === 0) break;
+    }
+    if (selected.size) throw new Error(`OMP transcript changed during replay: ${sessionFile}`);
+    return entries;
+  } finally {
+    await file.close();
   }
-  return chain.toReversed();
 }
 
 function mapEntryMessage(entry: OmpSessionEntry): OmpAgentMessage | null {