first commit

explainers/run-agent-loop-lab.mjs

@@ -0,0 +1,444 @@
+import assert from "node:assert/strict";
+import { execFileSync, spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { once } from "node:events";
+import {
+  appendFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync,
+  realpathSync, renameSync, rmSync, writeFileSync
+} from "node:fs";
+import { createServer } from "node:http";
+import { createRequire } from "node:module";
+import { homedir, tmpdir } from "node:os";
+import { basename, join } from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
+import { fileURLToPath, pathToFileURL } from "node:url";
+
+const self = fileURLToPath(import.meta.url);
+const provider = "em-coding-messages";
+const modelName = "DeepSeek-V4-Flash";
+const fixtureFiles = ["orders.mjs", "orders.test.mjs"];
+const request = `给这个订单列表加状态筛选。filterOrders(orders) 仍返回全部，传入 "pending" 或 "shipped" 时只返回匹配项，保持顺序。保留原测试和 before 钩子，用两个独立的 test(...) 分别测试 pending 和 shipped。
+先用 read_file 看 orders.mjs 和 orders.test.mjs，再用 write_file 修改。最后调用 start_tests 启动真实测试，收到任务 ID 后简短汇报已启动并结束当前轮；完成事件稍后投递。`;
+const followUp = "请报告刚收到的测试结果。再用 repo_diff 读取这两个文件此刻的真实工作区差异。如果工具返回 EXECUTOR_OFFLINE，就说明差异需要等设备重连，然后结束本轮。";
+
+const [action = "record", firstArgument, phase] = process.argv.slice(2);
+
+function recorder(directory, lane) {
+  return (kind, summary, data = null, route = [lane, lane]) => {
+    appendFileSync(join(directory, "events.jsonl"), JSON.stringify({
+      at: Date.now(), pid: process.pid, lane, kind, route, summary, data
+    }) + "\n");
+  };
+}
+
+function fixture(directory) {
+  const repo = join(directory, "repo");
+  mkdirSync(repo, { recursive: true });
+  writeFileSync(join(repo, "orders.mjs"), `export function filterOrders(orders) {
+  return orders;
+}
+`);
+  writeFileSync(join(repo, "orders.test.mjs"), `import assert from "node:assert/strict";
+import { existsSync } from "node:fs";
+import { before, test } from "node:test";
+import { setTimeout } from "node:timers/promises";
+import { filterOrders } from "./orders.mjs";
+
+const orders = [
+  { id: 101, status: "pending" },
+  { id: 102, status: "shipped" },
+  { id: 103, status: "pending" },
+];
+
+// The lab releases this gate after observing the Loop process state.
+before(async () => {
+  const gate = process.env.AGENT_LAB_TEST_GATE;
+  if (!gate) return;
+  const deadline = Date.now() + 60_000;
+  while (!existsSync(gate)) {
+    assert(Date.now() < deadline, "Lab did not release the test gate");
+    await setTimeout(50);
+  }
+});
+
+test("defaults to all orders", () => {
+  assert.deepEqual(filterOrders(orders), orders);
+});
+`);
+  writeFileSync(join(repo, "notes.txt"), "Release notes\n");
+  git(repo, ["init", "-q"]);
+  git(repo, ["add", "."]);
+  git(repo, ["-c", "user.name=Agent Loop Lab", "-c", "user.email=lab@example.invalid",
+    "-c", "commit.gpgsign=false", "-c", "core.hooksPath=/dev/null", "commit", "-qm", "Order list fixture"]);
+  appendFileSync(join(repo, "notes.txt"), "Uncommitted local release note\n");
+  return repo;
+}
+
+function git(repo, args) {
+  return execFileSync("git", ["-C", repo, ...args], { encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] });
+}
+
+function snapshot(repo) {
+  return {
+    status: git(repo, ["status", "--short"]),
+    diff: git(repo, ["diff", "--", ...fixtureFiles]),
+    notes: readFileSync(join(repo, "notes.txt"), "utf8")
+  };
+}
+
+function makeExecutor(directory) {
+  const repo = join(directory, "repo");
+  const log = recorder(directory, "executor");
+  let job;
+  return async ({ id, name, input }) => {
+    log("tool.execute", "执行器开始 " + name, { id, name, input });
+    if (name === "read_file" || name === "write_file") {
+      if (!fixtureFiles.includes(input.path)) throw Error("Path outside the fixture allowlist");
+      const path = join(repo, input.path);
+      if (name === "read_file") return readFileSync(path, "utf8");
+      if (typeof input.content !== "string" || input.content.length > 8000) throw Error("Invalid file content");
+      writeFileSync(path, input.content);
+      log("git.snapshot", "写入后 Git 工作区变化", snapshot(repo));
+      return "Wrote " + input.path;
+    }
+    if (name === "repo_diff") return snapshot(repo).diff;
+    if (name !== "start_tests") throw Error("Unknown fixture tool");
+    if (job) return JSON.stringify(job);
+    const child = spawn(process.execPath, ["--test", "--test-reporter=tap", "orders.test.mjs"], {
+      cwd: repo, env: { PATH: process.env.PATH, AGENT_LAB_TEST_GATE: join(directory, "release-tests") },
+      stdio: ["ignore", "pipe", "pipe"]
+    });
+    job = { jobId: "orders-tests-1", pid: child.pid, status: "running", command: "node --test --test-reporter=tap orders.test.mjs" };
+    let stdout = "";
+    let stderr = "";
+    child.stdout.on("data", chunk => { stdout += chunk; });
+    child.stderr.on("data", chunk => { stderr += chunk; });
+    child.on("close", (code, signal) => {
+      const completed = { ...job, status: "finished", code, signal, stdout, stderr };
+      log("test.finish", "本机测试进程退出", completed);
+      writeFileSync(join(directory, "test-result.json"), JSON.stringify(completed));
+    });
+    log("test.start", "本机启动 Node 测试进程", job);
+    return JSON.stringify(job);
+  };
+}
+
+async function serve(directory) {
+  const log = recorder(directory, "executor");
+  const execute = makeExecutor(directory);
+  const server = createServer(async (req, res) => {
+    if (req.method !== "POST" || req.url !== "/tool") {
+      res.writeHead(404).end();
+      return;
+    }
+    try {
+      const chunks = [];
+      for await (const chunk of req) chunks.push(chunk);
+      const call = JSON.parse(Buffer.concat(chunks));
+      log("rpc.receive", "daemon 收到工具请求", call, ["loop", "executor"]);
+      const text = await execute(call);
+      log("rpc.send", "daemon 回传工具结果", { id: call.id, name: call.name, text }, ["executor", "loop"]);
+      res.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ text }));
+    } catch {
+      res.writeHead(400).end(JSON.stringify({ text: "Fixture tool rejected the request" }));
+    }
+  });
+  const portFile = join(directory, "port");
+  server.listen(existsSync(portFile) ? Number(readFileSync(portFile, "utf8")) : 0, "127.0.0.1");
+  await once(server, "listening");
+  log("executor.start", "本机 daemon 监听 loopback", { port: server.address().port });
+  writeFileSync(portFile, String(server.address().port));
+  process.send({ ready: true });
+}
+
+function sanitizedPayload(payload) {
+  const { model, max_tokens, stream, system, messages, tools } = payload;
+  const scrub = value => {
+    if (Array.isArray(value)) return value.filter(item => !["thinking", "redacted_thinking"].includes(item?.type)).map(scrub);
+    if (!value || typeof value !== "object") return value;
+    return Object.fromEntries(Object.entries(value)
+      .filter(([key]) => !["signature", "metadata"].includes(key))
+      .map(([key, child]) => [key, scrub(child)]));
+  };
+  return scrub({ model, max_tokens, stream, system, messages, tools });
+}
+
+async function runLoop(directory, mode) {
+  const piCli = realpathSync(execFileSync("which", ["pi"], { encoding: "utf8" }).trim());
+  const piModule = new URL("../index.js", pathToFileURL(piCli));
+  const { Type } = await import(pathToFileURL(createRequire(piModule).resolve("typebox")).href);
+  const {
+    createAgentSession, DefaultResourceLoader, defineTool, ModelRuntime, SessionManager, SettingsManager
+  } = await import(piModule.href);
+  const log = recorder(directory, "loop");
+  const repo = join(directory, "repo");
+  const sessions = join(directory, "state");
+  const config = join(homedir(), ".pi", "agent");
+  const sdkDir = join(directory, "pi-sdk");
+  mkdirSync(sdkDir, { recursive: true });
+  mkdirSync(sessions, { recursive: true });
+  const settingsManager = SettingsManager.inMemory({
+    compaction: { enabled: false }, retry: { enabled: false }
+  });
+  const modelRuntime = await ModelRuntime.create({
+    authPath: join(sdkDir, "auth.json"),
+    modelsPath: join(config, "models.json"),
+    modelsStorePath: join(sdkDir, "models-store.json"),
+    allowModelNetwork: false
+  });
+  const model = modelRuntime.getModel(provider, modelName);
+  assert(model, "Configured V4 Flash model is unavailable");
+  const resourceLoader = new DefaultResourceLoader({
+    cwd: sdkDir, agentDir: sdkDir, settingsManager,
+    noExtensions: true, noSkills: true, noPromptTemplates: true, noThemes: true, noContextFiles: true,
+    systemPromptOverride: () => "You are a coding agent in an isolated demo repository. Follow the requested tool sequence and report observed results. Do not invent outputs.",
+    extensionFactories: [pi => {
+      pi.on("before_provider_request", event => {
+        log("model.request", "Pi 组装并发送 Anthropic Messages 请求", sanitizedPayload(event.payload), ["loop", "gateway"]);
+      });
+      pi.on("after_provider_response", event => {
+        log("model.http", "收到网关 HTTP 响应头", { status: event.status }, ["gateway", "loop"]);
+      });
+    }]
+  });
+  await resourceLoader.reload();
+  const rpc = mode !== "baseline";
+  const execute = rpc ? null : makeExecutor(directory);
+  const definitions = [
+    ["read_file", "Read a file from the working tree", Type.Object({ path: Type.String() })],
+    ["write_file", "Replace the contents of a file in the working tree", Type.Object({ path: Type.String(), content: Type.String() })],
+    ["start_tests", "Start the repository's Node test command and return its job ID immediately", Type.Object({})],
+    ["repo_diff", "Read current git diff for orders.mjs and orders.test.mjs", Type.Object({})]
+  ];
+  const customTools = definitions.map(([name, description, parameters]) => defineTool({
+    name, label: name, description, parameters,
+    execute: async (id, input) => {
+      const call = { id, name, input };
+      log("tool.dispatch", "Loop 派发 " + name, call, ["loop", "executor"]);
+      let text;
+      let resultOrigin = "executor";
+      if (rpc) {
+        const port = readFileSync(join(directory, "port"), "utf8");
+        try {
+          const response = await fetch("http://127.0.0.1:" + port + "/tool", {
+            method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(call),
+            signal: AbortSignal.timeout(5000)
+          });
+          ({ text } = await response.json());
+        } catch (error) {
+          const code = error.cause?.code || error.name;
+          log("rpc.offline", "Loop 无法连到本机 daemon", { id, name, code }, ["loop", "executor"]);
+          text = "EXECUTOR_OFFLINE: current workspace could not be read (" + code + ")";
+          resultOrigin = "loop";
+        }
+      } else {
+        text = await execute(call);
+      }
+      log("tool.result", resultOrigin === "executor" ? "工具结果交回 Loop" : "Loop 把连接失败转换成工具结果",
+        { id, name, text }, [resultOrigin, "loop"]);
+      return { content: [{ type: "text", text }], details: {} };
+    }
+  }));
+  const receipt = join(directory, "session-path");
+  const restored = existsSync(receipt);
+  // In the split topology Pi's cwd does not point to the repository.
+  const cwd = rpc ? sdkDir : repo;
+  const sessionManager = restored
+    ? SessionManager.open(readFileSync(receipt, "utf8"), sessions, cwd)
+    : SessionManager.create(cwd, sessions);
+  const { session } = await createAgentSession({
+    cwd, agentDir: sdkDir, resourceLoader, settingsManager, modelRuntime, sessionManager,
+    model, thinkingLevel: "off", noTools: "builtin", customTools
+  });
+  log(restored ? "session.restore" : "loop.start",
+    restored ? "新 Loop 进程恢复同一个 Session" : "启动 Pi SDK Loop", {
+      sessionId: session.sessionId, existingMessages: session.messages.length, mode
+    }, restored ? ["session", "loop"] : ["loop", "loop"]);
+  session.subscribe(event => {
+    if (event.type !== "message_end" || event.message.role !== "assistant") return;
+    const { stopReason, usage, content } = event.message;
+    log("model.reply", "模型返回 " + stopReason, {
+      stopReason, usage, content: content.filter(block => ["text", "toolCall"].includes(block.type))
+    }, ["gateway", "loop"]);
+  });
+  const prompt = async (text, origin = "user") => {
+    log("turn.input", "Loop 接收新输入", { text }, [origin, "loop"]);
+    await session.prompt(text);
+    assert.notEqual(session.messages.at(-1)?.stopReason, "error", "Pi reported a provider error");
+    assert.notEqual(session.messages.at(-1)?.stopReason, "aborted", "Pi run was aborted");
+    assert(session.sessionFile && existsSync(session.sessionFile), "Pi session was not persisted");
+    writeFileSync(receipt, session.sessionFile);
+    const entries = sessionManager.getEntries();
+    log("session.save", "Pi Session JSONL 已落盘", {
+      file: basename(session.sessionFile), sessionId: session.sessionId, entries: entries.length,
+      roles: entries.filter(entry => entry.type === "message").map(entry => entry.message.role)
+    }, ["loop", "session"]);
+  };
+  try {
+    if (mode === "baseline" || mode === "initial") {
+      await prompt(request);
+      if (mode === "baseline") {
+        writeFileSync(join(directory, "release-tests"), "");
+        await prompt(await completionMessage(directory) + "\n\n" + followUp, "controller");
+      }
+    } else if (mode === "offline") {
+      await prompt(await completionMessage(directory) + "\n\n" + followUp, "controller");
+    } else if (mode === "reconnect") {
+      await prompt("设备已重连。继续上轮挂起的请求，调用 repo_diff，返回当前实际 diff 摘要。", "controller");
+    }
+  } finally {
+    session.dispose();
+    log("loop.exit", "Loop 主动退出", { mode });
+  }
+}
+
+async function until(predicate, label, timeout = 70_000) {
+  const deadline = Date.now() + timeout;
+  while (!predicate()) {
+    if (Date.now() >= deadline) throw Error("Timed out: " + label);
+    await delay(50);
+  }
+}
+
+async function completionMessage(directory) {
+  const resultFile = join(directory, "test-result.json");
+  await until(() => existsSync(resultFile), "test completion");
+  const result = JSON.parse(readFileSync(resultFile, "utf8"));
+  return "本机完成事件 " + JSON.stringify({ jobId: result.jobId, exitCode: result.code, stdout: result.stdout });
+}
+
+function startChild(command, directory, mode = "") {
+  const child = spawn(process.execPath, [self, command, directory, mode], {
+    env: { ...process.env, PI_OFFLINE: "1" },
+    stdio: ["ignore", "ignore", "inherit", "ipc"]
+  });
+  const watchdog = setTimeout(() => child.kill(), 120_000);
+  const done = once(child, "exit").then(([code, signal]) => {
+    clearTimeout(watchdog);
+    return { code, signal, pid: child.pid };
+  });
+  return { child, done };
+}
+
+async function childLoop(directory, mode) {
+  const result = await startChild("loop", directory, mode).done;
+  assert.equal(result.code, 0, "Pi Loop child failed in phase " + mode);
+  return result;
+}
+
+async function startExecutor(directory) {
+  const execution = startChild("executor", directory);
+  await Promise.race([
+    once(execution.child, "message"),
+    execution.done.then(() => { throw Error("Executor exited before startup"); })
+  ]);
+  return execution;
+}
+
+async function recordRun(root, mode) {
+  const directory = join(root, mode);
+  const repo = fixture(directory);
+  const log = recorder(directory, "controller");
+  log("git.snapshot", "实验开始时的工作区", snapshot(repo), ["executor", "executor"]);
+  let daemon;
+  try {
+    if (mode === "baseline") {
+      await childLoop(directory, "baseline");
+    } else {
+      daemon = await startExecutor(directory);
+      const loop = await childLoop(directory, "initial");
+      const events = readEvents(directory);
+      const test = events.find(event => event.kind === "test.start");
+      assert(test?.data?.pid, "Model did not launch the tests");
+      assert(!existsSync(join(directory, "test-result.json")), "Tests finished before observing Loop exit");
+      process.kill(test.data.pid, 0);
+      log("controller.observed", "确认 Loop 已退出，本机测试 PID 仍存活", {
+        loopPid: loop.pid, executorPid: daemon.child.pid, testPid: test.data.pid, testGate: "released after observation"
+      });
+      writeFileSync(join(directory, "release-tests"), "");
+      await completionMessage(directory);
+      daemon.child.kill();
+      const stopped = await daemon.done;
+      log("executor.stop", "停止 daemon，注入设备离线", stopped, ["executor", "executor"]);
+      await childLoop(directory, "offline");
+      assert(readEvents(directory).some(event => event.kind === "rpc.offline" && event.data.name === "repo_diff"),
+        "Model did not attempt to read diff while offline");
+      daemon = await startExecutor(directory);
+      await childLoop(directory, "reconnect");
+    }
+    const final = snapshot(repo);
+    log("git.snapshot", "实验完成后的工作区", final, ["executor", "executor"]);
+    assert(final.diff.includes("orders.mjs") && final.diff.includes("orders.test.mjs"), "Model did not change both files");
+    const test = JSON.parse(readFileSync(join(directory, "test-result.json"), "utf8"));
+    assert.equal(test.code, 0, "Fixture tests failed");
+    const passes = Number(test.stdout.match(/^# pass (\d+)$/m)?.[1]);
+    assert(passes >= 3, "Expected at least three passed test cases; recorded " + passes);
+    assert(final.notes.includes("Uncommitted local release note"), "Unrelated local edit changed");
+    const events = readEvents(directory);
+    const requests = events.filter(event => event.kind === "model.request");
+    const responses = events.filter(event => event.kind === "model.http");
+    assert(requests.length && requests.length === responses.length && responses.every(event => event.data.status === 200),
+      "Expected one successful gateway HTTP response per observed request");
+    const instances = events.filter(event => ["loop.start", "session.restore"].includes(event.kind));
+    assert.equal(new Set(instances.map(event => event.data.sessionId)).size, 1, "Session ID changed");
+    assert.equal(new Set(instances.map(event => event.pid)).size, mode === "split" ? 3 : 1, "Unexpected Loop instances");
+    assert(events.some(event => event.kind === "tool.result" && event.data.name === "repo_diff" && event.data.text === final.diff),
+      "Model did not read the final diff");
+    return { mode, events };
+  } finally {
+    if (daemon?.child.exitCode === null && daemon.child.signalCode === null) {
+      daemon.child.kill();
+      await daemon.done;
+    }
+  }
+}
+
+function readEvents(directory) {
+  return readFileSync(join(directory, "events.jsonl"), "utf8").trim().split("\n").map(line => JSON.parse(line));
+}
+
+async function record(output) {
+  assert(output, "Usage: node run-agent-loop-lab.mjs record <trace.json|existing lab HTML>");
+  const root = mkdtempSync(join(tmpdir(), "agent-loop-lab-"));
+  try {
+    const runs = [];
+    for (const mode of ["baseline", "split"]) {
+      console.error("Running " + mode + " with Pi SDK and " + modelName);
+      runs.push(await recordRun(root, mode));
+      console.error("Verified " + mode + " (" + runs.at(-1).events.length + " observed events)");
+    }
+    const trace = { capturedAt: new Date().toISOString(), model: modelName, protocol: "Anthropic Messages", runs };
+    if (output.endsWith(".html")) embedTrace(trace, output);
+    else writeFileSync(output, JSON.stringify(trace, null, 2));
+  } finally {
+    rmSync(root, { recursive: true, force: true });
+  }
+}
+
+function embedTrace(trace, htmlPath) {
+  const html = readFileSync(htmlPath, "utf8");
+  const section = /(<script type="application\/json" id="trace-data">)[\s\S]*?(<\/script>)/;
+  assert(section.test(html), "HTML lacks the lab trace-data script");
+  const serialized = JSON.stringify(trace).replaceAll("<", "\\u003c");
+  const temporary = htmlPath + "." + randomUUID() + ".tmp";
+  try {
+    writeFileSync(temporary, html.replace(section, (_, start, end) => start + serialized + end), { flag: "wx" });
+    renameSync(temporary, htmlPath);
+  } finally {
+    rmSync(temporary, { force: true });
+  }
+}
+
+try {
+  if (action === "record") await record(firstArgument);
+  else if (action === "embed") embedTrace(JSON.parse(readFileSync(firstArgument, "utf8")), phase);
+  else if (action === "executor") await serve(firstArgument);
+  else if (action === "loop") {
+    await runLoop(firstArgument, phase);
+    process.disconnect?.();
+  } else throw Error("Unknown lab action");
+} catch (error) {
+  console.error(error.message);
+  process.exitCode = 1;
+  process.disconnect?.();
+}