Agent02_上下文,缓存与长记忆

流式输出与中断

流式逐字输出，以及让中断过的会话能继续。

问题

你给 agent 一个多步任务，它闷头执行半分钟，屏幕上什么都没有——不知道是卡住了还是在干活。加上流式输出（逐字打印）后，多半还会加 Ctrl+C 中断——跑错方向的任务得能停。然后新问题出现：中断之后再发一条消息，API 直接报 400 拒收。

先补一个协议知识：模型想用工具时，会在回复（assistant 消息）里带一批 tool_calls，每个有唯一 id；协议要求每个 id 后面必须跟一条对应的 tool 消息（工具执行结果）。

看一下 Ctrl+C 那一刻的消息序列：

user:      "把测试跑一遍，顺便看下 README"
assistant: tool_calls: [call_test → run_shell, call_read → read_file]
tool:      (call_test 的结果 —— 第一个工具跑完了)
                     ← Ctrl+C 落在这里，call_read 永远没有结果

call_read 没有对应结果——原样发出去，服务端拒收（400，报错原文见文末）。而且流是中途切断的，call_read 的参数可能停在 JSON 中间（{"path":"READ）。怎么处理这个残缺序列，决定了中断后会话能否继续。

解决方案

流式侧：SSE 按行缓冲解析（凑齐完整的一行才处理一行），tool_calls 的分片按 index 装配，parse 留到拼完之后。中断侧：一个 AbortController 贯穿 HTTP 层与工具循环，已装配的半截消息照常保留在历史里；对 Ctrl+C 留下的悬空 tool_call，给每个没有结果的 tool_call 回填一条合成 tool 消息，把序列配平——会话就能继续。

把这段话翻译一下

流式侧（数据接收层）

问题：大模型是逐字（逐token）返回的，tool_calls（工具调用指令）可能被拆成多个网络包发回来。

做法：系统不着急解析零碎数据，而是按SSE（流式协议）的换行符缓存数据，凑齐一行再处理。同时，根据index（索引编号）把属于同一个工具调用的碎片拼装起来，等拼完整了再统一解析成JSON。这样做是为了防止读到一半的数据导致解析报错。

中断侧（请求取消层）

问题：用户按了Ctrl+C，前端会通过AbortController（浏览器/Node的终止信号）中断HTTP请求。此时，后端可能刚执行完第一个工具（call_test），正准备执行第二个工具（call_read）——这个call_read就成了“悬空”指令（有指令但没执行结果）。

做法：系统不暴力丢弃这半截assistant消息，而是保留它，并补一条合成的tool消息，内容写“用户中断，未运行”。

历史配平层（会话修复）

硬性规则：在OpenAI等API格式中，assistant消息里的每个tool_calls，后面必须跟对应数量的tool结果消息，否则接口会报400错误（参数错误）。

做法：通过补上合成的结果，让历史记录中的assistant指令和tool结果数量配平。这样，下一轮用户继续说话时，系统读取的历史是完整合法的，模型不会“失忆”或报错，对话就能无缝继续。

运行

不需要 API key：

1	`node notes/s05_streaming_interrupt/demo.mjs`

#!/usr/bin/env node
// 不需要 API key 的流式与中断演示：
//   场景一：把一段 SSE 字节流故意切得支离破碎（掐在 data: 行中间、JSON 中间、
//           多字节字符中间），喂给解析器 + 装配器，看 tool_calls 分片怎么归队。
//   场景二：伪造一串被 Ctrl+C 撕成半截的 messages，跑修复函数，
//           看悬空的 tool_call 怎么被回填、断裂的参数怎么被修好。
//
//   node s05_streaming_interrupt/demo.mjs

import { sseJsonEvents, createAssembler, repairDanglingToolCalls, INTERRUPT_NOTE } from "./stream.mjs";

// ─── 场景一：SSE 分片装配 ────────────────────────────────────────────────

console.log("━━━ 场景一：SSE 分片装配（tool_calls 按 index 归队）━━━");

// 服务端"本来想发"的事件序列：先流两段文本，再交错地流两个工具调用 ——
// call 0 的 arguments 被切成 3 片，中间还插进来 call 1 的分片。
const events = [
  { choices: [{ delta: { role: "assistant" } }] },
  { choices: [{ delta: { content: "我来" } }] },
  { choices: [{ delta: { content: "查一下。" } }] },
  { choices: [{ delta: { tool_calls: [{ index: 0, id: "call_ls", type: "function", function: { name: "run_shell", arguments: "" } }] } }] },
  { choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '{"comm' } }] } }] },
  { choices: [{ delta: { tool_calls: [{ index: 1, id: "call_read", type: "function", function: { name: "read_file", arguments: '{"path":' } }] } }] },
  { choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: 'and":"ls' } }] } }] },
  { choices: [{ delta: { tool_calls: [{ index: 1, function: { arguments: '"a.txt"}' } }] } }] },
  { choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: ' -la"}' } }] } }] },
  { choices: [{ delta: {}, finish_reason: "tool_calls" }] },
];
const wire = events.map((e) => `data: ${JSON.stringify(e)}\n\n`).join("") + "data: [DONE]\n\n";

// 网络不看语义：整段字节流每 31 字节掐一刀 —— 必然有分片停在 data: 行
// 中间、JSON 字符串中间、甚至"我来"这种多字节字符的字节中间。
const bytes = new TextEncoder().encode(wire);
const chunks = [];
for (let i = 0; i < bytes.length; i += 31) chunks.push(bytes.slice(i, i + 31));
console.log(`  线路上：${events.length + 1} 个事件被切成 ${chunks.length} 个原始分片，比如：`);
const rawPreview = (u8) => JSON.stringify(new TextDecoder("utf-8", { fatal: false }).decode(u8));
console.log(`    分片[0] = ${rawPreview(chunks[0])}`);
console.log(`    分片[1] = ${rawPreview(chunks[1])}`);

async function* fakeBody() {
  for (const chunk of chunks) yield chunk;
}

const assembler = createAssembler();
let live = "";
for await (const event of sseJsonEvents(fakeBody())) {
  live += assembler.feed(event); // 真实 agent 在这里 process.stdout.write，打字机效果
}
const msg = assembler.message();
console.log(`  实时打印的文本：${JSON.stringify(live)}`);
console.log(`  装配出的 tool_calls（arguments 拼完才 parse）：`);
for (const tc of msg.tool_calls) {
  console.log(`    ${tc.id} → ${tc.function.name}(${JSON.stringify(JSON.parse(tc.function.arguments))})`);
}

// ─── 场景二：中断后的消息序列修复 ────────────────────────────────────────

console.log("\n━━━ 场景二：Ctrl+C 撕裂的消息序列，修复后能继续对话 ━━━");

// 复盘一次真实的中断现场：模型发起了两个工具调用，第一个跑完了，
// 用户在第二个开跑前按了 Ctrl+C —— 而且流是被掐断的，第二个调用的
// arguments 断在 JSON 中间。
const messages = [
  { role: "user", content: "把测试跑一遍，顺便看下 README" },
  {
    role: "assistant",
    content: "好，我先跑测试，再看 README。",
    tool_calls: [
      { id: "call_test", type: "function", function: { name: "run_shell", arguments: '{"command":"npm test"}' } },
      { id: "call_read", type: "function", function: { name: "read_file", arguments: '{"path":"READ' } }, // ← 断在半截
    ],
  },
  { role: "tool", tool_call_id: "call_test", content: "命令失败（exit 1）：\n1 test failed" },
  // ← Ctrl+C 落在这里：call_read 没有 tool 结果。这串 messages 直接发出去，
  //   OpenAI 兼容后端一律 400（每个 tool_call 必须跟一条 tool 结果消息）。
];

const shape = (msgs) =>
  msgs
    .map((m) => (m.role === "tool" ? `tool(${m.tool_call_id})` : m.tool_calls ? `assistant+${m.tool_calls.length}calls` : m.role))
    .join(" → ");
console.log(`  修复前：${shape(messages)}`);
console.log(`    悬空：call_read 没有 tool 结果；参数断裂：${JSON.stringify(messages[1].tool_calls[1].function.arguments)}`);

const filled = repairDanglingToolCalls(messages);

console.log(`  修复后：${shape(messages)}（回填 ${filled} 条）`);
console.log(`    合成结果：${JSON.stringify(messages.find((m) => m.tool_call_id === "call_read").content)}`);
console.log(`    参数修复：${JSON.stringify(messages[1].tool_calls[1].function.arguments)}`);
console.log(`  再跑一遍修复（应当无操作、幂等）：回填 ${repairDanglingToolCalls(messages)} 条`);

console.log(`
结论：
  · SSE 分片和事件边界无关 —— 按行缓冲 + TextDecoder(stream) 之后，
    掐在任何位置的分片都能装回原样；tool_calls 靠 index 归队，
    arguments 是碎 JSON 字符串，拼完才能 parse。
  · 中断很容易，中断后还能继续对话才是工程：悬空的 tool_call 回填
    合成结果（"${INTERRUPT_NOTE}"），
    断裂的参数修成 {}，消息序列重新配平 —— 下一句话不会撞上 400。
`);

输出（SSE 字节流每 31 字节切一次，故意切在 data: 行和 JSON 中间）：

━━━ 场景一：SSE 分片装配（tool_calls 按 index 归队）━━━
  线路上：11 个事件被切成 30 个原始分片，比如：
    分片[0] = "data: {\"choices\":[{\"delta\":{\"ro"
    分片[1] = "le\":\"assistant\"}}]}\n\ndata: {\"ch"
  实时打印的文本："我来查一下。"
  装配出的 tool_calls（arguments 拼完才 parse）：
    call_ls → run_shell({"command":"ls -la"})
    call_read → read_file({"path":"a.txt"})

━━━ 场景二：Ctrl+C 撕裂的消息序列，修复后能继续对话 ━━━
  修复前：user → assistant+2calls → tool(call_test)
    悬空：call_read 没有 tool 结果；参数断裂："{\"path\":\"READ"
  修复后：user → assistant+2calls → tool(call_read) → tool(call_test)（回填 1 条）
    合成结果："(用户中断了执行，该工具未运行——不是工具失败，需要时可以重新调用)"
    参数修复："{}"
  再跑一遍修复（应当无操作、幂等）：回填 0 条

结论：
  · SSE 分片和事件边界无关 —— 按行缓冲 + TextDecoder(stream) 之后，
    掐在任何位置的分片都能装回原样；tool_calls 靠 index 归队，
    arguments 是碎 JSON 字符串，拼完才能 parse。
  · 中断很容易，中断后还能继续对话才是工程：悬空的 tool_call 回填
    合成结果（"(用户中断了执行，该工具未运行——不是工具失败，需要时可以重新调用)"），
    断裂的参数修成 {}，消息序列重新配平 —— 下一句话不会撞上 400。

运行 AGENT_API_KEY=sk-xxx node notes/s05_streaming_interrupt/agent.mjs，是已经做了流式输出和中断消息处理的agent代码

#!/usr/bin/env node
// s05 —— Ctrl+C 之后：s03 的 agent + 流式输出 + 可恢复的中断。
//
// 在 s03 基底上的全部变化：
//   · chat() 改为 stream:true，手工解析 SSE（见 stream.mjs），文本增量实时打印
//   · 每轮一个 AbortController：第一次 Ctrl+C 中断本轮，第二次退出进程
//   · 中断后调用 repairDanglingToolCalls 修复消息序列 —— 悬空的 tool_call
//     回填合成结果，下一句话不会再撞上 400
//
// 运行方式与 s03 相同：AGENT_API_KEY=sk-xxx node agent.mjs

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";
import { sseJsonEvents, createAssembler, repairDanglingToolCalls } from "./stream.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;
const MODEL = process.env.AGENT_MODEL ?? "deepseek-chat";

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。任何 OpenAI 兼容的 key 都行（DeepSeek / Kimi / GLM / OpenRouter / 本地 Ollama）。");
  process.exit(1);
}

const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（与 s03 相同）─────────────────────────────────────────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        return `命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`;
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) return `编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`;
      if (text.indexOf(old_string, first + 1) !== -1)
        return `编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`;
      // 不能用 text.replace(old, new)：new_string 里的 $$ / $& 会被 JS 当替换模式展开，静默写坏文件。
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return `未知工具：${call.function.name}`;
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return `工具参数不是合法 JSON：${err.message}`;
  }
  try {
    return tool.handler(args);
  } catch (err) {
    return `工具执行出错：${err.message}`;
  }
}

// ─── 流式 chat：SSE 手工解析，文本增量实时打印 ───────────────────────────

async function chat(messages, signal) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...messages],
      tools: TOOLS,
      stream: true,
    }),
    signal, // 中断贯穿到 HTTP 层：abort 会掐断连接，流当场停
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);

  const assembler = createAssembler();
  let printed = false;
  try {
    for await (const event of sseJsonEvents(res.body)) {
      const textDelta = assembler.feed(event);
      if (textDelta) {
        process.stdout.write(textDelta); // 用户看到的"打字机"就是这一行
        printed = true;
      }
    }
  } catch (err) {
    // abort 掐断流时，body 迭代器会抛错（不同 Node 版本的错误类型不完全一致），
    // 以 signal 为准：确实是中断就吞掉，装配器里已有的半截消息照常返回。
    if (!signal.aborted) throw err;
  }
  if (printed) process.stdout.write("\n");
  return { message: assembler.message(), aborted: signal.aborted };
}

// ─── 主循环：中断可以落在任何缝隙里，落点之后统一修复 ────────────────────

let activeTurn = null; // 当前这轮的 AbortController；null = 没有轮次在跑

function finishInterrupt(messages) {
  const filled = repairDanglingToolCalls(messages);
  console.log(
    `\n\x1b[31m⏹ 本轮已中断${filled ? `，回填了 ${filled} 条合成工具结果` : ""}。会话可以继续。\x1b[0m`,
  );
}

async function runTurn(messages) {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;
  const controller = new AbortController();
  activeTurn = controller;

  try {
    while (true) {
      if (!budget.canContinue()) {
        const stop = budget.exhaustedStop();
        console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
        return;
      }

      // 中断也可能落在建连阶段（fetch 还没返回响应头）：此时 AbortError 从
      // await fetch 本身抛出，chat 里包住 body 迭代的 try/catch 接不到它。
      let result;
      try {
        result = await chat(messages, controller.signal);
      } catch (err) {
        if (!controller.signal.aborted) throw err;
        return finishInterrupt(messages); // 一个增量都没收到，修复后直接收场
      }
      const { message: msg, aborted } = result;
      // 半截消息也要进历史：已经流出来的文字用户看到了，历史里没有的话，
      // 模型下一轮就会"失忆"，答非所问。空壳（没内容没调用）才丢弃。
      if (msg.content || msg.tool_calls?.length) messages.push(msg);
      if (aborted) return finishInterrupt(messages);
      if (!msg.tool_calls?.length) return;

      const records = [];
      for (const call of msg.tool_calls) {
        // 中断可以落在两次工具执行的缝隙里 —— 每次派发前重查，
        // 已中断就跳过剩余调用（它们的悬空 tool_call 由修复函数回填）。
        if (controller.signal.aborted) break;
        const output = dispatch(call);
        messages.push({ role: "tool", tool_call_id: call.id, content: output });
        let input = {};
        try {
          input = JSON.parse(call.function.arguments || "{}");
        } catch { /* 坏参数已作为失败回填 */ }
        records.push({
          name: call.function.name,
          input,
          status: FAILURE_RE.test(output) ? "failed" : "completed",
          output,
        });
      }
      if (controller.signal.aborted) return finishInterrupt(messages);

      const stop = budget.recordTurn(records);
      if (!stop) continue;

      if (isRecoverable(stop) && !repaired) {
        repaired = true;
        console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
        messages.push({ role: "user", content: repairPrompt(stop) });
        continue;
      }
      console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }
  } finally {
    activeTurn = null;
  }
}

// ─── Ctrl+C：第一次中断本轮，第二次退出 ──────────────────────────────────

function onInterrupt() {
  if (activeTurn && !activeTurn.signal.aborted) {
    activeTurn.abort();
    console.log("\n\x1b[33m⚠ 正在中断本轮…（再按一次 Ctrl+C 退出）\x1b[0m");
  } else {
    console.log("\n再见。");
    process.exit(0);
  }
}

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
// 细节：终端在 readline 手里时处于 raw 模式，Ctrl+C 不会产生进程信号，
// 而是触发 rl 的 'SIGINT' 事件；stdin 不是 TTY（比如管道喂入）时才走
// process 的 'SIGINT'。两头都接到同一个处理函数上。
rl.on("SIGINT", onInterrupt);
process.on("SIGINT", onInterrupt);

const messages = [];

console.log(
  `s05 agent 已上线（${MODEL}，流式）。工具：${Object.keys(REGISTRY).join("、")}。` +
    `任务跑着的时候按 Ctrl+C 试试 —— 中断之后接着聊，不会 400。`,
);

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;
  messages.push({ role: "user", content: line });
  await runTurn(messages);
}
// 流式与中断 —— 本章的机制模块，零依赖。
//
// 三件事：
//   1. sseJsonEvents：手工解析 OpenAI 兼容的 SSE 流（data: 行、[DONE]、按行缓冲）
//   2. createAssembler：把 delta 分片装配回一条完整的 assistant 消息
//      （content 逐段拼接；tool_calls 按 index 聚合，arguments 是碎 JSON 字符串分片）
//   3. repairDanglingToolCalls：中断后修复消息序列 —— 给悬空的 tool_call
//      回填合成结果，会话才能继续（协议要求每个 tool_call 必须有对应的 tool 消息）
//
// 对应真实产品 Reina 的 packages/providers/src/tool-pairing.ts
// （codex 同款的双向配平：悬空 call 合成占位输出，孤儿结果降级为文本）。

/** 逐事件解析 SSE 字节流。body 是任何异步可迭代的字节块序列
 *  （fetch 的 response.body 就是），每拿到一条 `data: {...}` 就 yield 出解析
 *  好的 JSON 对象，遇到 `data: [DONE]` 结束。
 *
 *  关键点是**按行缓冲**：TCP/代理切分数据从不看语义，一个 chunk 可能停在
 *  半行、甚至半个多字节字符的中间 —— 所以字节先过 TextDecoder(stream 模式)，
 *  文本攒进缓冲区，凑出完整的一行才处理一行。 */
export async function* sseJsonEvents(body) {
  const decoder = new TextDecoder();
  let buffer = "";
  for await (const chunk of body) {
    buffer += decoder.decode(chunk, { stream: true });
    let newline;
    while ((newline = buffer.indexOf("\n")) !== -1) {
      const line = buffer.slice(0, newline).replace(/\r$/, "");
      buffer = buffer.slice(newline + 1);
      // 空行是事件分隔符；": xxx" 开头的是注释行（有些网关拿它当心跳）——都跳过。
      if (!line.startsWith("data:")) continue;
      const payload = line.slice(5).trim();
      if (payload === "[DONE]") return;
      yield JSON.parse(payload);
    }
  }
}

/** delta 装配器：喂进一个个流式事件，最后拿回一条完整的 assistant 消息。
 *
 *  content 好办 —— 字符串逐段拼。tool_calls 阴得多：
 *    · 一次 delta 可能只带 id，可能只带函数名，也可能只带 arguments 的几个字符；
 *    · arguments 是被切碎的 JSON 字符串分片 —— 拼完之前 parse 必炸，
 *      所以只做 += ，把 parse 留给拼完之后的人；
 *    · 并行工具调用时多个 call 交错到达，靠 delta 里的 index 归队。 */
export function createAssembler() {
  let content = "";
  const toolCalls = []; // index -> 聚合中的 tool_call
  let finishReason = null;

  return {
    /** 喂一个 SSE 事件，返回本次新到的文本增量（调用方拿去实时打印）。 */
    feed(event) {
      const choice = event.choices?.[0];
      if (!choice) return ""; // 有些兼容端在最后补一个只带 usage 的空 choices 事件
      if (choice.finish_reason) finishReason = choice.finish_reason;
      const delta = choice.delta ?? {};
      const textDelta = delta.content ?? "";
      content += textDelta;
      for (const tc of delta.tool_calls ?? []) {
        const slot = (toolCalls[tc.index] ??= { id: "", type: "function", function: { name: "", arguments: "" } });
        if (tc.id) slot.id = tc.id;
        if (tc.function?.name) slot.function.name += tc.function.name;
        if (tc.function?.arguments) slot.function.arguments += tc.function.arguments;
      }
      return textDelta;
    },

    /** 流结束（或被中断）后取出装配结果。中断时拿到的就是"半截消息"——
     *  已经流出来的部分都在，这正是修复函数的输入。 */
    message() {
      const msg = { role: "assistant", content };
      const calls = toolCalls.filter(Boolean);
      if (calls.length > 0) msg.tool_calls = calls;
      return msg;
    },

    finishReason: () => finishReason,
  };
}

export const INTERRUPT_NOTE = "(用户中断了执行，该工具未运行——不是工具失败，需要时可以重新调用)";

/** 中断后修复消息序列。OpenAI 协议的硬性要求：assistant 消息里的每个
 *  tool_call，后面都必须跟一条对应 tool_call_id 的 tool 消息 —— 否则下一次
 *  API 调用直接 400。中断恰好会把序列撕成半截：tool_calls 已经进了历史，
 *  工具却没跑（或只跑了一半的批次）。
 *
 *  修复动作有两个：
 *    1. 给每个悬空的 tool_call 回填一条合成 tool 结果（文案写明"用户中断、
 *       未运行"，防止模型误以为工具失败而盲目重试）；
 *    2. arguments 可能断在 JSON 中间（流被掐断时最后一个分片没到齐），
 *       断裂的参数修成 "{}" —— 挑剔的后端会校验历史里的这段字符串。
 *
 *  返回回填的条数。对完整的序列是无操作，随便多跑几遍都安全（幂等）。 */
export function repairDanglingToolCalls(messages, note = INTERRUPT_NOTE) {
  const answered = new Set(messages.filter((m) => m.role === "tool").map((m) => m.tool_call_id));
  let repaired = 0;
  for (let i = messages.length - 1; i >= 0; i--) {
    const msg = messages[i];
    if (msg.role !== "assistant" || !msg.tool_calls?.length) continue;
    for (const tc of msg.tool_calls) {
      try {
        JSON.parse(tc.function.arguments || "{}");
      } catch {
        tc.function.arguments = "{}";
      }
    }
    const missing = msg.tool_calls.filter((tc) => !answered.has(tc.id));
    if (missing.length === 0) continue;
    // 回填插在 assistant 消息紧后面，和真实 tool 结果保持同一块 —— 有些后端
    // 要求 tool 消息紧跟发起它的 assistant 消息，中间不能夹别的角色。
    messages.splice(i + 1, 0, ...missing.map((tc) => ({ role: "tool", tool_call_id: tc.id, content: note })));
    repaired += missing.length;
  }
  return repaired;
}

① SSE 解析：按行缓冲，而不是按 chunk 处理

SSE（Server-Sent Events）是流式输出的传输方式：stream: true 时服务端保持连接打开，把回答切成小片逐个推送，每片是文本流里的一行：

data: {"choices":[{"delta":{"content":"我来"}}]}

data: {"choices":[{"delta":{"content":"查一下"}}]}

data: [DONE]

data: 开头是数据行，空行是分隔符，data: [DONE] 表示结束。看起来逐行 parse 即可，但你收到的单位不是”行”：网络传输切分数据时不管语义，一个 chunk 可能停在 data: {"cho 中间，甚至停在一个 UTF-8 多字节字符中间。所以必须按行缓冲，凑齐一行才处理一行（stream.mjs）：

buffer += decoder.decode(chunk, { stream: true }); // stream 模式：不完整的多字节字符先保留
let newline;
while ((newline = buffer.indexOf("\n")) !== -1) {  // 凑齐完整的一行才处理一行
  const line = buffer.slice(0, newline).replace(/\r$/, "");
  buffer = buffer.slice(newline + 1);
  if (!line.startsWith("data:")) continue;          // 空行、": 心跳" 注释行，都跳过
  // ...
}

② tool_calls 分片：按 index 装配

文本增量直接拼接即可。tool_calls 的增量是被切碎的 JSON 字符串：第一片带 id 和函数名，后续片只带参数的几个字符；并行调用时多个 call 的分片交错到达，靠 index 归位：

for (const tc of delta.tool_calls ?? []) {
  const slot = (toolCalls[tc.index] ??= { id: "", type: "function", function: { name: "", arguments: "" } });
  if (tc.id) slot.id = tc.id;
  if (tc.function?.name) slot.function.name += tc.function.name;
  if (tc.function?.arguments) slot.function.arguments += tc.function.arguments;
}

这段代码的大概含义是这样

先建一个空壳子

{
  "id": "",
  "type": "function",
  "function": {
    "name": "",
    "arguments": ""
  }
}

等其中的参数逐步传输拼接进来，最后一起JSON.parse

常见错误：每收到一片就 JSON.parse——必然失败，{"comm 不是合法 JSON。装配阶段只拼接，parse 留到拼完之后。

③ 中断信号：一个 AbortController 贯穿 HTTP 层与工具循环

中断不能只设布尔标志——正在传输的 HTTP 流不会检查标志。把 AbortController 的 signal 传给 fetch，abort 时连接立即断开；断开会让读流代码抛错，确认是主动中断（signal.aborted）就吞掉错误，已装配的半截消息照常返回——用户看到的内容必须留在历史里，否则模型下一轮缺上下文。

中断也可能落在两次工具执行之间，每次派发前再检查：

for (const call of msg.tool_calls) {
  if (controller.signal.aborted) break; // 剩余调用不跑，悬空部分交给修复函数
  // ...
}

Ctrl+C 的策略：第一次中断本轮，第二次退出进程——“停下这一轮”和”退出程序”是两个意图，分开处理（readline 监听细节见文末）。

④ 悬空 tool_call 的修复：回填合成结果

残缺序列有三种处理方式：

做法	结果
把半截 assistant 消息整个丢掉	不会 400，但用户看到的那半截回答不在历史里，下一轮模型缺失上下文
原样保留，直接发	400
保留 + 回填合成 tool 结果	序列配平，会话继续

回填就是给每个没有结果的 tool_call 补一条假的 tool 消息。文案是写给模型看的界面（s02），要说清两点：

1	`(用户中断了执行，该工具未运行——不是工具失败，需要时可以重新调用)`

说”不是失败”，模型不会误判为工具故障而绕路；说”可以重新调用”，用户说”继续”时它知道从哪接。顺带把断在 JSON 中间的参数改为 {}（部分后端会校验它）。修复函数幂等——对完好序列无操作，重复执行安全。

整体来看流机制和中断处理的具体实现

async function chat(messages, signal) {
  const res = await fetch(..., {
    stream: true,  // 👈 开启流式
    signal,        // 👈 传入 AbortController，用于中断
  });

  const assembler = createAssembler(); // 👈 创建装配器
  let printed = false;

  for await (const event of sseJsonEvents(res.body)) {
    const textDelta = assembler.feed(event); // 喂事件，拿文本增量
    if (textDelta) {
      process.stdout.write(textDelta); // 👈 "打字机"效果
      printed = true;
    }
  }

  if (printed) process.stdout.write("\n");
  return { message: assembler.message(), aborted: signal.aborted };
}

关键新概念：

**sseJsonEvents**：手工解析 data: {...} 格式的 SSE 事件流
**createAssembler**：把零散的 delta 碎片拼成完整消息
**signal**：AbortController 的信号，用于中断 HTTP 请求
返回值从”完整消息”变成了”消息 + 是否中断”

let activeTurn = null; // 当前轮次的 AbortController

function finishInterrupt(messages) {
  const filled = repairDanglingToolCalls(messages);
  console.log(`⏹ 本轮已中断${filled ? `，回填了 ${filled} 条合成工具结果` : ""}`);
}

async function runTurn(messages) {
  const controller = new AbortController();
  activeTurn = controller; // 注册到全局

  try {
    while (true) {
      // 使用 controller.signal 调用 chat
      let result;
      try {
        result = await chat(messages, controller.signal);
      } catch (err) {
        if (!controller.signal.aborted) throw err;
        return finishInterrupt(messages); // 👈 中断后的修复入口
      }
      
      const { message: msg, aborted } = result;
      
      // 半截消息也要进历史（用户已经看到了流出的文字）
      if (msg.content || msg.tool_calls?.length) messages.push(msg);
      
      if (aborted) return finishInterrupt(messages); // 👈 中断后修复
      
      // ... 执行工具 ...
      for (const call of msg.tool_calls) {
        if (controller.signal.aborted) break; // 👈 检查是否已中断
        const output = dispatch(call);
        messages.push({ role: "tool", tool_call_id: call.id, content: output });
      }
      
      if (controller.signal.aborted) return finishInterrupt(messages);
      // ...
    }
  } finally {
    activeTurn = null;
  }
}

中断处理流程：

用户按 Ctrl+C → onInterrupt() 被触发
activeTurn.abort() 中断 HTTP 请求
chat() 中的 signal.aborted 变为 true，流被掐断
进入 finishInterrupt() → 调用 repairDanglingToolCalls(messages)
修复完成后，控制权返回主循环，用户可以继续输入

export function repairDanglingToolCalls(messages, note = INTERRUPT_NOTE) {
  // 1. 收集所有已有结果的 tool_call_id
  const answered = new Set(
    messages.filter((m) => m.role === "tool").map((m) => m.tool_call_id)
  );
  
  let repaired = 0;
  for (let i = messages.length - 1; i >= 0; i--) {
    const msg = messages[i];
    if (msg.role !== "assistant" || !msg.tool_calls?.length) continue;
    
    // 2. 修复断裂的 arguments（流被掐断时 JSON 可能不完整）
    for (const tc of msg.tool_calls) {
      try {
        JSON.parse(tc.function.arguments || "{}");
      } catch {
        tc.function.arguments = "{}"; // 👈 替换为合法空 JSON
      }
    }
    
    // 3. 找出没有对应 tool 消息的 call
    const missing = msg.tool_calls.filter((tc) => !answered.has(tc.id));
    if (missing.length === 0) continue;
    
    // 4. 在 assistant 消息后面插入合成 tool 消息
    messages.splice(i + 1, 0, ...missing.map((tc) => ({
      role: "tool",
      tool_call_id: tc.id,
      content: note // "(用户中断了执行，该工具未运行...)"
    })));
    repaired += missing.length;
  }
  return repaired;
}

这个函数其实就是填充了之前我们提到过的JSON代码壳子，能返回完整的json去parse，让模型正常处理数据

举例说明：

中断前的历史：

1
2
3

[0] user: "读一下 package.json"
[1] assistant: "好的，我来读" + tool_calls: [call_A, call_B]  // call_A 已执行，call_B 悬空
[2] tool: call_A 的结果

中断后调用 repairDanglingToolCalls 后：

[0] user: "读一下 package.json"
[1] assistant: "好的，我来读" + tool_calls: [call_A, call_B]
[2] tool: call_A 的结果
[3] tool: call_B 的结果 ← 合成的 "(用户中断了执行，该工具未运行)"

这样历史就”配平”了，下一轮调用 API 不会再报 400 错误。

练习

中断有时过重——用户只是想补一句”顺便用 –verbose 跑”，不想丢弃正在生成的回答。codex 和 Reina 都支持 steer：不打断当前流，把用户插话排队，等本轮迭代提交后作为下一条 user 消息注入。给本章 agent 加上这个能力：轮次进行中输入的文字不触发中断、进入队列（提示：难点在插入位置——工具结果和插话的先后顺序，想想为什么插在 tool 结果之后比之前安全）。

有研究过中断机制在代码层面的逻辑，显然这里插入会话时是在 tool 结果之后比较安全的。这里输入的流转是

用户输入->模型分析->tool_calls，content->调用tool->返回填充好的tools调用结果(JSON代码块)
[
  { role: "user", content: "读取 package.json" },           // 用户提问
  { role: "assistant", content: "", tool_calls: [...] },    // 模型决定调用工具
  { role: "tool", tool_call_id: "call_123", content: "..." }, // 工具执行结果
  { role: "assistant", content: "这个项目叫 my-app" },      // 模型基于结果回答
]

在这一轮完成后插入用户的补充需求是比较合理的。如果在tool 结果之前插入会导致导致严重的会话逻辑错乱，模型没有人那么智能，他们是按照固定的程序执行的

OpenAI/DeepSeek 等 API 有一个严格约束：**assistant** 消息中的 **tool_calls** 必须紧跟着对应的 **tool** 结果消息，中间不能插入其他角色。

# ✅ 正确配平
assistant: [tool_call: A, tool_call: B]
tool: A 的结果
tool: B 的结果

# ❌ 错误（中间插入了 user）
assistant: [tool_call: A, tool_call: B]
tool: A 的结果
user: "顺便加个 --verbose"  ← 插入
tool: B 的结果

大部分 API 后端会直接返回 400 错误，因为违反了”tool 消息必须紧跟 assistant 消息”的协议要求。

在代码层面的改动思路主要是

在runTurn主循环函数添加插入会话函数，要在本轮任务执行完成后插入用户消息队列的输入
把当前代码中 readline.question() 同步阻塞设计(任务执行时无法接受输入)，改为异步输入模式

这里其实我不是很会写，就直接把ai写好的东西贴出来吧

// steer.mjs —— 用户插话排队机制

export class SteerQueue {
  constructor() {
    this.pending = [];      // 排队中的用户消息
    this.isSteering = false; // 是否正在处理排队
    this.isActiveTurn = false; // 是否有轮次在运行
  }

  /** 轮次开始时调用，标记运行中 */
  turnStart() {
    this.isActiveTurn = true;
  }

  /** 轮次结束时调用，标记空闲 */
  turnEnd() {
    this.isActiveTurn = false;
  }

  /** 用户输入时调用：轮次运行中→排队，空闲时→立即处理 */
  enqueue(content) {
    const msg = { role: "user", content: content.trim() };
    if (this.isActiveTurn) {
      this.pending.push(msg);
      console.log(`\x1b[36m⏳ 已排队（${this.pending.length} 条），将在本轮结束后处理\x1b[0m`);
      return { queued: true, msg };
    } else {
      console.log(`\x1b[32m▶ 立即处理\x1b[0m`);
      return { queued: false, msg };
    }
  }

  /** 轮次结束后调用：取出所有排队消息，返回是否还有排队 */
  flush() {
    if (this.pending.length === 0) return { messages: [], hasMore: false };
    const msgs = this.pending.splice(0);
    console.log(`\x1b[35m📨 注入 ${msgs.length} 条排队消息\x1b[0m`);
    return { messages: msgs, hasMore: true };
  }

  /** 检查是否有排队 */
  hasPending() {
    return this.pending.length > 0;
  }

  /** 获取排队数量 */
  count() {
    return this.pending.length;
  }
}
#!/usr/bin/env node
// s05-steer —— 在 s05 基础上增加 steer（排队插话）能力

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";
import { sseJsonEvents, createAssembler, repairDanglingToolCalls } from "./stream.mjs";
import { SteerQueue } from "./steer.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;
const MODEL = process.env.AGENT_MODEL ?? "deepseek-chat";

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。");
  process.exit(1);
}

const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（与 s05 相同）─────────────────────────────────────────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        return `命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`;
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) return `编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`;
      if (text.indexOf(old_string, first + 1) !== -1)
        return `编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`;
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return `未知工具：${call.function.name}`;
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return `工具参数不是合法 JSON：${err.message}`;
  }
  try {
    return tool.handler(args);
  } catch (err) {
    return `工具执行出错：${err.message}`;
  }
}

// ─── 流式 chat（与 s05 相同）─────────────────────────────────────────────

async function chat(messages, signal) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...messages],
      tools: TOOLS,
      stream: true,
    }),
    signal,
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);

  const assembler = createAssembler();
  let printed = false;
  try {
    for await (const event of sseJsonEvents(res.body)) {
      const textDelta = assembler.feed(event);
      if (textDelta) {
        process.stdout.write(textDelta);
        printed = true;
      }
    }
  } catch (err) {
    if (!signal.aborted) throw err;
  }
  if (printed) process.stdout.write("\n");
  return { message: assembler.message(), aborted: signal.aborted };
}

// ─── 主循环：支持 Steer（排队插话）──────────────────────────────────────

async function runTurn(messages, steer) {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;
  const controller = new AbortController();

  // 标记轮次开始
  steer.turnStart();

  try {
    while (true) {
      // ── 检查排队：如果队列不为空，先注入排队消息 ──
      if (steer.hasPending()) {
        const { messages: queued, hasMore } = steer.flush();
        if (queued.length > 0) {
          messages.push(...queued);
          // 注入后继续循环，让模型先处理插话
          // 注意：不重置 budget，插话不算新轮次
          continue;
        }
      }

      // ── 预算检查 ──
      if (!budget.canContinue()) {
        const stop = budget.exhaustedStop();
        console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
        return;
      }

      // ── 调用模型 ──
      let result;
      try {
        result = await chat(messages, controller.signal);
      } catch (err) {
        if (!controller.signal.aborted) throw err;
        // 中断时修复悬空 tool_call（仍保留此能力，但 steer 模式下更少触发）
        return finishInterrupt(messages, steer);
      }

      const { message: msg, aborted } = result;

      // 半截消息也要进历史
      if (msg.content || msg.tool_calls?.length) messages.push(msg);
      
      if (aborted) {
        return finishInterrupt(messages, steer);
      }

      // ── 没有工具调用 → 本轮结束 ──
      if (!msg.tool_calls?.length) {
        // 检查是否有排队（可能在模型输出期间用户插话了）
        if (steer.hasPending()) {
          const { messages: queued, hasMore } = steer.flush();
          if (queued.length > 0) {
            messages.push(...queued);
            // 继续循环，不返回
            continue;
          }
        }
        return;
      }

      // ── 执行工具 ──
      const records = [];
      for (const call of msg.tool_calls) {
        // 中断检测（但 steer 模式下不会主动 abort）
        if (controller.signal.aborted) break;

        const output = dispatch(call);
        messages.push({ role: "tool", tool_call_id: call.id, content: output });

        let input = {};
        try {
          input = JSON.parse(call.function.arguments || "{}");
        } catch { /* */ }
        records.push({
          name: call.function.name,
          input,
          status: FAILURE_RE.test(output) ? "failed" : "completed",
          output,
        });

        // ⭐ 关键：每个工具执行后检查是否有排队
        // 用户可能在工具执行过程中插话
        if (steer.hasPending()) {
          console.log(`\x1b[36m⏸ 检测到排队消息，暂停工具执行，先处理插话\x1b[0m`);
          // 不 break，让外层循环检查队列
          // 但当前工具已执行完成，下个循环会先注入排队消息
        }
      }

      // 如果有排队，不记录 budget，先处理排队
      if (steer.hasPending()) {
        const { messages: queued, hasMore } = steer.flush();
        if (queued.length > 0) {
          messages.push(...queued);
          continue; // 下一轮先处理插话
        }
      }

      // ── 检查中断 ──
      if (controller.signal.aborted) {
        return finishInterrupt(messages, steer);
      }

      // ── 记录本轮 ──
      const stop = budget.recordTurn(records);
      if (!stop) continue;

      if (isRecoverable(stop) && !repaired) {
        repaired = true;
        console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
        messages.push({ role: "user", content: repairPrompt(stop) });
        continue;
      }
      console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }
  } finally {
    steer.turnEnd();
  }
}

// ─── 中断处理（保留）─────────────────────────────────────────────────────

function finishInterrupt(messages, steer) {
  const filled = repairDanglingToolCalls(messages);
  console.log(
    `\n\x1b[31m⏹ 本轮已中断${filled ? `，回填了 ${filled} 条合成工具结果` : ""}。会话可以继续。\x1b[0m`,
  );
  // 检查是否有排队（中断期间用户可能也输入了）
  if (steer.hasPending()) {
    console.log(`\x1b[36m📨 有 ${steer.count()} 条排队消息，将在下次轮次中处理\x1b[0m`);
  }
}

// ─── 主入口：改造输入处理 ──────────────────────────────────────────────

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
const messages = [];
const steer = new SteerQueue();

// 是否正在运行轮次（用于区分"轮次中"和"空闲"）
let isRunning = false;

// Ctrl+C：改为进入"steer 模式"，不强制中断
function onInterrupt() {
  if (isRunning) {
    // 轮次运行中 → 提示用户可以直接输入排队
    console.log("\n\x1b[33m📝 Steer 模式已激活——直接输入你的补充指令，将在本轮结束后处理\x1b[0m");
    console.log("\x1b[33m   （再按一次 Ctrl+C 强制退出）\x1b[0m");
    // 注意：不 abort，让轮次继续跑
    // 用户接下来的输入会通过 rl.on('line') 进入队列
  } else {
    console.log("\n再见。");
    process.exit(0);
  }
}

rl.on("SIGINT", onInterrupt);
process.on("SIGINT", onInterrupt);

// ⭐ 核心改造：用户输入处理
rl.on("line", async (input) => {
  const text = input.trim();
  if (!text) return;

  const result = steer.enqueue(text);

  if (result.queued) {
    // 排队了，不立即执行
    console.log(`\x1b[36m📌 已排队（当前队列 ${steer.count()} 条）\x1b[0m`);
    return;
  }

  // 空闲状态 → 直接作为 user 消息加入历史并启动轮次
  if (isRunning) {
    // 理论上不会进这里（因为 isRunning 为 true 时 enqueue 返回 queued: true）
    return;
  }

  messages.push(result.msg);
  isRunning = true;
  try {
    await runTurn(messages, steer);
  } finally {
    isRunning = false;
    // 检查是否有排队消息在运行期间被加入
    if (steer.hasPending()) {
      console.log(`\x1b[36m📨 有 ${steer.count()} 条排队消息待处理，输入任意内容触发处理\x1b[0m`);
    }
  }
});

// 启动时提示
console.log(
  `\x1b[32ms05-steer agent 已上线（${MODEL}，流式 + 排队插话）。\x1b[0m`
);
console.log(`\x1b[33m💡 轮次运行中按 Ctrl+C 进入 Steer 模式，直接输入补充指令即可排队\x1b[0m`);
console.log(`\x1b[33m   再次按 Ctrl+C 强制退出\x1b[0m`);

// 等待用户输入（readline 已经通过 'line' 事件处理）
// 但需要防止进程退出
await new Promise(() => {});

这里主要改动的是主循环函数和主入口(输入处理代码)

实测的时候发现代码处理还是有一些问题，第一次按下ctrl+C无法停止输出

思考题：合成结果的文案，”(用户中断了执行，该工具未运行)” 与 codex 风格的单词 “aborted”，各会把模型引向什么行为？构造一个”中断后用户说『继续』”的场景，推演两种文案下模型的下一步动作差异。

`"(用户中断了执行，该工具未运行)"`	“User interrupted execution, tool was not run”	中立陈述：告诉模型发生了什么，但不引导模型做任何事
`"aborted"`	“aborted”	状态标签：像一个错误码/状态码，暗示”这个操作异常终止了”

使用”aborted”，模型是无法理解当前场景具体发生了什么事。实测中发现，使用”aborted”时，在按下ctrl+c中断后，再跟ai说继续，ai不能理解具体要继续什么事情

与真实产品对照（延伸阅读）

先补几个正文省略的细节。悬空 tool_call 的报错原文：OpenAI 大意是 “An assistant message with ‘tool_calls’ must be followed by tool messages responding to each ‘tool_call_id’”；Anthropic 的版本是 “tool_use ids were found without tool_result blocks”。”parse 留到拼完之后”之所以顺手，是因为 s02 的分层里 dispatch 本来就在那时才 parse——分层带来的便利。还有一个 readline 细节：终端在 readline 手里时处于 raw 模式，Ctrl+C 不产生进程信号，而是触发 rl 的 'SIGINT' 事件——所以 rl.on("SIGINT") 和 process.on("SIGINT")（非 TTY 时）都要监听。

Reina（本系列对照的生产级 agent）的对应机制在 packages/providers/src/tool-pairing.ts：normalizeToolPairing 做双向配平——除了本章的”悬空 call 合成占位输出”（合成文案同样说明”可能被中断丢失，需要时重跑”），还处理反方向的”孤儿结果”：一条 tool 结果找不到发起它的 call，同样会被后端拒收（”No tool call found for function call output with call_id …”）。codex 的做法是把孤儿直接丢弃；Reina 把它降级为普通文本消息——因为 Reina 的孤儿消息里常含有真实信息，直接丢弃会损失内容。这套修复在生产里静默执行，但 REINA_STRICT_TOOL_PAIRING=1 时会直接抛错——配平失守说明上游某个不变量被破坏，开发环境里应当尽早暴露。

引擎侧的中断在 packages/core/src/engine.ts：interrupt() 除了 abort，还会立刻换上一个新的 AbortController 并清空待处理队列——这曾引出一个隐蔽 bug：换新之后，工具批次里尚未执行的调用读到的是新 controller 的未中断 signal，会照常执行；所以 Reina 在批次内每次派发前检查的是 session.interrupted 标志位，而不是 signal。另外中断不只有 Ctrl+C 一种：进程崩溃、断电也是中断——recoverInterruptedTurn() 在重新加载会话时检测”卡在 running 状态的工具调用”，统一标记失败并回填错误文案。这依赖会话落盘，见 s08。

上下文压缩

历史只增不减，迟早超过模型一次能读的上限。本章实现压缩器，并解决它自带的隐蔽问题。

本章代码 = s03 的最小 agent 循环 + compaction.mjs 压缩器（每轮结束检查触发）。

问题

你让 agent”把 utils/date.js 的 formatDate 改成支持时区，改完跑测试”。五十多轮工具调用后，API 突然拒收：This model's maximum context length is 131072 tokens...——历史只增不减，超过了模型一次能读的上限（上下文窗口）。

解法是压缩：把旧消息换成摘要。但压缩自带一个隐蔽问题：摘要是模型写的，转述必然走样——第一次压缩，你的指令变成”用户在重构日期工具函数”；再压一次，变成”用户在优化项目代码”——最后 agent 停下来问你想做什么。它没报错，只是把指令转述丢了。

解决方案

压缩不是”全部换成摘要”：用模型生成的结构化摘要替换中段历史，同时逐字保留启动本轮任务的用户消息、原样保留最近的尾部（模型正在用的工作记忆）。既然转述会走样，有些内容必须逐字保留过压缩。

触发时机用服务商报告的 usage 判断，不自己数 token；摘要让模型按栏目填写，不是笼统总结；摘要模型调用失败时降级为纯字符串处理的提取式摘要——压缩不能因摘要失败而毁掉会话。

运行

免 key 演示：

1	`node notes/s06_compaction/demo.mjs`

#!/usr/bin/env node
// s06 免 key 演示 —— 伪造一段长对话，看压缩器怎么做决定：
//   场景一：触发判定（信 usage，不信估算）
//   场景二：切片决策 + 结构化摘要（摘要模型用替身，不调 API）
//   场景三：摘要调用失败 → 提取式降级，会话不陪葬
//   场景四：回退/分支后，压缩产物（摘要）是丢是留
//
// 运行：node s06_compaction/demo.mjs

import { shouldCompact, compactSplitIndex, compactMessages } from "./compaction.mjs";

// ─── 伪造一段长会话 ──────────────────────────────────────────────────────
// 形状刻意贴近真实：先有一段"上一个任务"的往返，然后用户发出当前任务
// （启动消息），后面跟一长串 assistant(tool_calls) / tool 交替。

const LAUNCH = "帮我把 utils/date.js 里的 formatDate 改成支持时区参数 tz，默认 UTC；改完跑 npm test，把挂掉的用例也修好。";

let seq = 0;
function toolRound(name, args, output) {
  const id = `c${++seq}`;
  return [
    { role: "assistant", content: "", tool_calls: [{ id, type: "function", function: { name, arguments: JSON.stringify(args) } }] },
    { role: "tool", tool_call_id: id, content: output },
  ];
}

const messages = [
  // ── 上一个任务：已经完结，是最该被压缩的部分 ──
  { role: "user", content: "这个仓库的测试是怎么组织的？大概讲讲。" },
  ...toolRound("run_shell", { command: "ls tests" }, "date.test.js\nhttp.test.js\nutils.test.js\n" + "fixture-….json\n".repeat(40)),
  ...toolRound("read_file", { path: "package.json" }, '  1\t{ "scripts": { "test": "node --test tests/" } }\n' + "  …\n".repeat(50)),
  { role: "assistant", content: "测试用 node --test 跑 tests/ 目录，按模块分文件，fixture 放同目录 JSON。……（此处省略三段介绍）" },
  // ── 当前任务：启动消息 + 10 轮工具往返 ──
  { role: "user", content: LAUNCH },
  ...toolRound("run_shell", { command: "rg -n formatDate src utils" }, "utils/date.js:12: export function formatDate(ts) {\ntests/date.test.js:8: formatDate(0)"),
  ...toolRound("read_file", { path: "utils/date.js" }, "  12\texport function formatDate(ts) {\n  13\t  return new Date(ts).toISOString().slice(0, 10);\n" + "  …\n".repeat(60)),
  ...toolRound("read_file", { path: "tests/date.test.js" }, "   8\tassert.equal(formatDate(0), '1970-01-01');\n" + "  …\n".repeat(40)),
  ...toolRound("edit_file", { path: "utils/date.js", old_string: "function formatDate(ts) {", new_string: "function formatDate(ts, tz = 'UTC') {" }, "已编辑 utils/date.js"),
  ...toolRound("run_shell", { command: "npm test" }, "FAIL tests/date.test.js\n  ✕ formats with timezone (invalid tz)\n" + "    at …\n".repeat(30)),
  ...toolRound("read_file", { path: "utils/date.js" }, "  12\texport function formatDate(ts, tz = 'UTC') {\n" + "  …\n".repeat(60)),
  ...toolRound("edit_file", { path: "utils/date.js", old_string: "toISOString().slice(0, 10)", new_string: "new Intl.DateTimeFormat('en-CA', { timeZone: tz }).format(d)" }, "已编辑 utils/date.js"),
  ...toolRound("run_shell", { command: "npm test" }, "FAIL tests/date.test.js\n  ✕ legacy format (expected '1970-01-01', got '1970-01-01,')\n" + "    at …\n".repeat(30)),
  ...toolRound("edit_file", { path: "tests/date.test.js", old_string: "formatDate(0)", new_string: "formatDate(0, 'UTC')" }, "已编辑 tests/date.test.js"),
  ...toolRound("run_shell", { command: "npm test" }, "PASS tests/date.test.js (12 tests)\n" + "  ✓ …\n".repeat(12)),
];

const LAUNCH_IDX = messages.findIndex((m) => m.content === LAUNCH);

const preview = (m) =>
  m.role === "assistant" && m.tool_calls?.length
    ? `assistant → ${m.tool_calls[0].function.name}(${m.tool_calls[0].function.arguments.slice(0, 40)}…)`
    : `${m.role.padEnd(9)} ${(m.content ?? "").replace(/\s+/g, " ").slice(0, 50)}…`;

// ─── 场景一：触发判定 ────────────────────────────────────────────────────

console.log("━━━ 场景一：触发判定（usage 对照 128k 窗口，阈值 75%） ━━━");
for (const usage of [
  { prompt_tokens: 58_000, completion_tokens: 1_200, total_tokens: 59_200 },
  { prompt_tokens: 98_400, completion_tokens: 1_800, total_tokens: 100_200 },
]) {
  const d = shouldCompact({ usage, contextWindow: 128_000, triggerPercent: 75, messageCount: messages.length });
  console.log(`  usage.total_tokens=${usage.total_tokens} → ${d.compact ? "🗜️ 触发压缩" : "✅ 不压"}：${d.why}`);
}

// ─── 场景二：切片决策 + 假摘要 ───────────────────────────────────────────

console.log("\n━━━ 场景二：切片决策（保什么 / 压什么 / 启动消息逐字保留） ━━━");
const naive = messages.length - 8;
const keepFrom = compactSplitIndex(messages, { keepRecent: 8 });
console.log(`  共 ${messages.length} 条消息。只看"尾部保底 8 条"，该从第 ${naive} 条切——`);
console.log(`  但那样启动任务的用户消息（第 ${LAUNCH_IDX} 条）就会被摘要转述掉。`);
console.log(`  分割点回拉到最后一条真实用户消息：实际 keepFrom = ${keepFrom}。逐条判决：`);
messages.forEach((m, i) => {
  const mark = i < keepFrom ? "🗜️ 压缩" : i === LAUNCH_IDX ? "📌 保留 ←启动消息，逐字" : "📌 保留";
  console.log(`    [${String(i).padStart(2)}] ${mark}  ${preview(m)}`);
});

// 摘要模型替身：真实实现是一次不带 tools 的 API 调用（见 agent.mjs）。
const fakeSummarize = async () =>
  [
    "1. 任务目标：用户原话——\"这个仓库的测试是怎么组织的？大概讲讲。\"（该任务已完成并已回答）",
    "2. 已完成：确认测试用 node --test 跑 tests/ 目录，按模块分文件。",
    "3. 未完成：无（此段落只覆盖被压缩的旧任务）。",
    "4. 涉及文件与命令：package.json、tests/；ls tests。",
    "5. 关键决定与坑：无。",
  ].join("\n");

const ok = await compactMessages(messages, { summarize: fakeSummarize, keepRecent: 8 });
console.log(`\n  压缩完成：${messages.length} 条 → ${ok.messages.length} 条（压掉 ${ok.dropped} 条，降级=${ok.degraded}）`);
console.log("  压缩后的消息形状：");
console.log(`    [0] 摘要消息（role=user）: ${ok.messages[0].content.replace(/\s+/g, " ").slice(0, 56)}…`);
console.log(`    [1] ${preview(ok.messages[1])}   ← 启动消息，一字未动`);
console.log(`    [2..${ok.messages.length - 1}] 当前任务的全部工具往返，原样保留`);

console.log("\n  —— 回拉不是无限的：把上限压到 1000 字符再切一次 ——");
const bounded = compactSplitIndex(messages, { keepRecent: 8, maxAnchorChars: 1000 });
console.log(`  当前任务的历史超出上限，放弃回拉，keepFrom = ${bounded}（启动消息进了压缩区）。`);
console.log("  此时兜底的是摘要 prompt 第 1 栏：\"逐字引用用户原话，禁止转述\"。");

// ─── 场景三：摘要失败 → 降级 ────────────────────────────────────────────

console.log("\n━━━ 场景三：摘要模型挂了（超时/限流），压缩绝不能毁掉会话 ━━━");
const fallback = await compactMessages(messages, {
  summarize: async () => { throw new Error("429 rate limited"); },
  keepRecent: 8,
});
console.log(`  摘要调用抛出 429 → 自动降级为提取式摘要（degraded=${fallback.degraded}），会话照常继续。`);
console.log("  降级摘要节选：");
for (const line of fallback.summary.split("\n").slice(0, 5)) console.log(`    ${line}`);
console.log(`    …（共 ${fallback.summary.split("\n").length} 行）`);

// ─── 场景四：回退/分支后，压缩产物是丢是留 ──────────────────────────────
//
// 聊天产品迟早要做"撤回 / 从这条消息创建分支"。此时会话有两份历史：
//   · 完整文字稿 transcript —— 渲染层展示用，从未被压缩
//   · 模型视图 modelView    —— 场景二压缩后的 [摘要 + 尾部]
// 压缩后的新消息双写进两份（真实引擎 appendMessage 同款）。
// 回退时最顺手的实现是"从完整文字稿重建、把摘要当缓存清掉"——坑就在这。

console.log("\n━━━ 场景四：回退/分支后，压缩产物（摘要）是丢是留？ ━━━");
const followUp = { role: "user", content: "顺手把 parseDate 也支持一下 tz 参数。" };
const transcript = [...messages, followUp]; // 完整文字稿（回退的裁剪对象）
const modelView = [...ok.messages, followUp]; // 压缩后的模型视图

const chars = (msgs) => msgs.reduce((n, m) => n + (m.content?.length ?? 0), 0);

// 用户在 followUp 处撤回（分支到它之前）：
const naiveFork = transcript.slice(0, transcript.indexOf(followUp));
console.log(`  幼稚分支（丢弃摘要，从文字稿重建）：${naiveFork.length} 条原文，约 ${chars(naiveFork)} 字符`);
console.log("    → 原文水位和压缩前一样高，下一轮触发判定立刻再压一次：白付一次摘要调用，");
console.log("      且新摘要 ≠ 旧摘要（保留的细节会洗牌）——用户看到\"从哪回退都触发压缩\"。");

// 正确做法：切点消息在压缩后视图里找得到 ⇒ 摘要只覆盖切点之前的内容 ⇒ 随分支保留
const cut = modelView.indexOf(followUp);
const reuseFork = modelView.slice(0, cut);
console.log(`  复用分支（裁剪压缩后视图）：${reuseFork.length} 条，约 ${chars(reuseFork)} 字符（[0] 仍是摘要消息）`);
console.log("    → 落在\"摘要 + 尾部\"水位，零额外压缩。");

// 反例：撤回进被压缩区（比如回到最早那条 user）——切点消息在 modelView 里找不到
const earliest = messages[0];
console.log(`  反例：回退到第 0 条（"${earliest.content.slice(0, 18)}…"）：`);
console.log(`    它在压缩后视图里${modelView.includes(earliest) ? "找得到" : "找不到"} ⇒ 旧摘要概括了刚被撤回的"未来"，`);
console.log("    复用会把撤回的内容从摘要里泄漏回去 ⇒ 这种情况才丢弃摘要、用原文重建。");
console.log("  判据一句话：切点消息在压缩后视图里找得到就保留摘要并裁剪视图；找不到才重建。");

四个场景：触发判定、切片决策、摘要失败降级、回退/分支后压缩产物的去留。

PS E:\AI_Agent> node 参考源码\learn-agent\notes\s06_compaction\demo.mjs
━━━ 场景一：触发判定（usage 对照 128k 窗口，阈值 75%） ━━━
  usage.total_tokens=59200 → ✅ 不压：59200 / 128000 tokens（阈值 96000），还有余量
  usage.total_tokens=100200 → 🗜️ 触发压缩：100200 tokens 已超过阈值 96000（窗口的 75%）

━━━ 场景二：切片决策（保什么 / 压什么 / 启动消息逐字保留） ━━━
  共 27 条消息。只看"尾部保底 8 条"，该从第 19 条切——
  但那样启动任务的用户消息（第 6 条）就会被摘要转述掉。
  分割点回拉到最后一条真实用户消息：实际 keepFrom = 6。逐条判决：
    [ 0] 🗜️ 压缩  user      这个仓库的测试是怎么组织的？大概讲讲。…
    [ 1] 🗜️ 压缩  assistant → run_shell({"command":"ls tests"}…)
    [ 2] 🗜️ 压缩  tool      date.test.js http.test.js utils.test.js fixture-….…
    [ 3] 🗜️ 压缩  assistant → read_file({"path":"package.json"}…)
    [ 4] 🗜️ 压缩  tool       1 { "scripts": { "test": "node --test tests/" } }…
    [ 5] 🗜️ 压缩  assistant 测试用 node --test 跑 tests/ 目录，按模块分文件，fixture 放同目录 JS…
    [ 6] 📌 保留 ←启动消息，逐字  user      帮我把 utils/date.js 里的 formatDate 改成支持时区参数 tz，默认 UTC…
    [ 7] 📌 保留  assistant → run_shell({"command":"rg -n formatDate src utils"}…)
    [ 8] 📌 保留  tool      utils/date.js:12: export function formatDate(ts) {…
    [ 9] 📌 保留  assistant → read_file({"path":"utils/date.js"}…)
    [10] 📌 保留  tool       12 export function formatDate(ts) { 13 return new…
    [11] 📌 保留  assistant → read_file({"path":"tests/date.test.js"}…)
    [12] 📌 保留  tool       8 assert.equal(formatDate(0), '1970-01-01'); … … …
    [13] 📌 保留  assistant → edit_file({"path":"utils/date.js","old_string":"fu…)
    [14] 📌 保留  tool      已编辑 utils/date.js…
    [15] 📌 保留  assistant → run_shell({"command":"npm test"}…)
    [16] 📌 保留  tool      FAIL tests/date.test.js ✕ formats with timezone (i…
    [17] 📌 保留  assistant → read_file({"path":"utils/date.js"}…)
    [18] 📌 保留  tool       12 export function formatDate(ts, tz = 'UTC') { ……
    [19] 📌 保留  assistant → edit_file({"path":"utils/date.js","old_string":"to…)
    [20] 📌 保留  tool      已编辑 utils/date.js…
    [21] 📌 保留  assistant → run_shell({"command":"npm test"}…)
    [22] 📌 保留  tool      FAIL tests/date.test.js ✕ legacy format (expected …
    [23] 📌 保留  assistant → edit_file({"path":"tests/date.test.js","old_string…)
    [24] 📌 保留  tool      已编辑 tests/date.test.js…
    [25] 📌 保留  assistant → run_shell({"command":"npm test"}…)
    [26] 📌 保留  tool      PASS tests/date.test.js (12 tests) ✓ … ✓ … ✓ … ✓ ……

  压缩完成：27 条 → 22 条（压掉 6 条，降级=false）
  压缩后的消息形状：
    [0] 摘要消息（role=user）: [上下文压缩] 更早的 6 条对话已被压缩，以下摘要是它们仅存的记忆： 1. 任务目标：用户原话——"这个仓库的…
    [1] user      帮我把 utils/date.js 里的 formatDate 改成支持时区参数 tz，默认 UTC…   ← 启动消息，一字未动
    [2..21] 当前任务的全部工具往返，原样保留

  —— 回拉不是无限的：把上限压到 1000 字符再切一次 ——
  当前任务的历史超出上限，放弃回拉，keepFrom = 19（启动消息进了压缩区）。
  此时兜底的是摘要 prompt 第 1 栏："逐字引用用户原话，禁止转述"。

━━━ 场景三：摘要模型挂了（超时/限流），压缩绝不能毁掉会话 ━━━
  摘要调用抛出 429 → 自动降级为提取式摘要（degraded=true），会话照常继续。
  降级摘要节选：
    （自动降级：摘要模型调用失败，以下为逐条提取的对话骨架）
    - user: 这个仓库的测试是怎么组织的？大概讲讲。
    - assistant 调用: run_shell
    - 工具结果: date.test.js http.test.js utils.test.js fixture-….json fixture-….json fixture-….json fixture-….json fixture-….json fixture-….json fixture-….json fixture-….json …(截断)
    - assistant 调用: read_file
    …（共 7 行）

━━━ 场景四：回退/分支后，压缩产物（摘要）是丢是留？ ━━━
  幼稚分支（丢弃摘要，从文字稿重建）：27 条原文，约 2818 字符
    → 原文水位和压缩前一样高，下一轮触发判定立刻再压一次：白付一次摘要调用，
      且新摘要 ≠ 旧摘要（保留的细节会洗牌）——用户看到"从哪回退都触发压缩"。
  复用分支（裁剪压缩后视图）：22 条，约 2079 字符（[0] 仍是摘要消息）
    → 落在"摘要 + 尾部"水位，零额外压缩。
  反例：回退到第 0 条（"这个仓库的测试是怎么组织的？大概讲讲…"）：
    它在压缩后视图里找不到 ⇒ 旧摘要概括了刚被撤回的"未来"，
    复用会把撤回的内容从摘要里泄漏回去 ⇒ 这种情况才丢弃摘要、用原文重建。
  判据一句话：切点消息在压缩后视图里找得到就保留摘要并裁剪视图；找不到才重建。

场景四的解读

这个场景讲的是对话压缩（摘要）机制在遇到“回退/分支”操作时，该怎么处理已有摘要，核心就解决一个问题：回退后，之前花钱生成的摘要还能不能用？

\1. 先搞懂几个概念

压缩/摘要：对话太长时，AI会把早期对话浓缩成一段摘要，省得超字数，也省调用费。
“压缩后视图”：当前对话的状态 = 一段摘要 + 最近几条完整原文。
“原文重建”：丢弃摘要，只用最原始的对话记录重新拼出上下文。

\2. 两种处理方式的对比（核心剧情）

假设你从某条消息回退，产生了分支：

“幼稚分支”（丢弃摘要）：不管三七二十一，把摘要删了，从原始文字稿重新搭。
- 结果：原文又变回一堆，字数超标，系统一看“哎呀又超了”，立刻再生成一次新摘要。
- 代价：白花一次摘要调用的钱，而且新摘要和旧摘要内容还不一样（细节会变），用户体验就是“怎么我一回退就卡顿/变内容”。
“复用分支”（保留摘要）：检查一下，发现回退到的那个点，还在当前摘要覆盖的范围内，那就直接把摘要留下来，只裁剪掉后面多余的原文。
- 结果：对话状态立刻变成“摘要 + 尾部”，字数刚好不超，零额外开销。

\3. 反例：什么时候必须丢弃？

你回退到了第0条消息（最开头的那句“这个仓库的测试是怎么组织的？大概讲讲…”）。
这条消息太老了，早就不在“压缩后视图”的尾部里了，只存在于旧摘要的概括中。
如果这时候还“复用摘要”，就等于把已经被概括进去的、后面发生的对话内容，偷偷带回到回退后的新分支里——这会造成信息泄漏（还没发生的“未来”被剧透回去了）。
所以这种情况必须丢弃摘要，用原文重建。

\4. 判据（一句话总结）

回退到的那个切点消息，如果在“压缩后视图”（摘要+尾部原文）里能找到，就保留摘要、裁剪视图；如果找不到，才丢弃摘要、从原文重建。

接上真实模型：

AGENT_API_KEY=sk-xxx node notes/s06_compaction/agent.mjs
# 可选：AGENT_CONTEXT_WINDOW=128000 AGENT_COMPACT_PERCENT=75
#!/usr/bin/env node
// s06 —— 上下文压缩：s03 的 agent（工具注册表 + 看门狗）+ 压缩器。
//
// 新增的全部内容：
//   · chat() 顺带返回服务商的 usage —— 触发判定的唯一依据
//   · 每轮工具执行结束后 maybeCompact()：超阈值就地压缩 messages
//   · 摘要调用失败自动降级为提取式摘要（compaction.mjs 内部兜底）
//
// 运行方式与 s03 相同：AGENT_API_KEY=sk-xxx node agent.mjs

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";
import { shouldCompact, compactMessages, SUMMARY_PROMPT } from "./compaction.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;
const MODEL = process.env.AGENT_MODEL ?? "deepseek-chat";
// 模型窗口没有任何 API 能查询，只能自己配（Reina 也是配在 models.json 里）。
// DeepSeek 128k；换模型记得改，配大了会在真窗口边界撞 "context too long"。
const CONTEXT_WINDOW = Number(process.env.AGENT_CONTEXT_WINDOW ?? 128_000);
const COMPACT_PERCENT = Number(process.env.AGENT_COMPACT_PERCENT ?? 75);

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。任何 OpenAI 兼容的 key 都行（DeepSeek / Kimi / GLM / OpenRouter / 本地 Ollama）。");
  process.exit(1);
}

const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（与 s03 相同）─────────────────────────────────────────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        return `命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`;
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) return `编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`;
      if (text.indexOf(old_string, first + 1) !== -1)
        return `编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`;
      // 不能用 text.replace(old, new)：new_string 里的 $$ / $& 会被 JS 当替换模式展开，静默写坏文件。
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return `未知工具：${call.function.name}`;
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return `工具参数不是合法 JSON：${err.message}`;
  }
  try {
    return tool.handler(args);
  } catch (err) {
    return `工具执行出错：${err.message}`;
  }
}

// ─── 模型调用：现在连 usage 一起带回来 ───────────────────────────────────

async function chat(messages) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...messages],
      tools: TOOLS,
    }),
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);
  const data = await res.json();
  // usage 是服务商的账本：prompt_tokens / completion_tokens / total_tokens。
  // 压缩触发只信它——本地估算对不上服务商的 tokenizer。
  return { message: data.choices[0].message, usage: data.usage };
}

// 摘要也是一次模型调用——但不带 tools（摘要不许干活），system 换成结构化
// 摘要指令。任何失败直接 throw，由 compactMessages 降级为提取式摘要。
async function summarizeViaModel(middleText) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [
        { role: "system", content: SUMMARY_PROMPT },
        { role: "user", content: `以下是即将被压缩掉的对话段（从旧到新）：\n\n${middleText}\n\n现在按五个小节输出摘要。` },
      ],
    }),
  });
  if (!res.ok) throw new Error(`摘要调用失败：API ${res.status}`);
  const data = await res.json();
  return data.choices[0].message.content;
}

// ─── 每轮结束检查：超阈值就地压缩 ────────────────────────────────────────

async function maybeCompact(messages, usage) {
  const decision = shouldCompact({
    usage,
    contextWindow: CONTEXT_WINDOW,
    triggerPercent: COMPACT_PERCENT,
    messageCount: messages.length,
  });
  if (!decision.compact) return;

  console.log(`\n\x1b[36m🗜️ 触发压缩：${decision.why}\x1b[0m`);
  const result = await compactMessages(messages, { summarize: summarizeViaModel });
  if (!result.compacted) {
    console.log("\x1b[36m   可压的前缀太小，本次跳过。\x1b[0m");
    return;
  }
  // 就地替换：外层循环和本函数共享同一个数组引用，不能换新数组。
  messages.splice(0, messages.length, ...result.messages);
  console.log(
    `\x1b[36m🗜️ 压缩完成：压掉 ${result.dropped} 条${result.degraded ? "（摘要降级为提取式）" : ""}，现存 ${messages.length} 条。\x1b[0m`,
  );
}

// ─── 主循环：s03 原样 + 每轮结束的压缩检查 ───────────────────────────────

async function runTurn(messages) {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;

  while (true) {
    if (!budget.canContinue()) {
      const stop = budget.exhaustedStop();
      console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }

    const { message: msg, usage } = await chat(messages);
    messages.push(msg);

    if (msg.content) console.log(`\n${msg.content}`);
    if (!msg.tool_calls?.length) {
      await maybeCompact(messages, usage); // 纯文本收尾的轮次也要检查
      return;
    }

    const records = [];
    for (const call of msg.tool_calls) {
      const output = dispatch(call);
      messages.push({ role: "tool", tool_call_id: call.id, content: output });
      let input = {};
      try {
        input = JSON.parse(call.function.arguments || "{}");
      } catch { /* 坏参数已作为失败回填 */ }
      records.push({
        name: call.function.name,
        input,
        status: FAILURE_RE.test(output) ? "failed" : "completed",
        output,
      });
    }

    // 每轮结束检查触发：此刻工具结果刚落地，是上下文长胖最快的时刻。
    // usage 来自本轮响应，等下一轮再看它就是旧账了。
    await maybeCompact(messages, usage);

    const stop = budget.recordTurn(records);
    if (!stop) continue;

    if (isRecoverable(stop) && !repaired) {
      repaired = true;
      console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
      messages.push({ role: "user", content: repairPrompt(stop) });
      continue;
    }
    console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
    return;
  }
}

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
const messages = [];

console.log(
  `s06 agent 已上线（${MODEL}，窗口 ${CONTEXT_WINDOW}，${COMPACT_PERCENT}% 触发压缩）。工具：${Object.keys(REGISTRY).join("、")}。Ctrl+C 退出。`,
);

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;
  messages.push({ role: "user", content: line });
  await runTurn(messages);
}

窗口大小在标准协议里没有字段可查，只能自己配置（见文末）。想观察压缩，把 AGENT_COMPACT_PERCENT 调到 5，让它连续读几个大文件。切片决策输出节选（真实运行）：

我这里没有触发场景二三四，只触发了场景一，贴一个其他师傅跑的

━━━ 场景二：切片决策（保什么 / 压什么 / 启动消息逐字保留） ━━━
  共 27 条消息。只看"尾部保底 8 条"，该从第 19 条切——
  但那样启动任务的用户消息（第 6 条）就会被摘要转述掉。
  分割点回拉到最后一条真实用户消息：实际 keepFrom = 6。逐条判决：
    [ 0] 🗜️ 压缩  user      这个仓库的测试是怎么组织的？大概讲讲。…
    ...
    [ 6] 📌 保留 ←启动消息，逐字  user      帮我把 utils/date.js 里的 formatDate 改成支持时区参数 tz，默认 UTC…
    [ 7] 📌 保留  assistant → run_shell({"command":"rg -n formatDate src utils"}…)
    ...
  压缩完成：27 条 → 22 条（压掉 6 条，降级=false）

━━━ 场景三：摘要模型挂了（超时/限流），压缩绝不能毁掉会话 ━━━
  摘要调用抛出 429 → 自动降级为提取式摘要（degraded=true），会话照常继续。

实现

核心的压缩上下文的mjs文件

// 上下文压缩 —— 让 agent 忘掉过程，但绝不忘掉任务。
//
// 从真实产品 Reina 的 packages/core/src/compaction.ts 简化移植，机制一致：
//   · 触发：用服务商返回的 usage 对照模型窗口，不做本地估算
//   · 切片：三段式 [被压缩的中段] + [启动任务的用户消息，逐字保留] + [最近尾部]
//   · 摘要：结构化 prompt（不是"总结一下"），失败时降级为提取式摘要
//   · 铁律：压缩绝不能因为压缩失败而毁掉会话

// ─── ① 触发决策：信服务商的账本，不信自己的估算 ───────────────────────────
//
// 本地数 token 对不上服务商的 tokenizer（中文误差尤其大，轻松差出 15%+），
// 估少了会在真正的窗口边界撞出 "context too long"。上一次 API 响应的
// usage 是服务商亲口报的数——它说多少就是多少。
// total_tokens = prompt + completion ≈ 下一轮请求要背的全部历史。
export function shouldCompact({ usage, contextWindow, triggerPercent = 75, messageCount, minMessages = 12 }) {
  if (!usage) return { compact: false, why: "还没有任何 API 响应，无从判断" };
  if (messageCount < minMessages) {
    return { compact: false, why: `消息太少（${messageCount} < ${minMessages}），压了也腾不出几个 token` };
  }
  const used = usage.total_tokens ?? (usage.prompt_tokens ?? 0) + (usage.completion_tokens ?? 0);
  const threshold = Math.floor((contextWindow * triggerPercent) / 100);
  if (used < threshold) {
    return { compact: false, why: `${used} / ${contextWindow} tokens（阈值 ${threshold}），还有余量`, used, threshold };
  }
  return { compact: true, why: `${used} tokens 已超过阈值 ${threshold}（窗口的 ${triggerPercent}%）`, used, threshold };
}

// ─── ② 切片决策：切哪里，比怎么摘要更重要 ─────────────────────────────────
//
// 返回分割点 keepFrom：messages[0, keepFrom) 被压缩，messages[keepFrom, ∞) 原样保留。
// 返回 0 = 不值得压。
//
// 两条规则，一条语法约束：
//   a) 尾部至少保留 keepRecent 条——最近的工具结果是模型正在用的工作记忆。
//   b) 分割点回拉到最后一条真实用户消息（role === "user"），让"用户到底让我
//      干嘛"逐字活过压缩。长任务里这条消息往往在几十条工具结果之前——不回拉，
//      它就会被摘要转述，而转述必然走样（Reina 真实修过的坑，见 README）。
//      注意"真实"二字：看门狗的纠偏 prompt、上一次压缩的摘要，也都是
//      role:"user" 塞进历史的——锚点若停在它们身上，真正的启动指令照样被
//      转述丢失。所以要按已知前缀把合成消息排除掉。
//      回拉有上限（maxAnchorChars）：保留区太大，压缩就腾不出空间了。
//   c) 语法约束：OpenAI 格式里 role:"tool" 消息必须紧跟它的 assistant
//      tool_calls 消息，切口不能落在一对中间，否则下一次请求直接 400。
// 循环自己塞进历史的 role:"user" 消息，都以这些前缀开头（见 compactMessages
// 和 loop-budget 的 repairPrompt）——它们不是"用户让我干嘛"，不能当锚点。
const SYNTHETIC_USER_PREFIXES = ["[上下文压缩]", "自动纠偏触发："];
const isRealUser = (m) =>
  m.role === "user" && !SYNTHETIC_USER_PREFIXES.some((p) => String(m.content ?? "").startsWith(p));

export function compactSplitIndex(messages, { keepRecent = 8, maxAnchorChars = 40_000 } = {}) {
  if (messages.length <= keepRecent) return 0;
  let keepFrom = messages.length - keepRecent;

  const lastUser = messages.findLastIndex(isRealUser);
  if (lastUser >= 0 && lastUser < keepFrom) {
    const anchoredChars = charsOf(messages.slice(lastUser));
    if (anchoredChars <= maxAnchorChars) keepFrom = lastUser;
    // 超限就不回拉——此时靠摘要 prompt 里"逐字引用用户原话"的条款兜底。
  }

  // 切口修正：不能让保留区以孤儿 tool 消息开头。向前多保留几条，
  // 直到把 tool 消息和它的 assistant tool_calls 划进同一侧。
  while (keepFrom > 0 && messages[keepFrom].role === "tool") keepFrom--;

  if (keepFrom <= 1) return 0; // 能压的太少，白白多花一次摘要调用
  return keepFrom;
}

// ─── ③ 摘要 prompt：结构化，不是"总结一下" ────────────────────────────────
//
// "总结一下"得到的是一段抒情散文，丢的恰好是接续任务最需要的硬信息。
// 逼模型按栏目填表，每一栏都对应"压缩后第一轮"会用到的东西。
// 第 1 栏的"逐字引用"是对切片规则 b) 的双保险：即使启动消息因超长
// 没能逐字保留，原话也还在摘要里。
export const SUMMARY_PROMPT = `你的任务是把一段即将被丢弃的对话压缩成结构化摘要。这份摘要是后续对话仅存的记忆，宁可啰嗦不可遗漏。只输出纯文本，不要调用任何工具。

必须包含以下小节：

1. 任务目标：用户让你做什么。逐字引用用户原话，禁止转述。
2. 已完成：做了哪些事、各自的结论。
3. 未完成 / 待办：接下来该做什么，按优先级排。
4. 涉及的文件与关键命令：完整路径和完整命令，逐字保留。
5. 关键决定与踩过的坑：为什么选了这条路，哪些路已被证明走不通。`;

// 被压缩的中段可能含 tool 消息和 assistant tool_calls——摘要请求不带 tools
// 参数，部分服务商会拒收这些结构。统一拍平成带标签的纯文本（Reina 的
// toSummarySourceMessages 同款处理），顺带把超长工具输出截到摘要够用的长度。
export function toSummarySource(middle) {
  const lines = [];
  for (const m of middle) {
    if (m.role === "tool") {
      lines.push(`[工具结果] ${clip(m.content, 1500)}`);
    } else if (m.role === "assistant") {
      if (m.content) lines.push(`[assistant] ${clip(m.content, 1500)}`);
      for (const call of m.tool_calls ?? []) {
        lines.push(`[assistant 调用工具] ${call.function.name}(${clip(call.function.arguments, 300)})`);
      }
    } else {
      lines.push(`[${m.role}] ${clip(m.content, 2000)}`);
    }
  }
  return lines.join("\n");
}

// ─── ④ 降级路径：摘要死了，会话不能陪葬 ───────────────────────────────────
//
// 摘要要调一次模型——而模型调用什么错都可能出（限流、超时、断网）。
// 此刻会话已经贴着窗口上限，"下一轮再试"往往等不起。提取式摘要不聪明，
// 但零依赖、永不失败：有损的记忆也比崩掉的会话强。
export function extractiveSummary(middle) {
  const lines = ["（自动降级：摘要模型调用失败，以下为逐条提取的对话骨架）"];
  for (const m of middle) {
    if (m.role === "tool") lines.push(`- 工具结果: ${clip(oneLine(m.content), 160)}`);
    else if (m.role === "assistant" && m.tool_calls?.length) {
      lines.push(`- assistant 调用: ${m.tool_calls.map((c) => c.function.name).join("、")}`);
    } else if (m.content) lines.push(`- ${m.role}: ${clip(oneLine(m.content), 240)}`);
  }
  return lines.join("\n");
}

// ─── 组装：执行一次压缩 ──────────────────────────────────────────────────
//
// summarize(middleText) 是调用方注入的异步函数（真 agent 里调 API，
// demo 里用替身）。返回新的 messages 数组，形状：
//   [摘要消息] + [保留尾部（以启动任务的用户消息开头，逐字未动）]
export async function compactMessages(messages, { summarize, keepRecent = 8, maxAnchorChars = 40_000 } = {}) {
  const keepFrom = compactSplitIndex(messages, { keepRecent, maxAnchorChars });
  if (keepFrom <= 0) return { compacted: false, messages };

  const middle = messages.slice(0, keepFrom);
  const tail = messages.slice(keepFrom);

  let summary;
  let degraded = false;
  try {
    summary = (await summarize(toSummarySource(middle)))?.trim();
    if (!summary) throw new Error("摘要为空");
  } catch {
    degraded = true;
    summary = extractiveSummary(middle);
  }

  const summaryMessage = {
    role: "user",
    content: `[上下文压缩] 更早的 ${middle.length} 条对话已被压缩，以下摘要是它们仅存的记忆：\n\n${summary}\n\n从中断处直接继续任务。不要复述摘要，不要重新自我介绍。`,
  };
  return { compacted: true, messages: [summaryMessage, ...tail], dropped: middle.length, degraded, summary };
}

// ─── 小工具 ──────────────────────────────────────────────────────────────

function charsOf(messages) {
  return messages.reduce((n, m) => n + (m.content?.length ?? 0), 0);
}

function clip(text, max) {
  if (typeof text !== "string") return "";
  return text.length <= max ? text : `${text.slice(0, max)}…(截断)`;
}

function oneLine(text) {
  return String(text ?? "").replace(/\s+/g, " ").trim();
}

① 触发时机：用服务商报告的 usage，不自己数 token

第一反应是自己算 token 数。但你没有服务商的分词器，本地估算（tiktoken 也一样）只是近似，中文误差可超 15%。估少了会在真实窗口边界撞出 context too long，来不及压缩。

正确做法零成本：每次响应都带 usage，是服务商报告的准确数字。total_tokens 约等于下一轮要携带的全部历史，超过窗口阈值（默认 75%）就压缩：

// 流式调用需在请求里加 stream_options: {"include_usage": true}，
// usage 才会出现在流的最后一片——不加则拿不到用量，压缩永远不会触发
const used = usage.total_tokens ?? (usage.prompt_tokens ?? 0) + (usage.completion_tokens ?? 0);
const threshold = Math.floor((contextWindow * triggerPercent) / 100);
if (used >= threshold) /* 触发压缩 */;

留 25% 余量：压缩本身还要调一次模型、摘要也占空间，得在仍有余地时动手。

流式数据传输过程是这样的

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"role":"assistant","content":"你"},"index":0}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"好"},"index":0}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"！"},"index":0}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":3,"total_tokens":13}}

data: [DONE]

② 压缩的形状：三段式，启动消息逐字保留

压缩不是”全部换成摘要”。正确形状是三段：

[system]                        ← 不动（它本来就不在 messages 里）
[中段历史]                       → 换成模型生成的结构化摘要
[启动本轮任务的用户消息]           → 逐字保留 ★
[最近的尾部消息]                  → 原样保留（模型正在用的工作记忆）

尾部是模型的工作记忆，压掉等于打断手头操作。启动消息逐字保留是本章核心：压缩总在工具循环中途触发，”最近 N 条”全是工具结果，原始指令恰好落在被压缩区——被转述后，模型从此在转述版指令上工作（真实产品踩过同坑）。

实现是把分割点回拉到最后一条真实用户消息：

// 默认从尾部保留 keepRecent 条
let keepFrom = messages.length - keepRecent;

// 找最后一条"真实的用户消息"
const lastUser = messages.findLastIndex(isRealUser);

// 如果这条真实用户消息在"保留区"之前（即它将被压缩掉）
if (lastUser >= 0 && lastUser < keepFrom) {
  // 检查把从 lastUser 开始到结尾的内容加起来，字符数是否超标
  if (charsOf(messages.slice(lastUser)) <= maxAnchorChars) {
    // 不超标 → 把分割点回拉到真实用户消息的位置
    keepFrom = lastUser;
  }
}
// 切口不能落在 assistant(tool_calls)/tool 一对中间，否则下轮请求 400
while (keepFrom > 0 && messages[keepFrom].role === "tool") keepFrom--;

注意 isRealUser：历史里的 user 消息不全是用户说的——s03 看门狗注入的纠偏消息、上次压缩留下的摘要也是 role:"user"。锚点停在它们身上，指令仍会丢，所以按已知前缀（[上下文压缩]、自动纠偏触发：）排除合成消息。

回拉有上限：启动消息若在几百条之前，全保留就腾不出空间——超限时放弃，由下一条兜底。

③ 摘要 prompt：按栏目填写，不是笼统总结

“总结一下上面的对话”得到的是泛泛叙述，丢的恰好是接续任务最需要的信息。让模型按栏目填写，每栏对应压缩后第一轮会用到的内容：

1. 任务目标：用户让你做什么。逐字引用用户原话，禁止转述。
2. 已完成：做了哪些事、各自的结论。
3. 未完成 / 待办：接下来该做什么，按优先级排。
4. 涉及的文件与关键命令：完整路径和完整命令，逐字保留。
5. 关键决定与踩过的坑：为什么选了这条路，哪些路已被证明走不通。

第 1 栏的”逐字引用”是决定②的双保险：即使启动消息因超长没能逐字保留，原话也还在摘要里。第 5 栏容易被忽略：不记录走不通的路，模型会把失败路径重走一遍。

④ 摘要失败时的降级

摘要要调一次模型，而调用可能限流、超时、断网。此刻会话已接近窗口上限，下轮再试来不及——压缩不能因摘要失败而毁掉会话。所以要一条不会失败的降级路径：提取式摘要，纯字符串处理，截每条消息首尾拼成骨架：

try {
  // 尝试智能压缩
  summary = await summarize(toSummarySource(middle));
} catch {
  degraded = true;                    // 标记：当前处于降级模式
  summary = extractiveSummary(middle); // 不用 AI，纯文本裁剪
}

有损的记忆也比崩溃的会话好。

⑤ 回退与分支：压缩产物随切点走，不是易失缓存

聊天产品迟早要做”撤回 / 从这条消息创建分支”。此时会话有两份历史：完整文字稿（渲染层展示用，从未被压缩）和模型视图（压缩后的 [摘要 + 尾部]，新消息双写进两份）。回退最顺手的实现，是从完整文字稿重建、把摘要当派生缓存清掉——反正超阈值还会再压。这个”反正”就是坑：会话长到压缩过，裁到切点的原文几乎必然仍超阈值，于是每次回退/分支都白付一次摘要调用，用户看到的是”从哪回退都触发压缩”。更隐蔽的是重摘要不幂等：新摘要保留的细节和旧摘要不同，回退一次，agent 的记忆就洗牌一次。

判据只有一句话：切点消息在压缩后视图里找得到 ⇒ 旧摘要只覆盖切点之前的内容 ⇒ 摘要随分支保留，把压缩后视图按切点裁剪即可；找不到（撤回进了被压缩区）⇒ 旧摘要概括了刚被撤回的”未来”，复用会把撤回的内容泄漏回去 ⇒ 这种情况才丢弃摘要、用原文重建：

const cut = modelView.findIndex((m) => m.id === cutMessageId);
if (cut >= 0) {
  branch.modelView = modelView.slice(0, cut);  // 摘要在 [0]，天然保留
  branch.summary = session.summary;            // 高频路径：零额外压缩
} else {
  branch.modelView = undefined;                // 低频路径：原文重建，
  branch.summary = undefined;                  // 下轮按需重新压缩
}

前者是高频路径——用户几乎总是撤回最近几条；后者才需要付重摘要的钱。值得一提 codex 的架构让这条判据不需要写出来：它的 fork 是对持久化事件流做前缀截断，而压缩本身就是流里的一条 Compacted 事件（带着替换历史）——切点在它之后，它自然留在前缀里；切点在它之前，它自然被截掉。快照式的 fork（复制状态对象、清空派生字段）没有这份免费午餐，两条分支都得手写，而且很容易全部写成第二条。

把这段话翻译的通俗易懂一点来看就是这样

场景设定

假设你有一个聊天界面，用户聊了100轮，中间触发过压缩（早期对话变成了摘要）。

现在系统里存着两份历史：

副本	内容	用途
完整文字稿	所有原始消息，一字不差	界面展示（渲染层）
模型视图	`[摘要] + [最近20条原文]`	发给 AI 模型用的（压缩后的）

用户发新消息时，两条都写（完整稿存着备查，模型视图用来跑 AI）。

问题：用户要”回退到第50条消息，从这里开个新分支”

最粗暴的做法（坑）

“从完整文字稿重建、把摘要当派生缓存清掉——反正超阈值还会再压”

翻译：

回退？简单！把压缩摘要删了，从完整文字稿重新搭模型视图
反正如果内容太多，压缩逻辑会自动再压一次

为什么是坑？

白花钱：回退到第50条，原文还有50条那么多，必然超过压缩阈值 → 立刻又触发一次压缩 → 白白多花一次摘要调用费
记忆洗牌：新生成的摘要和旧摘要不一样（AI 每次总结细节不同）。用户回退一次，AI 对早期对话的”记忆”就变一次 → 体验很差
用户体验：用户发现”怎么我无论回退到哪，都要卡一下（压缩）”

正确做法（判据）

切点消息在压缩后视图里找得到 ⇒ 保留摘要；找不到 ⇒ 丢弃摘要

情况A：回退到最近的消息（高频场景）

1 2	`压缩后视图 = [摘要(覆盖1-80条)] + [81,82,83,84,85] ← 尾部5条用户回退到第83条`

判断：第83条在压缩后视图里 ✅ 找得到

操作：

1 2	`branch.modelView = modelView.slice(0, cut); // 保留 [摘要] + [81,82,83] branch.summary = session.summary; // 摘要直接复用`

结果：零额外开销，瞬间完成。用户撤回最近几条消息是最常见操作，这个路径覆盖了90%+的场景。

情况B：回退到很早的消息（低频场景）

1 2	`压缩后视图 = [摘要(覆盖1-80条)] + [81,82,83,84,85] 用户回退到第20条`

判断：第20条在压缩后视图里 ❌ 找不到（它被压缩进摘要里了）

操作：

1
2
3

branch.modelView = undefined;   // 从完整文字稿重建
branch.summary = undefined;     // 丢弃旧摘要
// 下一轮对话时，系统发现内容太多，自动重新压缩

结果：需要重新生成摘要，花一次调用费。但这是低频操作（很少有人回退到很早期的消息），可以接受。

为什么情况B必须丢弃摘要？

“旧摘要概括了刚被撤回的’未来’，复用会把撤回的内容泄漏回去”

举例：

原始对话：1-100条，摘要概括了1-80条（包括”用户说要分析代码”）
用户回退到第20条，撤回了21-100条（包括”分析代码”这个后续指令）
如果复用旧摘要，摘要里仍然包含”用户说要分析代码” → 这相当于把被撤回的未来内容带到了新分支 → 信息泄漏

所以必须丢弃旧摘要，从第20条原文重新开始。

设计亮点

1. Codex 架构的”免费午餐”

“codex 的架构让这条判据不需要写出来”

Codex 的做法：对话历史是一串不可变事件流，压缩本身也是一条事件（Compacted），记录了”我用摘要替换了1-80条”。

回退到第20条 → 对事件流做前缀截断，Compacted 事件在截断点之后 → 自动被截掉
回退到第83条 → Compacted 在截断点之前 → 自动保留

判据自动生效，不需要写一行代码。

对话历史 = 一条一条追加的事件流（不可变）

[事件1] 用户: "你好"
[事件2] AI: "你好！"
[事件3] 用户: "帮我分析代码"
[事件4] AI: tool_call: read_file
[事件5] tool: 返回 file1 内容
[事件6] AI: tool_call: read_file
[事件7] tool: 返回 file2 内容
...（100条事件）
[事件80] 压缩事件: "我把1-70条替换成摘要：[摘要内容]"
[事件81] 用户: "继续分析"
[事件82] AI: "我发现了一个bug..."

关键点： Compacted（压缩）本身也是一条事件，记录着”我用摘要替换了哪些历史”。

2. 快照式架构的”坑”

“快照式的 fork（复制状态对象、清空派生字段）没有这份免费午餐”

你的项目架构：每次分支都复制一份完整状态（session 对象），摘要和模型视图是派生字段。

// 一个可变的 JavaScript 对象
session = {
  messages: [...],      // 完整消息列表
  summary: "摘要内容",  // 压缩生成的摘要（派生字段）
  modelView: [summary, ...尾部消息]  // 发给AI的视图（派生字段）
}

没有事件流那种”自动截断”的能力
必须手写判据决定摘要留还是丢
很容易全部写成”丢弃摘要”（因为最简单），导致所有回退都触发重新压缩

一句话总结

回退时，如果切点在压缩后的可见区（摘要+尾部）里 → 复用摘要，零成本；如果切点被压缩进摘要里了 → 丢弃摘要，从原文重建（因为复用会泄漏被撤回的内容）。前者是日常操作，后者是低频操作。

完整流程图

用户回退到消息 X
    │
    ▼
在"压缩后视图"里找 X
    │
    ├─ 找到了（X在摘要之后）→ 保留摘要，裁剪尾部 → ✅ 零开销
    │
    └─ 找不到（X在摘要里）→ 丢弃摘要，从原文重建 → ⚠️ 花一次摘要钱

练习

1. 本章每次压缩都从头生成摘要。改成滚动摘要：把上一次的摘要作为输入传给摘要模型，并在 prompt 里加一条”上一份摘要中仍然相关的部分逐字复制，不要改写”。想想为什么”逐字复制”比”合并改写”更重要（提示：和启动消息逐字保留是同一个道理——转述会累积走样）。

逐字复制是压缩上下文形成摘要机制的关键，上一轮的摘要是由准确的SUMMARY_PROMPT指导下生成的结果，已经是压缩过的，如果不逐字复制难免产生未知结果，例如只是转述或者概述，任务描述在一次次滚动中逐渐模糊

为什么 “逐字复制” 比 “合并改写” 重要？

防止信息衰减：在“传话游戏”中，每一次“合并改写”都是对已有信息的一次有损转述。多轮后，任务目标会变得模糊。
保护关键指令：SUMMARY_PROMPT 的第一条要求就是“逐字引用用户原话”。如果旧摘要中的关键指令（例如“用 Python 实现一个 Web 服务器”）被改写成“写一个后端程序”，任务的核心就丢失了。

重构 compactMessages 为滚动模式

你需要修改 compaction.mjs 中的 compactMessages 函数，增加 previousSummary 参数，并调整摘要生成逻辑。

// compaction.mjs

// 修改 compactMessages 函数，增加参数 previousSummary
export async function compactMessages(
  messages, 
  { 
    summarize, 
    keepRecent = 8, 
    maxAnchorChars = 40000,
    previousSummary = null // 新增：接收上一次的摘要内容
  } = {}
) {
  const keepFrom = compactSplitIndex(messages, { keepRecent, maxAnchorChars });
  if (keepFrom <= 0) return { compacted: false, messages };

  const middle = messages.slice(0, keepFrom);
  const tail = messages.slice(keepFrom);

  let summary;
  let degraded = false;
  try {
    // 将上一份摘要和新内容一起传给摘要模型
    const sourceForSummary = buildRollingSummarySource({
      previousSummary,
      newContent: toSummarySource(middle),
    });

    summary = (await summarize(sourceForSummary))?.trim();
    if (!summary) throw new Error("摘要为空");
  } catch {
    degraded = true;
    // 降级模式也需要处理滚动摘要
    summary = extractiveSummary(middle, previousSummary); 
  }

  // ... 后面的组装逻辑不变 ...
  const summaryMessage = {
    role: "user",
    content: `[上下文压缩] 更早的 ${middle.length} 条对话已被压缩，以下摘要是它们仅存的记忆：\n\n${summary}\n\n从中断处直接继续任务。不要复述摘要，不要重新自我介绍。`,
  };
  return { compacted: true, messages: [summaryMessage, ...tail], dropped: middle.length, degraded, summary };
}

辅助函数：构建滚动摘要的输入源

你需要一个函数来组织传给摘要模型的内容，核心是将上一份摘要作为”只读”的参考材料，而不是让它参与改写。

// compaction.mjs

// 构建滚动摘要的输入源
function buildRollingSummarySource({ previousSummary, newContent }) {
  let source = '';

  if (previousSummary) {
    // 关键指令：指示模型必须逐字复制旧摘要中的相关内容
    source += `【上一轮压缩生成的摘要（必须逐字保留其中仍适用的部分）】\n`;
    source += `${previousSummary}\n\n`;
    source += `【以下是最新一轮被压缩的对话内容】\n`;
  }

  source += newContent;
  return source;
}

调用方（agent.mjs）的改动

在 maybeCompact 函数中，你需要持久化摘要，以便在下次压缩时传入。

// agent.mjs

// 在文件顶部增加一个全局变量来保存上一次的摘要
let lastSummary = null;

async function maybeCompact(messages, usage) {
  const decision = shouldCompact({
    usage,
    contextWindow: CONTEXT_WINDOW,
    triggerPercent: COMPACT_PERCENT,
    messageCount: messages.length,
  });
  if (!decision.compact) return;

  console.log(`\n\x1b[36m🗜️ 触发压缩：${decision.why}\x1b[0m`);

  // 传入上一次的摘要
  const result = await compactMessages(messages, { 
    summarize: summarizeViaModel,
    previousSummary: lastSummary // 传入
  });

  if (!result.compacted) {
    console.log("\x1b[36m   可压的前缀太小，本次跳过。\x1b[0m");
    return;
  }

  // 更新摘要
  lastSummary = result.summary; // 保存本次生成的摘要

  messages.splice(0, messages.length, ...result.messages);
  console.log(
    `\x1b[36m🗜️ 压缩完成：压掉 ${result.dropped} 条${result.degraded ? "（摘要降级为提取式）" : ""}，现存 ${messages.length} 条。\x1b[0m`,
  );
}

2. 压缩后的摘要消息每轮都会重发一遍。它的内容是稳定的吗？如果你在摘要里加上”压缩于 {当前时间}”，会发生什么？（这与成本有关，下一章展开。）

1.压缩后的摘要内容是稳定的吗？

答案是：绝对不稳定（非确定性）。

即使输入完全相同的对话历史，大模型在两次独立压缩中生成的摘要几乎不可能完全相同。原因有三：

概率采样：LLM生成时即使temperature=0，浮点运算误差和不同硬件（GPU/CPU）也可能导致logits微妙差异，进而影响选词。
信息重排损失：压缩本质是“有损编码”。每轮新消息加入后，模型对旧信息的“重要性评分”会动态变化。上一轮认为重要的细节（如“用户提到喜欢蓝色”），在新消息（“但后来买了红色”）出现后，可能被下一轮压缩直接丢弃。
累积漂移：如果压缩的是“压缩后的摘要+新消息”，那么摘要中的微小错误会在下一轮被当作事实参与压缩，导致错误逐级放大（幻觉固化）。

\2. 如果在摘要里加上“压缩于 {当前时间}”，会发生什么？

会发生三件极具启发意义的事：

现象	具体后果
① 伪稳定性	时间戳作为“锚点文字”被固定下来，后续压缩时模型会倾向于保留这个固定短语。你会误以为摘要稳定了，但实际上只有时间戳本身稳定，它前后的实质内容依然在漂移。
② 污染注意力权重	模型会把这个时间戳视为“特殊标记”，在注意力计算中分配一部分权重给它。这会挤占对实际业务关键信息（如用户ID、核心诉求）的注意力，导致摘要质量轻微下降。
③ 暴露重复发送的致命伤	如果每轮都重发带新时间戳的摘要，模型会认为“时间变化”是重要信息。当你问“我们聊了多久”时，模型可能会根据摘要里的多个时间戳算出错误时长，甚至把“压缩时间”误解为“事件发生时间”。

更危险的隐含问题

“每轮都重发”这个行为本身，比摘要不稳定更致命。

假设每轮上下文长度是10k tokens，压缩后是1k。你把1k摘要+新用户问题发过去，模型回复后，你又把新的1k摘要（基于旧摘要生成）发过去。

这会导致：

信息熵递减：每一轮压缩都会丢失约5%-15%的细节（取决于模型）。经过10轮，原始信息保留率可能低于 0.910≈35%0.910≈35%。
自我引用循环：摘要开始描述“上一版的摘要说了什么”，而非原始对话事实。

Prompt 缓存

本章讲怎么保持请求开头稳定以命中前缀缓存，以及怎么测量命中率。

本章代码 = s03 的最小 agent 循环 + 每轮打印 token 用量与缓存命中率（免 key 演示：字节级找出缓存断点 + 只靠 usage 数字的浪费计数器）。

问题

跑一个 30 轮的任务，账单会吓你一跳：agent 每轮都把越来越长的对话历史全量重发，第 1 轮发 2k token，第 30 轮就是 100k，累计发送的输入是历史长度的几十倍——花钱大头全在输入侧。

功能相同的两个实现，账单可以差出近 7 倍——不命中缓存的原因往往只是 system prompt 里的一行 当前时间：...。

解决方案

服务商为此提供了前缀缓存：这次请求的开头和上次逐字节相同的部分，单价降到约十分之一。原理：模型收到请求后要把整个 prompt 从头算一遍（prefill，占输入侧几乎全部算力）；开头和上次相同时，中间结果（KV cache）可直接复用，几乎不花算力。第 30 轮请求里前 99% 与上一轮相同——命中打一折，不命中全价。以 30 轮、每轮 60k、累计输入 1.8M token 为例：

全部未命中：1.8M × 全价；
稳定 95% 命中：1.8M × (5% × 全价 + 95% × 一折) ≈ 0.145 × 全价，省 85%。

（DeepSeek/OpenAI 的缓存自动生效；Anthropic 要显式打 cache_control 断点，否则命中率恒为 0，计价见文末。）做法是三条纪律保持前缀逐字节稳定，加一套仪表测量命中率。

运行

免 key 演示：

1	`node notes/s07_prompt_cache/demo.mjs`

#!/usr/bin/env node
// s07 免 key 演示 —— 前缀缓存从哪里断，纯字符串就能算出来：
//   实验 A：时间戳写进 system → 缓存在第一行就断
//   实验 B：system 稳定，时间戳放最后一条用户消息 → 上一轮请求 100% 复用
//   实验 C：违反 append-only（"好心"截短旧工具输出）→ 断点跳回历史中间
//   实验 D：浪费计数器 → 只用 provider 的 usage 数字，把每次击穿折成 token 并归因
//
// 服务商按"与上次请求逐字节相同的前缀"给缓存价（实际按 token/块粒度对齐，
// DeepSeek 是 64 token 一块；这里用字符近似，原理相同）。
//
// 运行：node s07_prompt_cache/demo.mjs

// ─── 把请求"线性化"：服务商看到的就是这样一条字节流 ─────────────────────
// system、tools、每条消息依次拼接。缓存能复用的 = 和上一条字节流相同的开头。

const TOOLS = JSON.stringify([
  { name: "run_shell", description: "执行一条 shell 命令" },
  { name: "read_file", description: "读取一个文本文件" },
]);

function serializeRequest(system, messages) {
  const parts = [`[system]\n${system}`, `[tools]\n${TOOLS}`];
  for (const m of messages) parts.push(`[${m.role}]\n${m.content}`);
  return parts.join("\n\n");
}

function commonPrefixLength(a, b) {
  let i = 0;
  while (i < a.length && i < b.length && a[i] === b[i]) i++;
  return i;
}

function report(label, prev, cur) {
  const n = commonPrefixLength(prev, cur);
  const reused = ((n / prev.length) * 100).toFixed(1);
  console.log(`  ${label}：公共前缀 ${n} 字符（上一轮请求共 ${prev.length} 字符 → 可复用 ${reused}%）`);
  if (n < prev.length) {
    const from = Math.max(0, n - 16);
    console.log(`    断点上下文: …${JSON.stringify(prev.slice(from, n + 12))} vs …${JSON.stringify(cur.slice(from, n + 12))}`);
  } else {
    console.log("    断点在上一轮请求的末尾之后——新增的只有追加的消息，这就是理想形状。");
  }
  return n;
}

// ─── 伪造一段工具循环：1 条用户消息 + 两轮工具往返 ───────────────────────

const TOOL_OUTPUT_1 = '   1\t{\n   2\t  "scripts": {\n   3\t    "test": "vitest run",\n   4\t    "build": "tsc -p ."\n   5\t  }\n' + '   …\t（省略 60 行依赖列表）\n'.repeat(12);

const round1 = [
  { role: "assistant", content: '（调用 read_file {"path":"package.json"}）' },
  { role: "tool", content: TOOL_OUTPUT_1 },
];
const round2 = [
  { role: "assistant", content: '（调用 run_shell {"command":"npx vitest list"}）' },
  { role: "tool", content: "test/date.test.ts\ntest/http.test.ts\ntest/utils.test.ts" },
];

// 演示用假时钟：每次调用走一秒，模拟"每轮请求时间都不同"。
let tick = 0;
const clock = () => `10:23:0${++tick}`;

// ─── 实验 A：时间戳写进 system（缓存杀手）────────────────────────────────

console.log("━━━ 实验 A：时间戳写进 system ━━━");
const badSystem = () => `当前时间：${clock()}。你是一个运行在用户终端里的编程助手。先观察真实世界再行动。`;
const userA = { role: "user", content: "看看 package.json 里有哪些 scripts，然后告诉我怎么跑测试。" };

const a1 = serializeRequest(badSystem(), [userA]);
const a2 = serializeRequest(badSystem(), [userA, ...round1]);
const a3 = serializeRequest(badSystem(), [userA, ...round1, ...round2]);
report("第 1↔2 轮", a1, a2);
report("第 2↔3 轮", a2, a3);
console.log("  断点永远卡在时间戳的秒位上 → 后面的 tools、全部历史，每一轮都按全价重算。");

// ─── 实验 B：system 稳定，时间戳放最后一条用户消息 ───────────────────────

console.log("\n━━━ 实验 B：system 稳定，时间戳放进用户消息尾部 ━━━");
const goodSystem = "你是一个运行在用户终端里的编程助手。先观察真实世界再行动。";
const userB = { role: "user", content: `看看 package.json 里有哪些 scripts，然后告诉我怎么跑测试。\n\n[本地时间：${clock()}]` };

const b1 = serializeRequest(goodSystem, [userB]);
const b2 = serializeRequest(goodSystem, [userB, ...round1]);
const b3 = serializeRequest(goodSystem, [userB, ...round1, ...round2]);
report("第 1↔2 轮", b1, b2);
report("第 2↔3 轮", b2, b3);
console.log("  时间没有消失——它随那条用户消息进了历史，但从此一个字节都不再变。");

// ─── 实验 C：违反 append-only（改写历史）────────────────────────────────

console.log("\n━━━ 实验 C：第 3 轮\"好心\"截短旧的工具输出 ━━━");
const trimmedRound1 = [
  round1[0],
  { role: "tool", content: TOOL_OUTPUT_1.slice(0, 80) + "…(为省上下文截短)" },
];
const c3 = serializeRequest(goodSystem, [userB, ...trimmedRound1, ...round2]);
const brk = report("第 2↔3 轮", b2, c3);
console.log(`  断点从请求末尾跳回第 ${brk} 字符（那条工具输出的中间）——`);
console.log(`  之后的 ${c3.length - brk} 字符全部退回全价。省了几十个字符的上下文，赔掉的是整段后续历史的缓存。`);
console.log("  真想省：要么当初就别让大输出进历史（s04 输出预算），要么整段压缩换摘要（s06）。");

// ─── 实验 D：浪费计数器——上一轮的 prompt，这一轮本该全是缓存读 ────────────
// A/B/C 拿着两轮请求的原文对比，能精确指出断点在哪个字节——但那是实验室条件：
// 生产里你不会留着每轮几十万字符的请求原文。生产里手上只有 provider 每轮
// 报回的三个数：prompt 总量、缓存读了多少、缓存写了多少。
// 好在"该命中多少"是可以推的：上一轮整个 prompt 刚被算过一遍，这一轮它的
// 每个 token 都该是缓存读。差出来的部分就是被重计费的浪费：
//   miss = min(上一轮 prompt, 这一轮 prompt) - 这一轮缓存读
// 再看闲置了多久：超过 TTL（Anthropic 默认 5 分钟）是缓存被淘汰（正常损耗）；
// 没超 TTL 就是前缀被改写了（实验 C 的病，去查 append-only）。

console.log("\n━━━ 实验 D：浪费计数器——只靠 usage 数字，量化每次击穿并归因 ━━━");

const TTL_MS = 5 * 60 * 1000; // 缓存保留时间
const NOISE = 1024; // 断点/块粒度造成的小额差异，不算浪费

function trackCacheUsage(prev, usage, now) {
  const reported = usage.cacheRead + usage.cacheWrite > 0;
  const next = { promptTokens: usage.prompt, at: now, reportedCache: (prev?.reportedCache ?? false) || reported };
  // 第一轮没有参照；从不报缓存字段的服务商，0 也说明不了什么
  if (!prev || (!reported && !prev.reportedCache)) return { next };
  const missed = Math.min(prev.promptTokens, usage.prompt) - usage.cacheRead;
  if (missed <= NOISE) return { next };
  const idleMs = now - prev.at;
  return { next, miss: { missed, idleMs, ttlExpired: idleMs > TTL_MS } };
}

// 一段会话的 usage 流水（provider 每轮报回的原始数字）：
const rounds = [
  { label: "第 1 轮（会话开始）　　　　", gap: 0, prompt: 8200, cacheRead: 0, cacheWrite: 8200 },
  { label: "第 2 轮（15 秒后）　　　　　", gap: 15e3, prompt: 11400, cacheRead: 8200, cacheWrite: 3200 },
  { label: "第 3 轮（20 秒后）　　　　　", gap: 20e3, prompt: 14100, cacheRead: 11400, cacheWrite: 2700 },
  { label: "第 4 轮（用户开会，38 分钟后）", gap: 38 * 60e3, prompt: 16300, cacheRead: 0, cacheWrite: 16300 },
  { label: "第 5 轮（10 秒后）　　　　　", gap: 10e3, prompt: 18000, cacheRead: 16300, cacheWrite: 1700 },
  { label: "第 6 轮（12 秒后，有人截短了老工具输出）", gap: 12e3, prompt: 17600, cacheRead: 5100, cacheWrite: 12500 },
];

let state, now = 0, waste = 0, misses = 0;
for (const r of rounds) {
  now += r.gap;
  const { next, miss } = trackCacheUsage(state, r, now);
  state = next;
  if (!miss) { console.log(`  ${r.label}：✓ 正常`); continue; }
  waste += miss.missed; misses++;
  const idle = miss.idleMs >= 60e3 ? `${Math.round(miss.idleMs / 60e3)} 分钟` : `${Math.round(miss.idleMs / 1e3)} 秒`;
  const verdict = miss.ttlExpired
    ? "闲置超过 TTL → 缓存被服务商淘汰（正常损耗，人回来了它就回来）"
    : "TTL 没过 → 前缀被改写了！去查 append-only（实验 C 的病）";
  console.log(`  ${r.label}：✗ 浪费 ${miss.missed} tok · 闲置 ${idle} · ${verdict}`);
}
console.log(`  会话累计：浪费 ${waste} tok / ${misses} 次 —— 放到 UI 上就是一行 "Cache waste"。`);
console.log("  注意第 4 轮和第 6 轮的数字长得几乎一样（cacheRead 掉下去了），");
console.log("  没有 idle 归因就分不清'正常损耗'和'代码有病'——量化的意义就在这一步。");

它会直接算出缓存断点位置（真实运行输出）：

━━━ 实验 A：时间戳写进 system ━━━
  第 1↔2 轮：公共前缀 21 字符（上一轮请求共 209 字符 → 可复用 10.0%）
    断点上下文: …"em]\n当前时间：10:23:01。你是一个运行在用户终" vs …"em]\n当前时间：10:23:02。你是一个运行在用户终"
  第 2↔3 轮：公共前缀 21 字符（上一轮请求共 594 字符 → 可复用 3.5%）
    断点上下文: …"em]\n当前时间：10:23:02。你是一个运行在用户终" vs …"em]\n当前时间：10:23:03。你是一个运行在用户终"
  断点永远卡在时间戳的秒位上 → 后面的 tools、全部历史，每一轮都按全价重算。

━━━ 实验 B：system 稳定，时间戳放进用户消息尾部 ━━━
  第 1↔2 轮：公共前缀 212 字符（上一轮请求共 212 字符 → 可复用 100.0%）
    断点在上一轮请求的末尾之后——新增的只有追加的消息，这就是理想形状。
  第 2↔3 轮：公共前缀 597 字符（上一轮请求共 597 字符 → 可复用 100.0%）
    断点在上一轮请求的末尾之后——新增的只有追加的消息，这就是理想形状。
  时间没有消失——它随那条用户消息进了历史，但从此一个字节都不再变。

━━━ 实验 C：第 3 轮"好心"截短旧的工具输出 ━━━
  第 2↔3 轮：公共前缀 353 字符（上一轮请求共 597 字符 → 可复用 59.1%）
    断点上下文: …"   \"build\": \"tsc -p .\"\n   5\t" vs …"   \"build\": \"tsc…(为省上下文截短)\n\n"
  断点从请求末尾跳回第 353 字符（那条工具输出的中间）——
  之后的 131 字符全部退回全价。省了几十个字符的上下文，赔掉的是整段后续历史的缓存。
  真想省：要么当初就别让大输出进历史（s04 输出预算），要么整段压缩换摘要（s06）。

━━━ 实验 D：浪费计数器——只靠 usage 数字，量化每次击穿并归因 ━━━
  第 1 轮（会话开始）　　　　：✓ 正常
  第 2 轮（15 秒后）　　　　　：✓ 正常
  第 3 轮（20 秒后）　　　　　：✓ 正常
  第 4 轮（用户开会，38 分钟后）：✗ 浪费 14100 tok · 闲置 38 分钟 · 闲置超过 TTL → 缓存被服务商淘汰（正常损耗，人回来了它就回来）
  第 5 轮（10 秒后）　　　　　：✓ 正常
  第 6 轮（12 秒后，有人截短了老工具输出）：✗ 浪费 12500 tok · 闲置 12 秒 · TTL 没过 → 前缀被改写了！去查 append-only（实验 C 的病）
  会话累计：浪费 26600 tok / 2 次 —— 放到 UI 上就是一行 "Cache waste"。
  注意第 4 轮和第 6 轮的数字长得几乎一样（cacheRead 掉下去了），
  没有 idle 归因就分不清'正常损耗'和'代码有病'——量化的意义就在这一步。

接上真实模型：

1	`AGENT_API_KEY=sk-xxx node notes/s07_prompt_cache/agent.mjs`

#!/usr/bin/env node
// s07 —— 缓存命中工程：s03 的 agent（工具注册表 + 看门狗）+ 缓存命中率仪表盘。
//
// 新增的全部内容：
//   · chat() 读回 usage，兼容两种缓存字段（DeepSeek / OpenAI）
//   · 每轮打印 prompt tokens、缓存命中/未命中、按"命中≈1折"估算省了多少钱
//   · 三条纪律落在代码里：system 字节稳定、tools 顺序稳定、messages 只追加
//
// 运行方式与 s03 相同：AGENT_API_KEY=sk-xxx node agent.mjs

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;
const MODEL = process.env.AGENT_MODEL ?? "deepseek-cache";

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。任何 OpenAI 兼容的 key 都行（DeepSeek / Kimi / GLM / OpenRouter / 本地 Ollama）。");
  process.exit(1);
}

// 纪律①：system prompt 必须逐轮字节稳定。cwd 和 platform 在进程生命周期内
// 不变，可以进来；时间戳、随机数、"今天是几号"是缓存杀手——
//
//   ❌ const SYSTEM = `当前时间：${new Date().toISOString()}\n你是…`;
//
// 这样写，每轮请求的第 6 个字符就开始不同，前缀缓存从那里断掉，
// 后面的一切（tools、全部历史）永远按全价计费。每轮会变的信息
// 放进"最后一条用户消息"随尾部走（见 demo.mjs 的对照实验）。
const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（与 s03 相同）─────────────────────────────────────────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        return `命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`;
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) return `编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`;
      if (text.indexOf(old_string, first + 1) !== -1)
        return `编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`;
      // 不能用 text.replace(old, new)：new_string 里的 $$ / $& 会被 JS 当替换模式展开，静默写坏文件。
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

// 纪律②：tools 数组也在前缀里。Object.entries 的顺序 = 注册表的书写顺序，
// 进程内每轮相同——别在运行时对它排序、增删或拼接动态描述。
const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return `未知工具：${call.function.name}`;
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return `工具参数不是合法 JSON：${err.message}`;
  }
  try {
    return tool.handler(args);
  } catch (err) {
    return `工具执行出错：${err.message}`;
  }
}

// ─── 缓存仪表盘：让省钱可见 ──────────────────────────────────────────────

// 兼容读两种 usage 字段：
//   DeepSeek：usage.prompt_cache_hit_tokens / usage.prompt_cache_miss_tokens
//   OpenAI：  usage.prompt_tokens_details.cached_tokens（未命中 = prompt - cached）
// 都没有就返回 undefined——有的服务商/本地模型根本不报缓存。
function readCacheUsage(usage = {}) {
  const prompt = usage.prompt_tokens ?? 0;
  let hit;
  if (typeof usage.prompt_cache_hit_tokens === "number") hit = usage.prompt_cache_hit_tokens;
  else if (typeof usage.prompt_tokens_details?.cached_tokens === "number") hit = usage.prompt_tokens_details.cached_tokens;
  if (hit === undefined) return { prompt };
  const miss = typeof usage.prompt_cache_miss_tokens === "number" ? usage.prompt_cache_miss_tokens : prompt - hit;
  return { prompt, hit, miss };
}

const sessionTotals = { prompt: 0, hit: 0 };

function printUsage(usage) {
  const u = readCacheUsage(usage);
  if (u.hit === undefined) {
    console.log(`\x1b[36m📊 prompt ${u.prompt} tokens（该服务商未返回缓存字段）\x1b[0m`);
    return;
  }
  sessionTotals.prompt += u.prompt;
  sessionTotals.hit += u.hit;
  const rate = u.prompt > 0 ? ((u.hit / u.prompt) * 100).toFixed(1) : "0.0";
  // 命中部分按约 1 折计费 → 每命中 1 token 省 0.9 个全价 token。
  const saved = u.prompt > 0 ? ((u.hit * 0.9) / u.prompt) * 100 : 0;
  const total = sessionTotals.prompt > 0 ? ((sessionTotals.hit / sessionTotals.prompt) * 100).toFixed(1) : "0.0";
  console.log(
    `\x1b[36m📊 prompt ${u.prompt} | 命中 ${u.hit}（${rate}%）| 未命中 ${u.miss} | 本轮输入费≈省 ${saved.toFixed(0)}% | 会话累计命中 ${total}%\x1b[0m`,
  );
}

// ─── 模型调用（与 s03 唯一的区别：读回 usage）────────────────────────────

async function chat(messages) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...messages],
      tools: TOOLS,
    }),
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);
  const data = await res.json();
  return { message: data.choices[0].message, usage: data.usage };
}

// ─── 主循环：s03 原样 + 每轮打印命中率 ───────────────────────────────────
// 纪律③就藏在这里：messages 只 push、从不改写。s03 的循环天生 append-only，
// 本章一行都不用改——难的不是做到，是别在后续迭代里破坏它
// （比如"帮模型省上下文"回头去截短旧的工具输出：省了 token，赔了缓存）。

async function runTurn(messages) {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;

  while (true) {
    if (!budget.canContinue()) {
      const stop = budget.exhaustedStop();
      console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }

    const { message: msg, usage } = await chat(messages);
    printUsage(usage);
    messages.push(msg);

    if (msg.content) console.log(`\n${msg.content}`);
    if (!msg.tool_calls?.length) return;

    const records = [];
    for (const call of msg.tool_calls) {
      const output = dispatch(call);
      messages.push({ role: "tool", tool_call_id: call.id, content: output });
      let input = {};
      try {
        input = JSON.parse(call.function.arguments || "{}");
      } catch { /* 坏参数已作为失败回填 */ }
      records.push({
        name: call.function.name,
        input,
        status: FAILURE_RE.test(output) ? "failed" : "completed",
        output,
      });
    }

    const stop = budget.recordTurn(records);
    if (!stop) continue;

    if (isRecoverable(stop) && !repaired) {
      repaired = true;
      console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
      messages.push({ role: "user", content: repairPrompt(stop) });
      continue;
    }
    console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
    return;
  }
}

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
const messages = [];

console.log(`s07 agent 已上线（${MODEL}）。每轮打印缓存命中率——盯着第二轮开始的 📊 行看。Ctrl+C 退出。`);

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;
  messages.push({ role: "user", content: line });
  await runTurn(messages);
}

给它一个多轮任务（比如”看看这个目录的结构，再读一下最大的那个文件”），观察每轮统计行：第一轮命中 0%，之后应迅速升到 90%+。再做对照实验：把 SYSTEM 第一行改成 `当前时间：${new Date().toISOString()}`，重跑同样的任务——命中率归零，直接看到这笔差价。

实现

① 三条纪律：保持前缀逐字节稳定

请求的开头依次是 system prompt、tools（工具定义）、历史消息，三条纪律分别对应：

system prompt 逐轮字节稳定。时间戳、随机数、”剩余预算 7 轮”这类每轮会变的内容会破坏缓存——变化在最开头，之后的 tools + 全部历史都按全价重算。会变的信息挪到最后一条用户消息：尾部本来就是新字节。
tools 数组顺序稳定。工具定义和 system 一起序列化在请求最前部。不要运行时排序、按条件增删、往 description 里拼动态内容。
messages 只追加，不改写（append-only）。任何一个字节被改动，缓存就断在那里，之后全部退回全价。常见错误是好心截短旧的工具输出——省几十 token，赔掉整段后续缓存（demo 实验 C）。

这三条纪律难的不是做到，而是在后续迭代中不被悄悄破坏——缓存击穿是静默的，没有报错，只体现在账单上。真实产品用回归测试守住它，见文末。

② 测量：读取 usage 里的命中数据

不打印命中率，就无法验证纪律是否生效。服务商在 usage（响应附带的用量字段）里报了账，但字段名不统一：

// DeepSeek：usage.prompt_cache_hit_tokens / usage.prompt_cache_miss_tokens
// OpenAI：  usage.prompt_tokens_details.cached_tokens（未命中 = prompt - cached）
let hit;
if (typeof usage.prompt_cache_hit_tokens === "number") hit = usage.prompt_cache_hit_tokens;
else if (typeof usage.prompt_tokens_details?.cached_tokens === "number") hit = usage.prompt_tokens_details.cached_tokens;

本章 agent 每轮打印一行统计（格式示意；第一轮命中 0% 正常——要有上一次请求才有可复用前缀）：

1	`📊 prompt 48213 \| 命中 47616（98.8%）\| 未命中 597 \| 本轮输入费≈省 89% \| 会话累计命中 91.2%`

健康的 agent 从第二轮起命中率应在 90% 以上，且随任务变长越来越高。异常时对照下表排查：

现场	诊断
每一轮都是 0%	前缀在最开头就断了——system 或 tools 里混进了每轮变化的字节（时间戳最常见）
一直 95%+，某轮突然跌到 50%	历史中段被改写了——检查是否有代码在回头修剪旧消息（违反 append-only）
压缩后的第一轮跌到接近 0%	预期行为：s06 把历史整段重写，缓存必然失效一次（system + tools 那一小段还能命中），用一次全价换之后每轮更短的前缀
命中率正常但账单没降	确认服务商确实支持前缀缓存并读对了字段——读不到字段时本章 agent 会明确提示，而不是打印一个假的 0

往尾部追加的机制天然无害（如 s03 看门狗的纠偏消息），危险的只有回头改写。

③ 记账：把每次击穿折成 token 数，并归因

②的统计行和排查表有一个前提：有人在看。但缓存击穿是静默、偶发的——TTL 过期只在用户走开又回来的那一轮出现，一个改写历史的 bug 可能只在特定分支触发。没人会每轮盯着命中率，等发现时只剩一张说不清原因的账单。

所以在每轮统计之上再加一层会话级记账。它不需要请求原文（demo 实验 A–C 那种字节对比是实验室条件，生产里没人保留每轮几十万字符的 payload），只吃 provider 每轮报回的 usage 数字，靠一个推断补出”应然”：上一轮的整个 prompt 刚被算过一遍，这一轮它的每个 token 都应该是缓存读。差值就是被重计费的浪费：

1	`miss = min(上一轮 promptTokens, 这一轮 promptTokens) - 这一轮 cacheRead`

配三条防误报纪律，缺一个数字就不可信：

噪声地板：断点/块粒度天然造成小额差异（文末”粒度与序列化细节”），1k token 以下不计；
首轮与不报账的服务商不计：第一轮没有参照；从不报缓存字段的服务商，cacheRead=0 说明不了任何事——但只要它报过一次，之后的 0 就是真未命中（用一个”报过账”的粘性标记区分两者）；
压缩边界重置：s06 把历史整段重写，那次击穿是买来的（用一次全价换之后更短的前缀），记进浪费会把预期成本和意外事故混在一起——压缩时把追踪器清零。

最后一步是归因。浪费数字本身分不清”正常损耗”还是”代码有病”，能分清的是闲置时长：距上一轮超过 TTL（Anthropic 默认 5 分钟），是缓存被服务商淘汰，人回来了它自然回来；没超 TTL 就是前缀被改写了，直接去查 append-only（②表格第二行的现场，现在有了数字和时间戳）。demo 实验 D 里第 4 轮和第 6 轮的 usage 数字长得几乎一样，判词完全相反——归因就差在 idleMs(闲置时长) 这一个维度上。

累计值（总浪费 token 数 + 击穿次数）放进会话状态持久化，在 UI 上常驻一行。它和②的关系是仪表盘和示波器的关系：②告诉你为什么断（对着排查表看现场），③告诉你断了多少钱（不看的时候也在记）。

TTL的具体含义：

TTL（Time To Live，生存时间）指的是服务商（如 Anthropic、DeepSeek）在缓存系统中为自动缓存（Auto-caching）设置的过期时间。

具体含义

当你连续发送多个请求时，服务商会把 system prompt + tools 定义 + 历史对话前缀缓存起来。但这个缓存不是永久有效的——它有有效期。

Anthropic 默认 TTL = 5 分钟，意思是：

如果你在 5 分钟内发起下一轮请求 → 缓存命中 ✅
如果你在 5 分钟后才发起下一轮请求 → 缓存已过期被清空，需要重新计算 ❌

DeepSeek 也有类似的机制，只是具体 TTL 值可能不同（通常也是几分钟量级）。

练习

加 `/cost` 命令

给本章 agent 加一个 /cost 命令：按你所用服务商的真实价目（命中价、未命中价、输出价）把会话累计花费折算成钱，并对比”如果全部未命中”的假想账单。成本可见，纪律才可验证。

这个可以根据从服务商获取到的usage计算，扩展sessionTotals，加两个函数计算真实花费和”如果全部未命中”的假想账单。但是核心问题是这个/cost命令应该加在哪里

看一下ai给出的关于/cost命令的处理。没想到这么简单，只是在主循环里加了if (isCostCommand(line))判断用户输入是否有/cost

全部改动

\1. 新增价目表配置（文件头部）

// ─── 价目表配置（按实际服务商修改）─────────────────────────────────────
// DeepSeek 缓存价（2026）：
//   - 命中：$0.014 / 1M tokens（约 1 折）
//   - 未命中：$0.14 / 1M tokens
//   - 输出：$0.28 / 1M tokens
// 如果用的是 OpenAI，替换成 OpenAI 的价格
const PRICING = {
  hitPerMTok: 0.014,    // 命中价（每百万token）
  missPerMTok: 0.14,    // 未命中价
  outputPerMTok: 0.28,  // 输出价
};

\2. 扩展会话统计（替换现有的sessionTotals）

// 原来的 sessionTotals 改成更完整的统计
const sessionStats = {
  prompt: 0,      // 总输入token
  hit: 0,         // 命中token
  miss: 0,        // 未命中token  
  output: 0,      // 输出token
  turns: 0,       // 轮数
};

// 计算实际花费的函数
function calculateCost(stats) {
  const hitCost = (stats.hit / 1_000_000) * PRICING.hitPerMTok;
  const missCost = (stats.miss / 1_000_000) * PRICING.missPerMTok;
  const outputCost = (stats.output / 1_000_000) * PRICING.outputPerMTok;
  return hitCost + missCost + outputCost;
}

// 计算"全部未命中"的假想花费
function calculateNoCacheCost(stats) {
  const allMissCost = ((stats.hit + stats.miss) / 1_000_000) * PRICING.missPerMTok;
  const outputCost = (stats.output / 1_000_000) * PRICING.outputPerMTok;
  return allMissCost + outputCost;
}

\3. 修改 printUsage 函数（累积统计数据）

function printUsage(usage) {
  const u = readCacheUsage(usage);
  // 累加输出token（不管有没有缓存字段都要加）
  sessionStats.output += usage.completion_tokens ?? 0;
  sessionStats.turns += 1;

  if (u.hit === undefined) {
    // 没缓存字段时，所有prompt都算未命中
    sessionStats.prompt += u.prompt;
    sessionStats.miss += u.prompt;
    console.log(`\x1b[36m📊 prompt ${u.prompt} tokens（该服务商未返回缓存字段）\x1b[0m`);
    return;
  }

  sessionStats.prompt += u.prompt;
  sessionStats.hit += u.hit;
  sessionStats.miss += u.miss;

  const rate = u.prompt > 0 ? ((u.hit / u.prompt) * 100).toFixed(1) : "0.0";
  const totalRate = sessionStats.prompt > 0 
    ? ((sessionStats.hit / sessionStats.prompt) * 100).toFixed(1) 
    : "0.0";

  // 本轮节省百分比（按价格算，不是按token数）
  const hitPrice = (u.hit / 1_000_000) * PRICING.hitPerMTok;
  const missPrice = (u.miss / 1_000_000) * PRICING.missPerMTok;
  const noCachePrice = ((u.hit + u.miss) / 1_000_000) * PRICING.missPerMTok;
  const savedPct = noCachePrice > 0 
    ? (((noCachePrice - hitPrice - missPrice) / noCachePrice) * 100).toFixed(0)
    : "0";

  console.log(
    `\x1b[36m📊 prompt ${u.prompt} | 命中 ${u.hit}（${rate}%）| 未命中 ${u.miss} | 本轮节省 ${savedPct}% | 会话累计命中 ${totalRate}%\x1b[0m`,
  );
}

\4. 新增 /cost 命令处理

// ─── /cost 命令：展示账单 ──────────────────────────────────────────────

function showCost() {
  const actual = calculateCost(sessionStats);
  const noCache = calculateNoCacheCost(sessionStats);
  const saved = noCache - actual;
  const savedPct = noCache > 0 ? ((saved / noCache) * 100).toFixed(1) : "0.0";

  console.log(`\x1b[1;36m
╔═══════════════════════════════════════════════════════════╗
║                      📊 会话账单                         ║
╠═══════════════════════════════════════════════════════════╣
║  总输入 tokens:  ${String(sessionStats.prompt).padStart(8)}                                   ║
║  其中命中:       ${String(sessionStats.hit).padStart(8)} (${((sessionStats.hit/sessionStats.prompt)*100).toFixed(1)}%)            ║
║  其中未命中:     ${String(sessionStats.miss).padStart(8)}                                     ║
║  总输出 tokens:  ${String(sessionStats.output).padStart(8)}                                   ║
║  总轮数:         ${String(sessionStats.turns).padStart(8)}                                   ║
╠═══════════════════════════════════════════════════════════╣
║  实际花费:       $${actual.toFixed(6).padStart(10)}                                   ║
║  若无缓存:       $${noCache.toFixed(6).padStart(10)}                                   ║
║  ───────────────────────────────────────────────────────  ║
║  💰 节省:        $${saved.toFixed(6).padStart(10)} (${savedPct}%)                    ║
╚═══════════════════════════════════════════════════════════╝
\x1b[0m`);
}

// 在 /cost 命令中触发
function isCostCommand(line) {
  return line.trim().toLowerCase() === '/cost';
}

\5. 修改主循环

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;

  // 处理 /cost 命令
  if (isCostCommand(line)) {
    showCost();
    continue;
  }

  messages.push({ role: "user", content: line });
  await runTurn(messages);
}

\6. 更新 runTurn 中的 usage 累积

完整文件

#!/usr/bin/env node
// s07 —— 缓存命中工程：s03 的 agent（工具注册表 + 看门狗）+ 缓存命中率仪表盘 + 账单统计。
//
// 新增的全部内容：
//   · chat() 读回 usage，兼容两种缓存字段（DeepSeek / OpenAI）
//   · 每轮打印 prompt tokens、缓存命中/未命中、按"命中≈1折"估算省了多少钱
//   · 三条纪律落在代码里：system 字节稳定、tools 顺序稳定、messages 只追加
//   · /cost 命令：显示会话累计账单，对比"全部未命中"的假想花费
//
// 运行方式与 s03 相同：AGENT_API_KEY=sk-xxx node agent.mjs

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;
const MODEL = process.env.AGENT_MODEL ?? "deepseek-cache";

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。任何 OpenAI 兼容的 key 都行（DeepSeek / Kimi / GLM / OpenRouter / 本地 Ollama）。");
  process.exit(1);
}

// ─── 价目表配置（按实际服务商修改）─────────────────────────────────────
// DeepSeek 缓存价（2026）：
//   - 命中：$0.014 / 1M tokens（约 1 折）
//   - 未命中：$0.14 / 1M tokens
//   - 输出：$0.28 / 1M tokens
// 如果用的是 OpenAI，替换成 OpenAI 的价格
const PRICING = {
  hitPerMTok: 0.014,    // 命中价（每百万token）
  missPerMTok: 0.14,    // 未命中价
  outputPerMTok: 0.28,  // 输出价
};

// 纪律①：system prompt 必须逐轮字节稳定。cwd 和 platform 在进程生命周期内
// 不变，可以进来；时间戳、随机数、"今天是几号"是缓存杀手——
//
//   ❌ const SYSTEM = `当前时间：${new Date().toISOString()}\n你是…`;
//
// 这样写，每轮请求的第 6 个字符就开始不同，前缀缓存从那里断掉，
// 后面的一切（tools、全部历史）永远按全价计费。每轮会变的信息
// 放进"最后一条用户消息"随尾部走（见 demo.mjs 的对照实验）。
const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（与 s03 相同）─────────────────────────────────────────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        return `命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`;
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) return `编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`;
      if (text.indexOf(old_string, first + 1) !== -1)
        return `编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`;
      // 不能用 text.replace(old, new)：new_string 里的 $$ / $& 会被 JS 当替换模式展开，静默写坏文件。
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

// 纪律②：tools 数组也在前缀里。Object.entries 的顺序 = 注册表的书写顺序，
// 进程内每轮相同——别在运行时对它排序、增删或拼接动态描述。
const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return `未知工具：${call.function.name}`;
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return `工具参数不是合法 JSON：${err.message}`;
  }
  try {
    return tool.handler(args);
  } catch (err) {
    return `工具执行出错：${err.message}`;
  }
}

// ─── 缓存仪表盘：让省钱可见 ──────────────────────────────────────────────

// 兼容读两种 usage 字段：
//   DeepSeek：usage.prompt_cache_hit_tokens / usage.prompt_cache_miss_tokens
//   OpenAI：  usage.prompt_tokens_details.cached_tokens（未命中 = prompt - cached）
// 都没有就返回 undefined——有的服务商/本地模型根本不报缓存。
function readCacheUsage(usage = {}) {
  const prompt = usage.prompt_tokens ?? 0;
  let hit;
  if (typeof usage.prompt_cache_hit_tokens === "number") hit = usage.prompt_cache_hit_tokens;
  else if (typeof usage.prompt_tokens_details?.cached_tokens === "number") hit = usage.prompt_tokens_details.cached_tokens;
  if (hit === undefined) return { prompt };
  const miss = typeof usage.prompt_cache_miss_tokens === "number" ? usage.prompt_cache_miss_tokens : prompt - hit;
  return { prompt, hit, miss };
}

// 会话统计（累计）
const sessionStats = {
  prompt: 0,      // 总输入token
  hit: 0,         // 命中token
  miss: 0,        // 未命中token  
  output: 0,      // 输出token
  turns: 0,       // 轮数
};

// 计算实际花费
function calculateCost(stats) {
  const hitCost = (stats.hit / 1_000_000) * PRICING.hitPerMTok;
  const missCost = (stats.miss / 1_000_000) * PRICING.missPerMTok;
  const outputCost = (stats.output / 1_000_000) * PRICING.outputPerMTok;
  return hitCost + missCost + outputCost;
}

// 计算"全部未命中"的假想花费
function calculateNoCacheCost(stats) {
  const allMissCost = ((stats.hit + stats.miss) / 1_000_000) * PRICING.missPerMTok;
  const outputCost = (stats.output / 1_000_000) * PRICING.outputPerMTok;
  return allMissCost + outputCost;
}

// 判断是否是 /cost 命令
function isCostCommand(line) {
  return line.trim().toLowerCase() === '/cost';
}

// 显示账单
function showCost() {
  const actual = calculateCost(sessionStats);
  const noCache = calculateNoCacheCost(sessionStats);
  const saved = noCache - actual;
  const savedPct = noCache > 0 ? ((saved / noCache) * 100).toFixed(1) : "0.0";
  const hitRate = sessionStats.prompt > 0 
    ? ((sessionStats.hit / sessionStats.prompt) * 100).toFixed(1) 
    : "0.0";
  
  console.log(`\x1b[1;36m
╔═══════════════════════════════════════════════════════════╗
║                      📊 会话账单                         ║
╠═══════════════════════════════════════════════════════════╣
║  总输入 tokens:  ${String(sessionStats.prompt).padStart(8)}                                   ║
║  其中命中:       ${String(sessionStats.hit).padStart(8)} (${hitRate}%)                         ║
║  其中未命中:     ${String(sessionStats.miss).padStart(8)}                                     ║
║  总输出 tokens:  ${String(sessionStats.output).padStart(8)}                                   ║
║  总轮数:         ${String(sessionStats.turns).padStart(8)}                                   ║
╠═══════════════════════════════════════════════════════════╣
║  实际花费:       $${actual.toFixed(6).padStart(10)}                                   ║
║  若无缓存:       $${noCache.toFixed(6).padStart(10)}                                   ║
║  ───────────────────────────────────────────────────────  ║
║  💰 节省:        $${saved.toFixed(6).padStart(10)} (${savedPct}%)                    ║
╚═══════════════════════════════════════════════════════════╝
\x1b[0m`);
}

function printUsage(usage) {
  const u = readCacheUsage(usage);
  // 累加输出token（不管有没有缓存字段都要加）
  sessionStats.output += usage.completion_tokens ?? 0;
  sessionStats.turns += 1;
  
  if (u.hit === undefined) {
    // 没缓存字段时，所有prompt都算未命中
    sessionStats.prompt += u.prompt;
    sessionStats.miss += u.prompt;
    console.log(`\x1b[36m📊 prompt ${u.prompt} tokens（该服务商未返回缓存字段）\x1b[0m`);
    return;
  }
  
  sessionStats.prompt += u.prompt;
  sessionStats.hit += u.hit;
  sessionStats.miss += u.miss;
  
  const rate = u.prompt > 0 ? ((u.hit / u.prompt) * 100).toFixed(1) : "0.0";
  const totalRate = sessionStats.prompt > 0 
    ? ((sessionStats.hit / sessionStats.prompt) * 100).toFixed(1) 
    : "0.0";
  
  // 本轮节省百分比（按价格算，不是按token数）
  const hitPrice = (u.hit / 1_000_000) * PRICING.hitPerMTok;
  const missPrice = (u.miss / 1_000_000) * PRICING.missPerMTok;
  const noCachePrice = ((u.hit + u.miss) / 1_000_000) * PRICING.missPerMTok;
  const savedPct = noCachePrice > 0 
    ? (((noCachePrice - hitPrice - missPrice) / noCachePrice) * 100).toFixed(0)
    : "0";
  
  console.log(
    `\x1b[36m📊 prompt ${u.prompt} | 命中 ${u.hit}（${rate}%）| 未命中 ${u.miss} | 本轮节省 ${savedPct}% | 会话累计命中 ${totalRate}%\x1b[0m`,
  );
}

// ─── 模型调用（与 s03 唯一的区别：读回 usage）────────────────────────────

async function chat(messages) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...messages],
      tools: TOOLS,
    }),
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);
  const data = await res.json();
  return { message: data.choices[0].message, usage: data.usage };
}

// ─── 主循环：s03 原样 + 每轮打印命中率 ───────────────────────────────────
// 纪律③就藏在这里：messages 只 push、从不改写。s03 的循环天生 append-only，
// 本章一行都不用改——难的不是做到，是别在后续迭代里破坏它
// （比如"帮模型省上下文"回头去截短旧的工具输出：省了 token，赔了缓存）。

async function runTurn(messages) {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;

  while (true) {
    if (!budget.canContinue()) {
      const stop = budget.exhaustedStop();
      console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }

    const { message: msg, usage } = await chat(messages);
    printUsage(usage);
    messages.push(msg);

    if (msg.content) console.log(`\n${msg.content}`);
    if (!msg.tool_calls?.length) return;

    const records = [];
    for (const call of msg.tool_calls) {
      const output = dispatch(call);
      messages.push({ role: "tool", tool_call_id: call.id, content: output });
      let input = {};
      try {
        input = JSON.parse(call.function.arguments || "{}");
      } catch { /* 坏参数已作为失败回填 */ }
      records.push({
        name: call.function.name,
        input,
        status: FAILURE_RE.test(output) ? "failed" : "completed",
        output,
      });
    }

    const stop = budget.recordTurn(records);
    if (!stop) continue;

    if (isRecoverable(stop) && !repaired) {
      repaired = true;
      console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
      messages.push({ role: "user", content: repairPrompt(stop) });
      continue;
    }
    console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
    return;
  }
}

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
const messages = [];

console.log(`s07 agent 已上线（${MODEL}）。每轮打印缓存命中率——盯着第二轮开始的 📊 行看。Ctrl+C 退出。`);
console.log(`输入 /cost 查看会话账单。`);

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;
  
  // 处理 /cost 命令
  if (isCostCommand(line)) {
    showCost();
    continue;
  }
  
  messages.push({ role: "user", content: line });
  await runTurn(messages);
}

思考题1

s06 的压缩把历史整段重写，缓存必然失效一次。压缩的阈值（75%）和缓存之间存在一个权衡：压得越早，缓存失效越频繁；压得越晚，每轮承担的未命中风险越大。如果服务商的缓存保留时间很短（Anthropic 默认 TTL 只有 5 分钟；DeepSeek 是小时级的闲置淘汰），这个权衡又会怎么变？（s09 讲子代理时会再回到”前缀即资产”这个视角。）

TTL 短（Anthropic 5 分钟）→ 晚压缩更优

为什么？

核心洞察：TTL 越短，你”未命中”的概率本身就在快速上升（用户稍微走开一下就过期了）。这时候再主动制造击穿（压缩）是雪上加霜。

因素	分析
闲置时间概率	用户对话间隔 > 5 分钟很常见（回消息、看代码、接电话）。每次间隔超 TTL，下一轮本来就是全价，压缩带来的”额外击穿”边际成本很低
压缩的”白费”风险	如果你压缩后用户立刻走了 10 分钟再回来，压缩那次全价白花了——因为回来时缓存已过期，压缩后的新前缀也没用上
性价比	既然 TTL 短导致缓存经常自然失效，压缩带来的”人为击穿”就是多此一举。不如拉高阈值，少压缩几次，让自然过期承担清理任务

最优策略（TTL 短）

阈值拉高（如 80-85%）+ 压缩时机选择在”刚发生过自然过期”之后

为什么选在自然过期后？

自然过期时缓存已清空，这轮本来就是全价
如果在这一轮末尾做压缩，压缩的代价被”掩盖”在全价轮里（反正这轮已经是全价了，顺便把历史整一下）
下一轮用户回来时，新压缩的前缀可以享受完整 TTL

// 伪代码：检测到自然过期后触发压缩
if (idleMs > TTL_MS) {
  // 这轮是全价，顺便检查是否需要压缩
  if (shouldCompress(messages)) {
    await compressHistory(messages); // 在本轮末尾做
  }
}

TTL 长（DeepSeek 小时级）→ 早压缩更优

为什么？

核心洞察：TTL 长意味着只要代码不乱改，缓存几乎一直有效。唯一会击穿缓存的就是压缩本身。

因素	分析
自然过期概率低	用户间隔 > 1 小时不常见。大部分轮次都能命中缓存
未命中成本逐轮累积	如果不压缩，每轮 prompt 越来越长，万一某次击穿（比如压缩），损失巨大
压缩收益高	压缩后每轮成本大幅下降，而因为 TTL 长，这个”新压缩前缀”可以用很久

最优策略（TTL 长）

阈值拉低（如 60-70%）+ 主动、高频压缩

为什么？

每次压缩虽然会击穿一次，但击穿后新的短前缀能享受很长 TTL
相当于”投资一次全价，换取未来很多轮的半价”
阈值越低，压缩越频繁，但每轮的平均上下文越短，单轮未命中的损失越小

对比总结

	TTL 短（5 分钟）	TTL 长（小时级）
自然过期频率	高	低
最优压缩阈值	高（80-85%）	低（60-70%）
压缩时机	被动（跟着自然过期走）	主动（定期/按阈值触发）
核心理念	少压，让自然过期清理	多压，用投资换长期收益
风险	长期不压缩 → 单轮未命中损失大	频繁压缩 → 击穿次数多

第三种策略：看 `idleMs` 动态决策

如果想让系统更智能，可以把压缩决策跟 idleMs 绑定：

function shouldCompress(messages, idleMs, TTL_MS) {
  const ratio = currentPromptLength(messages) / compressedLength(messages);

  if (TTL_MS < 10 * 60 * 1000) { // 短 TTL
    // 只在自然过期后压缩，阈值拉高
    if (idleMs > TTL_MS) return ratio > 0.85;
    return false; // 活跃时不压
  } else { // 长 TTL
    // 主动压缩，阈值拉低
    return ratio > 0.65;
  }
}

这样：

TTL 短 → 压缩是”顺便的”，不主动制造击穿
TTL 长 → 压缩是”主动投资”，尽早享受收益

一句话回答

TTL 越短，越应该晚压缩、少压缩，把清理工作交给自然过期；TTL 越长，越应该早压缩、多压缩，用主动击穿换取长期成本下降。 核心是把”人为击穿”尽量安排在”已经自然击穿”的轮次里，避免额外损失。

思考题2

③的归因只有两档（TTL / 前缀改写）。给”前缀改写”再往下分：在开发模式下保留最近两轮请求的序列化文本，
击穿且 TTL 未过时自动跑一次实验 A–C 那样的字节 diff，把断点落进 system / tools / 历史第几条消息，
直接报”凶手是谁”。想清楚为什么这只能是开发模式的功能——生产里留全量 payload 的代价是什么？

先回顾一下”③ 记账”在做什么

③ 的功能是：发现缓存没命中（击穿了），然后判断是”正常”还是”异常”。

判断逻辑只有两档：

判断结果	条件	含义
正常损耗	闲置时间 > TTL（超过5分钟）	缓存被服务商自动清掉了，不是谁的错
前缀改写	闲置时间 ≤ TTL（5分钟内）	缓存本应在，但没了，说明有人改了什么东西

问题来了：”前缀改写”太笼统了

“前缀改写”只是告诉你”有人改了东西”，但没告诉你具体改了哪里。

可能是：

可能性	举例
system prompt 变了	你给 system 里加了个时间戳
tools 定义变了	你给某个工具加了新参数
某条历史消息被改了	你为了”省 token”把旧消息截断了
工具输出变了	这次执行结果跟上次不一样

这些都是”前缀改写”，但根因完全不同，修法也完全不同。

题目给的方案：开发模式下做”字节 diff”

意思是：如果击穿了，且 TTL 没过期（说明不该击穿），自动对比”上一轮请求”和”本轮请求”的原始文本，逐字节找出哪里不同。

// 伪代码：两轮请求的 payload 对比
const prev = JSON.stringify(上一轮请求); // system + tools + 全部历史
const curr = JSON.stringify(本轮请求);

const diff = byteByByteDiff(prev, curr);
// 输出：断点在 system 第 15 个字符
// 或：断点在 messages[3].content 第 200 个字符
// 或：tools[0].function.parameters 变了

这样你就能直接报出**”凶手是谁”**——是 system 变了，还是哪条历史消息被改了。

但为什么只能在”开发模式”下做？

因为要做字节 diff，你必须保留每一轮请求的完整原文（几十万字符的 payload），才能对比。

生产环境如果这样做，代价是什么？

代价	具体说明
存储爆炸	每轮存一份完整 payload，会话越长 payload 越大。1000个用户 × 每天100轮 × 每轮50KB = 每天5TB
隐私风险	payload 里有用户对话原文、代码、API key、密码——你把它们全存下来了，合规审计过不去
调试噪音	99% 的 diff 是”用户说了新的话”或”助手新增了回复”，是正常推进，不是 bug。你从一万行 diff 里找真正的问题，等于没用
性能开销	每轮做全量字节对比，大会话下会拖慢响应速度

会话持久化与恢复

本章把会话写到磁盘，崩溃后原样恢复，做法是追加式事件日志 + 重放。

本章代码 = s03 的最小 agent 循环 + store.mjs（事件日志与重放）。

问题

agent 干了半小时的活：几十轮对话、二十次工具调用。你按错了 Ctrl+C，或终端误关、笔记本没电——重新打开，它什么都不记得。前几章的 agent 都这样：messages 只存在内存里。

常见的第一反应是”存成 JSON 文件，每次变化整个重写”。这个方案有隐患——demo 场景一真实写磁盘、真实模拟崩溃，展示了这个现场：

━━━ 场景一：全量重写 JSON，崩溃在写到一半时 ━━━
  磁盘上留下 124/207 字节的 session.json，尝试恢复：
  ✗ JSON.parse 失败：Unterminated string in JSON at position 124 (line 7 column 27)
  ✗ 整个会话报废 —— 不止最后一条消息，之前的历史也一起没了。

解决方案

落盘的不是状态快照，而是事件：每发生一件事就往 JSONL 文件末尾追加一行，已写下的字节永远不被触碰；恢复时逐行重放事件，把 messages 重新推导出来。崩溃的影响从”整个文件可能写坏”缩小到”最后一行可能写了一半”——跳过那行即可：丢一条消息，不丢整个会话。

运行

免 key 演示：

1	`node notes/s08_persistence/demo.mjs`

#!/usr/bin/env node
// 不需要 API key 的落盘演示：在临时目录里真写 JSONL、真模拟崩溃、真重放。
//
//   node s08_persistence/demo.mjs
//
// 三个场景：
//   一、"每次全量重写 JSON"在崩溃时留下半个文件 —— 整个会话报废
//   二、追加式事件日志：写几个事件 → 重放 → 重建出 messages 和会话配置
//   三、最后一行写了一半（撕裂写入）—— 跳过那一行，会话照常恢复

import { appendFileSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { appendEvent, createSession, replaySession, sessionPath } from "./store.mjs";

const dir = mkdtempSync(path.join(tmpdir(), "s08-demo-"));

// ─── 场景一：全量重写 + 崩溃 = 半个文件 ─────────────────────────────────

console.log("━━━ 场景一：全量重写 JSON，崩溃在写到一半时 ━━━");

const snapshot = {
  id: "ses_snapshot",
  model: "deepseek-chat",
  messages: [
    { role: "user", content: "帮我修一下登录页的报错" },
    { role: "assistant", content: "我先看看代码。" },
  ],
};
const full = JSON.stringify(snapshot, null, 2);
// 模拟崩溃：写盘写到 60% 时断电 —— 磁盘上只有前 60% 的字节。
const torn = full.slice(0, Math.floor(full.length * 0.6));
const snapPath = path.join(dir, "session.json");
writeFileSync(snapPath, torn);
console.log(`  磁盘上留下 ${torn.length}/${full.length} 字节的 session.json，尝试恢复：`);
try {
  JSON.parse(readFileSync(snapPath, "utf8"));
  console.log("  （不可能走到这里）");
} catch (err) {
  console.log(`  ✗ JSON.parse 失败：${err.message.split("\n")[0]}`);
  console.log("  ✗ 整个会话报废 —— 不止最后一条消息，之前的历史也一起没了。");
}

// ─── 场景二：追加式事件日志 → 重放重建 ──────────────────────────────────

console.log("\n━━━ 场景二：追加式事件日志（JSONL）→ 重放重建 ━━━");

const meta = createSession(dir, { model: "deepseek-chat" });
appendEvent(dir, meta.id, { type: "message", message: { role: "user", content: "帮我修一下登录页的报错" } });
appendEvent(dir, meta.id, {
  type: "message",
  message: {
    role: "assistant",
    content: null,
    tool_calls: [{ id: "call_1", type: "function", function: { name: "read_file", arguments: '{"path":"login.js"}' } }],
  },
});
appendEvent(dir, meta.id, {
  type: "tool_call",
  record: { id: "call_1", name: "read_file", input: { path: "login.js" }, status: "completed" },
});
appendEvent(dir, meta.id, { type: "message", message: { role: "tool", tool_call_id: "call_1", content: "   1\tconst token = nul;" } });
appendEvent(dir, meta.id, { type: "message", message: { role: "assistant", content: "找到了：nul 是笔误，应该是 null。" } });

const restored = replaySession(dir, meta.id);
console.log(`  重放 ${sessionPath(dir, meta.id)}：`);
console.log(`  模型 = ${restored.meta.model}（来自 session_meta，不是恢复时的默认值）`);
console.log(`  重建出 ${restored.messages.length} 条消息：`);
for (const m of restored.messages) {
  const body = m.content ?? `<调用 ${m.tool_calls?.map((c) => c.function.name).join("、")}>`;
  console.log(`    [${m.role}] ${body}`);
}
console.log(`  工具调用记录：${restored.toolCalls.map((r) => `${r.name}(${r.status})`).join("、")} ← 结构化 status，不用再解析报错文案`);

// ─── 场景三：最后一行撕裂 —— 丢一行，不丢会话 ───────────────────────────

console.log("\n━━━ 场景三：崩溃把最后一行写了一半 ━━━");

// 模拟崩溃：一条正常事件只落了前半截字节（没有换行、JSON 没闭合）。
const half = JSON.stringify({ ts: new Date().toISOString(), type: "message", message: { role: "user", content: "再帮我加个记住密码" } });
appendFileSync(sessionPath(dir, meta.id), half.slice(0, Math.floor(half.length / 2)));
console.log(`  往同一个 .jsonl 追加了半行（${Math.floor(half.length / 2)}/${half.length} 字节），再次重放：`);

const again = replaySession(dir, meta.id);
console.log(`  ✓ 恢复出 ${again.messages.length} 条消息，跳过 ${again.skipped} 行损坏数据`);
console.log(`  ✓ 只丢了崩溃瞬间那一条，之前的全部历史完好。`);

console.log(`
结论：
  · 全量重写 JSON：崩在写盘中间 = 新旧两份都没了，恢复无从谈起。
  · 追加式 JSONL：已写下的字节永远不被触碰，崩溃的爆炸半径被压缩到"最后一行"。
  · 恢复 = 重放：状态是从事件流里推出来的，会话配置（模型）也在流里，一起回来。
（演示文件在 ${dir}，可以 cat 看看每一行长什么样。）
`);

三个场景，全部真实写磁盘、真实模拟崩溃（场景一的输出见上文”问题”）。

接上真实模型，启动时分岔：带 --resume <id> 就重放恢复，否则开新会话并打印 id：

AGENT_API_KEY=sk-xxx node agent.mjs
# 新会话 ses_mr47xdu8_sgf8（落盘于 …/.sessions/ses_mr47xdu8_sgf8.jsonl）
# 下次续上：node agent.mjs --resume ses_mr47xdu8_sgf8

AGENT_API_KEY=sk-xxx node agent.mjs --resume ses_mr47xdu8_sgf8
# 已恢复会话 ses_mr47xdu8_sgf8：14 条消息，5 次工具调用
#!/usr/bin/env node
// s08 —— 会话落盘与恢复：s03 的 agent + 追加式事件日志（store.mjs）。
//
// 新增的全部内容：
//   · 每条消息进 messages 数组的同时 append 一行事件到 .sessions/<id>.jsonl
//   · 工具调用带结构化 status 落盘（回收 s03 靠报错文案前缀判失败的临时方案）
//   · 启动带 --resume <id> → 重放事件重建 messages，续上次的会话接着聊
//   · 会话用的模型记录在 session_meta 里，恢复时以它为准，不取当前默认
//
// 运行：AGENT_API_KEY=sk-xxx node agent.mjs             # 新会话，打印会话 id
//       AGENT_API_KEY=sk-xxx node agent.mjs --resume <id>  # 断了接上

import readline from "node:readline/promises";
import { execSync } from "node:child_process";
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { LoopBudget, isRecoverable, repairPrompt } from "./loop-budget.mjs";
import { appendEvent, createSession, listSessionIds, replaySession, sessionPath } from "./store.mjs";

const BASE_URL = process.env.AGENT_BASE_URL ?? "https://api.deepseek.com/v1";
const API_KEY = process.env.AGENT_API_KEY;

if (!API_KEY) {
  console.error("缺少 AGENT_API_KEY。任何 OpenAI 兼容的 key 都行（DeepSeek / Kimi / GLM / OpenRouter / 本地 Ollama）。");
  process.exit(1);
}

// ─── 会话：新建或恢复 ─────────────────────────────────────────────────────

const SESSIONS_DIR = path.join(process.cwd(), ".sessions");
const resumeAt = process.argv.indexOf("--resume");
const resumeId = resumeAt !== -1 ? process.argv[resumeAt + 1] : undefined;

let meta;
const messages = [];

if (resumeId) {
  const restored = replaySession(SESSIONS_DIR, resumeId);
  if (!restored) {
    console.error(`找不到会话 ${resumeId}。可用的会话：${listSessionIds(SESSIONS_DIR).join("、") || "（无）"}`);
    process.exit(1);
  }
  meta = restored.meta;
  messages.push(...restored.messages);
  console.log(
    `已恢复会话 ${meta.id}：${restored.messages.length} 条消息，${restored.toolCalls.length} 次工具调用` +
      (restored.skipped ? `（跳过 ${restored.skipped} 行损坏数据）` : ""),
  );
} else {
  // 模型在"创建时"读一次环境变量，然后冻结进 session_meta ——
  // 恢复这个会话的永远是它，而不是恢复那一刻的默认值。
  meta = createSession(SESSIONS_DIR, { model: process.env.AGENT_MODEL ?? "deepseek-chat" });
  console.log(`新会话 ${meta.id}（落盘于 ${sessionPath(SESSIONS_DIR, meta.id)}）`);
  console.log(`下次续上：node agent.mjs --resume ${meta.id}`);
}

const MODEL = meta.model; // 会话粒度的配置，来自会话记录

/** 消息只从这里进数组：内存 + 磁盘一步完成，两边永远一致。 */
function pushMessage(message) {
  messages.push(message);
  appendEvent(SESSIONS_DIR, meta.id, { type: "message", message });
}

const SYSTEM = `你是一个运行在用户终端里的编程助手。
优先用专用工具（read_file / write_file / edit_file）操作文件；run_shell 用于其余一切。
先观察真实世界再行动，不要凭空猜测文件内容。
当前目录：${process.cwd()}
操作系统：${process.platform}`;

// ─── 工具注册表（s02/s03 的四件套，失败改为 throw —— 见 dispatch）────────

const REGISTRY = {
  run_shell: {
    description: "在用户的终端里执行一条 shell 命令，返回 stdout 和 stderr。",
    parameters: {
      type: "object",
      properties: { command: { type: "string", description: "要执行的命令" } },
      required: ["command"],
    },
    handler: ({ command }) => {
      console.log(`\x1b[33m  $ ${command}\x1b[0m`);
      try {
        const out = execSync(command, {
          encoding: "utf8",
          timeout: 30_000,
          maxBuffer: 1024 * 1024,
          stdio: ["ignore", "pipe", "pipe"],
        });
        return out.trim() || "(命令执行成功，无输出)";
      } catch (err) {
        // 失败 = 抛异常。报错文案原样保留（错误即信息，模型看得到），
        // 但"这次调用失败了"这个事实由 dispatch 转成结构化 status。
        throw new Error(`命令失败（exit ${err.status ?? "?"}）：\n${err.stdout ?? ""}${err.stderr ?? err.message}`);
      }
    },
  },

  read_file: {
    description: "读取一个文本文件，返回带行号的内容。",
    parameters: {
      type: "object",
      properties: { path: { type: "string", description: "文件路径（相对或绝对）" } },
      required: ["path"],
    },
    handler: ({ path: p }) => {
      console.log(`\x1b[33m  read ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8"); // 读不到 → 自然抛出，dispatch 记 failed
      const CAP = 50_000;
      const body = text.length > CAP ? text.slice(0, CAP) + `\n…(截断，共 ${text.length} 字符)` : text;
      return body
        .split("\n")
        .map((line, i) => `${String(i + 1).padStart(4)}\t${line}`)
        .join("\n");
    },
  },

  write_file: {
    description: "写入一个文件（整体覆盖）。父目录不存在时自动创建。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        content: { type: "string", description: "完整的文件内容" },
      },
      required: ["path", "content"],
    },
    handler: ({ path: p, content }) => {
      console.log(`\x1b[33m  write ${p} (${content.length} 字符)\x1b[0m`);
      mkdirSync(path.dirname(path.resolve(p)), { recursive: true });
      writeFileSync(p, content);
      return `已写入 ${p}`;
    },
  },

  edit_file: {
    description:
      "对文件做一次精确替换。old_string 必须在文件中出现且仅出现一次（带上足够的上下文来保证唯一），否则会失败。",
    parameters: {
      type: "object",
      properties: {
        path: { type: "string", description: "文件路径" },
        old_string: { type: "string", description: "要被替换的原文（必须唯一匹配）" },
        new_string: { type: "string", description: "替换成的新文本" },
      },
      required: ["path", "old_string", "new_string"],
    },
    handler: ({ path: p, old_string, new_string }) => {
      console.log(`\x1b[33m  edit ${p}\x1b[0m`);
      const text = readFileSync(p, "utf8");
      const first = text.indexOf(old_string);
      if (first === -1) throw new Error(`编辑失败：old_string 在 ${p} 中找不到。请先 read_file 确认原文。`);
      if (text.indexOf(old_string, first + 1) !== -1)
        throw new Error(`编辑失败：old_string 在 ${p} 中出现多次。请带上更多上下文让它唯一。`);
      // 不能用 text.replace(old, new)：new_string 里的 $$ / $& 会被 JS 当替换模式展开，静默写坏文件。
      writeFileSync(p, text.slice(0, first) + new_string + text.slice(first + old_string.length));
      return `已编辑 ${p}`;
    },
  },
};

const TOOLS = Object.entries(REGISTRY).map(([name, t]) => ({
  type: "function",
  function: { name, description: t.description, parameters: t.parameters },
}));

// s03 靠 FAILURE_RE 前缀猜成败，本章拆掉这个脚手架：
// 成败在执行的那一刻就确定（return = completed，throw = failed），
// dispatch 把它变成结构化字段，落盘之后审计和重放都不用再解析报错文案。
function dispatch(call) {
  const tool = REGISTRY[call.function.name];
  if (!tool) return { status: "failed", output: `未知工具：${call.function.name}` };
  let args;
  try {
    args = JSON.parse(call.function.arguments || "{}");
  } catch (err) {
    return { status: "failed", output: `工具参数不是合法 JSON：${err.message}` };
  }
  try {
    return { status: "completed", output: tool.handler(args) };
  } catch (err) {
    return { status: "failed", output: err.message }; // 错误依旧作为文本回给模型
  }
}

// ─── 主循环：s03 原样 + 每一步落盘 ───────────────────────────────────────

async function chat(msgs) {
  const res = await fetch(`${BASE_URL}/chat/completions`, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: MODEL,
      messages: [{ role: "system", content: SYSTEM }, ...msgs],
      tools: TOOLS,
    }),
  });
  if (!res.ok) throw new Error(`API ${res.status}：${await res.text()}`);
  const data = await res.json();
  return data.choices[0].message;
}

async function runTurn() {
  const budget = new LoopBudget({ baseSteps: 12 });
  let repaired = false;

  while (true) {
    if (!budget.canContinue()) {
      const stop = budget.exhaustedStop();
      console.log(`\n\x1b[31m⛔ ${stop.message}（第 ${stop.turnCount} 轮）\x1b[0m`);
      return;
    }

    const msg = await chat(messages);
    pushMessage(msg); // 助手消息（含 tool_calls）落盘

    if (msg.content) console.log(`\n${msg.content}`);
    if (!msg.tool_calls?.length) return;

    const records = [];
    for (const call of msg.tool_calls) {
      const { status, output } = dispatch(call);
      pushMessage({ role: "tool", tool_call_id: call.id, content: output }); // 工具结果落盘
      let input = {};
      try {
        input = JSON.parse(call.function.arguments || "{}");
      } catch { /* 坏参数已作为失败回填 */ }
      const record = { name: call.function.name, input, status, output };
      records.push(record);
      // 结构化审计记录：真实产品的 ToolCallRecord 还带 permission / preview /
      // outputPath 等字段，这里只落最小集。output 已在 tool 消息里，不重复存。
      appendEvent(SESSIONS_DIR, meta.id, {
        type: "tool_call",
        record: { id: call.id, name: record.name, input, status },
      });
    }

    const stop = budget.recordTurn(records);
    if (!stop) continue;

    if (isRecoverable(stop) && !repaired) {
      repaired = true;
      console.log(`\n\x1b[35m🟡 看门狗触发（${stop.reason}），注入纠偏 prompt…\x1b[0m`);
      pushMessage({ role: "user", content: repairPrompt(stop) }); // 纠偏也是历史的一部分
      continue;
    }
    console.log(`\n\x1b[31m⛔ ${stop.message}（reason=${stop.reason}，第 ${stop.turnCount} 轮）\x1b[0m`);
    return;
  }
}

const rl = readline.createInterface({ input: process.stdin, output: process.stdout });

console.log(`s08 agent 已上线（${MODEL}，来自会话记录）。工具：${Object.keys(REGISTRY).join("、")}。Ctrl+C 随便按 —— 会话在盘上。`);

while (true) {
  const line = (await rl.question("\n你> ")).trim();
  if (!line) continue;
  pushMessage({ role: "user", content: line });
  await runTurn();
}

验收：跟它聊两轮、让它读个文件，Ctrl+C 退出，再 --resume 回来问”刚才聊到哪了”——它应该答得上来。s03 看门狗注入的纠偏消息也走 pushMessage（见下文”实现”）：恢复出的会话必须和退出前一致。

实现

// 会话落盘 —— 追加式事件日志（JSONL）+ 重放恢复。
//
// 从真实产品 Reina 的 packages/core/src/rollout.ts 简化移植，机制一致：
//   · 一个会话一个 <id>.jsonl，每发生一件事 append 一行 { ts, type, ... }
//   · 只追加、绝不回写已有字节 —— 崩溃最坏丢最后一行，不会损坏整个会话
//   · 恢复 = 逐行重放（replay）重建状态；坏行（半截写入）跳过，其余照常
//   · 会话粒度的配置（用的什么模型）记在第一行 session_meta 里，随重放恢复
//
// 事件类型（本章只用三种，真实产品有二十多种）：
//   session_meta —— 会话头：id、model、createdAt（永远是第一行）
//   message      —— 一条对话消息（user / assistant / tool，原样保存）
//   tool_call    —— 一次工具调用的结构化记录 { id, name, input, status }

import { appendFileSync, mkdirSync, readFileSync, readdirSync } from "node:fs";
import path from "node:path";

export function sessionPath(dir, id) {
  return path.join(dir, `${id}.jsonl`);
}

/** 新建会话：生成 id，把 session_meta 作为第一行写下去。
 *  model 在这里被"冻结"进会话 —— 以后恢复时以这一行为准，
 *  而不是取恢复那一刻的环境默认值。 */
export function createSession(dir, { model }) {
  const id = `ses_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 6)}`;
  const meta = { id, model, createdAt: new Date().toISOString() };
  appendEvent(dir, id, { type: "session_meta", meta });
  return meta;
}

/** 追加一个事件。appendFileSync 用 "a" 标志打开 —— 操作系统层面的
 *  O_APPEND：每次写都落在文件末尾，永远不触碰已有内容。
 *  （Reina 为了性能和写入顺序保证，会常驻一个文件句柄并用 Promise
 *  链串行化写入；示例版每次重开，崩溃安全性完全一样。） */
export function appendEvent(dir, id, event) {
  mkdirSync(dir, { recursive: true });
  const line = JSON.stringify({ ts: new Date().toISOString(), ...event });
  appendFileSync(sessionPath(dir, id), line + "\n");
}

/** 恢复 = 重放：逐行读事件，重建 { meta, messages, toolCalls }。
 *  解析失败的行（多半是崩溃时写了一半的最后一行）跳过并计数 ——
 *  丢一行，不丢整个会话。文件不存在或没有 session_meta 返回 null。 */
export function replaySession(dir, id) {
  let text;
  try {
    text = readFileSync(sessionPath(dir, id), "utf8");
  } catch {
    return null;
  }
  let meta = null;
  const messages = [];
  const toolCalls = [];
  let skipped = 0;
  for (const raw of text.split("\n")) {
    const line = raw.trim();
    if (!line) continue;
    let event;
    try {
      event = JSON.parse(line);
    } catch {
      skipped++; // 半截行：跳过，别让一行坏数据毁掉全部历史
      continue;
    }
    switch (event.type) {
      case "session_meta":
        meta = event.meta;
        break;
      case "message":
        messages.push(event.message);
        break;
      case "tool_call": {
        // 按 id upsert：同一次调用的后续更新（如 running → completed）覆盖前一条。
        const i = toolCalls.findIndex((r) => r.id === event.record.id);
        if (i === -1) toolCalls.push(event.record);
        else toolCalls[i] = { ...toolCalls[i], ...event.record };
        break;
      }
      default:
        // 未知事件类型（来自新版本写的日志）：忽略，老代码也能加载新会话。
        break;
    }
  }
  if (!meta) return null;
  return { meta, messages, toolCalls, skipped };
}

/** 列出目录下已有的会话 id（给 --resume 提示用）。 */
export function listSessionIds(dir) {
  let entries;
  try {
    entries = readdirSync(dir);
  } catch {
    return [];
  }
  return entries.filter((f) => f.endsWith(".jsonl")).map((f) => f.slice(0, -".jsonl".length));
}

① 为什么”每次全量重写 JSON”不可靠

全量重写是会每次打开文件清空所有旧内容再开始重新写入

写文件不是原子操作。writeFileSync(f, bigJson) 在操作系统层面是”打开（清空旧内容）→ 分块写入字节”。崩溃落在中间，磁盘上就是半个 JSON：旧版本已清空，新版本没写完——JSON.parse 失败，整个会话（包括崩溃前完好的部分）一起丢失。演示场景一就是这个现场。

而且越用越危险：会话越长，重写窗口越大，中招概率越高——最有价值的长会话恰恰最容易被写坏。

追加式日志（JSONL：一行一个 JSON）从结构上消除了这个问题：

1
2
3

{"ts":"…","type":"session_meta","meta":{"id":"ses_x","model":"deepseek-chat",…}}
{"ts":"…","type":"message","message":{"role":"user","content":"帮我修一下登录页"}}
{"ts":"…","type":"tool_call","record":{"id":"call_1","name":"read_file","status":"completed",…}}

每发生一件事（用户消息/助手消息/工具结果/压缩边界）就在末尾追加一行，已写下的字节永远不被触碰。崩溃的影响被限制在”最后一行可能写了一半”——恢复时跳过那行即可：丢一条消息，不丢整个会话。这不是额外的容错，是”只追加”结构自带的性质。

② 恢复 = 重放：状态是事件流的推导结果

落盘的是事件，不是状态。恢复时逐行读事件，把 messages 数组重新推导出来：

for (const line of text.split("\n")) {
  if (!line.trim()) continue; // 文件末尾的换行会产生空串，先跳过
  let event;
  try { event = JSON.parse(line); } catch { skipped++; continue; } // 半截行：跳过
  switch (event.type) {
    case "session_meta": meta = event.meta; break;
    case "message":      messages.push(event.message); break;
    case "tool_call":    /* 按 id upsert 进 toolCalls */ break;
    default: break; // 未知类型忽略 —— 老代码也能加载新版本写的日志
  }
}

这个结构带来两个不显眼但重要的自由度：坏行可以跳过（容错）；未知事件类型可以忽略（向前兼容——新版本程序写的日志，旧程序照样能加载它认识的部分）。

写的时候逐次添加事件，读的时候逐行读逐行重放

③ 会话粒度的配置也在流里

恢复会话时，模型配置从哪来？很多实现顺手用当前的环境变量或全局默认。这是错的：用什么模型是这个会话自己的属性，创建时就冻结进第一行 session_meta，恢复时以它为准：

1
2
3

meta = createSession(SESSIONS_DIR, { model: process.env.AGENT_MODEL ?? "deepseek-chat" });
// ……第二天恢复时：
const MODEL = restored.meta.model; // 来自会话记录，不是今天的默认值

配置跟着会话走，界面显示和实际请求才不会不一致（真实产品踩过事故，见文末）。

session_meta 是会话的”出生证明”——创建时把模型、system prompt 等配置写死，恢复时以它为准，而不是用今天的环境变量。否则昨天用 deepseek 跑的会话，今天可能被 gpt-4 续写，账目和能力都对不上。

④ 工具调用带结构化 status 落盘

s03 有个临时方案：靠报错文案的开头文字（FAILURE_RE）判断工具调用是否成功。文案是写给模型看的，随时会改——拿它当机器判据太脆。

旧代码如下

// s03 的做法：看报错文案开头
const FAILURE_RE = /^(命令失败|编辑失败|工具执行出错|未知工具|工具参数)/;

// 执行工具
const output = dispatch(call);

// 判断是否失败：用正则匹配 output 的开头文字
const status = FAILURE_RE.test(output) ? "failed" : "completed";

本章移除这个脚手架：成败在执行那一刻确定，作为结构化字段落盘。

约定很简单：handler return = completed，throw = failed；报错文案原样回给模型（错误即信息），但”失败了”这个事实走字段：

try {
  return { status: "completed", output: tool.handler(args) };
} catch (err) {
  return { status: "failed", output: err.message };
}

落盘的 tool_call 事件带着这个 status。从此审计、重放、监督逻辑都读字段，不再解析文案。

映射到 agent 代码

旧做法（s03）：

// 执行工具
const output = dispatch(call);  // output 是一段文字

// 判断是否失败：看文字开头是不是 "命令失败" 那几个字
const status = /^(命令失败|编辑失败)/.test(output) ? "failed" : "completed";

新做法：

// 执行工具时，直接返回一个对象，明确告诉成功还是失败
function executeTool(tool, args) {
  try {
    const result = tool.handler(args);
    return { status: "completed", output: result };
  } catch (err) {
    return { status: "failed", output: err.message };
  }
}

落盘时的对比

旧做法（s03）落盘：

{
  "type": "tool_call",
  "tool_call": {
    "id": "call_123",
    "name": "read_file",
    "output": "编辑失败：old_string 在文件中找不到"  // ← 状态藏在文字里
  }
}

恢复时想统计”失败了多少次”？你要用正则去匹配 output 字符串——一旦文案改了，统计就坏了。

新做法落盘：

{
  "type": "tool_call",
  "tool_call": {
    "id": "call_123",
    "name": "read_file",
    "status": "failed",  // ← 单独字段，机器直接读
    "output": "old_string 在文件中找不到"  // ← 纯信息，给模型看
  }
}

恢复时想统计”失败了多少次”？

1 2	`const failures = toolCalls.filter(tc => tc.status === "failed"); // 直接、准确、不受文案影响`

接进你的 agent

agent.mjs 是 s03 的 agent + 落盘。关键改动两处。第一，消息只从一个入口进数组——内存和磁盘一步完成：

function pushMessage(message) {
  messages.push(message);
  appendEvent(SESSIONS_DIR, meta.id, { type: "message", message });
}

第二，启动时分岔：带 --resume <id> 就重放恢复，否则开新会话并打印 id（运行命令见上文”运行”）。

练习

追加`compacted` 事件和对应的重放逻辑

给 store.mjs 加一个 compacted 事件和对应的重放逻辑：记录”从第 N 条消息之前已被压缩为摘要 S”，重放时用摘要替换被压缩的区间。s06 的压缩机制落盘之后，才算完整闭环。

这里主要需要思考的点是如何放进去这个compacted事件，现在JSONL 的承诺是”已写下的字节永远不被触碰”。如果删除或替换已有行，就违背了这个承诺，崩溃时可能损坏整个文件。只能在末尾追加事件

那么我们这里实现的思路是追加一条覆盖指令，用摘要覆盖之前压缩区间记录下来的事件

原有事件：
第1行: {"type":"session_meta",...}
第2行: {"type":"message","message":{"role":"user","content":"你好"}}
第3行: {"type":"message","message":{"role":"assistant","content":"你好！"}}
第4行: {"type":"message","message":{"role":"user","content":"帮我读文件"}}
第5行: {"type":"message","message":{"role":"assistant","content":"文件内容是..."}}
...（中间100条消息）
第105行: {"type":"message","message":{"role":"user","content":"继续"}}

压缩后追加：
第106行: {"type":"compacted","record":{"fromIndex":1,"toIndex":105,"summary":"用户问了关于文件读取和配置修改的问题，助手已帮助完成..."}}

重放时：
  读到第1-5行 → 正常加入 messages
  读到第6-105行 → 正常加入 messages
  读到第106行 (compacted) → 把 messages[1] 到 messages[105] 替换成一条摘要消息
  结果：messages 变成 [第1行 (meta), 第106行 (摘要), 第105行之后的新消息]

实现方案

1. 新增 `appendCompacted` 函数

// store.mjs

/**
 * 追加一个压缩事件。
 * @param {string} dir - 会话目录
 * @param {string} id - 会话ID
 * @param {object} params
 * @param {number} params.fromIndex - 压缩起始索引（messages 数组中的位置）
 * @param {number} params.toIndex - 压缩结束索引（含）
 * @param {string} params.summary - 压缩摘要
 * @param {number} params.originalLength - 被压缩的原始 token 数（可选，用于统计）
 */
export function appendCompacted(dir, id, { fromIndex, toIndex, summary, originalLength }) {
  const record = {
    fromIndex,
    toIndex,
    summary,
    originalLength,
    compressedAt: new Date().toISOString(),
  };
  appendEvent(dir, id, { type: "compacted", record });
}

2. 修改 `replaySession` 函数

export function replaySession(dir, id) {
  let text;
  try {
    text = readFileSync(sessionPath(dir, id), "utf8");
  } catch {
    return null;
  }
  let meta = null;
  const messages = [];
  const toolCalls = [];
  const compactions = []; // 存储所有压缩事件
  let skipped = 0;

  for (const raw of text.split("\n")) {
    const line = raw.trim();
    if (!line) continue;
    let event;
    try {
      event = JSON.parse(line);
    } catch {
      skipped++;
      continue;
    }
    switch (event.type) {
      case "session_meta":
        meta = event.meta;
        break;
      case "message":
        messages.push(event.message);
        break;
      case "tool_call": {
        const i = toolCalls.findIndex((r) => r.id === event.record.id);
        if (i === -1) toolCalls.push(event.record);
        else toolCalls[i] = { ...toolCalls[i], ...event.record };
        break;
      }
      case "compacted": {
        // 不立即执行压缩，先存起来，等所有事件读完后统一处理
        compactions.push(event.record);
        break;
      }
      default:
        break;
    }
  }

  if (!meta) return null;

  // --- 执行压缩重放 ---
  // 按 fromIndex 从后往前排序，避免索引偏移问题
  const sortedCompactions = compactions.sort((a, b) => b.fromIndex - a.fromIndex);

  for (const comp of sortedCompactions) {
    const { fromIndex, toIndex, summary } = comp;
    // 检查索引是否有效
    if (fromIndex < 0 || toIndex >= messages.length || fromIndex > toIndex) {
      console.warn(`压缩区间 [${fromIndex}, ${toIndex}] 无效，跳过`);
      continue;
    }
    // 用摘要消息替换被压缩的区间
    const summaryMessage = {
      role: "assistant",
      content: `[历史摘要] ${summary}`,
      // 可选：记录这是压缩产生的
      _compacted: true,
      _fromIndex: fromIndex,
      _toIndex: toIndex,
    };
    messages.splice(fromIndex, toIndex - fromIndex + 1, summaryMessage);
  }

  return { meta, messages, toolCalls, skipped, compactions };
}

3. 使用示例

// agent.mjs 中的压缩逻辑

import { appendCompacted, replaySession } from './store.mjs';

// 判断是否需要压缩（比如消息超过 50 条）
function shouldCompact(messages) {
  // 简单策略：消息数 > 50 且未压缩过
  return messages.length > 50;
}

// 执行压缩
function compactHistory(sessionDir, sessionId, messages) {
  // 决定压缩范围：保留最近 10 条，压缩前面的
  const keepRecent = 10;
  const fromIndex = 1; // 跳过 session_meta 对应的消息？实际看你的消息结构
  const toIndex = messages.length - keepRecent - 1;

  if (toIndex <= fromIndex) return; // 没什么可压缩的

  // 生成摘要（调用 LLM 或简单拼接）
  const summary = "用户进行了多轮对话，涉及文件操作、代码修改等...";

  // 追加压缩事件
  appendCompacted(sessionDir, sessionId, {
    fromIndex,
    toIndex,
    summary,
    originalLength: messages.slice(fromIndex, toIndex + 1).reduce((sum, m) => sum + (m.content?.length || 0), 0),
  });

  // 内存中的 messages 也要同步压缩，不然下次 append 会出错
  const summaryMessage = {
    role: "assistant",
    content: `[历史摘要] ${summary}`,
    _compacted: true,
  };
  messages.splice(fromIndex, toIndex - fromIndex + 1, summaryMessage);
}

思考题：

demo 场景三里半截行恰好在文件末尾，跳过它显然安全。但如果坏行出现在文件中间（比如磁盘坏块），跳过一条 message 可能让后面的 tool 消息变成”孤儿”（tool_call_id 对不上助手消息）——API 会拒绝这样的序列。恢复时该怎么检测并修剪这种断链？（提示：s05 处理 Ctrl+C 留下的残缺消息序列用的是同一套办法。）

给缺失信息配平或者修剪掉残缺信息

与真实产品对照（延伸阅读）

本章机制对应 Reina（本系列对照的生产级 agent）的 packages/core/src/rollout.ts（参照 openai/codex 的 rollout recorder 建模）：每个会话一个 .reina/sessions/<id>.jsonl，每次状态变化 append 一行 { ts, type, ... }。示例版三种事件类型，生产版二十多种（message / tool_call / tool_update / compacted / usage / todos……）。几个值得参考的生产细节：

不变量写在注释里：”Writes are O_APPEND only. No code path ever rewrites an existing byte”——跨进程的并发写者也无法互相覆盖历史。进程内则常驻一个文件句柄、用 Promise 链串行化所有 append（示例版每次重开文件，崩溃安全性一样，性能差一些）。
重放跳过坏行：loadRolloutAsSession 对每行单独 JSON.parse，失败（”likely a torn final write from a crash”）就跳过并告警 skipped N malformed line(s)——和本章 demo 场景三相同。
工具调用是结构化记录：packages/protocol/src/index.ts 的 ToolCallRecord 带 status: "pending_approval" | "running" | "completed" | "rejected" | "failed"——比示例版的两态多出审批流和运行中；还有 outputPath 指向 .reina/tool_outputs/ 下的完整输出存档。
决定③的真实事故：Reina 曾在加载旧会话时，模型选择器显示新会话的默认值，而不是会话真正在用的（重放恢复出来的）模型——一个绑定了订阅的会话看起来像在用普通 API key，请求 401，用户以为是配置错误，排查了很久。模型配置随会话重放之外，还有一个细节：config 事件对 model 是整体替换而非浅合并——浅合并会让上一个模型的 baseUrl 泄漏到切换后的模型上，Reina 注释里记着一次真实事故：kimi 切 codex 后残留的 baseUrl 把请求路由到了错误的主机。
仅有的”全量重写”出现在迁移旧格式时（migrateJsonSnapshotToJsonl），而且写法是先写临时文件再 rename 进位——rename 在同一文件系统上是原子的，崩溃也不会留下半个 jsonl。

另一个可观察的例子：Claude Code 的会话也是 JSONL（~/.claude/projects/<项目>/**.jsonl），--resume 的底层就是同一套重放事件流。

AI_Agent

#AI_Agent

Agent02_上下文,缓存与长记忆

http://huang-d1.github.io/2026/09/03/02-上下文，缓存与记忆/

作者

huangdi

发布于

2026年9月3日

许可协议

Agent03_工具溢出防护，渐进式披露与MCP 上一篇

Agent01_运行简单逻辑与最小实现下一篇

Agent02_上下文,缓存与长记忆

流式输出与中断

问题

解决方案

运行

① SSE 解析：按行缓冲，而不是按 chunk 处理

② tool_calls 分片：按 index 装配

③ 中断信号：一个 AbortController 贯穿 HTTP 层与工具循环

④ 悬空 tool_call 的修复：回填合成结果

整体来看流机制和中断处理的具体实现

练习

与真实产品对照（延伸阅读）

上下文压缩

问题

解决方案

运行

实现

① 触发时机：用服务商报告的 usage，不自己数 token

② 压缩的形状：三段式，启动消息逐字保留

③ 摘要 prompt：按栏目填写，不是笼统总结

④ 摘要失败时的降级

⑤ 回退与分支：压缩产物随切点走，不是易失缓存

场景设定

问题：用户要”回退到第50条消息，从这里开个新分支”

最粗暴的做法（坑）

正确做法（判据）

为什么情况B必须丢弃摘要？

设计亮点

1. Codex 架构的”免费午餐”

2. 快照式架构的”坑”

一句话总结

完整流程图

练习

2. 压缩后的摘要消息每轮都会重发一遍。它的内容是稳定的吗？如果你在摘要里加上”压缩于 {当前时间}”，会发生什么？（这与成本有关，下一章展开。）

Prompt 缓存

问题

解决方案

运行

实现

① 三条纪律：保持前缀逐字节稳定

② 测量：读取 usage 里的命中数据

③ 记账：把每次击穿折成 token 数，并归因

练习

加 /cost 命令

思考题1

TTL 短（Anthropic 5 分钟）→ 晚压缩更优

TTL 长（DeepSeek 小时级）→ 早压缩更优

第三种策略：看 idleMs 动态决策

思考题2

会话持久化与恢复

问题

解决方案

运行

实现

① 为什么”每次全量重写 JSON”不可靠

② 恢复 = 重放：状态是事件流的推导结果

③ 会话粒度的配置也在流里

④ 工具调用带结构化 status 落盘

接进你的 agent

练习

追加compacted 事件和对应的重放逻辑

实现方案

1. 新增 appendCompacted 函数

2. 修改 replaySession 函数

3. 使用示例

思考题：

与真实产品对照（延伸阅读）

加 `/cost` 命令

第三种策略：看 `idleMs` 动态决策

追加`compacted` 事件和对应的重放逻辑

1. 新增 `appendCompacted` 函数

2. 修改 `replaySession` 函数