pi coding-agent 实现教程

    第 06 章:会话持久化 —— JSONL 树

    重启就失忆的 agent 没法干活。这一章实现会话持久化。pi 的做法优雅得值得一字不差地学:append-only 的 JSONL 文件,每条记录带 id/parentId,会话历史因此是一棵树——分支不需要新文件,换个 parentId 追加就行。pi 的 /tree、/fork、/clone 全部建立在这个结构上。

    为什么是树,而不是线性日志

    想回到三天前的某个回答、换个方向继续——线性日志只能开新会话或者截断历史,树则让你就地分叉:

            ┌─ msg:A(我错了,重来)
msg:Q ──┤
        └─ msg:A'(换个问法)─ msg:B ── 当前分支

    旧分支完好无损地留在同一个文件里。pi 的会话格式(docs/session-format.md)正是如此:每行一个带 type 的 JSON 对象,id/parentId 串成树;文件头部是 session 类型的记录(带版本号,cwd 等)。恢复会话 = 从根沿选定分支走到叶子,把沿途的消息收集起来。

    记录类型

    mini-pi 只需要三种记录(第 07 章会用到第三种):

    src/main.ts(新增)
    interface SessionEntry {
  id: string;
  parentId: string | null;
  timestamp: number;
  type: "session" | "message" | "compaction";
  /** type=session:工作目录 */
  cwd?: string;
  /** type=message:消息本体(OpenAI wire 格式) */
  message?: Message;
  /** type=compaction:摘要与压缩信息(第 07 章) */
  summary?: string;
  compactedCount?: number;
  firstKeptId?: string;
}

    注意一个简化:我们的 message 直接就是 OpenAI wire 格式,恢复后可以直接喂回给模型。pi 存的是自己的 AgentMessage(含 usage、stopReason、自定义消息类型等),调 LLM 前再经 convertToLlm 转换——抽象层级不同,但"落盘格式 ≠ 调用格式"的思想一样:落盘要留住一切,调用时按需转换。

    Session:追加与重建

    src/main.ts(新增)
    const SESSION_DIR = join(homedir(), ".mini-pi", "sessions");

function newId(): string {
  return crypto.randomUUID().slice(0, 8);
}

function readEntries(file: string): SessionEntry[] {
  return readFileSync(file, "utf8")
    .split("\n")
    .filter(Boolean)
    .map((l) => JSON.parse(l) as SessionEntry);
}

class Session {
  readonly file: string;
  private leafId: string;
  /** 消息对象 → 会话条目 id,压缩时需要知道被保留消息的第一条 id */
  readonly entryIds = new Map<Message, string>();

  private constructor(file: string, leafId: string) {
    this.file = file;
    this.leafId = leafId;
  }

  static create(): Session {
    mkdirSync(SESSION_DIR, { recursive: true });
    const stamp = new Date().toISOString().replace(/[:.]/g, "-");
    const file = join(SESSION_DIR, `${stamp}-${basename(process.cwd())}.jsonl`);
    const header: SessionEntry = {
      id: newId(),
      parentId: null,
      timestamp: Date.now(),
      type: "session",
      cwd: process.cwd(),
    };
    appendFileSync(file, JSON.stringify(header) + "\n");
    return new Session(file, header.id);
  }

  /** 找当前目录最近一次的会话,重建"最新分支"上的上下文 */
  static loadLatest(): { session: Session; messages: Message[]; compacted: number } | null {
    if (!existsSync(SESSION_DIR)) return null;
    const files = readdirSync(SESSION_DIR)
      .filter((f) => f.endsWith(".jsonl"))
      .map((f) => join(SESSION_DIR, f))
      .sort((a, b) => statSync(b).mtimeMs - statSync(a).mtimeMs);
    for (const file of files) {
      const entries = readEntries(file);
      const header = entries.find((e) => e.type === "session");
      if (header?.cwd !== process.cwd()) continue;
      const rebuilt = rebuildContext(entries);
      const session = new Session(file, rebuilt.leafId);
      for (const [msg, id] of rebuilt.entryIds) session.entryIds.set(msg, id);
      return { session, messages: rebuilt.messages, compacted: rebuilt.compacted };
    }
    return null;
  }

  append(entry: {
    type: SessionEntry["type"];
    message?: Message;
    summary?: string;
    compactedCount?: number;
    firstKeptId?: string;
  }): string {
    const id = newId();
    const full: SessionEntry = { ...entry, id, parentId: this.leafId, timestamp: Date.now() };
    appendFileSync(this.file, JSON.stringify(full) + "\n");
    this.leafId = id;
    if (entry.message) this.entryIds.set(entry.message, id);
    return id;
  }
}

    rebuildContext 是恢复的核心:建 children 表 → 从 header 出发,每个分叉选时间戳最新的孩子(这就是"最新分支")→ 沿途收集消息;遇到 compaction 记录则插入摘要消息并跳过被压缩的旧消息(下一章填坑):

    src/main.ts(新增)
    function rebuildContext(entries: SessionEntry[]): {
  messages: Message[];
  leafId: string;
  entryIds: Map<Message, string>;
  compacted: number;
} {
  const children = new Map<string | null, SessionEntry[]>();
  for (const e of entries) {
    const list = children.get(e.parentId) ?? [];
    list.push(e);
    children.set(e.parentId, list);
  }
  const header = entries.find((e) => e.type === "session");
  if (!header) throw new Error("Invalid session file: no session header");

  const path: SessionEntry[] = [];
  let current: SessionEntry | undefined = header;
  while (current) {
    path.push(current);
    const kids: SessionEntry[] = children.get(current.id) ?? [];
    current = kids.sort((a, b) => a.timestamp - b.timestamp).at(-1);
  }

  const messages: Message[] = [];
  const entryIds = new Map<Message, string>();
  let skipUntilId: string | null = null;
  let compacted = 0;
  for (const e of path) {
    if (e.type === "compaction" && e.summary) {
      messages.push({ role: "user", content: `[Earlier conversation summary]\n${e.summary}` });
      skipUntilId = e.firstKeptId ?? null;
      compacted += e.compactedCount ?? 0;
      continue;
    }
    if (e.type === "message" && e.message) {
      if (skipUntilId) {
        if (e.id !== skipUntilId) continue;
        skipUntilId = null;
      }
      messages.push(e.message);
      entryIds.set(e.message, e.id);
    }
  }
  return { messages, leafId: path.at(-1)!.id, entryIds, compacted };
}

    接线:每条消息落盘

    runAgent 和 REPL 里每产生一条消息(user / assistant / tool)就 session.append(...);main() 处理 --continue:

    src/main.ts(main 更新)
    const wantContinue = process.argv.includes("-c") || process.argv.includes("--continue");
let session: Session;
let messages: Message[] = [];
if (wantContinue) {
  const loaded = Session.loadLatest();
  if (loaded) {
    session = loaded.session;
    messages = loaded.messages;
    console.log(
      `\x1b[2m[resumed ${basename(loaded.session.file)}: ${messages.length} messages` +
        (loaded.compacted > 0 ? `, ${loaded.compacted} older compacted` : "") +
        "]\x1b[0m",
    );
  } else {
    session = Session.create();
  }
} else {
  session = Session.create();
}
console.log(`\x1b[2m[session: ${session.file}]\x1b[0m`);

    REPL 里用户发消息时也顺手落盘:

    src/main.ts(REPL 的 line 处理内)
    const userMsg: Message = { role: "user", content: text };
messages.push(userMsg);
session.append({ type: "message", message: userMsg });

    验收

    1. 正常对话两轮,退出。打开 ~/.mini-pi/sessions/ 里最新的 .jsonl,看到 header + 一串 message 记录,parentId 首尾相接。
    2. npx tsx src/main.ts --continue——输出 [resumed …: N messages],接着上次的话题继续聊,它记得。
    3. 思考(不用做):如果要在 mini-pi 上实现 pi 的 /tree,需要改什么?——只需要一个"从任意节点开始继续"的选择器,append 的 parentId 指过去,分支自然产生。树结构把分叉从"特殊功能"变成了"免费属性"。

    本章要点

    • append-only JSONL + id/parentId = 就地分支、崩溃安全(每行写完即持久)、可直接阅读的会话格式。
    • 恢复 = 从根沿选定分支走一遍;--continue 选"最新分支"即可覆盖 90% 场景。
    • 落盘格式与 LLM 调用格式可以不同;pi 存 AgentMessage 并在调用前 convertToLlm,我们存 wire 格式本身。
    • pi 的会话文件按工作目录归档(~/.pi/agent/sessions/--<path>--/),支持 /tree 可视化跳转、HTML 导出等,见设计思想:harness 层。

    下一章:第 07 章:上下文压缩。