Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
5 changes: 4 additions & 1 deletion docs/architecture.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
5 changes: 4 additions & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 的目标:

- 保存稳定、未来有用的信息
Expand Down
43 changes: 29 additions & 14 deletions src/lib/domain/rollout.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -21,6 +22,32 @@ interface RolloutMetaWithMtime {
mtimeMs: number;
}

async function* readRolloutLines(filePath: string): AsyncGenerator<string> {
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<string> {
try {
return await fs.realpath(input);
Expand Down Expand Up @@ -175,13 +202,7 @@ async function sortRolloutsByMtimeThenPath(metas: RolloutMeta[]): Promise<Rollou
}

export async function readRolloutMeta(filePath: string): Promise<RolloutMeta | null> {
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;
Expand Down Expand Up @@ -309,12 +330,6 @@ export async function findLatestProjectRollout(
}

export async function parseRolloutEvidence(filePath: string): Promise<RolloutEvidence | null> {
const raw = await fs.readFile(filePath, "utf8");
const lines = raw
.split("\n")
.map((line) => line.trim())
.filter(Boolean);

const callOutputs = new Map<string, string>();
const toolCalls: RolloutToolCall[] = [];
const userMessages: string[] = [];
Expand All @@ -326,7 +341,7 @@ export async function parseRolloutEvidence(filePath: string): Promise<RolloutEvi
let forkedFromSessionId: string | undefined;
let seenPrimaryMeta = false;

for (const line of lines) {
for await (const line of readRolloutLines(filePath)) {
let item: JsonLine;
try {
item = JSON.parse(line) as JsonLine;
Expand Down
32 changes: 32 additions & 0 deletions test/rollout-stream.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
import { constants as bufferConstants } from "node:buffer";
import { createReadStream } from "node:fs";
import type { ReadStream } from "node:fs";
import { Readable } from "node:stream";
import { expect, it, vi } from "vitest";
import { readRolloutMeta } from "../src/lib/domain/rollout.js";

vi.mock("node:fs", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs")>();
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);
}
);
99 changes: 99 additions & 0 deletions test/rollout.test.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -15,19 +16,117 @@ 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<string> {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), prefix));
tempDirs.push(dir);
return dir;
}

async function writeLargeSparseRollout(filePath: string, prefix: string): Promise<void> {
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");
Expand Down
Loading