Files
InfoRSI/eval/diagnosis/run.ts
T

348 lines
13 KiB
TypeScript

import { execFileSync } from "node:child_process";
import {
appendFileSync,
existsSync,
mkdirSync,
readFileSync,
writeFileSync,
} from "node:fs";
import { loadavg, freemem, totalmem } from "node:os";
import { join, resolve } from "node:path";
import { fullAccounts, FULL_STOP } from "../lib/full.ts";
import { jsonFile } from "../lib/forward.ts";
import {
sameContainerThroughResult,
type ContainerSnapshot,
} from "../lib/forward-evidence.ts";
import { Board } from "../../src/materials/index.ts";
import { ofKind, readLedger } from "../lib/ledger.ts";
import { launch, sleep, waitForLedger, writeConfig } from "../lib/study.ts";
import { watchTimeline } from "../lib/timeline.ts";
// 任务书 18 第 3.1 节:从头真实跑一份诊断开、消融关的完整研究,到余额用完停在"导出还没写"。三份同时跑,由父会话各起一个进程;
// 容器名带研究编号,这里只认本研究的,不和别的研究互相核对。
// node eval/diagnosis/run.ts <研究编号> <绝对输出目录> [--kill-after-first-diagnosis-start] [--resume]
const [name, output, ...extra] = process.argv.slice(2);
// 选项:--kill-after-first-diagnosis-start(杀一次)、--resume(模型服务一时调不通而停下后接着跑,同一个研究目录)。
const OPTIONS = ["--kill-after-first-diagnosis-start", "--resume"];
const kill = extra.includes("--kill-after-first-diagnosis-start");
const resume = extra.includes("--resume");
if (
!name ||
!output ||
extra.some((x) => !OPTIONS.includes(x)) ||
new Set(extra).size !== extra.length ||
output !== resolve(output)
) {
throw new Error(
"用法:node eval/diagnosis/run.ts <研究编号> <绝对输出目录> [--kill-after-first-diagnosis-start] [--resume]",
);
}
const out = output;
const studyDir = join(out, "study");
let config: string | null = null;
if (resume) {
if (!existsSync(studyDir)) throw new Error(`没有可续跑的研究:${studyDir}`);
} else {
if (existsSync(out)) throw new Error(`输出目录已存在,不覆盖现场:${out}`);
mkdirSync(out, { recursive: true });
config = writeConfig(
"eval/diagnosis/research.json",
join(out, "research.json"),
name,
null,
);
}
const daemonTime = () =>
execFileSync("docker", ["info", "--format", "{{.SystemTime}}"], {
encoding: "utf8",
}).trim();
const inspect = (
container: string,
): ContainerSnapshot & {
Name: string;
Created: string;
State: {
Running: boolean;
StartedAt: string;
FinishedAt: string;
Status: string;
};
} =>
JSON.parse(
execFileSync("docker", ["inspect", container], { encoding: "utf8" }),
)[0];
/** 事件点的重快照(进程列表、容器列表),照任务书 16 的 machine.jsonl 做法。 */
function observe(label: string) {
appendFileSync(
join(out, "machine.jsonl"),
`${JSON.stringify({
at: new Date().toISOString(),
label,
processes: execFileSync("ps", ["-eo", "pid,ppid,lstart,args"], {
encoding: "utf8",
}),
containers: execFileSync("docker", ["ps", "--no-trunc"], {
encoding: "utf8",
}),
})}\n`,
);
}
/** 运行期间的轻量负载采样,几秒一条(任务书 17 第 3.1 节:和每次运行的 CPU/墙钟比对照,只列事实)。 */
function sample() {
appendFileSync(
join(out, "machine.jsonl"),
`${JSON.stringify({ at: new Date().toISOString(), label: "sample", load: loadavg(), memory: { total: totalmem(), free: freemem() } })}\n`,
);
}
function start(path: string | null) {
observe(path === null ? "resume" : "start");
const study = launch(path, studyDir, {});
appendFileSync(
join(out, "launches.jsonl"),
`${JSON.stringify({ event: "start", pid: study.child.pid, at: new Date().toISOString(), config: path })}\n`,
);
const interrupt = () => study.child.kill("SIGINT");
process.on("SIGINT", interrupt);
study.done.then((end) => {
process.off("SIGINT", interrupt);
appendFileSync(
join(out, "launches.jsonl"),
`${JSON.stringify({ event: "end", pid: study.child.pid, at: new Date().toISOString(), ...end })}\n`,
);
});
return study;
}
const sampler = setInterval(sample, 5000);
sampler.unref();
let active = start(config);
// 杀过一次以后(kill.json 已写),模型服务出错等停下再续跑时不再杀;最后的三条检查照 kill.json 和杀时账本做。
const killPending = kill && !existsSync(join(out, "kill.json"));
try {
await waitForLedger(studyDir, active.child);
let watching = watchTimeline(
studyDir,
active.child,
join(out, "timeline.jsonl"),
);
if (killPending) {
// 盯住账本:第一个用途为 diagnosis 的运行确认启动之后、结果落账之前杀控制进程(照 eval/forward/run.ts 的 kill 做法)。
let hit = false;
while (active.child.exitCode === null && active.child.signalCode === null) {
const records = readLedger(studyDir);
const plan = ofKind(records, "runPlanned").find(
(r) => r.content.purpose === "diagnosis",
);
if (plan !== undefined) {
const run = plan.content.run;
if (ofKind(records, "runResult").some((r) => r.content.run === run)) {
throw new Error(
"第一个诊断运行已经有结果了,未打中;保留现场,不改杀别的轮",
);
}
const started = ofKind(records, "runStarted").find(
(r) => r.content.run === run,
);
if (started !== undefined) {
// 容器名是正式入口写在结构文档里的约定(inforsi-<研究编号>-<运行编号>),不 import flow 内部文件。
const container = `inforsi-${name}-${run}`;
// 从杀前划定历史事件区间,覆盖恢复启动到目标结果落账;不把 spawn 当成恢复已执行。
const lifecycleSince = daemonTime();
const before = inspect(container);
jsonFile(join(out, "container-before.json"), before);
if (!before.State.Running)
throw new Error("第一个诊断容器已不在运行,未打中");
jsonFile(join(out, "kill.json"), {
run,
cell: plan.content.object,
round: plan.content.round,
planSeq: plan.seq,
startSeq: started.seq,
beforeKillSeq: records.at(-1)!.seq,
pid: active.child.pid,
at: new Date().toISOString(),
container,
});
if (!active.child.kill("SIGKILL")) throw new Error("未发出 SIGKILL");
const killed = await active.done;
await watching;
if (killed.signal !== "SIGKILL")
throw new Error("控制进程不是以 SIGKILL 结束");
const atKill = readLedger(studyDir);
jsonFile(join(out, "ledger-at-kill.json"), atKill);
if (ofKind(atKill, "runResult").some((r) => r.content.run === run)) {
throw new Error("杀后核对发现目标运行已经写结果,未打中;保留现场");
}
jsonFile(join(out, "container-after-kill.json"), inspect(container));
observe("after-kill");
jsonFile(
join(out, "container-before-resume.json"),
inspect(container),
);
active = start(null);
watching = watchTimeline(
studyDir,
active.child,
join(out, "timeline.jsonl"),
);
let targetResult;
while (true) {
targetResult = ofKind(readLedger(studyDir), "runResult").find(
(r) => r.content.run === run,
);
if (
targetResult !== undefined ||
active.child.exitCode !== null ||
active.child.signalCode !== null
)
break;
await sleep(100);
}
const lifecycleUntil = daemonTime();
// Docker CLI 按时间回读,不依赖事件订阅的启动时序;原文全存,stats 核对缓存上限和目标 die。
writeFileSync(
join(out, "container-events.jsonl"),
execFileSync(
"docker",
[
"events",
"--since",
lifecycleSince,
"--until",
lifecycleUntil,
"--format",
"{{json .}}",
],
{ encoding: "utf8" },
),
);
jsonFile(join(out, "container-events-window.json"), {
since: lifecycleSince,
until: lifecycleUntil,
run,
resultSeq: targetResult?.seq ?? null,
});
if (targetResult === undefined)
throw new Error(
"恢复进程没有交出目标结果,生命周期证据不完整;保留现场",
);
hit = true;
break;
}
}
await sleep(100);
}
if (!hit)
throw new Error(
`没有打中第一个诊断运行:${JSON.stringify(await active.done)}`,
);
}
await watching;
const end = await active.done;
jsonFile(join(out, "outcome.json"), end);
observe("end");
const records = readLedger(studyDir);
const accounts = fullAccounts(records);
jsonFile(join(out, "accounts.json"), accounts);
// 杀进程的三条检查(任务书 17 第 3.1 节):①恢复按容器名接着等同一个容器、没有重跑;②只扣一次;③结果照常写入、结算和后续步骤走完。
if (existsSync(join(out, "kill.json"))) {
const killSeq = (
JSON.parse(
readFileSync(join(out, "ledger-at-kill.json"), "utf8"),
) as Array<{ seq: number }>
).at(-1)!.seq;
const evidence = JSON.parse(
readFileSync(join(out, "kill.json"), "utf8"),
) as { run: string; cell: string; round: number; container: string };
const snapshots = (["before", "after-kill", "before-resume"] as const).map(
(part) =>
JSON.parse(
readFileSync(join(out, `container-${part}.json`), "utf8"),
) as ContainerSnapshot,
);
const lifecycle = readFileSync(join(out, "container-events.jsonl"), "utf8")
.split("\n")
.filter(Boolean)
.map((line) => JSON.parse(line));
const lifecycleChecks = sameContainerThroughResult(
snapshots,
lifecycle,
evidence.container,
);
const target = accounts.runs.find(
(r) => r.plan.content.run === evidence.run,
)!;
const cellPlans = accounts.runs.filter(
(r) =>
r.plan.content.purpose === "diagnosis" &&
r.plan.content.object === evidence.cell,
);
const board = new Board(records);
const settlement = board.diagnosisSettlement(evidence.round);
const settlementSeq = ofKind(records, "diagnosisSettlement").find(
(r) => r.content.round === evidence.round,
)!.seq;
// 选中的每个实验:核验不通过拿掉的记在 dropped,其余都在逐实验判分里有一行。
const chosen = board.decision("selectDiagnosis", evidence.round)!.output.chosen;
const perExperimentRecorded =
settlement !== undefined &&
chosen.every(
(e) =>
settlement.dropped.some((d) => d.experiment === e) ||
settlement.judged.experiments.some((x) => x.id === e),
);
const report = {
evidence,
killSeq,
snapshotMeaning: "杀前、杀后、恢复启动前;没有把快照当成恢复完成的证据",
lifecycleChecks,
checks: {
sameContainerNoRerun:
Object.values(lifecycleChecks).every(Boolean) &&
cellPlans.length === 1 &&
target.starts.length === 1,
chargedOnce: target.starts.length === 1,
resultSettledAndFinished:
target.results.length === 1 &&
target.results[0]!.seq > killSeq &&
settlement !== undefined &&
settlement.run === evidence.run &&
perExperimentRecorded &&
ofKind(records, "stageOpen").some(
(o) =>
o.content.stage === "proposal" &&
o.content.round === evidence.round &&
o.seq > settlementSeq,
) &&
accounts.halt !== null &&
accounts.halt.reason === FULL_STOP,
},
};
jsonFile(join(out, "kill-report.json"), report);
}
if (
end.code !== 0 ||
!accounts.checks.haltedAtExport ||
!accounts.checks.chargedEqualsStarts
) {
throw new Error(
`没有走到"导出还没写"或账目不平,保留现场;查明原因后用 npm run eval:study -- --resume --dir ${studyDir} 续跑:${JSON.stringify(end)}`,
);
}
console.log(`研究走完:${studyDir}。仍需 export、stats 和独立质量审阅。`);
} finally {
clearInterval(sampler);
if (active.child.exitCode === null && active.child.signalCode === null) {
active.child.kill("SIGINT");
await active.done;
}
}