// Declarative run config — spaCy's `config.cfg` for leveling.
//
// One JSON document fully describes a leveling run: how to resolve the
// project, which blocks (by versioned registry code) with which params, and
// how scorer outputs combine into the objective. `runFromConfig` builds a
// live run from it; the run carries back the *filled* config (every default
// expanded), which is what gets persisted as provenance — reproducibility
// comes from saving the expanded config with the artifact, not from
// re-deriving defaults at load time.
//
// The block list reuses the existing PipelineGraph format rather than
// inventing a second serializable pipeline shape: the graph is the config's
// `pipeline` section, and instanceIds double as constraint instance names
// for failure attribution.

import { z } from "zod";

import { analyzeRun, type RunAnalysis } from "../level-core/analyzeRun.ts";
import { RunAnalysisError } from "../level-core/errors.ts";
import { materialize } from "../level-core/materialize.ts";
import { resolveCalendar } from "../level-core/resolveCalendar.ts";
import {
  concatStreams,
  streamFromFactory,
} from "../level-core/scheduleStream.ts";
import type {
  Collector,
  Constraint,
  Project,
  ResolvedProject,
  Schedule,
  ScheduleStream,
  Scorer,
  Search,
} from "../level-core/types.ts";
import { assertNeverCollector } from "../level-core/types.ts";
import { weightedScorer } from "../level-core/weightedScorer.ts";
import type { RunProvenance } from "../schema/provenance.ts";
import { LEVELSET_VERSION } from "../version.ts";

import {
  compileGraph,
  GraphValidationError,
  type CompileTarget,
} from "./compile.ts";
import { selectGate, type ResolvedGateLimit } from "./gate.ts";
import {
  pipelineGraphSchema,
  validateGraph,
  type GraphIssue,
  type ValidateGraphResult,
} from "./graph.ts";
import type { BlockRegistry } from "./registry.ts";
import { standardRegistry } from "./standard.ts";
import {
  expandSweepCells,
  SWEEP_ENABLED_TARGET,
  SWEEP_POINT_ANNOTATION,
  SweepError,
  sweepCombinationCount,
  sweepSchema,
  type SweepPoint,
  type SweepStats,
} from "./sweep.ts";

// WorkUnit as JSON (mirrors the WorkUnit interface in level-core/types.ts).
export const workUnitSchema = z.object({
  id: z.number().int(),
  location: z.string().optional(),
  productType: z.string().optional(),
  serial: z.string().optional(),
  taskUniqueIds: z.array(z.number().int()),
});

export const weightedSumTierSchema = z.strictObject({
  weightedSum: z
    .record(z.string(), z.number())
    .refine(
      (weights) => Object.keys(weights).length > 0,
      "weightedSum must name at least one scorer",
    )
    .refine(
      (weights) => Object.values(weights).some((weight) => weight !== 0),
      "weightedSum must contain at least one non-zero weight",
    ),
});
export const lexicographicTierSchema = z.union([
  z.string(),
  weightedSumTierSchema,
]);
export type LexicographicTier = z.infer<typeof lexicographicTierSchema>;

// Framework-owned sections are strict (a typo'd key is an error, not a
// silently-ignored one); the pipeline section keeps the graph format's own
// validation.
export const runConfigSchema = z.strictObject({
  formatVersion: z.literal(1),
  /** levelset version that filled this config; stamped by fillRunConfig. */
  levelset: z.string().optional(),
  /** Recorded for stochastic searches/transformers; serialSGS ignores it. */
  seed: z.number().int().optional(),
  resolve: z
    .strictObject({
      /** ISO date string for ResolveOptions.epoch. */
      epoch: z
        .string()
        .refine(
          (s) => !Number.isNaN(new Date(s).getTime()),
          "not a parseable date",
        )
        .optional(),
      horizonDays: z.number().int().positive().optional(),
      workUnits: z.array(workUnitSchema).optional(),
    })
    .optional(),
  pipeline: pipelineGraphSchema,
  /** Objective weights keyed by scorer instanceId. Missing = 1, null =
   *  compute and report but don't optimize on it. */
  weights: z.record(z.string(), z.number().nullable()).optional(),
  /** Tiered selection: each tier is either one scorer instanceId or a
   *  weighted sum of scorer instanceIds. Tiers compare lexicographically —
   *  the second tier only breaks ties on the first. When present, `best()`
   *  selects by these tiers and top-level `weights` remain report-only. */
  objective: z
    .strictObject({
      lexicographic: z.array(lexicographicTierSchema).min(1),
    })
    .optional(),
  /** Bounded parameter grid evaluated above the unchanged flat pipeline. */
  sweep: sweepSchema.optional(),
});
export type RunConfig = z.infer<typeof runConfigSchema>;

export interface WeightedScorerEntry {
  readonly instanceId: string;
  readonly scorer: Scorer;
  readonly weight: number | null;
}

/** One collector block's output, resolved and executed against the search
 *  stream. Labeled by the graph `instanceId` so callers can tell two
 *  collectors of the same kind apart. */
export interface CollectorResult {
  readonly instanceId: string;
  readonly kind: Collector["kind"];
  readonly schedules: readonly Schedule[];
}

/** A compiled, enabled collector paired with its graph instanceId — the
 *  spec still references scorers by instanceId; `collectors()` binds them. */
interface CompiledCollector {
  readonly instanceId: string;
  readonly collector: Collector;
}

export interface LevelingRun {
  /** The filled config: every block's params expanded through its schema
   *  defaults, levelset version stamped. Persist this, not the input. */
  readonly config: RunConfig;
  readonly resolved: ResolvedProject;
  /** Static analysis result (warnings only — errors throw in runFromConfig). */
  readonly analysis: RunAnalysis;
  readonly constraints: readonly Constraint[];
  readonly scorers: readonly WeightedScorerEntry[];
  readonly search: Search;
  /** The combined weighted objective (falls back to makespan when the
   *  config declares no weighted scorers). */
  readonly objective: Scorer;
  /** Lazy stream of candidate schedules from the configured search. */
  stream(): ScheduleStream;
  /** Drain the stream and keep the best schedule by the objective. */
  best(): Promise<Schedule | null>;
  /** Run every enabled collector block against a fresh search stream,
   *  returning each one's selected schedule set, each result labeled by its
   *  graph instanceId. Empty when the pipeline declares no collectors. */
  collectors(): Promise<readonly CollectorResult[]>;
  /** Every scorer's raw value, keyed by instanceId — the reported view. */
  scores(schedule: Schedule): Record<string, number>;
  /** Write the schedule back onto the source project, embedding run
   *  provenance (the filled config, engine version, achieved scores,
   *  warnings, and search annotations) as the file's `levelset` block. */
  materialize(schedule: Schedule): Project;
}

/**
 * Referential-integrity checks over a *validated* graph: every weights key,
 * every `objective.lexicographic` tier, and every collector's scorer
 * reference (`KBest.by` / `Pareto.over` / `Gate.limits[].scorer`) must name a
 * scorer-family block instance. Returns the issues (empty when clean) rather
 * than throwing, so both `fillRunConfig` and `parseConf` can surface them the
 * same way and never drift. `validated` must be the successful result of
 * `validateGraph(registry, config.pipeline)` — its `parsedParams` are what the
 * collector `apply()` reads.
 */
export function checkRunConfigRefs(
  config: RunConfig,
  registry: BlockRegistry,
  validated: Extract<ValidateGraphResult, { ok: true }>,
): GraphIssue[] {
  const scorerIds = new Set(
    validated.graph.blocks
      .filter((b) => registry.get(b.code)?.family === "scorer")
      .map((b) => b.instanceId),
  );
  // Collector blocks reference scorers by instanceId (KBest.by / Pareto.over);
  // like weights and lexicographic tiers those refs must land on scorer-family
  // instances. Read the refs off the *compiled* Collector spec (via apply)
  // rather than duck-typing the parsed params: the closed Collector ADT
  // guarantees by/over are string[], so a custom collector that names its refs
  // under a different param still gets layer-1 validation and a non-array
  // by/over can't leak through as a raw TypeError. apply() is pure and
  // compileGraph re-applies it later, so calling it here costs nothing.
  const compileIssues: GraphIssue[] = [];
  const collectorRefs = validated.graph.blocks
    .filter((b) => registry.get(b.code)?.family === "collector")
    .flatMap((b) => {
      const def = registry.get(b.code)!;
      const params = validated.parsedParams.get(b.instanceId);
      let collector: Collector;
      try {
        collector = (def.block as { apply(input: unknown): Collector }).apply(
          params,
        );
      } catch (err) {
        compileIssues.push({
          instanceId: b.instanceId,
          code: b.code,
          message: `collector "${b.instanceId}" failed to compile its spec: ${
            err instanceof Error ? err.message : String(err)
          }`,
        });
        return [];
      }
      return collectorScorerRefs(collector).map((k) => ({
        k,
        where: `collector "${b.instanceId}" scorer reference`,
      }));
    });
  if (compileIssues.length > 0) return compileIssues;

  const badRefs = [
    ...Object.keys(config.weights ?? {}).map((k) => ({
      k,
      where: "weights key",
    })),
    ...(config.objective?.lexicographic ?? []).flatMap((tier) =>
      lexicographicTierRefs(tier).map((k) => ({
        k,
        where: "objective.lexicographic entry",
      })),
    ),
    ...collectorRefs,
  ].filter(({ k }) => !scorerIds.has(k));
  return badRefs.map(({ k, where }) => ({
    instanceId: k,
    message: `${where} "${k}" does not reference a scorer-family block instance`,
  }));
}

/** Validate a config and expand it: block params parsed through each
 *  block's input schema (defaults applied), levelset version stamped.
 *  Throws GraphValidationError / ZodError on invalid input. */
export function fillRunConfig(
  config: unknown,
  registry?: BlockRegistry,
): RunConfig {
  const reg = registry ?? standardRegistry();
  const parsed = runConfigSchema.parse(config);
  const validated = validateGraph(reg, parsed.pipeline);
  if (!validated.ok) throw new GraphValidationError(validated.issues);

  const refIssues = checkRunConfigRefs(parsed, reg, validated);
  if (refIssues.length > 0) throw new GraphValidationError(refIssues);

  const filled: RunConfig = {
    ...parsed,
    levelset: LEVELSET_VERSION,
    pipeline: {
      ...validated.graph,
      blocks: validated.graph.blocks.map((instance) => ({
        ...instance,
        params: validated.parsedParams.get(instance.instanceId) as Record<
          string,
          unknown
        >,
      })),
    },
  };
  const sweepIssues = validateSweepConfig(filled, reg, validated);
  if (sweepIssues.length > 0) throw new GraphValidationError(sweepIssues);
  return filled;
}

/** Cross-field sweep validation that depends on the filled graph and block schemas. */
export function validateSweepConfig(
  config: RunConfig,
  registry: BlockRegistry,
  validated: Extract<ValidateGraphResult, { ok: true }>,
): GraphIssue[] {
  if (!config.sweep) return [];
  const issues: GraphIssue[] = [];
  const instanceById = new Map(
    config.pipeline.blocks.map((block) => [block.instanceId, block]),
  );
  const axisIds = new Set<string>();

  for (const axis of config.sweep.axes) {
    if (axisIds.has(axis.id)) {
      issues.push({ message: `sweep axis id "${axis.id}" is duplicated` });
    }
    axisIds.add(axis.id);

    const keyedTargets = axis.targets.filter(
      (target) => target.valueKey !== undefined,
    );
    const isZip = keyedTargets.length === axis.targets.length;
    if (keyedTargets.length > 0 && !isZip) {
      issues.push({
        message: `sweep axis "${axis.id}" mixes keyed and unkeyed targets`,
      });
      continue;
    }

    const referencedKeys = new Set(
      keyedTargets.map((target) => target.valueKey!),
    );
    if (isZip) {
      axis.values.forEach((point, pointIndex) => {
        if (
          point === null ||
          Array.isArray(point) ||
          typeof point !== "object"
        ) {
          issues.push({
            message: `sweep axis "${axis.id}" zip point ${pointIndex} must be a record`,
          });
          return;
        }
        const pointKeys = Object.keys(point);
        for (const key of referencedKeys) {
          if (!Object.hasOwn(point, key)) {
            issues.push({
              message: `sweep axis "${axis.id}" zip point ${pointIndex} is missing valueKey "${key}"`,
            });
          }
        }
        for (const key of pointKeys) {
          if (!referencedKeys.has(key)) {
            issues.push({
              message: `sweep axis "${axis.id}" zip point ${pointIndex} has unreferenced key "${key}"`,
            });
          }
        }
      });
    }

    for (const target of axis.targets) {
      const instance = instanceById.get(target.instanceId);
      if (!instance) {
        issues.push({
          instanceId: target.instanceId,
          message: `sweep axis "${axis.id}" targets unknown instance "${target.instanceId}"`,
        });
        continue;
      }

      const parsedParams = validated.parsedParams.get(instance.instanceId) as
        | Record<string, unknown>
        | undefined;
      if (
        target.paramPath !== SWEEP_ENABLED_TARGET &&
        resolveSweepDotPath(parsedParams, target.paramPath) === undefined
      ) {
        issues.push({
          instanceId: target.instanceId,
          code: instance.code,
          message: `sweep axis "${axis.id}" paramPath "${target.paramPath}" does not resolve in instance "${target.instanceId}"`,
        });
        continue;
      }

      const def = registry.get(instance.code)!;
      if (def.family !== "constraint" && def.family !== "scorer") {
        issues.push({
          instanceId: target.instanceId,
          code: instance.code,
          message: `sweep axis "${axis.id}" targets ${def.family} instance "${target.instanceId}"; only constraint and scorer instances are sweepable`,
        });
        continue;
      }
      if (
        def.family === "scorer" &&
        target.paramPath === SWEEP_ENABLED_TARGET
      ) {
        issues.push({
          instanceId: target.instanceId,
          code: instance.code,
          message: `sweep axis "${axis.id}" cannot toggle scorer instance "${target.instanceId}"; scorer membership must stay fixed across cells`,
        });
        continue;
      }
      axis.values.forEach((point, pointIndex) => {
        if (
          isZip &&
          (point === null || Array.isArray(point) || typeof point !== "object")
        )
          return;
        const value =
          target.valueKey === undefined
            ? point
            : (point as Record<string, unknown>)[target.valueKey];
        const valueLabel =
          target.valueKey === undefined ? "" : ` valueKey "${target.valueKey}"`;
        if (target.paramPath === SWEEP_ENABLED_TARGET) {
          if (typeof value !== "boolean") {
            issues.push({
              instanceId: target.instanceId,
              code: instance.code,
              message: `sweep axis "${axis.id}"${valueLabel} point ${pointIndex} must deliver a boolean to $enabled`,
            });
          }
          return;
        }

        const candidate = structuredClone(parsedParams!);
        setSweepDotPath(candidate, target.paramPath, structuredClone(value));
        const result = def.block.schema.input.safeParse(candidate);
        if (!result.success) {
          issues.push({
            instanceId: target.instanceId,
            code: instance.code,
            path: target.paramPath.split("."),
            message: `sweep axis "${axis.id}"${valueLabel} point ${pointIndex} delivers an invalid value to "${target.instanceId}.${target.paramPath}": ${result.error.issues.map((issue) => issue.message).join("; ")}`,
          });
        }
      });
    }
  }

  const combinations = sweepCombinationCount(config.sweep);
  if (combinations > config.sweep.maxCombinations) {
    issues.push({
      message: `sweep has ${combinations} combinations, exceeding maxCombinations ${config.sweep.maxCombinations}`,
    });
  }
  return issues;
}

function resolveSweepDotPath(value: unknown, path: string): unknown {
  let current = value;
  for (const segment of path.split(".")) {
    if (current === null || typeof current !== "object") return undefined;
    current = (current as Record<string, unknown>)[segment];
  }
  return current;
}

function setSweepDotPath(
  target: Record<string, unknown>,
  path: string,
  value: unknown,
): void {
  const segments = path.split(".");
  let current = target;
  for (let i = 0; i < segments.length - 1; i++) {
    current = current[segments[i]!] as Record<string, unknown>;
  }
  current[segments.at(-1)!] = value;
}

interface CompiledRun {
  constraints: Constraint[];
  scorers: WeightedScorerEntry[];
  search: Search;
  collectors: CompiledCollector[];
}

function runTarget(config: RunConfig): CompileTarget<CompiledRun> {
  return {
    fromFamilies(families): CompiledRun {
      const constraints: Constraint[] = [];
      for (const compiled of families.constraints) {
        const emitted = Array.isArray(compiled.output)
          ? (compiled.output as Constraint[])
          : [compiled.output as Constraint];
        emitted.forEach((constraint, index) => {
          constraints.push({
            ...constraint,
            name:
              constraint.name ??
              (emitted.length === 1
                ? compiled.instance.instanceId
                : `${compiled.instance.instanceId}[${String(index)}]`),
            enabled: compiled.instance.enabled,
          });
        });
      }

      const scorers: WeightedScorerEntry[] = [];
      const weights = config.weights ?? {};
      for (const compiled of families.scorers) {
        if (compiled.instance.enabled === false) continue;
        const id = compiled.instance.instanceId;
        scorers.push({
          instanceId: id,
          scorer: compiled.output as Scorer,
          // `null` is a deliberate value (report, don't optimize) — only a
          // genuinely absent key falls back to weight 1.
          weight: Object.hasOwn(weights, id)
            ? (weights[id] as number | null)
            : 1,
        });
      }

      if (families.search.instance.enabled === false) {
        throw new GraphValidationError([
          {
            instanceId: families.search.instance.instanceId,
            message:
              "the search block cannot be disabled — a run needs exactly one active search",
          },
        ]);
      }
      const search = families.search.output as Search;
      if (typeof search.run !== "function") {
        throw new GraphValidationError([
          {
            instanceId: families.search.instance.instanceId,
            code: families.search.instance.code,
            message:
              "search block did not produce a runnable Search (SearchTransformers need an inner search and are not yet supported in run configs)",
          },
        ]);
      }

      // Disabled collectors stay in the graph (what-if toggling) but don't
      // run — same present-but-inactive lever as scorers.
      const collectors: CompiledCollector[] = [];
      for (const compiled of families.collectors) {
        if (compiled.instance.enabled === false) continue;
        collectors.push({
          instanceId: compiled.instance.instanceId,
          collector: compiled.output as Collector,
        });
      }

      return { constraints, scorers, search, collectors };
    },
  };
}

export interface RunFromConfigOptions {
  readonly registry?: BlockRegistry | undefined;
}

/**
 * Build a live leveling run from a declarative config: fill the config,
 * resolve the project, compile the blocks, and statically analyze the run.
 * Throws RunAnalysisError when analysis finds errors (unknown references,
 * infeasible deadline/release pairs, cycles); analysis warnings ride on
 * `run.analysis` and into provenance.
 */
export function runFromConfig(
  project: Project,
  config: unknown,
  opts?: RunFromConfigOptions,
): LevelingRun {
  const registry = opts?.registry ?? standardRegistry();
  const filled = fillRunConfig(config, registry);

  const resolved = resolveCalendar(project, {
    ...(filled.resolve?.epoch !== undefined
      ? { epoch: new Date(filled.resolve.epoch) }
      : {}),
    ...(filled.resolve?.horizonDays !== undefined
      ? { horizonDays: filled.resolve.horizonDays }
      : {}),
    workUnits: filled.resolve?.workUnits,
  });

  const { constraints, scorers, search, collectors } = compileGraph(
    registry,
    filled.pipeline,
    runTarget(filled),
  );

  // fillRunConfig checked lexicographic ids reference scorer blocks; they
  // must also be *active* (a disabled scorer can't drive selection).
  const activeScorerIds = new Set(scorers.map((s) => s.instanceId));
  const inactiveLex = (filled.objective?.lexicographic ?? [])
    .flatMap(lexicographicTierRefs)
    .filter((id) => !activeScorerIds.has(id));
  if (inactiveLex.length > 0) {
    throw new GraphValidationError(
      inactiveLex.map((id) => ({
        instanceId: id,
        message: `objective.lexicographic entry "${id}" references a disabled scorer instance`,
      })),
    );
  }

  // Same active-scorer requirement for collector references: fillRunConfig
  // proved they point at scorer instances; an *enabled* collector can only
  // select against *enabled* scorers. Distinguish disabled from truly absent:
  // the ref exists in the full scorer set (compiled away by `enabled: false`)
  // versus never a scorer instance at all.
  const allScorerIds = new Set(
    filled.pipeline.blocks
      .filter((b) => registry.get(b.code)?.family === "scorer")
      .map((b) => b.instanceId),
  );
  const inactiveCollectorRefs = collectors.flatMap((c) =>
    collectorScorerRefs(c.collector)
      .filter((id) => !activeScorerIds.has(id))
      .map((id) => ({ instanceId: c.instanceId, id })),
  );
  if (inactiveCollectorRefs.length > 0) {
    throw new GraphValidationError(
      inactiveCollectorRefs.map(({ instanceId, id }) => ({
        instanceId,
        message: allScorerIds.has(id)
          ? `collector "${instanceId}" references a disabled scorer instance "${id}"`
          : `collector "${instanceId}" references scorer instance "${id}", which does not exist`,
      })),
    );
  }

  const analysis = analyzeRun(resolved, constraints, {
    scorers: scorers.map((s) => s.scorer),
    search,
  });
  if (!analysis.ok) throw new RunAnalysisError(analysis.issues);

  // The base graph is analyzed once. Sweep cells only substitute schema-checked
  // parameter values, then compile through the same exactly-one-search path.
  const sweepCells = filled.sweep
    ? expandSweepCells(filled).map((cell) => ({
        point: cell.point,
        compiled: compileGraph(
          registry,
          cell.config.pipeline,
          runTarget(cell.config),
        ),
      }))
    : [];
  const sweepStatsBySchedule = new WeakMap<Schedule, SweepStats>();
  const sweepScorersBySchedule = new WeakMap<
    Schedule,
    ReadonlyMap<string, Scorer>
  >();
  // Named rather than inlined into the .map() call: oxfmt 0.59 and 0.63
  // disagree on whether to hug an arrow body against `.map(`, and this is the
  // only construct in the repo where they differ. A downstream consumer vendors
  // this library and pins the older one.
  const withSweepScorer = (
    entry: WeightedScorerEntry,
  ): WeightedScorerEntry => ({
    ...entry,
    scorer: {
      ...entry.scorer,
      score(schedule: Schedule): number {
        return (
          sweepScorersBySchedule.get(schedule)?.get(entry.instanceId) ??
          entry.scorer
        ).score(schedule);
      },
    },
  });
  const runScorers = filled.sweep ? scorers.map(withSweepScorer) : scorers;

  const objective = runScorers.some((s) => s.weight !== null && s.weight !== 0)
    ? weightedScorer(runScorers, "objective")
    : {
        name: "Makespan",
        direction: "min" as const,
        score: (s: Schedule) => s.makespan,
      };

  const run: LevelingRun = {
    config: filled,
    resolved,
    analysis,
    constraints,
    scorers: runScorers,
    search,
    objective,
    stream(): ScheduleStream {
      if (filled.sweep) {
        const stats: SweepStats = {
          combinations: sweepCells.length,
          feasibleCells: 0,
          skipped: [],
        };
        const factories = sweepCells.map(
          ({ point, compiled }) =>
            () =>
              streamFromFactory(() =>
                runSweepCell(
                  compiled.search,
                  resolved,
                  compiled.constraints,
                  point,
                  stats,
                  sweepStatsBySchedule,
                  compiled.scorers,
                  sweepScorersBySchedule,
                ),
              ),
        );
        const merged = concatStreams(factories);
        return streamFromFactory(async function* () {
          for await (const schedule of merged) yield schedule;
          if (stats.feasibleCells === 0) {
            throw new SweepError(
              `no feasible sweep cell among ${stats.combinations} combinations — loosen the grid or scheduling windows`,
            );
          }
        });
      }
      return streamFromFactory(() =>
        stripReturn(search.run(resolved, constraints)),
      );
    },
    async best(): Promise<Schedule | null> {
      const lex = filled.objective?.lexicographic;
      if (lex) {
        const byId = new Map(runScorers.map((s) => [s.instanceId, s.scorer]));
        return run
          .stream()
          .bestByLex(lex.map((tier) => compileLexicographicTier(tier, byId)));
      }
      return run.stream().bestBy(objective);
    },
    async collectors(): Promise<readonly CollectorResult[]> {
      const byId = new Map(runScorers.map((s) => [s.instanceId, s.scorer]));
      // Each collector drains its own fresh stream (stream() rebuilds the
      // generator), so they run independently.
      return Promise.all(
        collectors.map(async (c): Promise<CollectorResult> => {
          let schedules: readonly Schedule[];
          switch (c.collector.kind) {
            case "KBest":
              schedules = await run.stream().kBestByLex(
                c.collector.by.map((id) => byId.get(id)!),
                c.collector.k,
              );
              break;
            case "Pareto":
              schedules = await run
                .stream()
                .paretoFrontier(c.collector.over.map((id) => byId.get(id)!));
              break;
            case "Gate": {
              // Gate selects over a materialized set (its limits carry bounds,
              // not just scorers), so collect once, then filter/fall back.
              const limits: ResolvedGateLimit[] = c.collector.limits.map(
                (l) => ({
                  scorer: byId.get(l.scorer)!,
                  max: l.max,
                  min: l.min,
                }),
              );
              schedules = selectGate(
                await run.stream().collect(),
                limits,
                c.collector.fallback,
              );
              break;
            }
            default:
              assertNeverCollector(c.collector);
          }
          return {
            instanceId: c.instanceId,
            kind: c.collector.kind,
            schedules,
          };
        }),
      );
    },
    scores(schedule: Schedule): Record<string, number> {
      const report: Record<string, number> = {};
      for (const entry of runScorers)
        report[entry.instanceId] = entry.scorer.score(schedule);
      return report;
    },
    materialize(schedule: Schedule): Project {
      const sweepStats = sweepStatsBySchedule.get(schedule);
      const winner = schedule.annotations.get(SWEEP_POINT_ANNOTATION);
      const provenance: RunProvenance = {
        levelsetVersion: LEVELSET_VERSION,
        generatedAt: new Date().toISOString(),
        search: search.name,
        runConfig: filled,
        ...(filled.seed !== undefined ? { seed: filled.seed } : {}),
        scores: run.scores(schedule),
        ...(analysis.issues.length > 0
          ? { warnings: analysis.issues.map((issue) => ({ ...issue })) }
          : {}),
        ...(schedule.annotations.size > 0
          ? { annotations: Object.fromEntries(schedule.annotations) }
          : {}),
        ...(filled.sweep && sweepStats && isSweepPoint(winner)
          ? {
              sweep: {
                combinations: sweepStats.combinations,
                feasibleCells: sweepStats.feasibleCells,
                winner,
                skipped: sweepStats.skipped.map(({ point, reason }) => ({
                  point,
                  reason,
                })),
              },
            }
          : {}),
      };
      return { ...materialize(schedule), levelset: provenance };
    },
  };
  return run;
}

/** The scorer instanceIds a collector references, whatever its kind. */
function collectorScorerRefs(collector: Collector): readonly string[] {
  switch (collector.kind) {
    case "KBest":
      return collector.by;
    case "Pareto":
      return collector.over;
    case "Gate":
      return collector.limits.map((l) => l.scorer);
    default:
      return assertNeverCollector(collector);
  }
}

function lexicographicTierRefs(tier: LexicographicTier): string[] {
  return typeof tier === "string" ? [tier] : Object.keys(tier.weightedSum);
}

function compileLexicographicTier(
  tier: LexicographicTier,
  scorersById: ReadonlyMap<string, Scorer>,
): Scorer {
  if (typeof tier === "string") return scorersById.get(tier)!;
  return weightedScorer(
    Object.entries(tier.weightedSum).map(([instanceId, weight]) => ({
      scorer: scorersById.get(instanceId)!,
      weight,
    })),
    `weightedSum(${Object.keys(tier.weightedSum).join(",")})`,
  );
}

// Search.run returns a Failure as the generator's return value;
// streamFromFactory expects a plain AsyncGenerator<Schedule>. Re-yield to
// drop the return value (the stream just ends, matching ScheduleStream's
// documented contract).
async function* stripReturn(
  gen: AsyncGenerator<Schedule, unknown>,
): AsyncGenerator<Schedule> {
  for await (const schedule of gen) yield schedule;
}

async function* runSweepCell(
  search: Search,
  resolved: ResolvedProject,
  constraints: readonly Constraint[],
  point: SweepPoint,
  stats: SweepStats,
  statsBySchedule: WeakMap<Schedule, SweepStats>,
  cellScorers: readonly WeightedScorerEntry[],
  scorersBySchedule: WeakMap<Schedule, ReadonlyMap<string, Scorer>>,
): AsyncGenerator<Schedule> {
  let emitted = false;
  try {
    for await (const schedule of search.run(resolved, constraints)) {
      if (!emitted) {
        emitted = true;
        stats.feasibleCells++;
      }
      const annotations = new Map(schedule.annotations);
      annotations.set(SWEEP_POINT_ANNOTATION, point);
      const tagged: Schedule = { ...schedule, annotations };
      statsBySchedule.set(tagged, stats);
      scorersBySchedule.set(
        tagged,
        new Map(cellScorers.map((entry) => [entry.instanceId, entry.scorer])),
      );
      yield tagged;
    }
    if (!emitted) stats.skipped.push({ point, reason: "infeasible" });
  } catch (error) {
    if (emitted) throw error;
    stats.skipped.push({
      point,
      reason: `failed: ${error instanceof Error ? error.message : String(error)}`,
    });
  }
}

function isSweepPoint(value: unknown): value is SweepPoint {
  return value !== null && typeof value === "object" && !Array.isArray(value);
}