复杂并行调度

// 在主线 10 的 types 模块里用 interface merging 追加字段，// 不要复制一份 PlanStep：否则 selectNextStep / reconcilePlan 会看到两个不兼容的计划类型。type StepParallelism = { reads: string[]; writes: string[]; sharedState: string[] };export interface PlanStep {  parallelism?: StepParallelism | undefined;}export function assertIndependent(stepA: PlanStep, stepB: PlanStep): void {  const adjacent = stepA.dependsOn.includes(stepB.id) || stepB.dependsOn.includes(stepA.id);  if (adjacent) throw new Error(`declared_dependency:${stepA.id}:${stepB.id}`);  const a = stepA.parallelism;  const b = stepB.parallelism;  if (!a || !b) throw new Error("step parallelism contract is missing");  for (const path of a.writes) {    if (b.writes.includes(path) || b.reads.includes(path)) throw new Error(`write_conflict:${path}`);  }  for (const path of b.writes) {    if (a.reads.includes(path)) throw new Error(`write_conflict:${path}`);  }  const shared = a.sharedState.filter((resource) => b.sharedState.includes(resource));  if (shared.length > 0) throw new Error(`shared_state:${shared[0]}`);}

export interface PlanLayer {  index: number;  stepIds: string[];}export function computeReadyLayers(plan: TaskPlan): PlanLayer[] {  const remaining = new Map(plan.steps.map((step) => [step.id, new Set(step.dependsOn)] as const));  const done = new Set(plan.steps.filter((step) => step.status === "completed").map((s) => s.id));  const layers: PlanLayer[] = [];  while (remaining.size > 0) {    const ready = [...remaining.entries()]      .filter(([, dependencies]) => [...dependencies].every((id) => done.has(id)))      .map(([id]) => id)      .sort(); // 稳定排序：同一图的层内顺序永远是字典序    if (ready.length === 0) throw new Error(`no ready step: ${[...remaining.keys()].join(",")}`);    for (const id of ready) {      remaining.delete(id);      done.add(id);    }    layers.push({ index: layers.length, stepIds: ready });  }  return layers;}

// SchedulerOptions 复用主线已有边界：plan / budget / store / lease / writeLease / trace / signalexport interface SchedulerOptions {  maxConcurrency: number;  plan: TaskPlan;  budget: AgentBudget;  store: RunStore;  lease: RunLease;  writeLease: WriteLease;  trace: TraceSink;  signal?: AbortSignal | undefined;}/** 已预占预算 = 正在运行的每个分支的估算之和；不足则不得启动。 */export function reserveBudget(  budget: AgentBudget,  reserved: Map<string, number>,  estimate: number,): boolean {  const inFlight = [...reserved.values()].reduce((sum, value) => sum + value, 0);  return estimate + inFlight <= budget.maxToolCalls - budget.toolCalls;}

export type PartialFailurePolicy = "fail_fast" | "continue_others";type BranchFailure = { stepId: string; recoverable: boolean; message: string };export async function runLayer(  layer: PlanLayer,  controller: AbortController,  options: SchedulerOptions,  runBranch: (stepId: string, signal: AbortSignal) => Promise<void>,): Promise<BranchFailure[]> {  const failures: BranchFailure[] = [];  const running = new Set<Promise<void>>();  for (const stepId of layer.stepIds) {    if (controller.signal.aborted) break;    const branch = runBranch(stepId, controller.signal).catch(async (error: unknown) => {      const failure: BranchFailure = {        stepId,        recoverable: !isFatal(error),        message: error instanceof Error ? error.message : String(error),      };      failures.push(failure);      recordBranchFailure(options, failure); // 写入 TraceSink：type=branch_failed      // 失败先落成 append-only 事件，再决定是否取消同层分支，绝不静默丢弃      await options.store.append({        runId: options.plan.goal,        expectedSequence: nextSequence(options.plan),        ownerId: options.lease.ownerId,        epoch: options.lease.epoch,        event: { type: "step_failed", stepId, message: failure.message },      });      if (!failure.recoverable) controller.abort(new Error("sibling_branch_failed"));    });    running.add(branch);  }  await Promise.allSettled(running);  return failures;}

export interface MergeKey {  layer: number;  stepId: string;}/** 归并顺序 = 层号升序，同层按 stepId 字典序；与完成时间无关。 */export function compareMergeKey(left: MergeKey, right: MergeKey): number {  return left.layer - right.layer || (left.stepId < right.stepId ? -1 : 1);}export interface BranchOutcome extends MergeKey {  evidence: string[];  events: AgentEvent[];}export function mergeLayerOutcomes(  plan: TaskPlan,  outcomes: BranchOutcome[],): { plan: TaskPlan; events: AgentEvent[] } {  const ordered = outcomes.toSorted((left, right) => compareMergeKey(left, right));  const events: AgentEvent[] = [];  let merged = plan;  for (const outcome of ordered) {    merged = updateStep(merged, outcome.stepId, (step) => ({      ...step,      status: "completed",      evidence: [...step.evidence, ...outcome.evidence],    }));    events.push(...outcome.events.toSorted((a, b) => a.sequence - b.sequence));  }  return { plan: merged, events };}

it("只把依赖已完成的步骤放进同一层", () => {  expect(computeReadyLayers(createIndependentPlan())).toEqual([    { index: 0, stepIds: ["inspect-a", "inspect-b"] },    { index: 1, stepIds: ["patch-a", "patch-b"] },    { index: 2, stepIds: ["validate"] },  ]);});it("预算不足时既不预占也不启动分支", () => {  const budget: AgentBudget = { maxSteps: 8, maxToolCalls: 3, modelSteps: 0, toolCalls: 0 };  const reserved = new Map([["inspect-a", 2], ["inspect-b", 2]]);  expect(reserveBudget(budget, reserved, 2)).toBe(false);});it("归并顺序与完成顺序无关", async () => {  const plan = createIndependentPlan();  const slowFirst = await runSchedule(plan, { settleOrder: ["patch-b", "patch-a"] });  const fastFirst = await runSchedule(plan, { settleOrder: ["patch-a", "patch-b"] });  expect(mergedEvidence(plan, slowFirst)).toEqual(mergedEvidence(plan, fastFirst));  expect(eventSequences(slowFirst)).toEqual(eventSequences(fastFirst));});