diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1636807..b5f814f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ jobs: uses: pnpm/action-setup@v4 - name: Setup Node - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: 20 cache: pnpm @@ -63,7 +63,7 @@ jobs: uses: pnpm/action-setup@v4 - name: Setup Node - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: ${{ matrix.node }} cache: pnpm @@ -98,7 +98,7 @@ jobs: uses: pnpm/action-setup@v4 - name: Setup Node - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: 22 cache: pnpm @@ -136,7 +136,7 @@ jobs: uses: pnpm/action-setup@v4 - name: Setup Node - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: ${{ matrix.node }} cache: pnpm diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a35353c..6acaab3 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -35,7 +35,7 @@ jobs: uses: pnpm/action-setup@v4 - name: Setup Node - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: 20 cache: pnpm diff --git a/docs/architecture.en.md b/docs/architecture.en.md index 9af0883..b7d4584 100644 --- a/docs/architecture.en.md +++ b/docs/architecture.en.md @@ -72,7 +72,7 @@ Important traits: The sync path turns rollout evidence into durable Markdown memory: -1. read the relevant rollout JSONL +1. stream the relevant rollout JSONL line by line without assembling the entire file into one string 2. parse user messages, tool calls, and tool outputs 3. let the extractor produce candidate memory operations 4. run contradiction review so conflicting candidates can be conservatively suppressed while explicit corrections still win @@ -81,6 +81,9 @@ The sync path turns rollout evidence into durable Markdown memory: 7. append durable sync audit entries that keep suppressed conflict candidates reviewer-visible 8. record lifecycle history sidecars that power `cam recall timeline` and archive-aware retrieval +Streaming still retains parsed messages, tool calls, and outputs; it does not bound individual line size or total evidence memory. +Malformed JSON lines remain skippable, while file read errors propagate to the caller. + This is where the repository currently handles: - automatic extraction diff --git a/docs/architecture.md b/docs/architecture.md index 1777f34..a8c6c76 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -70,7 +70,7 @@ flowchart TD sync path 的职责是把“值得长期保存的信息”写回 durable memory： -1. 读取相关 rollout JSONL +1. 按行流式读取相关 rollout JSONL，避免将整个文件拼成单个字符串 2. 解析 user messages、tool calls、tool outputs 3. 由 extractor 生成 candidate memory operations 4. 经过 contradiction review，对冲突 candidate 做保守 suppress @@ -79,6 +79,9 @@ sync path 的职责是把“值得长期保存的信息”写回 durable memory 7. 追加 durable sync audit，显式暴露 reviewer 信息 8. 记录 lifecycle history，为 `cam recall timeline` 与归档检索提供 sidecar 线索 +流式读取仍会保留解析出的 messages、tool calls 与 outputs；它不限制单行大小或 evidence 的总内存。 +损坏的 JSON 行继续跳过，文件读取错误会向上返回。 + 当前 extractor 的目标： - 保存稳定、未来有用的信息 diff --git a/src/lib/domain/rollout.ts b/src/lib/domain/rollout.ts index 0137432..175177b 100644 --- a/src/lib/domain/rollout.ts +++ b/src/lib/domain/rollout.ts @@ -1,3 +1,4 @@ +import { createReadStream } from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; import os from "node:os"; @@ -21,6 +22,32 @@ interface RolloutMetaWithMtime { mtimeMs: number; } +async function* readRolloutLines(filePath: string): AsyncGenerator { + const input = createReadStream(filePath, { encoding: "utf8" }); + let pending = ""; + try { + for await (const chunk of input) { + const lines = chunk.split("\n"); + const tail = lines.pop() ?? ""; + for (const line of lines) { + const complete = pending + line; + pending = ""; + const trimmed = complete.trim(); + if (trimmed) { + yield trimmed; + } + } + pending += tail; + } + const trimmed = pending.trim(); + if (trimmed) { + yield trimmed; + } + } finally { + input.destroy(); + } +} + async function normalizeFsPath(input: string): Promise { try { return await fs.realpath(input); @@ -175,13 +202,7 @@ async function sortRolloutsByMtimeThenPath(metas: RolloutMeta[]): Promise { - const raw = await fs.readFile(filePath, "utf8"); - const lines = raw - .split("\n") - .map((line) => line.trim()) - .filter(Boolean); - - for (const line of lines) { + for await (const line of readRolloutLines(filePath)) { let item: JsonLine; try { item = JSON.parse(line) as JsonLine; @@ -309,12 +330,6 @@ export async function findLatestProjectRollout( } export async function parseRolloutEvidence(filePath: string): Promise { - const raw = await fs.readFile(filePath, "utf8"); - const lines = raw - .split("\n") - .map((line) => line.trim()) - .filter(Boolean); - const callOutputs = new Map(); const toolCalls: RolloutToolCall[] = []; const userMessages: string[] = []; @@ -326,7 +341,7 @@ export async function parseRolloutEvidence(filePath: string): Promise { + const actual = await importOriginal(); + return { ...actual, createReadStream: vi.fn(actual.createReadStream) }; +}); + +it.skipIf(bufferConstants.MAX_STRING_LENGTH > 512 * 1024 * 1024)( + "reads metadata after a representable line without combining it with subsequent lines", + async () => { + const ignored = JSON.stringify({ type: "ignored", payload: {} }); + const lineLength = bufferConstants.MAX_STRING_LENGTH - 32; + const line = ignored + " ".repeat(lineLength - ignored.length); + const metadata = JSON.stringify({ + type: "session_meta", payload: { + id: "near-limit-session", timestamp: "2026-03-14T00:00:00.000Z", cwd: process.cwd() + } + }); + // Control the pending-line boundary without repeatedly copying a huge file. + const input = Readable.from([line, "\n" + metadata + "\n"]); + vi.mocked(createReadStream).mockReturnValueOnce(input as ReadStream); + + const meta = await readRolloutMeta("synthetic-rollout.jsonl"); + expect(meta?.sessionId).toBe("near-limit-session"); + expect(input.destroyed).toBe(true); + } +); diff --git a/test/rollout.test.ts b/test/rollout.test.ts index 7d6ebce..af49c38 100644 --- a/test/rollout.test.ts +++ b/test/rollout.test.ts @@ -1,3 +1,4 @@ +import { constants as bufferConstants } from "node:buffer"; import fs from "node:fs/promises"; import realFs from "node:fs"; import os from "node:os"; @@ -15,6 +16,10 @@ import { const tempDirs: string[] = []; const originalSessionsDir = process.env.CAM_CODEX_SESSIONS_DIR; +const rolloutReaders = [ + { name: "metadata reader", parse: readRolloutMeta }, + { name: "evidence parser", parse: parseRolloutEvidence } +]; async function tempDir(prefix: string): Promise { const dir = await fs.mkdtemp(path.join(os.tmpdir(), prefix)); @@ -22,12 +27,106 @@ async function tempDir(prefix: string): Promise { return dir; } +async function writeLargeSparseRollout(filePath: string, prefix: string): Promise { + const fileSize = bufferConstants.MAX_STRING_LENGTH + 4096; + const handle = await fs.open(filePath, "w"); + try { + await handle.truncate(fileSize); + await handle.write(prefix); + // Keep invalid JSONL lines small while the total file exceeds the string limit. + for (let offset = 1024 * 1024; offset < fileSize; offset += 1024 * 1024) { + await handle.write("\n", offset, "utf8"); + } + } finally { + await handle.close(); + } +} + afterEach(async () => { process.env.CAM_CODEX_SESSIONS_DIR = originalSessionsDir; await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); }); describe("rollout helpers", () => { + it.skipIf(bufferConstants.MAX_STRING_LENGTH > 512 * 1024 * 1024)( + "reads session metadata from a rollout larger than the JavaScript string limit", + async () => { + const projectDir = await tempDir("cam-rollout-large-meta-"); + const rolloutPath = path.join(projectDir, "rollout-large.jsonl"); + await writeLargeSparseRollout(rolloutPath, JSON.stringify({ + type: "session_meta", + payload: { id: "large-session", timestamp: "2026-03-14T00:00:00.000Z", cwd: projectDir } + }) + "\n"); + + const meta = await readRolloutMeta(rolloutPath); + expect(meta).toMatchObject({ + sessionId: "large-session", + cwd: realFs.realpathSync.native(projectDir) + }); + } + ); + + it("preserves UTF-8, CRLF, blank lines, tool outputs, and an unterminated final line across chunks", async () => { + const projectDir = await tempDir("cam-rollout-chunks-"); + const rolloutPath = path.join(projectDir, "rollout-chunks.jsonl"); + const metaLine = JSON.stringify({ + type: "session_meta", + payload: { id: "chunked-session", timestamp: "2026-03-14T00:00:00.000Z", cwd: projectDir } + }); + const messagePrefix = '{"type":"event_msg","payload":{"type":"user_message","message":"'; + const message = "x".repeat(65535 - Buffer.byteLength(metaLine + "\r\n" + messagePrefix)) + "é🍵"; + await fs.writeFile(rolloutPath, [ + metaLine, + JSON.stringify({ type: "event_msg", payload: { type: "user_message", message } }), + "", + "not valid JSON", + JSON.stringify({ type: "response_item", payload: { + type: "function_call", name: "exec_command", call_id: "chunk-call", arguments: "{}" + } }), + JSON.stringify({ type: "response_item", payload: { + type: "function_call_output", call_id: "chunk-call", output: "complete 🍵" + } }), + JSON.stringify({ type: "event_msg", payload: { type: "agent_message", message: "last line" } }) + ].join("\r\n"), "utf8"); + + const evidence = await parseRolloutEvidence(rolloutPath); + expect(evidence).toMatchObject({ + sessionId: "chunked-session", + userMessages: [message], + agentMessages: ["last line"], + toolCalls: [{ callId: "chunk-call", name: "exec_command", arguments: "{}", output: "complete 🍵" }] + }); + }); + + it.each(rolloutReaders)("propagates missing-file errors from $name", async ({ parse }) => { + const projectDir = await tempDir("cam-rollout-missing-"); + await expect(parse(path.join(projectDir, "missing.jsonl"))).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it.each(rolloutReaders)("propagates read errors from $name", async ({ parse }) => { + const projectDir = await tempDir("cam-rollout-directory-"); + await expect(parse(projectDir)).rejects.toMatchObject({ code: "EISDIR" }); + }); + + it.skipIf(bufferConstants.MAX_STRING_LENGTH > 512 * 1024 * 1024)( + "parses evidence from a rollout larger than the JavaScript string limit", + async () => { + const projectDir = await tempDir("cam-rollout-large-evidence-"); + const rolloutPath = path.join(projectDir, "rollout-large.jsonl"); + await writeLargeSparseRollout(rolloutPath, [ + JSON.stringify({ + type: "session_meta", + payload: { id: "large-session", timestamp: "2026-03-14T00:00:00.000Z", cwd: projectDir } + }), + JSON.stringify({ type: "event_msg", payload: { type: "user_message", message: "remember pnpm" } }), + "" + ].join("\n")); + + const evidence = await parseRolloutEvidence(rolloutPath); + expect(evidence).toMatchObject({ sessionId: "large-session", userMessages: ["remember pnpm"] }); + } + ); + it("skips corrupted JSONL lines without crashing", async () => { const projectDir = await tempDir("cam-rollout-corrupt-"); const rolloutPath = path.join(projectDir, "rollout-corrupt.jsonl");