1
2
3
4

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

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293

#!/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
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102

#!/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。
`);

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

// 循环预算 —— agent 的防空转看门狗。
//
// 从真实产品 Reina 的 packages/core/src/loop-budget.ts 简化移植，机制一致：
//   · 软预算 + 硬顶（4 倍）：有进展就自动续期，真正失控才熔断
//   · 三个行为探测器：复读机 / 原地踏步 / 连环报错
//   · 熔断不是死刑：可恢复的停止先给模型一次"自我纠偏"的机会
//
// 演示阈值调小了方便观察（生产值见 README）。

export class LoopBudget {
  constructor({
    baseSteps = 12,
    hardMaxSteps,
    stagnationLimit = 5,
    repeatedActionLimit = 4,
    consecutiveErrorLimit = 3,
  } = {}) {
    this.baseSteps = baseSteps;
    this.hardMaxSteps = hardMaxSteps ?? baseSteps * 4;
    this.stagnationLimit = stagnationLimit;
    this.repeatedActionLimit = repeatedActionLimit;
    this.consecutiveErrorLimit = consecutiveErrorLimit;
    this.seenActions = new Map(); // "工具名:参数指纹" -> 出现次数
    this.noProgressTurns = 0;
    this.consecutiveErrorTurns = 0;
    this.budget = this.baseSteps;
    this.turns = 0;
  }

  canContinue() {
    return this.turns < this.budget;
  }

  /** 每执行完一轮工具调用喂一次。返回 undefined = 继续，返回 stop 对象 = 熔断。
   *  records: [{ name, input, status: "completed"|"failed", output }] */
  recordTurn(records) {
    this.turns++;
    const counts = records.map((r) => this.#recordAction(r));
    const hasProgress = records.some((r, i) => isProgress(r, counts[i]));
    const hasRepeated = counts.some((c) => c >= this.repeatedActionLimit);
    const onlyErrors = records.length > 0 && records.every((r) => r.status === "failed");

    // 两个计数器：有进展就清零，说明 agent 还活着；持续无进展/持续报错才累积。
    this.noProgressTurns = hasProgress ? 0 : this.noProgressTurns + 1;
    this.consecutiveErrorTurns = onlyErrors ? this.consecutiveErrorTurns + 1 : 0;

    if (hasRepeated && !hasProgress) return this.#stop("repeated_action");
    if (this.consecutiveErrorTurns >= this.consecutiveErrorLimit) return this.#stop("consecutive_errors");
    if (this.noProgressTurns >= this.stagnationLimit) return this.#stop("no_progress");

    // 自动续期：有进展、且预算只剩 2 轮，就再给一份 baseSteps（封顶 hardMax）。
    // 勤奋的 agent 不该被一刀切的上限打断，失控的 agent 也不该无限烧钱。
    if (hasProgress && this.turns >= this.budget - 2 && this.budget < this.hardMaxSteps) {
      this.budget = Math.min(this.hardMaxSteps, this.budget + this.baseSteps);
    }
    return undefined;
  }

  exhaustedStop() {
    return this.#stop(this.budget >= this.hardMaxSteps ? "hard_max_steps" : "max_steps");
  }

  #recordAction(record) {
    // 参数做稳定序列化（key 排序），结构相同的调用无论字段顺序都算同一个动作。
    const key = `${record.name}:${stableStringify(record.input)}`;
    const count = (this.seenActions.get(key) ?? 0) + 1;
    this.seenActions.set(key, count);
    return count;
  }

  #stop(reason) {
    return {
      reason,
      message: MESSAGES[reason],
      turnCount: this.turns,
      maxSteps: this.budget,
      hardMaxSteps: this.hardMaxSteps,
    };
  }
}

/** 什么算"进展"？——写操作永远算；读操作只有第一次算。
 *  第二次跑同一条命令、读同一个文件，世界没有变化，不算进展。 */
function isProgress(record, actionCount) {
  if (record.status !== "completed") return false;
  if (["write_file", "edit_file"].includes(record.name)) return true;
  return Boolean(record.output?.trim()) && actionCount === 1;
}

export const MESSAGES = {
  no_progress: "连续多轮没有新进展，暂停。",
  repeated_action: "同一个工具动作被反复执行，暂停。",
  consecutive_errors: "连续多轮全部报错，暂停。",
  max_steps: "达到本轮工具预算，暂停。",
  hard_max_steps: "达到硬性上限，强制停止。",
};

/** 行为异常（复读/停滞/连环报错）可以给模型一次自我纠偏的机会；
 *  预算耗尽（max_steps 系）说明该歇了，直接交还用户。 */
export function isRecoverable(stop) {
  return ["no_progress", "repeated_action", "consecutive_errors"].includes(stop.reason);
}

/** 纠偏 prompt：告诉模型它被暂停的原因和当前状态，并明确指令 ——
 *  别重复原动作；先总结、找到卡点、换一条路；实在不行就问用户。 */
export function repairPrompt(stop) {
  return [
    `自动纠偏触发：${stop.message}`,
    `循环状态：原因=${stop.reason}，已用 ${stop.turnCount} 轮，预算 ${stop.maxSteps}，硬顶 ${stop.hardMaxSteps}。`,
    "不要再重复同样的工具调用或失败的命令。先总结目前发生了什么、找出卡点，换一条不同的路。如果任务被阻塞或有歧义，向用户提一个具体的问题，而不是继续空转。",
  ].join("\n");
}

function stableStringify(value) {
  if (value === null || typeof value !== "object") return JSON.stringify(value);
  if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]`;
  return `{${Object.entries(value)
    .sort(([a], [b]) => a.localeCompare(b))
    .map(([k, v]) => `${JSON.stringify(k)}:${stableStringify(v)}`)
    .join(",")}}`;
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// 流式与中断 —— 本章的机制模块，零依赖。
//
// 三件事：
//   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;
}

1
2
3
4
5

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

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

data: [DONE]

1
2
3
4
5
6
7
8

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;          // 空行、": 心跳" 注释行，都跳过
  // ...
}

1
2
3
4
5
6

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;
}

1
2
3
4

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

1

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

1
2
3
4
5

user: 运行测试并读取 README
assistant: tool_calls [call_test, call_read]
tool: (call_test 结果 —— npm test 实际跑完了)
tool: (call_read 的合成结果 ← 两种文案在这里分叉)
user: 继续

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

let queue = [];        // 插话队列
let inTurn = false;    // 是否有轮次在跑
let idleResolver = null;

rl.on("line", (line) => {
  line = line.trim();
  if (!line) return;
  if (inTurn) {
    queue.push(line);                 // 轮次进行中 → 排队，不打断
    console.log("\x1b[36m（已排队，将在当前工具执行完后处理）\x1b[0m");
  } else if (idleResolver) {
    idleResolver(line);               // 空闲 → 正常走下一轮
  }
});

function askIdle() {
  return new Promise((resolve) => { idleResolver = resolve; });
}

1
2
3
4
5
6
7
8

while (true) {
  const line = queue.length ? queue.shift() : (await askIdle());
  if (!line) continue;
  inTurn = true;
  messages.push({ role: "user", content: line });
  await runTurn(messages);
  inTurn = false;
}

1
2
3
4

assistant: tool_calls [call_A, call_B]     ← 本轮迭代
tool: (call_A 结果)
tool: (call_B 结果)
            ↑ 插话要插在这里，而不是更早

1
2
3

assistant: tool_calls [call_A]        ← 还没执行
user: "顺便用 --verbose 跑"            ← 插在这里 ← ❌
tool: (call_A 结果)

1
2

// 工具批次执行完、全部 tool 结果入列之后 —— 此刻 assistant+tool 配对完整
while (queue.length) messages.push({ role: "user", content: queue.shift() });

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285

#!/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);
}