445 lines
20 KiB
JavaScript
445 lines
20 KiB
JavaScript
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?.();
|
||
}
|