fix: async transcript I/O to unblock gateway event loop (#75595)

amknight · web-flow · commit 6f819280a302 · 2026-05-01T22:03:24.000+10:00
* fix: async transcript I/O to unblock gateway event loop Two related fixes for event-loop starvation caused by synchronous file operations on session transcript files during gateway hot paths. ## sessions.list: yield between transcript reads (#75330) Extract filterAndSortSessionEntries() from listSessionsFromStore() and add a new listSessionsFromStoreAsync() that yields to the event loop via setImmediate every 10 session rows. The sessions.list RPC handler now uses the async version. The synchronous version is kept for callers that need it (sessions- resolve visibility checks, embedded backends, subagent tools). The dominant blocker is readSessionTitleFieldsFromTranscript(), which performs fs.statSync + fs.openSync + fs.readSync (head) + fs.readSync (tail) for every session row that requests derived titles or last- message previews. With 100+ sessions, this blocks the event loop for 32-64 seconds, starving WebSocket heartbeats, channel I/O, and concurrent RPC. ## session compaction: async file copy (#75414) Add captureCompactionCheckpointSnapshotAsync() using fs.promises for stat, copyFile, and unlink instead of fsSync equivalents. Switch both compact.ts and compact.queued.ts to the async version. The synchronous copyFileSync of large transcript files (20MB+ observed in production) was blocking the event loop for the entire copy duration — one reporter measured a 43-minute event loop block from a single compaction checkpoint capture. Refs: #75330, #75414 * test: cover async transcript I/O responsiveness * fix: avoid sync checkpoint metadata reads
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -46,6 +46,7 @@ Docs: https://docs.openclaw.ai
 - Gateway/config: cap oversized plugin-owned schemas in the full `config.schema` response so large installed plugin sets cannot balloon Gateway RSS or crash schema clients. Thanks @vincentkoc.
 - Plugins/update: skip ClawHub and marketplace plugin updates when the bundled version is newer than the recorded installed version, so `openclaw update` no longer overwrites working bundled plugins with older external packages. Fixes #75447. Thanks @amknight.
 - Gateway/sessions: use bounded tail reads for sessions-list transcript usage fallbacks and cap bulk title/last-message hydration, keeping large session stores responsive when rows request derived previews. Thanks @vincentkoc.
+- Gateway/sessions: yield during bulk transcript title/preview hydration and copy compaction checkpoints asynchronously, keeping the Gateway event loop responsive for large session stores and large transcripts. Refs #75330 and #75414. Thanks @amknight.
 - Gateway/chat: bound chat-history transcript reads to the requested display window so large session logs no longer OOM the Gateway when clients ask for a small history page. Thanks @vincentkoc.
 - Voice Call/Twilio: honor stored pre-connect TwiML before realtime webhook shortcuts and reject DTMF sequences outside conversation mode, so Meet PIN entry cannot be skipped or silently dropped. Thanks @donkeykong91 and @PfanP.
 - Docs/sandboxing: clarify that sandbox setup scripts (`sandbox-setup.sh`, `sandbox-common-setup.sh`, `sandbox-browser-setup.sh`) are only available from a source checkout, and add inline `docker build` commands for npm-installed users so sandbox image setup works without cloning the repo. Fixes #75485. Thanks @amknight.
diff --git a/src/agents/pi-embedded-runner/compact.queued.ts b/src/agents/pi-embedded-runner/compact.queued.ts
@@ -3,9 +3,10 @@ import { ensureContextEnginesInitialized } from "../../context-engine/init.js";
 import { resolveContextEngine } from "../../context-engine/registry.js";
 import type { ContextEngineRuntimeContext } from "../../context-engine/types.js";
 import {
-  captureCompactionCheckpointSnapshot,
+  captureCompactionCheckpointSnapshotAsync,
   cleanupCompactionCheckpointSnapshot,
   persistSessionCompactionCheckpoint,
+  readSessionLeafIdFromTranscriptAsync,
   resolveSessionCompactionCheckpointReason,
   type CapturedCompactionCheckpointSnapshot,
 } from "../../gateway/session-compaction-checkpoints.js";
@@ -115,8 +116,7 @@ export async function compactEmbeddedPiSession(
         // are notified regardless of which engine is active.
         const engineOwnsCompaction = contextEngine.info.ownsCompaction === true;
         checkpointSnapshot = engineOwnsCompaction
-          ? captureCompactionCheckpointSnapshot({
-              sessionManager: SessionManager.open(params.sessionFile),
+          ? await captureCompactionCheckpointSnapshotAsync({
               sessionFile: params.sessionFile,
             })
           : null;
@@ -200,7 +200,7 @@ export async function compactEmbeddedPiSession(
             try {
               const postLeafId =
                 postCompactionLeafId ??
-                SessionManager.open(postCompactionSessionFile).getLeafId() ??
+                (await readSessionLeafIdFromTranscriptAsync(postCompactionSessionFile)) ??
                 undefined;
               const storedCheckpoint = await persistSessionCompactionCheckpoint({
                 cfg: params.config,
diff --git a/src/agents/pi-embedded-runner/compact.ts b/src/agents/pi-embedded-runner/compact.ts
@@ -11,7 +11,7 @@ import { isAcpRuntimeSpawnAvailable } from "../../acp/runtime/availability.js";
 import type { ThinkLevel } from "../../auto-reply/thinking.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import {
-  captureCompactionCheckpointSnapshot,
+  captureCompactionCheckpointSnapshotAsync,
   cleanupCompactionCheckpointSnapshot,
   persistSessionCompactionCheckpoint,
   resolveSessionCompactionCheckpointReason,
@@ -822,7 +822,7 @@ export async function compactEmbeddedPiSessionDirect(
             : undefined,
         allowedToolNames,
       });
-      checkpointSnapshot = captureCompactionCheckpointSnapshot({
+      checkpointSnapshot = await captureCompactionCheckpointSnapshotAsync({
         sessionManager,
         sessionFile: params.sessionFile,
       });
diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts
@@ -72,7 +72,7 @@ import {
 import { reactivateCompletedSubagentSession } from "../session-subagent-reactivation.js";
 import {
   archiveFileOnDisk,
-  listSessionsFromStore,
+  listSessionsFromStoreAsync,
   loadCombinedSessionStoreForGateway,
   loadGatewaySessionRow,
   loadSessionEntry,
@@ -650,7 +650,7 @@ export const sessionsHandlers: GatewayRequestHandlers = {
     const cfg = context.getRuntimeConfig();
     const { storePath, store } = loadCombinedSessionStoreForGateway(cfg);
     const modelCatalog = await loadOptionalSessionsListModelCatalog(context);
-    const result = listSessionsFromStore({
+    const result = await listSessionsFromStoreAsync({
       cfg,
       storePath,
       store,
diff --git a/src/gateway/server.sessions.list-changed.test.ts b/src/gateway/server.sessions.list-changed.test.ts
@@ -157,6 +157,76 @@ test("sessions.list uses the gateway model catalog for effective thinking defaul
   );
 });
 
+test("sessions.list yields before responding during bulk transcript hydration", async () => {
+  const { dir } = await createSessionStoreDir();
+  const entries: Record<string, ReturnType<typeof sessionStoreEntry>> = {};
+  const now = Date.now();
+  for (let i = 0; i < 11; i += 1) {
+    const sessionId = `sess-list-yield-${i}`;
+    entries[`bulk-${i}`] = sessionStoreEntry(sessionId, { updatedAt: now - i });
+    await fs.writeFile(
+      path.join(dir, `${sessionId}.jsonl`),
+      [
+        JSON.stringify({ type: "session", version: 1, id: sessionId }),
+        JSON.stringify({ message: { role: "user", content: `title ${i}` } }),
+        JSON.stringify({ message: { role: "assistant", content: `last ${i}` } }),
+      ].join("\n"),
+      "utf-8",
+    );
+  }
+  await writeSessionStore({ entries });
+
+  const respond = vi.fn();
+  const sessionsHandlers = await getSessionsHandlers();
+  const { getRuntimeConfig } = await getGatewayConfigModule();
+  const request = sessionsHandlers["sessions.list"]({
+    req: {
+      type: "req",
+      id: "req-sessions-list-yield",
+      method: "sessions.list",
+      params: {
+        includeDerivedTitles: true,
+        includeLastMessage: true,
+        limit: 11,
+      },
+    },
+    params: {
+      includeDerivedTitles: true,
+      includeLastMessage: true,
+      limit: 11,
+    },
+    respond,
+    client: null,
+    isWebchatConnect: () => false,
+    context: {
+      getRuntimeConfig,
+      loadGatewayModelCatalog: async () => [],
+      logGateway: {
+        debug: vi.fn(),
+      },
+    } as never,
+  });
+
+  await Promise.resolve();
+  await Promise.resolve();
+
+  expect(respond).not.toHaveBeenCalled();
+  await request;
+  expect(respond).toHaveBeenCalledWith(
+    true,
+    expect.objectContaining({
+      sessions: expect.arrayContaining([
+        expect.objectContaining({
+          key: "agent:main:bulk-0",
+          derivedTitle: "title 0",
+          lastMessagePreview: "last 0",
+        }),
+      ]),
+    }),
+    undefined,
+  );
+});
+
 test("sessions.list does not block on slow model catalog discovery", async () => {
   await createSessionStoreDir();
   await writeSessionStore({
diff --git a/src/gateway/session-compaction-checkpoints.test.ts b/src/gateway/session-compaction-checkpoints.test.ts
@@ -4,13 +4,15 @@ import os from "node:os";
 import path from "node:path";
 import type { AssistantMessage, UserMessage } from "@mariozechner/pi-ai";
 import { SessionManager } from "@mariozechner/pi-coding-agent";
-import { afterEach, describe, expect, test } from "vitest";
+import { afterEach, describe, expect, test, vi } from "vitest";
 import type { OpenClawConfig } from "../config/types.openclaw.js";
 import {
   captureCompactionCheckpointSnapshot,
+  captureCompactionCheckpointSnapshotAsync,
   cleanupCompactionCheckpointSnapshot,
   MAX_COMPACTION_CHECKPOINT_SNAPSHOT_BYTES,
   persistSessionCompactionCheckpoint,
+  readSessionLeafIdFromTranscriptAsync,
 } from "./session-compaction-checkpoints.js";
 
 const tempDirs: string[] = [];
@@ -85,6 +87,144 @@ describe("session-compaction-checkpoints", () => {
     expect(fsSync.existsSync(sessionFile!)).toBe(true);
   });
 
+  test("async capture stores the copied pre-compaction transcript without sync copy", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-async-"));
+    tempDirs.push(dir);
+
+    const session = SessionManager.create(dir, dir);
+    session.appendMessage({
+      role: "user",
+      content: "before async compaction",
+      timestamp: Date.now(),
+    });
+    session.appendMessage({
+      role: "assistant",
+      content: [{ type: "text", text: "async working on it" }],
+      api: "responses",
+      provider: "openai",
+      model: "gpt-test",
+      timestamp: Date.now(),
+    } as AssistantMessage);
+
+    const sessionFile = session.getSessionFile();
+    const leafId = session.getLeafId();
+    expect(sessionFile).toBeTruthy();
+    expect(leafId).toBeTruthy();
+
+    const originalBefore = await fs.readFile(sessionFile!, "utf-8");
+    const copyFileSyncSpy = vi.spyOn(fsSync, "copyFileSync");
+    const sessionManagerOpenSpy = vi.spyOn(SessionManager, "open");
+    try {
+      const snapshot = await captureCompactionCheckpointSnapshotAsync({
+        sessionManager: session,
+        sessionFile: sessionFile!,
+      });
+
+      expect(copyFileSyncSpy).not.toHaveBeenCalled();
+      expect(sessionManagerOpenSpy).not.toHaveBeenCalled();
+      expect(snapshot).not.toBeNull();
+      expect(snapshot?.leafId).toBe(leafId);
+      expect(snapshot?.sessionFile).not.toBe(sessionFile);
+      expect(snapshot?.sessionFile).toContain(".checkpoint.");
+      expect(fsSync.existsSync(snapshot!.sessionFile)).toBe(true);
+      expect(await fs.readFile(snapshot!.sessionFile, "utf-8")).toBe(originalBefore);
+
+      session.appendCompaction("checkpoint summary", leafId!, 123, { ok: true });
+
+      expect(await fs.readFile(snapshot!.sessionFile, "utf-8")).toBe(originalBefore);
+      expect(await fs.readFile(sessionFile!, "utf-8")).not.toBe(originalBefore);
+
+      await cleanupCompactionCheckpointSnapshot(snapshot);
+
+      expect(fsSync.existsSync(snapshot!.sessionFile)).toBe(false);
+      expect(fsSync.existsSync(sessionFile!)).toBe(true);
+    } finally {
+      copyFileSyncSpy.mockRestore();
+      sessionManagerOpenSpy.mockRestore();
+    }
+  });
+
+  test("async capture derives session metadata without synchronous SessionManager.open", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-async-metadata-"));
+    tempDirs.push(dir);
+
+    const session = SessionManager.create(dir, dir);
+    session.appendMessage({
+      role: "user",
+      content: "derive checkpoint metadata",
+      timestamp: Date.now(),
+    });
+    session.appendMessage({
+      role: "assistant",
+      content: "metadata derived",
+      api: "responses",
+      provider: "openai",
+      model: "gpt-test",
+      timestamp: Date.now(),
+    } as unknown as AssistantMessage);
+
+    const sessionFile = session.getSessionFile();
+    const sessionId = session.getSessionId();
+    const leafId = session.getLeafId();
+    expect(sessionFile).toBeTruthy();
+    expect(sessionId).toBeTruthy();
+    expect(leafId).toBeTruthy();
+    await fs.appendFile(sessionFile!, "\nnot-json\n", "utf-8");
+
+    const copyFileSyncSpy = vi.spyOn(fsSync, "copyFileSync");
+    const sessionManagerOpenSpy = vi.spyOn(SessionManager, "open");
+    let snapshot: Awaited<ReturnType<typeof captureCompactionCheckpointSnapshotAsync>> = null;
+    try {
+      expect(await readSessionLeafIdFromTranscriptAsync(sessionFile!)).toBe(leafId);
+      snapshot = await captureCompactionCheckpointSnapshotAsync({
+        sessionFile: sessionFile!,
+      });
+
+      expect(copyFileSyncSpy).not.toHaveBeenCalled();
+      expect(sessionManagerOpenSpy).not.toHaveBeenCalled();
+      expect(snapshot).not.toBeNull();
+      expect(snapshot?.sessionId).toBe(sessionId);
+      expect(snapshot?.leafId).toBe(leafId);
+      expect(snapshot?.sessionFile).not.toBe(sessionFile);
+      expect(snapshot?.sessionFile).toContain(".checkpoint.");
+    } finally {
+      await cleanupCompactionCheckpointSnapshot(snapshot);
+      copyFileSyncSpy.mockRestore();
+      sessionManagerOpenSpy.mockRestore();
+    }
+  });
+
+  test("async capture skips oversized pre-compaction transcripts without sync copy", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-async-oversized-"));
+    tempDirs.push(dir);
+
+    const session = SessionManager.create(dir, dir);
+    session.appendMessage({
+      role: "user",
+      content: "before compaction",
+      timestamp: Date.now(),
+    });
+    const sessionFile = session.getSessionFile();
+    expect(sessionFile).toBeTruthy();
+    await fs.appendFile(sessionFile!, "x".repeat(128), "utf-8");
+
+    const copyFileSyncSpy = vi.spyOn(fsSync, "copyFileSync");
+    try {
+      const snapshot = await captureCompactionCheckpointSnapshotAsync({
+        sessionManager: session,
+        sessionFile: sessionFile!,
+        maxBytes: 64,
+      });
+
+      expect(snapshot).toBeNull();
+      expect(copyFileSyncSpy).not.toHaveBeenCalled();
+      expect(MAX_COMPACTION_CHECKPOINT_SNAPSHOT_BYTES).toBeGreaterThan(64);
+      expect(fsSync.readdirSync(dir).filter((file) => file.includes(".checkpoint."))).toEqual([]);
+    } finally {
+      copyFileSyncSpy.mockRestore();
+    }
+  });
+
   test("capture skips oversized pre-compaction transcripts", async () => {
     const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-checkpoint-oversized-"));
     tempDirs.push(dir);
diff --git a/src/gateway/session-compaction-checkpoints.ts b/src/gateway/session-compaction-checkpoints.ts
diff --git a/src/gateway/session-utils.test.ts b/src/gateway/session-utils.test.ts
diff --git a/src/gateway/session-utils.ts b/src/gateway/session-utils.ts
