feat: flow 第 2 部分,第 r 轮开轮到选批(任务书 14 第二块)

- next-step 按"第 r 轮"列出提案、第一次预测、挑刺、回应与统一、第二次预测的任务
  (一人对一个对象),判重、冻结协议、选批三步,走到 implementation 停下(任务书 15)
- steps/dispatch.ts:同阶段一人任务先后跑、不同人同时跑;提案、挑刺按 proposalId、critiqueId 分份
- steps/round.ts:判重两两问 Jev、连通成组、并进留池候选;冻结协议在回应阶段关闭后一个事务写全;
  选批照 rules 的 BatchInput 拼,种子用 decisionSeed 版本 1(选格预留版本 2)
- materials 导出 Board(构造改收记录数组,给纯函数用);讨论原文里已作废的入口编号就地标"(已作废)"
- 删 SOURCES.md 里参考项目实测事实那一段(用户 2026-10-03 同意)
- 测试:下一步做什么 21 条、判重连组 4 条、冻结与选批 2 条、作废标注 1 条;npm test 437 通过

全新上下文审阅一轮(只看三个程序步骤与恢复):没有会污染账本的问题。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-03 13:59:54 -04:00
parent 2c6fdce1ca
commit f98d664a16
17 changed files with 835 additions and 133 deletions
+13 -2
View File
@@ -8,7 +8,9 @@ import { EVENTS, nextStep, type Next } from "./next-step.ts";
import { ledgerFile } from "./paths.ts";
import { recover } from "./recover.ts";
import { baselineRun, saveBaseline } from "./steps/baseline.ts";
import { assignEntries, dispatchStage } from "./steps/round-zero.ts";
import { dispatchStage } from "./steps/dispatch.ts";
import { dedupe, freezeProtocols, selectBatch } from "./steps/round.ts";
import { assignEntries } from "./steps/round-zero.ts";
import { freezeConfig, halt, openStudy, USER_STOPPED, type Study, type StudyEnding } from "./study.ts";
// 主循环:读账本 → 问下一步 → 做这一步 → 回到开头(任务书 12 第 2.4 节)。每做完一步都重新读账本。
@@ -83,7 +85,7 @@ async function doStep(s: Study, next: Exclude<Next, { do: "freezeConfig" }>): Pr
s.ledger.append("event", { event: EVENTS.skipped, reason: next.reason, objects: [next.step] });
return null;
case "openStage":
s.ledger.append("stageOpen", { stageId: next.stageId, stage: next.stage, round: 0, roster: next.roster });
s.ledger.append("stageOpen", { stageId: next.stageId, stage: next.stage, round: next.round, roster: next.roster });
return null;
case "closeStage":
s.ledger.append("stageClose", { stageId: next.stageId });
@@ -93,6 +95,15 @@ async function doStep(s: Study, next: Exclude<Next, { do: "freezeConfig" }>): Pr
case "assignEntries":
assignEntries(s);
return null;
case "dedupe":
await dedupe(s, next.round);
return null;
case "freezeProtocols":
freezeProtocols(s, next.round);
return null;
case "selectBatch":
selectBatch(s, next.round);
return null;
case "halt":
return halt(s.ledger, next.reason, next.objects);
}
+43 -23
View File
@@ -10,13 +10,16 @@ import type { Study } from "./study.ts";
// 最后写"模型调用结束",前后顺序由这个入口保证(方法.md 第九部分"先登记、再执行、再登记结果")。
// 提交收不收只由 submissions 判;形状不合格的由 agents 报上来,这里直接记退回("给 flow")。
/** 一个研究者任务:哪种提交、谁、哪个阶段、对象;known 是交上来之后才知道的东西(submissions 的 KnownOf)。 */
/** 一个研究者任务:哪种提交、谁、哪个阶段、对象。 */
export type ModelTask<K extends KindOf<"researcher">> = {
kind: K;
actor: string;
stageId: string;
object: string | null;
known(content: Submission<K>): KnownOf<K>;
/** 一个任务里能交几份的种类(提案、挑刺):第 n 份的编号(materials 的 proposalId、critiqueId),收下时分配;只交一份的为 null。 */
item: ((n: number) => string) | null;
/** 交上来之后才知道的东西(submissions 的 KnownOf)。others 是本任务已收下、而且不是这次要替换的那几份的份数。 */
known(content: Submission<K>, others: number): KnownOf<K>;
};
/** 派一次任务,等会话结束,交回调用编号和结局。 */
@@ -81,41 +84,58 @@ function researcherModel(s: Study, actor: string) {
return models[i]!;
}
/**
* 一份提交收不收。这个任务只有一份提交(第零轮每人每阶段一个任务、一份提交):
* 回给模型的提交编号就是这一份的位置,不分版本;替换时写它,接在最新一版后面。
* 第一次交以后不写 replaces 又交,或 replaces 写的不是这一份,都退回并说明该怎么写。
*/
/** 一份提交收不收:先定写到哪一份,形状合格的再交给 submissions 判。 */
async function submit<K extends KindOf<"researcher">>(s: Study, t: ModelTask<K>, material: TaskMaterial, attempt: SubmitAttempt<K>): Promise<SubmitReceipt> {
const { ledger } = s;
const slot = { stageId: t.stageId, author: t.actor, kind: t.kind, target: t.object, item: null };
const reject = (submitted: Json, reason: string, jev: JevJudgment[] | null): SubmitReceipt => {
ledger.append("rejection", { stageId: t.stageId, author: t.actor, kind: t.kind, target: t.object, submitted, reason, jev });
return { accepted: false, reason };
};
if (attempt.shape === "invalid") return reject(attempt.raw, attempt.reason, null);
const id = receiptId(t);
const latest = ledger.latestSubmission(slot)?.content.version ?? 0;
const content = attempt.content as unknown as Json;
if (attempt.replaces === null && latest > 0) {
return reject(content, `本阶段你已经交过这一份(提交编号 ${id})。要改就在 replaces 里写 ${id} 整份替换,不要另交`, null);
}
if (attempt.replaces !== null && (attempt.replaces !== id || latest === 0)) {
const which = latest === 0 ? "你本阶段还没有收下的提交,第一次交时不要带 replaces" : `要替换就写 ${id}`;
return reject(content, `replaces 写的 ${attempt.replaces} 不是你本阶段交过的提交:${which}`, null);
}
const slot = slotOf(s, t, attempt.replaces);
if ("refuse" in slot) return reject(content, slot.refuse, null);
const verdict = await receive({
kind: t.kind,
content: attempt.content,
checklist: material.checklist,
known: t.known(attempt.content),
known: t.known(attempt.content, slot.others),
jevModel: s.frozen.config.models.jev,
} as Incoming);
if (!verdict.accepted) return reject(content, verdict.reason, verdict.jev);
ledger.append("submission", { ...slot, version: latest + 1, content: attempt.content, jev: verdict.jev } as never);
return { accepted: true, id };
const { stageId, actor: author, kind, object: target } = t;
ledger.append("submission", { stageId, author, kind, target, item: slot.item, version: slot.version, content: attempt.content, jev: verdict.jev } as never);
return { accepted: true, id: slot.id };
}
/** 回给模型的提交编号:阶段、本人、提交种类、对象。 */
const receiptId = (t: { stageId: string; actor: string; kind: string; object: string | null }): string =>
`${t.stageId}.${t.actor}.${t.kind}.${t.object ?? "none"}`;
/**
* 这次提交写到哪一份:回给模型的提交编号、记录的 item、版本、本任务其余已收下几份。
* 提交编号:能交几份的种类就是这一份的编号,只交一份的是"阶段.本人.种类.对象"。
* 不写 replaces 是交新的一份,只交一份的种类已经交过就退回并说明该怎么写;写了 replaces 就必须是本任务已收下的某一份。
*/
function slotOf<K extends KindOf<"researcher">>(
s: Study,
t: ModelTask<K>,
replaces: string | null,
): { id: string; item: string | null; version: number; others: number } | { refuse: string } {
const mine = s.ledger
.records()
.flatMap((r) => (r.kind === "submission" && r.content.stageId === t.stageId && r.content.author === t.actor && r.content.kind === t.kind && r.content.target === t.object ? [r.content.item] : []));
const items = [...new Set(mine)];
const idOf = (item: string | null): string => item ?? `${t.stageId}.${t.actor}.${t.kind}.${t.object ?? "none"}`;
if (replaces === null) {
if (t.item === null && items.length > 0) {
return { refuse: `本阶段你已经交过这一份(提交编号 ${idOf(null)})。要改就在 replaces 里写 ${idOf(null)} 整份替换,不要另交` };
}
const item = t.item === null ? null : t.item(items.length + 1);
return { id: idOf(item), item, version: 1, others: items.length };
}
const at = items.findIndex((item) => idOf(item) === replaces);
if (at < 0) {
const which = items.length === 0 ? "你本阶段还没有收下的提交,第一次交时不要带 replaces" : `能替换的提交编号:${items.map(idOf).join("、")}`;
return { refuse: `replaces 写的 ${replaces} 不是你本阶段交过的提交:${which}` };
}
const item = items[at] as string | null;
const latest = s.ledger.latestSubmission({ stageId: t.stageId, author: t.actor, kind: t.kind, target: t.object, item })!;
return { id: replaces, item, version: latest.content.version + 1, others: items.length - 1 };
}
+159 -41
View File
@@ -1,6 +1,8 @@
import type { StudyStepId } from "../config/index.ts";
import type { KindOf } from "../agents/index.ts";
import type { RoundStepId, StudyStepId } from "../config/index.ts";
import type { Content, LedgerRecord } from "../ledger/index.ts";
import { ORIGINAL_BASELINE, researcherId } from "../materials/index.ts";
import { Board, ORIGINAL_BASELINE, researcherId } from "../materials/index.ts";
import { compareIds } from "../rules/index.ts";
import type { Stage } from "../vocab/index.ts";
import { readFrozen, type Frozen } from "./frozen.ts";
@@ -19,16 +21,18 @@ const ROUND_ZERO_STAGES = {
claim: { stage: "claim", kind: "claim" },
} as const satisfies Partial<Record<StudyStepId, { stage: Stage; kind: string }>>;
type RoundZeroStep = keyof typeof ROUND_ZERO_STAGES;
export type RoundZeroKind = (typeof ROUND_ZERO_STAGES)[RoundZeroStep]["kind"];
/** 阶段编号由 flow 起:第零轮是 r0-<步骤>,一个步骤一个阶段。 */
export const stageIdOf = (step: RoundZeroStep): string => `r0-${step}`;
/** 第 r 轮的研究者阶段:步骤名就是阶段名(vocab STAGES)。 */
type RoundStageStep = Extract<RoundStepId, "proposal" | "firstPrediction" | "critique" | "response" | "secondPrediction">;
/** 阶段编号由 flow 起:r<轮>-<步骤>,一个步骤一个阶段。第 r 轮按配置跳过的步骤也用这个编号记"按配置跳过"。 */
export const stageIdOf = (round: number, step: string): string => `r${round}-${step}`;
/** 基线运行编号:没启动就起下一个编号重跑(docs/结构.md 第四部分)。 */
export const baselineRunId = (n: number): string => `base-${n}`;
/** flow 写的事件种类(账本记录 event)。 */
export const EVENTS = {
/** 按配置跳过的步骤,objects 是步骤编号,reason 是 config 给的原因。 */
/** 按配置跳过的步骤,objects 是步骤编号(第 r 轮的步骤写 r<轮>-<步骤>),reason 是 config 给的原因。 */
skipped: "skippedByConfig",
/** 研究停下,reason 是停下的原因,objects 是相关的调用或运行。 */
halted: "halted",
@@ -36,15 +40,21 @@ export const EVENTS = {
runStatusUnknown: "runStatusUnknown",
} as const;
/** 一个研究者任务:一个人对一个对象(候选、原提案;提案和第零轮的任务没有对象)交一种提交。 */
export type TaskRef = { actor: string; kind: KindOf<"researcher">; object: string | null };
export type Next =
| { do: "freezeConfig" }
| { do: "saveBaseline" }
| { do: "baselineRun"; run: string }
| { do: "skip"; step: StudyStepId; reason: string }
| { do: "openStage"; stageId: string; stage: Stage; roster: string[] }
| { do: "skip"; step: string; reason: string }
| { do: "openStage"; stageId: string; stage: Stage; round: number; roster: string[] }
| { do: "closeStage"; stageId: string }
| { do: "dispatch"; stageId: string; kind: RoundZeroKind; actors: string[] }
| { do: "dispatch"; stageId: string; tasks: TaskRef[] }
| { do: "assignEntries" }
| { do: "dedupe"; round: number }
| { do: "freezeProtocols"; round: number }
| { do: "selectBatch"; round: number }
| { do: "halt"; reason: string; objects: string[] };
export function nextStep(records: readonly LedgerRecord[]): Next {
@@ -53,19 +63,20 @@ export function nextStep(records: readonly LedgerRecord[]): Next {
if (frozen === null) return { do: "freezeConfig" };
const view = new View(records);
for (const step of frozen.steps.study) {
if (step.status === "skip") {
// 按配置跳过的步骤 → 一条"按配置跳过"事件,写明步骤和原因。
if (!view.events(EVENTS.skipped).some((e) => e.objects[0] === step.id)) return { do: "skip", step: step.id, reason: step.reason };
continue;
}
const next = stepNext(step.id, view, frozen);
const next =
step.status === "skip" ? skipNext(step.id, step.reason, view) : step.id === "rounds" ? roundsNext(view, frozen) : studyStepNext(step.id, view, frozen);
if (next !== null) return next;
}
throw new Error("步骤清单走完了却没有停下:rounds 一步应当停下");
}
/** 一个要执行的步骤还要做什么;做完了交回 null。 */
function stepNext(step: StudyStepId, view: View, frozen: Frozen): Next | null {
/** 按配置跳过的步骤 → 一条"按配置跳过"事件,写明步骤和原因。 */
function skipNext(step: string, reason: string, view: View): Next | null {
return view.events(EVENTS.skipped).some((e) => e.objects[0] === step) ? null : { do: "skip", step, reason };
}
/** 整份研究的一个要执行的步骤还要做什么;做完了交回 null。 */
function studyStepNext(step: Exclude<StudyStepId, "rounds">, view: View, frozen: Frozen): Next | null {
switch (step) {
case "freezeConfig":
return null;
@@ -74,18 +85,107 @@ function stepNext(step: StudyStepId, view: View, frozen: Frozen): Next | null {
case "entryDraft":
case "discussion1":
case "discussion2":
case "claim":
return stageNext(step, view, frozen);
case "claim": {
const { stage, kind } = ROUND_ZERO_STAGES[step];
return stageNext(stageIdOf(0, step), stage, 0, (roster) => roster.map((actor) => ({ actor, kind, object: null })), view, frozen);
}
case "assignEntries":
// 分配 → 第 0 轮的入口分配决定(decision,种类 entries)。
return view.of("decision").some((d) => d.decision === "entries" && d.round === 0) ? null : { do: "assignEntries" };
case "rounds":
return { do: "halt", reason: "第 1 轮还没写(任务书 13)", objects: [] };
return view.board.decision("entries", 0) === undefined ? { do: "assignEntries" } : null;
case "export":
throw new Error("走到了导出:rounds 一步应当先停下");
}
}
/**
* 一轮接一轮(结构.md 第三部分"每一轮"):第一个没做完的轮次里第一个没做完的步骤。
* 每一步做完的记录都带轮次,做完的轮次整轮跳过;现在走到实现一步就停下,所以循环一定会返回。
*/
function roundsNext(view: View, frozen: Frozen): Next {
for (let round = 1; ; round++) {
for (const step of frozen.steps.round) {
const next = step.status === "skip" ? skipNext(stageIdOf(round, step.id), step.reason, view) : roundStepNext(step.id, round, view, frozen);
if (next !== null) return next;
}
}
}
/** 第 round 轮的一个要执行的步骤还要做什么;做完了交回 null。 */
function roundStepNext(step: RoundStepId, round: number, view: View, frozen: Frozen): Next | null {
switch (step) {
case "openRound":
case "carryOverPool":
// 不写记录:本轮名单和提案上限就是冻结的配置(方法.md 4.3 ①),留池候选由 materials 从记录算出(Board.carried)。
return null;
case "proposal":
case "firstPrediction":
case "critique":
case "response":
case "secondPrediction":
return stageNext(stageIdOf(round, step), step, round, (roster) => roundTasks(step, round, roster, view.board), view, frozen);
case "dedupe":
// 判重合并 → 本轮的判重合并决定(decision,种类 dedup)。
return view.board.decision("dedup", round) === undefined ? { do: "dedupe", round } : null;
case "freezeProtocol": {
// 冻结协议 → 本轮每个新候选都有冻结协议记录(protocol)。全部候选在同一个事务里写;
// 这一步排在回应阶段关闭之后,materials 把冻结协议的出现当作"已公开",早写会泄露。
if (frozen.steps.round.some((s) => s.id === "response" && s.status === "skip")) {
return { do: "halt", reason: "配置关掉了挑刺与回应时怎么冻结协议还没写:submissions 的 protocolBody 只收回应和统一", objects: [] };
}
const frozenIds = new Set(view.board.protocols().map((p) => p.candidate));
return freshOf(view.board, round).every((c) => frozenIds.has(c.candidate)) ? null : { do: "freezeProtocols", round };
}
case "selectBatch":
// 选批 → 本轮的选批决定(decision,种类 batch)。
return view.board.decision("batch", round) === undefined ? { do: "selectBatch", round } : null;
case "implementation":
return { do: "halt", reason: "实现还没写(任务书 15)", objects: [] };
default:
throw new Error(`走到了第 ${round} 轮的 ${step}:implementation 一步应当先停下`);
}
}
/** 第 round 轮判重合并出的新候选,以及各合并了哪几条原提案。 */
export function freshOf(board: Board, round: number): Array<{ candidate: string; proposals: string[] }> {
return board.decision("dedup", round)!.output.fresh;
}
/** 第 round 轮的候选池:新候选加留池候选(方法.md 4.3 ①⑦)。第二次预测和选批都用这一份。 */
export function poolOf(board: Board, round: number): string[] {
return [...freshOf(board, round).map((c) => c.candidate), ...board.carried(round)];
}
/**
* 第 r 轮一个研究者阶段的任务(任务书 14 第 2.3 节)。同一个人的几个任务按这里的先后跑。
* - 提案:每人一个任务,没有对象;
* - 第一次预测:每人 × 本轮每个新候选(含自己的;留池候选不做,方法.md 4.3 ④);
* - 挑刺:每人 × 本轮每条不是自己写的原提案(监督者定,照参考项目);
* - 回应与统一:每个新候选一个任务,一条原提案的由作者交回应,几条合成的由其中编号最小的作者交统一(方法.md 4.3 ③⑥);
* - 第二次预测:每人 × 候选池里每个候选。
*/
function roundTasks(step: RoundStageStep, round: number, roster: readonly string[], board: Board): TaskRef[] {
const each = (kind: TaskRef["kind"], objects: readonly string[]): TaskRef[] =>
roster.flatMap((actor) => objects.map((object) => ({ actor, kind, object })));
switch (step) {
case "proposal":
return roster.map((actor) => ({ actor, kind: "proposal", object: null }));
case "firstPrediction":
return each("firstPrediction", freshOf(board, round).map((c) => c.candidate));
case "critique": {
const proposals = board.published("proposal").filter((p) => p.round === round).sort((a, b) => compareIds(a.item!, b.item!));
return roster.flatMap((actor) => proposals.filter((p) => p.author !== actor).map((p) => ({ actor, kind: "critique" as const, object: p.item! })));
}
case "response":
return freshOf(board, round).map((c) => ({
actor: board.authorsOf(c.candidate)[0]!,
kind: c.proposals.length === 1 ? "response" : "unification",
object: c.candidate,
}));
case "secondPrediction":
return each("secondPrediction", poolOf(board, round));
}
}
/**
* 基线 → 基线代码记录(对象 ORIGINAL_BASELINE),而且最近一次基线运行的结果不是"没启动"。
* 没启动的起新编号重跑;判为不算数的停下(方法.md 4.1)。
@@ -104,28 +204,34 @@ function baselineNext(view: View): Next | null {
}
/**
* 研究者阶段 → 这个阶段的关闭记录(stageClose)。没开就开;名单上的人都有被收下的提交才关;
* 没交上的人按最近一次调用怎么结束的来定派不派(任务书 12 第 2.3 节的表)。能派的先全派,派完还有人停在"没交"才停下。
* 研究者阶段 → 这个阶段的关闭记录(stageClose)。没开就开,名单是全部研究者(本轮名单就是冻结的配置);
* 每个任务(一个人对一个对象)都有被收下的提交才关;没做完的任务按它最近一次调用怎么结束的来定派不派
* (任务书 12 第 2.3 节的表)。能派的先全派,派完还有任务停在"没交"才停下。
*/
function stageNext(step: RoundZeroStep, view: View, frozen: Frozen): Next | null {
const stageId = stageIdOf(step);
const { stage, kind } = ROUND_ZERO_STAGES[step];
const open = view.of("stageOpen").find((s) => s.stageId === stageId);
function stageNext(
stageId: string,
stage: Stage,
round: number,
tasksOf: (roster: readonly string[]) => TaskRef[],
view: View,
frozen: Frozen,
): Next | null {
const open = view.board.stages.get(stageId);
if (open === undefined) {
return { do: "openStage", stageId, stage, roster: frozen.config.models.researchers.map((_, i) => researcherId(i)) };
return { do: "openStage", stageId, stage, round, roster: frozen.config.models.researchers.map((_, i) => researcherId(i)) };
}
if (view.of("stageClose").some((s) => s.stageId === stageId)) return null;
const missing = open.roster.filter((a) => !view.of("submission").some((s) => s.stageId === stageId && s.author === a));
if (missing.length === 0) return { do: "closeStage", stageId };
const actors: string[] = [];
if (open.closed) return null;
const todo = tasksOf(open.roster).filter((t) => !view.done(stageId, t));
if (todo.length === 0) return { do: "closeStage", stageId };
const tasks: TaskRef[] = [];
const stuck: Array<{ call: string; reason: string }> = [];
for (const actor of missing) {
const last = view.lastCall(stageId, actor);
for (const t of todo) {
const last = view.lastCall(stageId, t);
// 没有调用、被用户停下、出错(含恢复时补写的"进程中断")都派;没交的只在开发入口开着时派(方法.md 第九部分"开发入口")。
if (last?.status === "notSubmitted" && !frozen.config.devEntry) stuck.push({ call: last.call, reason: last.reason });
else actors.push(actor);
else tasks.push(t);
}
if (actors.length > 0) return { do: "dispatch", stageId, kind, actors };
if (tasks.length > 0) return { do: "dispatch", stageId, tasks };
return {
do: "halt",
reason: `交不出合格提交(方法.md 4.3 ②:按内部错误停下):${stuck.map((s) => `调用 ${s.call},${s.reason}`).join(";")}`,
@@ -133,12 +239,14 @@ function stageNext(step: RoundZeroStep, view: View, frozen: Frozen): Next | null
};
}
/** 按种类取记录内容的小工具。 */
/** 按种类取记录内容的小工具,加上 materials 读账本的那一份(Board)。 */
class View {
readonly #records: readonly LedgerRecord[];
readonly board: Board;
constructor(records: readonly LedgerRecord[]) {
this.#records = records;
this.board = new Board(records);
}
of<K extends LedgerRecord["kind"]>(kind: K): Array<Content<K>> {
@@ -149,9 +257,19 @@ class View {
return this.of("event").filter((e) => e.event === event);
}
/** 某人在某阶段最近一次模型调用的结束记录,没有调用为 null。开始了没结束的,恢复入口应当已经补写。 */
lastCall(stageId: string, actor: string): Content<"callEnded"> | null {
const started = this.of("callStarted").filter((c) => c.stageId === stageId && c.actor === actor).at(-1);
/** 一个任务做完了:这个人在这个阶段对这个对象有一份被收下的提交(任务书 14 第 2.3 节)。 */
done(stageId: string, t: TaskRef): boolean {
return this.of("submission").some((s) => s.stageId === stageId && s.author === t.actor && s.target === t.object);
}
/**
* 一个任务最近一次模型调用的结束记录,没有调用为 null。"模型调用开始"里存的材料带着对象,按它认出是哪个任务。
* 开始了没结束的,恢复入口应当已经补写。
*/
lastCall(stageId: string, t: TaskRef): Content<"callEnded"> | null {
const started = this.of("callStarted")
.filter((c) => c.stageId === stageId && c.actor === t.actor && (c.materials as { object: string | null }).object === t.object)
.at(-1);
if (started === undefined) return null;
const ended = this.of("callEnded").find((e) => e.call === started.call);
if (ended === undefined) throw new Error(`模型调用 ${started.call} 开始了还没有结束记录:恢复入口应当已经补写`);
+65
View File
@@ -0,0 +1,65 @@
import { Board, critiqueId, proposalId } from "../../materials/index.ts";
import { runModelTask, type ModelTask } from "../model-task.ts";
import type { Next, TaskRef } from "../next-step.ts";
import { halt, USER_STOPPED, type Study, type StudyEnding } from "../study.ts";
// 派一个研究者阶段里要派的任务(任务书 12 第 2.6 节、任务书 14 第 2.3 节):
// 同一个人的几个任务按先后跑,不同的人同时跑,各自一个会话。
/**
* 派这些任务。有任务没交上(没交或出错)、或有会话抛出内部错误后,不再开新的会话;
* 已经在跑的不打断(别人已经在花钱了),等它们全部结束后再停下或把错误抛出去(任务书 12 第 2.4 节)。
*/
export async function dispatchStage(s: Study, next: Extract<Next, { do: "dispatch" }>): Promise<StudyEnding | null> {
const byActor = new Map<string, TaskRef[]>();
for (const t of next.tasks) byActor.set(t.actor, [...(byActor.get(t.actor) ?? []), t]);
let halting = false;
const missed: Array<{ call: string; text: string }> = [];
const runActor = async (tasks: readonly TaskRef[]): Promise<void> => {
for (const t of tasks) {
if (halting || s.stop.aborted) return;
const { call, ending } = await runModelTask(s, modelTask(s, next.stageId, t));
if (ending.status === "notSubmitted" || ending.status === "failed") {
halting = true;
missed.push({ call, text: `${ending.status === "failed" ? "出错" : "没交"}:${ending.reason}` });
}
}
};
const settled = await Promise.allSettled(
[...byActor.values()].map((tasks) =>
runActor(tasks).catch((error: unknown) => {
halting = true;
throw error;
}),
),
);
const failure = settled.find((r) => r.status === "rejected");
if (failure !== undefined) throw failure.reason;
if (s.stop.aborted) return USER_STOPPED;
if (missed.length === 0) return null;
return halt(s.ledger, `阶段 ${next.stageId} 有任务没交上,等这次派发的会话全部结束后停下:${missed.map((m) => `调用 ${m.call},${m.text}`).join(";")}`, missed.map((m) => m.call));
}
/**
* 一个任务交上来之后才知道的东西和分份编号(submissions 的 KnownOf、任务书 14 第 2.3 节):
* 提案是本人本阶段已收下几条,按 proposalId 分份;挑刺按 critiqueId 分份;回应是作者那条原提案;其余没有。
*/
function modelTask(s: Study, stageId: string, t: TaskRef): ModelTask<TaskRef["kind"]> {
const base = { kind: t.kind, actor: t.actor, stageId, object: t.object };
const board = new Board(s.ledger.records());
switch (t.kind) {
case "proposal": {
const round = board.stage(stageId).round;
return { ...base, item: (n) => proposalId(round, t.actor, n), known: (_, others) => ({ accepted: others }) } as ModelTask<"proposal">;
}
case "critique":
return { ...base, item: (n) => critiqueId(t.object!, t.actor, n), known: () => null };
case "response": {
// 回应的候选只由一条原提案而来(几条合成的交统一)。
const original = board.proposal(board.proposalsOf(t.object!)[0]!).content;
return { ...base, item: null, known: () => ({ original }) } as ModelTask<"response">;
}
default:
return { ...base, item: null, known: () => null };
}
}
+2 -24
View File
@@ -1,31 +1,9 @@
import { finalEntries } from "../../materials/index.ts";
import { assignEntries as assign } from "../../rules/index.ts";
import type { Json } from "../../vocab/index.ts";
import { runModelTask } from "../model-task.ts";
import type { Next } from "../next-step.ts";
import { halt, USER_STOPPED, type Study, type StudyEnding } from "../study.ts";
import type { Study } from "../study.ts";
// 第零轮(方法.md 4.2):三个研究者阶段派任务,认领之后由程序分配入口。
/**
* 派一个研究者阶段里要派的人,同时跑,各自一个会话(任务书 12 第 2.6 节)。第零轮的提交种类交上来之后没有要另算的东西。
* 有人没交上(没交或出错),等这次派发的会话全部结束后停下,不打断别人(第 2.4 节)。
* 某个会话抛出内部错误时,也等别人结束再把它抛出去。
*/
export async function dispatchStage(s: Study, next: Extract<Next, { do: "dispatch" }>): Promise<StudyEnding | null> {
const settled = await Promise.allSettled(
next.actors.map((actor) => runModelTask(s, { kind: next.kind, actor, stageId: next.stageId, object: null, known: () => null })),
);
const failure = settled.find((r) => r.status === "rejected");
if (failure !== undefined) throw failure.reason;
if (s.stop.aborted) return USER_STOPPED;
const results = settled.map((r) => (r as PromiseFulfilledResult<Awaited<ReturnType<typeof runModelTask>>>).value);
const missed = results.flatMap(({ call, ending }) =>
ending.status === "notSubmitted" || ending.status === "failed" ? [{ call, text: `${ending.status === "failed" ? "出错" : "没交"}:${ending.reason}` }] : [],
);
if (missed.length === 0) return null;
return halt(s.ledger, `阶段 ${next.stageId} 有人没交上,等这次派发的会话全部结束后停下:${missed.map((m) => `调用 ${m.call},${m.text}`).join(";")}`, missed.map((m) => m.call));
}
// 第零轮(方法.md 4.2):三个研究者阶段由 dispatch.ts 派任务,认领之后由程序分配入口。
/**
* 分配(方法.md 4.2"分配规则"):每人最后一份认领和最终有效的入口交给 rules 的 assignEntries,写一条入口分配决定。
+138
View File
@@ -0,0 +1,138 @@
import { judgeSameChange, type DedupFactor } from "../../jev/index.ts";
import { Board, candidateId, freezeProtocol, type DedupOutput } from "../../materials/index.ts";
import { compareIds, decisionSeed, runBalance, runBatchRule, type BatchInput, type BatchRuleName } from "../../rules/index.ts";
import { protocolBody } from "../../submissions/index.ts";
import type { JevJudgment, Json } from "../../vocab/index.ts";
import { freshOf, poolOf, stageIdOf } from "../next-step.ts";
import type { Study } from "../study.ts";
// 第 r 轮里程序自己做的三步(方法.md 4.3 ③⑥⑧):判重合并、冻结协议、选批。每步写一条记录(冻结协议一个事务),
// 写下之前被打断就从账本重新算一遍;写下之后恢复时照原样用,不重算(方法.md 第九部分)。
/**
* 同一轮里几个要随机的决定各用一个"版本"算种子(方法.md 第九部分):选批和选格都用随机规则时,
* 两次抽取不会用同一串随机数。选格由 flow 第 4 部分接上。
*/
const SEED_VERSIONS = { batch: 1, cells: 2 } as const;
/** 一对提案(或提案与留池候选)的 Jev 判断,原样存进判重合并决定的输入。 */
export type SamePair = { a: string; b: string; judgment: JevJudgment };
/**
* ③ 判重合并:本轮新提案两两问一次 Jev,每条新提案再和每个留池候选问一次(留池候选看它冻结协议里的因子),
* 按顺序发。全部问完才写决定:中途被打断就整轮重问,判重不扣运行次数。
*/
export async function dedupe(s: Study, round: number): Promise<void> {
const board = new Board(s.ledger.records());
const proposals = board.published("proposal").filter((p) => p.round === round).sort((a, b) => compareIds(a.item!, b.item!));
const carried = board.carried(round);
const model = s.frozen.config.models.jev;
const factorsOf = new Map<string, readonly DedupFactor[]>([
...proposals.map((p) => [p.item!, p.content.factors] as const),
...carried.map((c) => [c, board.protocol(c).factors] as const),
]);
const asks = [
...proposals.flatMap((p, i) => proposals.slice(i + 1).map((q) => [p.item!, q.item!] as const)),
...proposals.flatMap((p) => carried.map((c) => [p.item!, c] as const)),
];
const pairs: SamePair[] = [];
for (const [a, b] of asks) pairs.push({ a, b, judgment: await judgeSameChange(model, factorsOf.get(a)!, factorsOf.get(b)!) });
const output = groupSameChange(round, proposals.map((p) => p.item!), carried, pairs);
s.ledger.append("decision", {
decision: "dedup",
round,
rule: "jevSameChange",
ruleVersion: 1,
params: { jevModel: model },
seed: null,
input: { pairs } as unknown as Json,
output: output as unknown as Json,
});
}
/**
* 判为相同的连成一组(监督者定:A 和 B 相同、B 和 C 相同,A、B、C 就是一个候选;方法没写这种情况)。
* 一组里有留池候选,整组并进它(方法.md 4.3 ③);一组连到两个留池候选按内部错误停下(真实运行里没出现过)。
* 其余每组是一个新候选,按组里编号最小的原提案排序,编号用 candidateId。每个留池候选都列出,没并进提案的是空数组。
*/
export function groupSameChange(round: number, proposals: readonly string[], carried: readonly string[], pairs: readonly SamePair[]): DedupOutput {
const parent = new Map([...proposals, ...carried].map((x) => [x, x]));
const root = (x: string): string => {
while (parent.get(x) !== x) x = parent.get(x)!;
return x;
};
for (const p of pairs) if (p.judgment.yes) parent.set(root(p.a), root(p.b));
const groups = new Map<string, string[]>();
for (const x of [...proposals, ...carried]) groups.set(root(x), [...(groups.get(root(x)) ?? []), x]);
const into = new Map(carried.map((c) => [c, [] as string[]]));
const fresh: string[][] = [];
for (const members of groups.values()) {
const old = members.filter((m) => into.has(m));
const added = members.filter((m) => !into.has(m)).sort(compareIds);
if (old.length > 1) {
throw new Error(`判重合并:提案 ${added.join("、")} 把留池候选 ${old.join("、")} 连成了一组,方法没写该并进哪一个(真实运行里没出现过),按内部错误停下`);
}
if (old.length === 1) into.get(old[0]!)!.push(...added);
else fresh.push(added);
}
fresh.sort((a, b) => compareIds(a[0]!, b[0]!));
return {
fresh: fresh.map((ids, i) => ({ candidate: candidateId(round, i + 1), proposals: ids })),
carried: [...into].sort(([a], [b]) => compareIds(a, b)).map(([candidate, ids]) => ({ candidate, proposals: ids })),
};
}
/**
* ⑥ 冻结协议:回应阶段关闭后,每个新候选取它最后一份回应或统一,用 protocolBody 拼主体,freezeProtocol 补上候选编号、
* 轮次、本批基线、合并了哪些原提案。全部候选在同一个事务里写:materials 把冻结协议的出现当作"已公开"。
* 留池候选沿用原来的冻结协议(方法.md 4.3 ①),不重写。
*/
export function freezeProtocols(s: Study, round: number): void {
const board = new Board(s.ledger.records());
const baseline = board.baselineOf(round);
s.ledger.transaction(() => {
for (const { candidate, proposals } of freshOf(board, round)) {
const answer = board.answerOf(candidate);
const body =
answer.kind === "response"
? protocolBody({ kind: "response", content: answer.content, original: board.proposal(proposals[0]!).content })
: protocolBody({ kind: "unification", content: answer.content });
s.ledger.append("protocol", { candidate, protocol: freezeProtocol(body, candidate, round, baseline, proposals) as unknown as Json });
}
});
}
/**
* ⑧ 选批(方法.md 6.4):池里每个候选、各人最后一份第二次预测,加本批基线分数、方向、ε、λ、每批上限、余额和种子,
* 交给步骤清单里 selectBatch 那一步的规则,写一条选批决定。
*/
export function selectBatch(s: Study, round: number): void {
const board = new Board(s.ledger.records());
const { config, facts, steps } = s.frozen;
const stageId = stageIdOf(round, "secondPrediction");
const predictions = board.published("secondPrediction").filter((p) => p.stageId === stageId);
const range = (researcher: string, candidate: string) => predictions.find((p) => p.author === researcher && p.target === candidate)!.content.range;
const seed = decisionSeed({ studyId: config.studyId, round, version: SEED_VERSIONS.batch });
const input: BatchInput = {
candidates: poolOf(board, round).map((id) => ({ id, ranges: board.stage(stageId).roster.map((researcher) => ({ researcher, range: range(researcher, id) })) })),
baseline: board.baselineScore(board.baselineOf(round)),
direction: facts.finalMetric.direction,
epsilon: facts.epsilon,
lambda: config.flow.openReserve,
batchCap: config.batchLimit,
balance: runBalance(config.runs, board.runStates()).balance,
seed,
};
const rule = steps.round.find((x) => x.id === "selectBatch")!.rule!.name as BatchRuleName;
const { rule: used, ...output } = runBatchRule(rule, input);
s.ledger.append("decision", {
decision: "batch",
round,
rule: used.name,
ruleVersion: used.version,
params: output.params,
seed,
input: input as unknown as Json,
output: output as unknown as Json,
});
}