refactor(gateway): remove sync session reader surface (#75909)

2026-05-06 09:10:45 +00:00 · 2026-05-02 03:49:36 +01:00
parent 8c8cf79687
commit ee8371d313
8 changed files with 85 additions and 68 deletions
--- a/src/agents/cli-runner.ts
+++ b/src/agents/cli-runner.ts
@@ -129,7 +129,7 @@ export async function runPreparedCliAgent(
  const hasAgentEndHooks = hookRunner?.hasHooks("agent_end") === true;
  const historyMessages =
    hasLlmInputHooks || hasAgentEndHooks
-      ? loadCliSessionHistoryMessages({
+      ? await loadCliSessionHistoryMessages({
          sessionId: params.sessionId,
          sessionFile: params.sessionFile,
          sessionKey: params.sessionKey,
--- a/src/agents/cli-runner/prepare.ts
+++ b/src/agents/cli-runner/prepare.ts
@@ -274,8 +274,8 @@ export async function prepareCliRunContext(
    );
  }
  let openClawHistoryMessages: unknown[] | undefined;
-  const loadOpenClawHistoryMessages = () => {
-    openClawHistoryMessages ??= loadCliSessionHistoryMessages({
+  const loadOpenClawHistoryMessages = async () => {
+    openClawHistoryMessages ??= await loadCliSessionHistoryMessages({
      sessionId: params.sessionId,
      sessionFile: params.sessionFile,
      sessionKey: params.sessionKey,
@@ -340,7 +340,7 @@ export async function prepareCliRunContext(
    const hookResult = await resolvePromptBuildHookResult({
      config: params.config ?? getRuntimeConfig(),
      prompt: params.prompt,
-      messages: loadOpenClawHistoryMessages(),
+      messages: await loadOpenClawHistoryMessages(),
      hookCtx: {
        runId: params.runId,
        agentId: sessionAgentId,
@@ -382,7 +382,7 @@ export async function prepareCliRunContext(
  const openClawHistoryPrompt = reusableCliSession.sessionId
    ? undefined
    : buildCliSessionHistoryPrompt({
-        messages: loadCliSessionReseedMessages({
+        messages: await loadCliSessionReseedMessages({
          sessionId: params.sessionId,
          sessionFile: params.sessionFile,
          sessionKey: params.sessionKey,
--- a/src/agents/cli-runner/session-history.test.ts
+++ b/src/agents/cli-runner/session-history.test.ts
@@ -64,7 +64,7 @@ describe("loadCliSessionHistoryMessages", () => {
    vi.unstubAllEnvs();
  });

-  it("reads the canonical session transcript instead of an arbitrary external path", () => {
+  it("reads the canonical session transcript instead of an arbitrary external path", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    const outsideDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-outside-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
@@ -82,7 +82,7 @@ describe("loadCliSessionHistoryMessages", () => {

    try {
      expect(
-        loadCliSessionHistoryMessages({
+        await loadCliSessionHistoryMessages({
          sessionId: "session-test",
          sessionFile: outsideFile,
          sessionKey: "agent:main:main",
@@ -95,7 +95,7 @@ describe("loadCliSessionHistoryMessages", () => {
    }
  });

-  it("keeps only the newest bounded history window", () => {
+  it("keeps only the newest bounded history window", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
    const sessionFile = createSessionTranscript({
@@ -108,7 +108,7 @@ describe("loadCliSessionHistoryMessages", () => {
    });

    try {
-      const history = loadCliSessionHistoryMessages({
+      const history = await loadCliSessionHistoryMessages({
        sessionId: "session-bounded",
        sessionFile,
        sessionKey: "agent:main:main",
@@ -125,7 +125,7 @@ describe("loadCliSessionHistoryMessages", () => {
    }
  });

-  it("rejects symlinked transcripts instead of following them outside the sessions directory", () => {
+  it("rejects symlinked transcripts instead of following them outside the sessions directory", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    const outsideDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-outside-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
@@ -147,7 +147,7 @@ describe("loadCliSessionHistoryMessages", () => {

    try {
      expect(
-        loadCliSessionHistoryMessages({
+        await loadCliSessionHistoryMessages({
          sessionId: "session-symlink",
          sessionFile: canonicalSessionFile,
          sessionKey: "agent:main:main",
@@ -160,7 +160,7 @@ describe("loadCliSessionHistoryMessages", () => {
    }
  });

-  it("drops oversized transcript files instead of loading them into hook payloads", () => {
+  it("drops oversized transcript files instead of loading them into hook payloads", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
    const sessionFile = path.join(
@@ -175,7 +175,7 @@ describe("loadCliSessionHistoryMessages", () => {

    try {
      expect(
-        loadCliSessionHistoryMessages({
+        await loadCliSessionHistoryMessages({
          sessionId: "session-oversized",
          sessionFile,
          sessionKey: "agent:main:main",
@@ -187,7 +187,7 @@ describe("loadCliSessionHistoryMessages", () => {
    }
  });

-  it("honors custom session store roots when resolving hook history transcripts", () => {
+  it("honors custom session store roots when resolving hook history transcripts", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    const customStoreDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-store-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
@@ -202,7 +202,7 @@ describe("loadCliSessionHistoryMessages", () => {

    try {
      expect(
-        loadCliSessionHistoryMessages({
+        await loadCliSessionHistoryMessages({
          sessionId: "session-custom-store",
          sessionFile,
          sessionKey: "agent:main:main",
@@ -226,7 +226,7 @@ describe("loadCliSessionReseedMessages", () => {
    vi.unstubAllEnvs();
  });

-  it("does not reseed fresh CLI sessions from raw transcript history before compaction", () => {
+  it("does not reseed fresh CLI sessions from raw transcript history before compaction", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
    const sessionFile = createSessionTranscript({
@@ -237,7 +237,7 @@ describe("loadCliSessionReseedMessages", () => {

    try {
      expect(
-        loadCliSessionReseedMessages({
+        await loadCliSessionReseedMessages({
          sessionId: "session-no-compaction",
          sessionFile,
          sessionKey: "agent:main:main",
@@ -249,7 +249,7 @@ describe("loadCliSessionReseedMessages", () => {
    }
  });

-  it("reseeds fresh CLI sessions from the latest compaction summary and post-compaction tail", () => {
+  it("reseeds fresh CLI sessions from the latest compaction summary and post-compaction tail", async () => {
    const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-state-"));
    vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
    const sessionFile = createSessionTranscript({
@@ -287,7 +287,7 @@ describe("loadCliSessionReseedMessages", () => {
    );

    try {
-      const reseed = loadCliSessionReseedMessages({
+      const reseed = await loadCliSessionReseedMessages({
        sessionId: "session-compacted",
        sessionFile,
        sessionKey: "agent:main:main",
--- a/src/agents/cli-runner/session-history.ts
+++ b/src/agents/cli-runner/session-history.ts
@@ -1,6 +1,6 @@
-import fs from "node:fs";
+import fsp from "node:fs/promises";
 import path from "node:path";
-import { SessionManager } from "@mariozechner/pi-coding-agent";
+import { migrateSessionEntries, parseSessionEntries } from "@mariozechner/pi-coding-agent";
 import {
  resolveSessionFilePath,
  resolveSessionFilePathOptions,
@@ -100,9 +100,9 @@ export function buildCliSessionHistoryPrompt(params: {
  ].join("\n");
 }

-function safeRealpathSync(filePath: string): string | undefined {
+async function safeRealpath(filePath: string): Promise<string | undefined> {
  try {
-    return fs.realpathSync(filePath);
+    return await fsp.realpath(filePath);
  } catch {
    return undefined;
  }
@@ -140,56 +140,58 @@ function resolveSafeCliSessionFile(params: {
  };
 }

-function loadCliSessionEntries(params: {
+async function loadCliSessionEntries(params: {
  sessionId: string;
  sessionFile: string;
  sessionKey?: string;
  agentId?: string;
  config?: OpenClawConfig;
-}): unknown[] {
+}): Promise<unknown[]> {
  try {
    const { sessionFile, sessionsDir } = resolveSafeCliSessionFile(params);
-    const entryStat = fs.lstatSync(sessionFile);
+    const entryStat = await fsp.lstat(sessionFile);
    if (!entryStat.isFile() || entryStat.isSymbolicLink()) {
      return [];
    }
-    const realSessionsDir = safeRealpathSync(sessionsDir) ?? path.resolve(sessionsDir);
-    const realSessionFile = safeRealpathSync(sessionFile);
+    const realSessionsDir = (await safeRealpath(sessionsDir)) ?? path.resolve(sessionsDir);
+    const realSessionFile = await safeRealpath(sessionFile);
    if (!realSessionFile || !isPathWithinBase(realSessionsDir, realSessionFile)) {
      return [];
    }
-    const stat = fs.statSync(realSessionFile);
+    const stat = await fsp.stat(realSessionFile);
    if (!stat.isFile() || stat.size > MAX_CLI_SESSION_HISTORY_FILE_BYTES) {
      return [];
    }
-    return SessionManager.open(realSessionFile).getEntries();
+    const entries = parseSessionEntries(await fsp.readFile(realSessionFile, "utf-8"));
+    migrateSessionEntries(entries);
+    return entries.filter((entry) => entry.type !== "session");
  } catch {
    return [];
  }
 }

-export function loadCliSessionHistoryMessages(params: {
+export async function loadCliSessionHistoryMessages(params: {
  sessionId: string;
  sessionFile: string;
  sessionKey?: string;
  agentId?: string;
  config?: OpenClawConfig;
-}): unknown[] {
-  const history = loadCliSessionEntries(params).flatMap((entry) => {
+}): Promise<unknown[]> {
+  const history = (await loadCliSessionEntries(params)).flatMap((entry) => {
    const candidate = entry as HistoryEntry;
    return candidate.type === "message" ? [candidate.message] : [];
  });
  return limitAgentHookHistoryMessages(history, MAX_CLI_SESSION_HISTORY_MESSAGES);
 }

-export function loadCliSessionReseedMessages(params: {
+export async function loadCliSessionReseedMessages(params: {
  sessionId: string;
  sessionFile: string;
  sessionKey?: string;
  agentId?: string;
  config?: OpenClawConfig;
-}): unknown[] {
-  const entries = loadCliSessionEntries(params);
+}): Promise<unknown[]> {
+  const entries = await loadCliSessionEntries(params);
  const latestCompactionIndex = entries.findLastIndex((entry) => {
    const candidate = entry as HistoryEntry;
    return candidate.type === "compaction" && typeof candidate.summary === "string";
--- a/src/auto-reply/reply/commands-export-session.test.ts
+++ b/src/auto-reply/reply/commands-export-session.test.ts
@@ -13,9 +13,6 @@ const hoisted = await vi.hoisted(async () => {
      injectedFiles: [],
      sandboxRuntime: { sandboxed: false, mode: "off" },
    })),
-    getEntriesMock: vi.fn(() => []),
-    getHeaderMock: vi.fn(() => null),
-    getLeafIdMock: vi.fn(() => null),
    writeFileSyncMock: vi.fn(),
    mkdirSyncMock: vi.fn(),
    existsSyncMock: vi.fn(() => true),
@@ -23,16 +20,6 @@ const hoisted = await vi.hoisted(async () => {
  };
 });

-vi.mock("@mariozechner/pi-coding-agent", () => ({
-  SessionManager: {
-    open: vi.fn(() => ({
-      getEntries: hoisted.getEntriesMock,
-      getHeader: hoisted.getHeaderMock,
-      getLeafId: hoisted.getLeafIdMock,
-    })),
-  },
-}));
-
 vi.mock("../../config/sessions/paths.js", () => ({
  resolveDefaultSessionStorePath: hoisted.resolveDefaultSessionStorePathMock,
  resolveSessionFilePath: hoisted.resolveSessionFilePathMock,
@@ -72,6 +59,23 @@ vi.mock("node:fs", async () => {
  };
 });

+vi.mock("node:fs/promises", async () => {
+  const actual = await vi.importActual<typeof import("node:fs/promises")>("node:fs/promises");
+  const mockedFsPromises = {
+    ...actual,
+    readFile: vi.fn(async (filePath: string, encoding?: BufferEncoding) => {
+      if (filePath === "/tmp/target-store/session.jsonl") {
+        return "";
+      }
+      return actual.readFile(filePath, encoding);
+    }),
+  };
+  return {
+    ...mockedFsPromises,
+    default: mockedFsPromises,
+  };
+});
+
 function makeParams(): HandleCommandsParams {
  return {
    cfg: {},
--- a/src/auto-reply/reply/commands-export-session.ts
+++ b/src/auto-reply/reply/commands-export-session.ts
@@ -1,8 +1,13 @@
 import fs from "node:fs";
+import fsp from "node:fs/promises";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
-import type { SessionEntry as PiSessionEntry, SessionHeader } from "@mariozechner/pi-coding-agent";
-import { SessionManager } from "@mariozechner/pi-coding-agent";
+import {
+  migrateSessionEntries,
+  parseSessionEntries,
+  type SessionEntry as PiSessionEntry,
+  type SessionHeader,
+} from "@mariozechner/pi-coding-agent";
 import type { ReplyPayload } from "../types.js";
 import {
  isReplyPayload,
@@ -116,6 +121,22 @@ function generateHtml(sessionData: SessionData): string {
  ].reduce((html, [name, value]) => replaceHtmlPlaceholder(html, name, value), template);
 }

+async function readSessionDataFromTranscript(sessionFile: string): Promise<{
+  header: SessionHeader | null;
+  entries: PiSessionEntry[];
+  leafId: string | null;
+}> {
+  const raw = await fsp.readFile(sessionFile, "utf-8");
+  const fileEntries = parseSessionEntries(raw);
+  migrateSessionEntries(fileEntries);
+  const header =
+    fileEntries.find((entry): entry is SessionHeader => entry.type === "session") ?? null;
+  const entries = fileEntries.filter((entry): entry is PiSessionEntry => entry.type !== "session");
+  const lastEntry = entries.at(-1);
+  const leafId = typeof lastEntry?.id === "string" ? lastEntry.id : null;
+  return { header, entries, leafId };
+}
+
 export async function buildExportSessionReply(params: HandleCommandsParams): Promise<ReplyPayload> {
  const args = parseExportCommandOutputPath(params.command.commandBodyNormalized, [
    "export-session",
@@ -135,10 +156,7 @@ export async function buildExportSessionReply(params: HandleCommandsParams): Pro
  }

  // 2. Load session entries
-  const sessionManager = SessionManager.open(sessionFile);
-  const entries = sessionManager.getEntries();
-  const header = sessionManager.getHeader();
-  const leafId = sessionManager.getLeafId();
+  const { entries, header, leafId } = await readSessionDataFromTranscript(sessionFile);

  // 3. Build full system prompt
  const { systemPrompt, tools } = await resolveCommandsSystemPromptBundle({
--- a/src/gateway/gateway-models.profiles.live.test.ts
+++ b/src/gateway/gateway-models.profiles.live.test.ts
@@ -50,7 +50,7 @@ import {
  shouldRetryToolReadProbe,
 } from "./live-tool-probe-utils.js";
 import { startGatewayServer } from "./server.impl.js";
-import { loadSessionEntry, readSessionMessages } from "./session-utils.js";
+import { loadSessionEntry, readSessionMessagesAsync } from "./session-utils.js";

 const ZAI_FALLBACK = isTruthyEnvValue(process.env.OPENCLAW_LIVE_GATEWAY_ZAI_FALLBACK);
 const REQUIRE_PROFILE_KEYS = isLiveProfileKeyModeEnabled();
@@ -1158,12 +1158,12 @@ function extractTranscriptMessageText(message: unknown): string {
    .trim();
 }

-function readSessionAssistantTexts(sessionKey: string, modelKey?: string): string[] {
+async function readSessionAssistantTexts(sessionKey: string, modelKey?: string): Promise<string[]> {
  const { storePath, entry } = loadSessionEntry(sessionKey);
  if (!entry?.sessionId) {
    return [];
  }
-  const messages = readSessionMessages(entry.sessionId, storePath, entry.sessionFile);
+  const messages = await readSessionMessagesAsync(entry.sessionId, storePath, entry.sessionFile);
  const assistantTexts: string[] = [];
  for (const message of messages) {
    if (!message || typeof message !== "object") {
@@ -1190,7 +1190,7 @@ async function waitForSessionAssistantText(params: {
  let lastHeartbeatAt = startedAt;
  let delayMs = 50;
  while (Date.now() - startedAt < GATEWAY_LIVE_PROBE_TIMEOUT_MS) {
-    const assistantTexts = readSessionAssistantTexts(params.sessionKey, params.modelKey);
+    const assistantTexts = await readSessionAssistantTexts(params.sessionKey, params.modelKey);
    if (assistantTexts.length > params.baselineAssistantCount) {
      const freshText = assistantTexts
        .slice(params.baselineAssistantCount)
@@ -1226,9 +1226,8 @@ async function requestGatewayAgentText(params: {
    content: string;
  }>;
 }) {
-  const baselineAssistantCount = readSessionAssistantTexts(
-    params.sessionKey,
-    params.modelKey,
+  const baselineAssistantCount = (
+    await readSessionAssistantTexts(params.sessionKey, params.modelKey)
  ).length;
  const accepted = await withGatewayLiveProbeTimeout(
    params.client.request("agent", {
--- a/src/gateway/session-utils.ts
+++ b/src/gateway/session-utils.ts
@@ -105,23 +105,17 @@ export {
  capArrayByJsonBytes,
  readFirstUserMessageFromTranscript,
  readLastMessagePreviewFromTranscript,
-  readLatestSessionUsageFromTranscript,
  readLatestSessionUsageFromTranscriptAsync,
-  readRecentSessionMessages,
  readRecentSessionMessagesAsync,
  readRecentSessionMessagesWithStatsAsync,
-  readRecentSessionMessagesWithStats,
  readRecentSessionTranscriptLines,
  readRecentSessionUsageFromTranscript,
  readSessionMessageCountAsync,
-  readSessionMessageCount,
  readSessionTitleFieldsFromTranscript,
  readSessionTitleFieldsFromTranscriptAsync,
  readSessionPreviewItemsFromTranscript,
  readSessionMessagesAsync,
-  readSessionMessages,
  visitSessionMessagesAsync,
-  visitSessionMessages,
  resolveSessionTranscriptCandidates,
 } from "./session-utils.fs.js";
 export { canonicalizeSpawnedByForAgent, resolveSessionStoreKey } from "./session-store-key.js";