155 lines
9.4 KiB
TypeScript
155 lines
9.4 KiB
TypeScript
import { writeFileSync } from "node:fs";
|
||
import { join } from "node:path";
|
||
import type { LedgerRecord } from "../../src/ledger/index.ts";
|
||
import { callId, type DedupOutput } from "../../src/materials/index.ts";
|
||
import { ofKind, readLedger, stages, type Of } from "../lib/ledger.ts";
|
||
import { launch, newEvalDir, sleep, waitForLedger, writeConfig } from "../lib/study.ts";
|
||
import { watchTimeline } from "../lib/timeline.ts";
|
||
|
||
// 真实运行里的杀进程检验（任务书 14 第 3.1 节）：和 run.ts 一样从头跑一份研究（真模型、真 Docker、真 Jev，要花钱），
|
||
// 一边跑一边看账本；第 1 轮第一次预测阶段开着、已有一部分第一次预测被收下、还有人在写时，SIGKILL 控制进程，
|
||
// 再用恢复入口（--resume）启动让它跑完，查四条：
|
||
// 1 被杀那次调用补了结束记录（出错、原因"进程中断"）；
|
||
// 2 交过的任务（人×候选）没被重派，没交的被重派；
|
||
// 3 没有重复收下的提交（同人同对象同阶段只有一份收下，除非是替换）；
|
||
// 4 第一次预测阶段照常关闭，往下走到选批后停下（原因"实现还没写"）。
|
||
// node eval/round-one/kill.ts <名字>
|
||
// 产物和 run.ts 一样（study、timeline.jsonl、outcome.json），另有 kill-report.json；这份研究也算 3 次真实运行之一。
|
||
|
||
const POLL_MS = 3000;
|
||
const KIND = "firstPrediction";
|
||
/** flow 写停下事件用的事件名（src/flow/next-step.ts 的 EVENTS.halted；flow 只导出 runStudy，这里照抄）。 */
|
||
const HALTED = "halted";
|
||
|
||
const name = process.argv[2];
|
||
if (name === undefined) throw new Error("要写检验名字，例如 r1-kill-1（只许小写字母、数字和 ._-，它也是研究编号）");
|
||
const out = newEvalDir(name);
|
||
const config = writeConfig("eval/round-zero/research.json", join(out, "research.json"), name, null);
|
||
const studyDir = join(out, "study");
|
||
const timeline = join(out, "timeline.jsonl");
|
||
|
||
function openCalls(r: LedgerRecord[]): Array<Of<"callStarted">> {
|
||
const ended = new Set(ofKind(r, "callEnded").map((e) => e.content.call));
|
||
return ofKind(r, "callStarted").filter((c) => !ended.has(c.content.call));
|
||
}
|
||
/** 第 1 轮第一次预测阶段（开着的或已关的）；还没开是 undefined。 */
|
||
const predictionStage = (r: LedgerRecord[]) => stages(r).find((s) => s.round === 1 && s.stage === KIND);
|
||
/** 第一次预测的任务：名单上每个人 × 本轮每个新候选（任务书 14 第 2.3 节）。新候选取第 1 轮的判重决定。 */
|
||
function tasks(r: LedgerRecord[], roster: string[]): Array<{ actor: string; candidate: string }> {
|
||
const dedup = ofKind(r, "decision").find((d) => d.content.decision === "dedup" && d.content.round === 1);
|
||
if (dedup === undefined) throw new Error("第一次预测阶段开了，账本里却没有第 1 轮的判重决定");
|
||
const fresh = (dedup.content.output as unknown as DedupOutput).fresh.map((g) => g.candidate);
|
||
return roster.flatMap((actor) => fresh.map((candidate) => ({ actor, candidate })));
|
||
}
|
||
const accepted = (r: LedgerRecord[], stageId: string, actor: string, candidate: string) =>
|
||
ofKind(r, "submission").some((s) => s.content.stageId === stageId && s.content.kind === KIND && s.content.author === actor && s.content.target === candidate);
|
||
|
||
/** 打中的条件：阶段开着没关，已有至少一份收下，还有任务没交，而且有人正在写（这样第 1、2 条都查得到东西）。 */
|
||
function hit(r: LedgerRecord[]): boolean {
|
||
const s = predictionStage(r);
|
||
if (s === undefined || s.closed) return false;
|
||
const all = tasks(r, s.roster);
|
||
const done = all.filter((t) => accepted(r, s.stageId, t.actor, t.candidate)).length;
|
||
return done > 0 && done < all.length && openCalls(r).some((c) => c.content.stageId === s.stageId);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 第一次启动，打中就杀
|
||
// ---------------------------------------------------------------------------
|
||
|
||
const first = launch(config, studyDir, {});
|
||
await waitForLedger(studyDir, first.child);
|
||
const watching = watchTimeline(studyDir, first.child, timeline);
|
||
let triggered: LedgerRecord | undefined;
|
||
while (first.child.exitCode === null && first.child.signalCode === null) {
|
||
const r = readLedger(studyDir);
|
||
if (hit(r)) {
|
||
first.child.kill("SIGKILL");
|
||
triggered = r.at(-1);
|
||
break;
|
||
}
|
||
await sleep(POLL_MS);
|
||
}
|
||
const firstEnd = await first.done;
|
||
await watching;
|
||
if (triggered === undefined) {
|
||
writeFileSync(join(out, "outcome.json"), `${JSON.stringify({ launches: [firstEnd] }, null, 2)}\n`);
|
||
throw new Error(`没打中：研究进程自己结束了（${JSON.stringify(firstEnd.outcome)}），账本最后一条是 ${JSON.stringify(readLedger(studyDir).at(-1))}`);
|
||
}
|
||
// 进程已死，此刻的账本就是被杀时的样子。
|
||
const atKill = readLedger(studyDir);
|
||
const killSeq = atKill.at(-1)!.seq;
|
||
const killedCalls = openCalls(atKill).map((c) => c.content.call);
|
||
const stage = predictionStage(atKill)!;
|
||
const taskList = tasks(atKill, stage.roster).map((t) => ({ ...t, submittedBeforeKill: accepted(atKill, stage.stageId, t.actor, t.candidate) }));
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 再启动，跑完
|
||
// ---------------------------------------------------------------------------
|
||
|
||
const second = launch(null, studyDir, {});
|
||
process.on("SIGINT", () => second.child.kill("SIGINT"));
|
||
await watchTimeline(studyDir, second.child, timeline);
|
||
const secondEnd = await second.done;
|
||
writeFileSync(join(out, "outcome.json"), `${JSON.stringify({ launches: [firstEnd, secondEnd] }, null, 2)}\n`);
|
||
const r = readLedger(studyDir);
|
||
|
||
// 1 被杀那次调用补了结束记录
|
||
const ends = ofKind(r, "callEnded");
|
||
const check1 = killedCalls.map((call) => {
|
||
const end = ends.find((e) => e.content.call === call);
|
||
return { call, end: end === undefined ? null : { seq: end.seq, ...end.content } };
|
||
});
|
||
|
||
// 2 交过的没重派、没交的重派了。一次调用属于哪个任务，用 materials 的调用编号规则对：同一任务第 n 次发起是 callId(…, n)。
|
||
const startedAfter = ofKind(r, "callStarted").filter((c) => c.seq > killSeq);
|
||
const attempts = ofKind(r, "callStarted").length;
|
||
const check2 = taskList.map((t) => {
|
||
const ids = new Set(Array.from({ length: attempts }, (_, i) => callId(stage.stageId, t.actor, KIND, t.candidate, i + 1)));
|
||
const again = startedAfter.filter((c) => ids.has(c.content.call)).map((c) => c.content.call);
|
||
return { ...t, dispatchedAfterRestart: again, ok: t.submittedBeforeKill === (again.length === 0) };
|
||
});
|
||
|
||
// 3 没有重复收下：第 1 轮每份提交（阶段、作者、种类、对象、编号）的各版版本号是 1..n；第一次预测里被杀前已收下的，被杀后不能再多出一版
|
||
const round1 = new Set(stages(r).filter((s) => s.round === 1).map((s) => s.stageId));
|
||
const slots = new Map<string, Array<{ seq: number; version: number }>>();
|
||
for (const s of ofKind(r, "submission")) {
|
||
const c = s.content;
|
||
if (!round1.has(c.stageId)) continue;
|
||
const key = JSON.stringify([c.stageId, c.author, c.kind, c.target, c.item]);
|
||
slots.set(key, [...(slots.get(key) ?? []), { seq: s.seq, version: c.version }]);
|
||
}
|
||
const check3 = [...slots].map(([slot, versions]) => {
|
||
const numbered = versions.map((v) => v.version).join(",") === versions.map((_, i) => i + 1).join(",");
|
||
const [stageId, , kind] = JSON.parse(slot) as [string, string, string];
|
||
const acrossKill = stageId === stage.stageId && kind === KIND && versions.some((v) => v.seq <= killSeq) && versions.some((v) => v.seq > killSeq);
|
||
return { slot, versions, replaced: versions.length > 1, ok: numbered && !acrossKill };
|
||
});
|
||
|
||
// 4 阶段照常关闭、往下走到选批后停下
|
||
const closed = ofKind(r, "stageClose").some((c) => c.content.stageId === stage.stageId);
|
||
const batch = ofKind(r, "decision").some((d) => d.content.decision === "batch" && d.content.round === 1);
|
||
const lastHalt = ofKind(r, "event").filter((e) => e.content.event === HALTED).at(-1);
|
||
|
||
const checks = {
|
||
"1 被杀那次调用补了结束记录（出错、进程中断）": {
|
||
pass: check1.length > 0 && check1.every((c) => c.end !== null && c.end.status === "failed" && c.end.reason.includes("进程中断")),
|
||
calls: check1,
|
||
},
|
||
"2 交过的任务没被重派，没交的被重派": { pass: check2.every((t) => t.ok), tasks: check2 },
|
||
"3 没有重复收下的提交（替换除外）": { pass: check3.every((s) => s.ok), problems: check3.filter((s) => !s.ok), replaced: check3.filter((s) => s.replaced) },
|
||
"4 阶段照常关闭，往下走到选批后停下": {
|
||
pass: closed && batch && lastHalt !== undefined && lastHalt.content.reason.includes("实现还没写"),
|
||
firstPredictionClosed: closed,
|
||
batchDecision: batch,
|
||
lastHalt: lastHalt?.content ?? null,
|
||
},
|
||
};
|
||
const report = { triggeredAt: { seq: triggered.seq, kind: triggered.kind }, killSeq, lastRecordAtKill: atKill.at(-1), tasksAtKill: taskList, outcomeAfterRestart: secondEnd.outcome, checks };
|
||
writeFileSync(join(out, "kill-report.json"), `${JSON.stringify(report, null, 2)}\n`);
|
||
console.log(`\n被杀时账本最后一条 #${killSeq} ${atKill.at(-1)!.kind}；被杀时正在进行的调用：${killedCalls.join("、") || "无"}`);
|
||
console.log(`被杀时第一次预测：${taskList.filter((t) => t.submittedBeforeKill).length}/${taskList.length} 个任务已交`);
|
||
console.log(`再启动后的结局：${JSON.stringify(secondEnd.outcome)}`);
|
||
for (const [k, v] of Object.entries(checks)) console.log(`${v.pass ? "通过" : "不通过"} ${k}`);
|
||
console.log(`报告：${join(out, "kill-report.json")}`);
|