perf: stream wire.jsonl reads to reduce peak memory consumption

.changeset/stream-wire-jsonl-reads.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Read session wire logs line-by-line instead of loading whole files into memory, cutting peak memory when serving session snapshots, history transcripts, and debug exports of long sessions.

packages/agent-core/src/services/message/transcript.ts

@@ -49,7 +49,7 @@
 import { readFile } from 'node:fs/promises';
 import path from 'node:path';
 
-import type { AgentRecord } from '../../agent/records';
+import { FileSystemAgentRecordPersistence, type AgentRecord } from '../../agent/records';
 import type { ContextMessage } from '../../agent/context';
 import type { ExecutableToolResult, LoopRecordedEvent } from '../../loop';
 import {
@@ -324,27 +324,16 @@ function rawToolResultContent(output: ExecutableToolResult['output']): ContentPa
 }
 
 /**
- * Parse a `wire.jsonl` file. A torn FINAL line (crash mid-flush) is dropped,
- * matching `FileSystemAgentRecordPersistence.read`; corruption anywhere else
- * throws so the caller can fall back to the live context view.
+ * Parse a `wire.jsonl` file. Streams the file line-by-line through the same
+ * crash-tolerant reader as `FileSystemAgentRecordPersistence.read` — a torn
+ * FINAL line (crash mid-flush) is dropped; corruption anywhere else throws so
+ * the caller can fall back to the live context view. A missing file yields no
+ * records (same as a fresh log).
  */
 export async function readWireRecords(wirePath: string): Promise<AgentRecord[]> {
-  const raw = await readFile(wirePath, 'utf8');
-  const lines = raw.split('\n');
   const records: AgentRecord[] = [];
-  for (let i = 0; i < lines.length; i++) {
-    let line = lines[i]!;
-    if (line.endsWith('\r')) line = line.slice(0, -1);
-    if (line.length === 0) continue;
-    try {
-      records.push(JSON.parse(line) as AgentRecord);
-    } catch (parseError) {
-      if (i === lines.length - 1) break;
-      throw new Error(
-        `wire.jsonl: corrupted line ${i + 1} in ${wirePath}: ${String(parseError)}`,
-        { cause: parseError },
-      );
-    }
+  for await (const record of new FileSystemAgentRecordPersistence(wirePath).read()) {
+    records.push(record);
   }
   return records;
 }

packages/agent-core/src/session/export/wire-scan.ts

@@ -1,4 +1,5 @@
-import { readFile } from 'node:fs/promises';
+import { createReadStream } from 'node:fs';
+import { createInterface } from 'node:readline';
 import { join } from 'pathe';
 
 export interface SessionWireScan {
@@ -9,50 +10,52 @@ export interface SessionWireScan {
 }
 
 export async function scanSessionWire(sessionDir: string): Promise<SessionWireScan> {
-  let raw: string;
-  try {
-    raw = await readFile(join(sessionDir, 'wire.jsonl'), 'utf-8');
-  } catch {
-    return {};
-  }
-
   let firstActivityMs: number | undefined;
   let lastActivityMs: number | undefined;
   let lastUserMessageMs: number | undefined;
   let firstUserInput: string | undefined;
 
-  for (const line of raw.split('\n')) {
-    const trimmed = line.trim();
-    if (trimmed.length === 0) continue;
-    let parsed: unknown;
-    try {
-      parsed = JSON.parse(trimmed) as unknown;
-    } catch {
-      continue;
-    }
-    if (typeof parsed !== 'object' || parsed === null) continue;
-    const record = parsed as {
-      type?: unknown;
-      time?: unknown;
-      userInput?: unknown;
-    };
-    const timeMs = typeof record.time === 'number' ? normalizeTimestampMs(record.time) : undefined;
-    if (timeMs !== undefined) {
-      firstActivityMs ??= timeMs;
-      lastActivityMs = timeMs;
-    }
-    if (record.type === 'turn_begin') {
+  try {
+    // Stream line-by-line: export-time scans must not hold the whole log
+    // (plus a per-line string array) in memory. A missing/unreadable file
+    // degrades to an empty scan, matching the old readFile catch-all.
+    const input = createReadStream(join(sessionDir, 'wire.jsonl'), { encoding: 'utf8' });
+    const lines = createInterface({ input, crlfDelay: Infinity });
+    for await (const line of lines) {
+      const trimmed = line.trim();
+      if (trimmed.length === 0) continue;
+      let parsed: unknown;
+      try {
+        parsed = JSON.parse(trimmed) as unknown;
+      } catch {
+        continue;
+      }
+      if (typeof parsed !== 'object' || parsed === null) continue;
+      const record = parsed as {
+        type?: unknown;
+        time?: unknown;
+        userInput?: unknown;
+      };
+      const timeMs = typeof record.time === 'number' ? normalizeTimestampMs(record.time) : undefined;
       if (timeMs !== undefined) {
-        lastUserMessageMs = timeMs;
+        firstActivityMs ??= timeMs;
+        lastActivityMs = timeMs;
       }
-      if (
-        firstUserInput === undefined &&
-        typeof record.userInput === 'string' &&
-        record.userInput.trim().length > 0
-      ) {
-        firstUserInput = record.userInput;
+      if (record.type === 'turn_begin') {
+        if (timeMs !== undefined) {
+          lastUserMessageMs = timeMs;
+        }
+        if (
+          firstUserInput === undefined &&
+          typeof record.userInput === 'string' &&
+          record.userInput.trim().length > 0
+        ) {
+          firstUserInput = record.userInput;
+        }
       }
     }
+  } catch {
+    return {};
   }
 
   return {

packages/kap-server/src/services/snapshot/snapshotReader.ts

@@ -18,6 +18,7 @@
  * resolve to empty / `'idle'` (a cold session owns no runtime interaction).
  */
 
+import { createReadStream } from 'node:fs';
 import { readFile, stat as fsStat } from 'node:fs/promises';
 import { join } from 'node:path';
 
@@ -282,32 +283,62 @@ interface ContextRecord {
 }
 
 /**
- * Parse a `wire.jsonl` file. A torn final line (crash mid-flush) is dropped;
- * corruption anywhere else throws so the route surfaces 50001. The leading
- * `metadata` envelope and any non-`context.*` record are returned as-is and
- * filtered by the reducer's `default` branch.
+ * Parse a `wire.jsonl` file. Streams the file line-by-line (chunk-buffered,
+ * never the whole file plus a per-line string array in memory — cold snapshot
+ * and transcript rebuilds read logs that can reach hundreds of MB). A torn
+ * final line (crash mid-flush) is dropped; corruption anywhere else throws so
+ * the route surfaces 50001. The leading `metadata` envelope and any
+ * non-`context.*` record are returned as-is and filtered by the reducer's
+ * `default` branch. Missing file rejects with ENOENT, same as `readFile`.
  */
 export async function readWireRecords(wirePath: string): Promise<ContextRecord[]> {
-  const raw = await readFile(wirePath, 'utf8');
-  const lines = raw.split('\n');
   const records: ContextRecord[] = [];
-  for (let i = 0; i < lines.length; i++) {
-    let line = lines[i]!;
-    if (line.endsWith('\r')) line = line.slice(0, -1);
-    if (line.length === 0) continue;
-    try {
-      records.push(JSON.parse(line) as ContextRecord);
-    } catch (parseError) {
-      if (i === lines.length - 1) break;
-      throw new Error(
-        `wire.jsonl: corrupted line ${i + 1} in ${wirePath}: ${String(parseError)}`,
-        { cause: parseError },
+  let buffered = '';
+  let lineNumber = 0;
+  const stream = createReadStream(wirePath, { encoding: 'utf8' });
+  for await (const chunk of stream) {
+    buffered += chunk;
+    let newlineIndex = buffered.indexOf('\n');
+    while (newlineIndex !== -1) {
+      const record = parseWireLine(
+        buffered.slice(0, newlineIndex),
+        ++lineNumber,
+        wirePath,
+        false,
       );
+      buffered = buffered.slice(newlineIndex + 1);
+      if (record !== undefined) records.push(record);
+      newlineIndex = buffered.indexOf('\n');
     }
   }
+  if (buffered.length > 0) {
+    const record = parseWireLine(buffered, ++lineNumber, wirePath, true);
+    if (record !== undefined) records.push(record);
+  }
   return records;
 }
 
+function parseWireLine(
+  raw: string,
+  lineNumber: number,
+  wirePath: string,
+  allowTruncated: boolean,
+): ContextRecord | undefined {
+  const line = raw.endsWith('\r') ? raw.slice(0, -1) : raw;
+  if (line.length === 0) return undefined;
+  try {
+    return JSON.parse(line) as ContextRecord;
+  } catch (parseError) {
+    // Tolerate a truncated trailing line — last write may have crashed
+    // mid-flush; everything before is still well-formed.
+    if (allowTruncated) return undefined;
+    throw new Error(
+      `wire.jsonl: corrupted line ${lineNumber} in ${wirePath}: ${String(parseError)}`,
+      { cause: parseError },
+    );
+  }
+}
+
 async function resolveBlobRef(
   url: string,
   blobsDir: string,