fix(doctor): stream heartbeat transcript scans without full-file reads (#110721)

* fix(doctor): stream heartbeat transcript scans with record caps

* fix(doctor): bound heartbeat recovery scans

Signed-off-by: sallyom <somalley@redhat.com>

---------

Signed-off-by: sallyom <somalley@redhat.com>
Co-authored-by: sallyom <somalley@redhat.com>

src/commands/doctor-heartbeat-main-session-repair.test-support.ts

@@ -11,6 +11,7 @@ type TranscriptHeartbeatSummary = {
 };
 
 type TestApi = {
+  TRANSCRIPT_RECORD_MAX_CHARS: number;
   moveHeartbeatMainSessionEntry(params: {
     store: Record<string, SessionEntry>;
     mainKey: string;
@@ -19,7 +20,11 @@ type TestApi = {
   resolveHeartbeatMainSessionRepairCandidate(params: {
     entry: SessionEntry | undefined;
     transcriptPath?: string;
-  }): { reason: "metadata" | "transcript"; summary?: TranscriptHeartbeatSummary } | null;
+  }):
+    | { reason: "metadata" | "transcript"; summary?: TranscriptHeartbeatSummary }
+    | { declineReason: "record-too-large"; reason?: undefined }
+    | null;
+  summarizeTranscriptHeartbeatMessages(transcriptPath: string): TranscriptHeartbeatSummary | null;
 };
 
 function getTestApi(): TestApi {
@@ -28,8 +33,13 @@ function getTestApi(): TestApi {
   ] as TestApi;
 }
 
+export const getTranscriptRecordMaxChars = (): number => getTestApi().TRANSCRIPT_RECORD_MAX_CHARS;
+
 export const moveHeartbeatMainSessionEntry: TestApi["moveHeartbeatMainSessionEntry"] = (params) =>
   getTestApi().moveHeartbeatMainSessionEntry(params);
 
 export const resolveHeartbeatMainSessionRepairCandidate: TestApi["resolveHeartbeatMainSessionRepairCandidate"] =
   (params) => getTestApi().resolveHeartbeatMainSessionRepairCandidate(params);
+
+export const summarizeTranscriptHeartbeatMessages: TestApi["summarizeTranscriptHeartbeatMessages"] =
+  (transcriptPath) => getTestApi().summarizeTranscriptHeartbeatMessages(transcriptPath);

src/commands/doctor-heartbeat-main-session-repair.ts

@@ -1,5 +1,6 @@
 /** Doctor repair for main sessions accidentally occupied by synthetic heartbeat transcripts. */
 import fs from "node:fs";
+import { StringDecoder } from "node:string_decoder";
 import { asNullableObjectRecord } from "@openclaw/normalization-core/record-coerce";
 import type { note } from "../../packages/terminal-core/src/note.js";
 import { isHeartbeatOkResponse, isHeartbeatUserMessage } from "../auto-reply/heartbeat-filter.js";
@@ -15,6 +16,12 @@ import type { OpenClawConfig } from "../config/types.openclaw.js";
 import { parseAgentSessionKey } from "../sessions/session-key-utils.js";
 import { clearTuiLastSessionPointers } from "../tui/tui-last-session.js";
 
+/** Chunk size for sync transcript scans. */
+const TRANSCRIPT_SCAN_CHUNK_BYTES = 64 * 1024;
+// Cap incomplete/complete JSONL records so a missing newline or huge line cannot
+// recreate full-file allocation after chunked reads. Oversized records fail closed.
+const TRANSCRIPT_RECORD_MAX_CHARS = 256 * 1024;
+
 type DoctorPrompterLike = {
   confirmRuntimeRepair: (params: {
     message: string;
@@ -38,6 +45,11 @@ type HeartbeatMainSessionRepairCandidate = {
   summary?: TranscriptHeartbeatSummary;
 };
 
+type HeartbeatMainSessionRepairDeclined = {
+  declineReason: "record-too-large";
+  reason?: undefined;
+};
+
 function countLabel(count: number, singular: string, plural = `${singular}s`): string {
   return `${count} ${count === 1 ? singular : plural}`;
 }
@@ -69,12 +81,49 @@ function parseTranscriptMessageLine(line: string): { role: string; content?: unk
   return { role, content: message.content };
 }
 
-function summarizeTranscriptHeartbeatMessages(
+function accumulateTranscriptHeartbeatMessage(
+  summary: TranscriptHeartbeatSummary,
+  line: string,
+): void {
+  const trimmed = line.trim();
+  if (!trimmed) {
+    return;
+  }
+  const message = parseTranscriptMessageLine(trimmed);
+  if (!message) {
+    return;
+  }
+  summary.inspectedMessages += 1;
+  if (message.role === "user") {
+    summary.userMessages += 1;
+    if (isHeartbeatUserMessage(message)) {
+      summary.heartbeatUserMessages += 1;
+    } else {
+      summary.nonHeartbeatUserMessages += 1;
+    }
+    return;
+  }
+  if (message.role === "assistant") {
+    summary.assistantMessages += 1;
+    if (isHeartbeatOkResponse(message)) {
+      summary.heartbeatOkAssistantMessages += 1;
+    }
+  }
+}
+
+/**
+ * Scans a transcript JSONL file in fixed-size chunks so doctor repair never loads
+ * the whole file into a single string (large poisoned heartbeat transcripts).
+ *
+ * Incomplete lines are retained only up to TRANSCRIPT_RECORD_MAX_CHARS; larger
+ * records decline classification so repair stays fail-closed.
+ */
+function scanTranscriptHeartbeatMessages(
   transcriptPath: string,
-): TranscriptHeartbeatSummary | null {
-  let raw: string;
+): TranscriptHeartbeatSummary | "record-too-large" | null {
+  let fd: number;
   try {
-    raw = fs.readFileSync(transcriptPath, "utf8");
+    fd = fs.openSync(transcriptPath, "r");
   } catch {
     return null;
   }
@@ -86,33 +135,50 @@ function summarizeTranscriptHeartbeatMessages(
     assistantMessages: 0,
     heartbeatOkAssistantMessages: 0,
   };
-  for (const line of raw.split(/\r?\n/)) {
-    const trimmed = line.trim();
-    if (!trimmed) {
-      continue;
-    }
-    const message = parseTranscriptMessageLine(trimmed);
-    if (!message) {
-      continue;
-    }
-    summary.inspectedMessages += 1;
-    if (message.role === "user") {
-      summary.userMessages += 1;
-      if (isHeartbeatUserMessage(message)) {
-        summary.heartbeatUserMessages += 1;
-      } else {
-        summary.nonHeartbeatUserMessages += 1;
+  try {
+    const decoder = new StringDecoder("utf8");
+    const chunk = Buffer.alloc(TRANSCRIPT_SCAN_CHUNK_BYTES);
+    let carry = "";
+    for (;;) {
+      const bytesRead = fs.readSync(fd, chunk, 0, chunk.length, null);
+      if (bytesRead <= 0) {
+        break;
       }
-    } else if (message.role === "assistant") {
-      summary.assistantMessages += 1;
-      if (isHeartbeatOkResponse(message)) {
-        summary.heartbeatOkAssistantMessages += 1;
+      carry += decoder.write(chunk.subarray(0, bytesRead));
+      let newline = carry.indexOf("\n");
+      while (newline >= 0) {
+        if (newline > TRANSCRIPT_RECORD_MAX_CHARS) {
+          return "record-too-large";
+        }
+        const line = carry.slice(0, newline).replace(/\r$/, "");
+        carry = carry.slice(newline + 1);
+        accumulateTranscriptHeartbeatMessage(summary, line);
+        newline = carry.indexOf("\n");
+      }
+      if (carry.length > TRANSCRIPT_RECORD_MAX_CHARS) {
+        return "record-too-large";
       }
     }
+    carry += decoder.end();
+    if (carry.length > TRANSCRIPT_RECORD_MAX_CHARS) {
+      return "record-too-large";
+    }
+    if (carry) {
+      accumulateTranscriptHeartbeatMessage(summary, carry.replace(/\r$/, ""));
+    }
+  } finally {
+    fs.closeSync(fd);
   }
   return summary.inspectedMessages > 0 ? summary : null;
 }
 
+function summarizeTranscriptHeartbeatMessages(
+  transcriptPath: string,
+): TranscriptHeartbeatSummary | null {
+  const scan = scanTranscriptHeartbeatMessages(transcriptPath);
+  return scan === "record-too-large" ? null : scan;
+}
+
 /**
  * Detects main-session entries that are safe to archive because they only contain heartbeat turns.
  *
@@ -122,7 +188,7 @@ function summarizeTranscriptHeartbeatMessages(
 function resolveHeartbeatMainSessionRepairCandidate(params: {
   entry: SessionEntry | undefined;
   transcriptPath?: string;
-}): HeartbeatMainSessionRepairCandidate | null {
+}): HeartbeatMainSessionRepairCandidate | HeartbeatMainSessionRepairDeclined | null {
   const { entry, transcriptPath } = params;
   if (!entry) {
     return null;
@@ -138,7 +204,10 @@ function resolveHeartbeatMainSessionRepairCandidate(params: {
   if (!transcriptPath) {
     return null;
   }
-  const summary = summarizeTranscriptHeartbeatMessages(transcriptPath);
+  const summary = scanTranscriptHeartbeatMessages(transcriptPath);
+  if (summary === "record-too-large") {
+    return { declineReason: "record-too-large" };
+  }
   if (!summary) {
     return null;
   }
@@ -195,8 +264,10 @@ if (process.env.VITEST || process.env.NODE_ENV === "test") {
   (globalThis as Record<PropertyKey, unknown>)[
     Symbol.for("openclaw.doctorHeartbeatMainSessionRepairTestApi")
   ] = {
+    TRANSCRIPT_RECORD_MAX_CHARS,
     moveHeartbeatMainSessionEntry,
     resolveHeartbeatMainSessionRepairCandidate,
+    summarizeTranscriptHeartbeatMessages,
   };
 }
 
@@ -234,6 +305,12 @@ export async function repairHeartbeatPoisonedMainSession(params: {
   if (!candidate) {
     return;
   }
+  if ("declineReason" in candidate) {
+    params.warnings.push(
+      `- Skipped heartbeat main-session recovery for ${mainKey}: the transcript contains a JSONL record larger than ${TRANSCRIPT_RECORD_MAX_CHARS} characters, so doctor left it unchanged.`,
+    );
+    return;
+  }
   const recoveredKey = resolveHeartbeatMainRecoveryKey({
     mainKey,
     store: params.store,
@@ -268,7 +345,7 @@ export async function repairHeartbeatPoisonedMainSession(params: {
       entry: currentEntry,
       transcriptPath,
     });
-    if (!currentCandidate) {
+    if (!currentCandidate || "declineReason" in currentCandidate) {
       return;
     }
     if (moveHeartbeatMainSessionEntry({ store: currentStore, mainKey, recoveredKey })) {

src/commands/doctor-state-integrity.test.ts

@@ -18,8 +18,10 @@ import {
   writeTuiLastSessionKey,
 } from "../tui/tui-last-session.js";
 import {
+  getTranscriptRecordMaxChars,
   moveHeartbeatMainSessionEntry,
   resolveHeartbeatMainSessionRepairCandidate,
+  summarizeTranscriptHeartbeatMessages,
 } from "./doctor-heartbeat-main-session-repair.test-support.js";
 import {
   detectStateIntegrityHealthIssues,
@@ -744,6 +746,104 @@ describe("doctor state integrity oauth dir checks", () => {
     );
   });
 
+  it("repairs a multi-chunk heartbeat transcript without loading it via readFileSync", async () => {
+    const cfg: OpenClawConfig = {};
+    setupSessionState(cfg, process.env, tempHome);
+    const sessionsDir = resolveSessionTranscriptsDirForAgent("main", process.env, () => tempHome);
+    const transcriptPath = path.join(sessionsDir, "large-heartbeat-session.jsonl");
+    const heartbeatLine = `${JSON.stringify({
+      message: { role: "user", content: HEARTBEAT_TRANSCRIPT_PROMPT },
+    })}\n${JSON.stringify({ message: { role: "assistant", content: "HEARTBEAT_OK" } })}\n`;
+    // >64 KiB so the sync scanner must read more than one chunk.
+    const repeats = Math.ceil((80 * 1024) / heartbeatLine.length);
+    fs.writeFileSync(transcriptPath, heartbeatLine.repeat(repeats));
+    expect(fs.statSync(transcriptPath).size).toBeGreaterThan(64 * 1024);
+
+    writeSessionStore(cfg, {
+      "agent:main:main": {
+        sessionId: "large-heartbeat-session",
+        updatedAt: Date.now(),
+      },
+    });
+
+    const readFileSyncSpy = vi.spyOn(fs, "readFileSync");
+    const confirmRuntimeRepair = vi.fn(async (params: { message: string }) =>
+      params.message.startsWith("Move heartbeat-owned main session"),
+    );
+    try {
+      await noteStateIntegrity(cfg, { confirmRuntimeRepair, note: noteMock });
+    } finally {
+      const transcriptReads = readFileSyncSpy.mock.calls.filter((call) => {
+        const target = call[0];
+        return typeof target === "string" && path.resolve(target) === path.resolve(transcriptPath);
+      });
+      readFileSyncSpy.mockRestore();
+      expect(transcriptReads).toEqual([]);
+    }
+
+    const summary = summarizeTranscriptHeartbeatMessages(transcriptPath);
+    expect(summary?.heartbeatUserMessages).toBe(repeats);
+    expect(summary?.nonHeartbeatUserMessages).toBe(0);
+    expect(summary?.userMessages).toBe(repeats);
+
+    const storePath = resolveStorePath(cfg.session?.store, { agentId: "main" });
+    const store = JSON.parse(fs.readFileSync(storePath, "utf8")) as Record<string, SessionEntry>;
+    expect(store["agent:main:main"]).toBeUndefined();
+    const recoveredKey = Object.keys(store).find((key) =>
+      key.startsWith("agent:main:heartbeat-recovered-"),
+    );
+    expect(recoveredKey).toBeDefined();
+    expect(store[recoveredKey!]?.sessionId).toBe("large-heartbeat-session");
+    expect(doctorChangesText()).toContain("Moved heartbeat-owned main session agent:main:main");
+  });
+
+  it("declines repair when a single JSONL record exceeds the scanner record cap", async () => {
+    const cfg: OpenClawConfig = {};
+    setupSessionState(cfg, process.env, tempHome);
+    const sessionsDir = resolveSessionTranscriptsDirForAgent("main", process.env, () => tempHome);
+    const transcriptPath = path.join(sessionsDir, "oversized-record-session.jsonl");
+    const maxChars = getTranscriptRecordMaxChars();
+    const oversizedRecord = `${"x".repeat(maxChars + 1)}\n`;
+    const heartbeatLine = `${JSON.stringify({
+      message: { role: "user", content: HEARTBEAT_TRANSCRIPT_PROMPT },
+    })}\n`;
+    fs.writeFileSync(transcriptPath, `${oversizedRecord}${heartbeatLine}`);
+
+    writeSessionStore(cfg, {
+      "agent:main:main": {
+        sessionId: "oversized-record-session",
+        updatedAt: Date.now(),
+      },
+    });
+
+    const confirmRuntimeRepair = vi.fn(async (params: { message: string }) =>
+      params.message.startsWith("Move heartbeat-owned main session"),
+    );
+    const readFileSyncSpy = vi.spyOn(fs, "readFileSync");
+    try {
+      await noteStateIntegrity(cfg, { confirmRuntimeRepair, note: noteMock });
+    } finally {
+      const transcriptReads = readFileSyncSpy.mock.calls.filter((call) => {
+        const target = call[0];
+        return typeof target === "string" && path.resolve(target) === path.resolve(transcriptPath);
+      });
+      readFileSyncSpy.mockRestore();
+      expect(transcriptReads).toEqual([]);
+    }
+
+    expect(summarizeTranscriptHeartbeatMessages(transcriptPath)).toBeNull();
+    expect(stateIntegrityText()).toContain(
+      "Skipped heartbeat main-session recovery for agent:main:main: the transcript contains a JSONL record larger than",
+    );
+    expect(hasRepairPromptMessage(confirmRuntimeRepair, "Move heartbeat-owned main session")).toBe(
+      false,
+    );
+    const storePath = resolveStorePath(cfg.session?.store, { agentId: "main" });
+    const store = JSON.parse(fs.readFileSync(storePath, "utf8")) as Record<string, SessionEntry>;
+    expect(store["agent:main:main"]?.sessionId).toBe("oversized-record-session");
+    expect(Object.keys(store).filter((key) => key.includes("heartbeat-recovered"))).toEqual([]);
+  });
+
   it("does not treat heartbeat-labeled routing metadata as heartbeat ownership", () => {
     const entry: SessionEntry = {
       sessionId: "session",

src/commands/doctor-state-integrity.ts

@@ -259,23 +259,38 @@ function addUserRwx(mode: number): number {
 }
 
 function countJsonlLines(filePath: string): number {
+  let fd: number;
   try {
-    const raw = fs.readFileSync(filePath, "utf-8");
-    if (!raw) {
-      return 0;
-    }
+    fd = fs.openSync(filePath, "r");
+  } catch {
+    return 0;
+  }
+  try {
+    const chunk = Buffer.alloc(64 * 1024);
     let count = 0;
-    for (const char of raw) {
-      if (char === "\n") {
-        count += 1;
+    let hasBytes = false;
+    let endsWithNewline = false;
+    for (;;) {
+      const bytesRead = fs.readSync(fd, chunk, 0, chunk.length, null);
+      if (bytesRead <= 0) {
+        break;
+      }
+      hasBytes = true;
+      endsWithNewline = chunk[bytesRead - 1] === 0x0a;
+      for (let index = 0; index < bytesRead; index += 1) {
+        if (chunk[index] === 0x0a) {
+          count += 1;
+        }
       }
     }
-    if (!raw.endsWith("\n")) {
+    if (hasBytes && !endsWithNewline) {
       count += 1;
     }
     return count;
   } catch {
     return 0;
+  } finally {
+    fs.closeSync(fd);
   }
 }