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?.();
}