/**
 * # session/harness/plan-fixture — freeze one authored Plan once, fan it out to N watchers (ADR-0032 §7)
 *
 * ADR-0032 §7 describes — in prose only — the paradigm the cascade exists to serve: a Fable strategist
 * authors the day's Plan+Brief+Mandate+View **once** (the few expensive frontier judgments), that authored
 * frame is **content-hashed and frozen** (`plan_fixture_sha`), and then **dozens/hundreds of small/fast
 * watchers** run against the byte-identical frozen plan+frame. This module is that paradigm in code, with
 * ZERO determinism-core change — it composes primitives that already exist:
 *
 *   - **freeze-once** — {@link capturePlanFixture} snapshots the strategist's OPEN turn
 *     ({@link CapturedTurns}[{@link OPEN_WAKE_KEY}], the armed Plan) plus its {@link Mandate}/{@link Brief}/View
 *     context, canonicalizes it (sorted keys, dropped `undefined` — the SAME discipline as the receipt/Blotter
 *     content-address) and content-hashes it via {@link sha256} → a stable {@link PlanFixture.planFixtureSha}.
 *   - **replay-many, zero model calls** — {@link recordedStrategistOf} rehydrates a fixture into a strategist
 *     tier backed by the EXISTING {@link recordedAgent}: the frozen OPEN turn is re-emitted byte-identically on
 *     every run with no provider call (ADR-0032 §6, "recordedAgent replays the cascade byte-identically").
 *   - **N-watcher fan-out** — {@link fanWatchers} builds one {@link cascadeAgent} per watcher config over the
 *     recorded strategist and the SAME frozen frame (tape+wakes+fill+mandate), runs each through the EXISTING
 *     `runSimulateSession`, and collects grade + {@link attributeCascade attribution} into per-ConfigId grid
 *     columns via the EXISTING {@link buildGrid}. One authored plan → N deterministic columns.
 *
 * ## The determinism claim the sha buys
 * `plan_fixture_sha` is a total function of the fixture's DATA (the OPEN turn + mandate + brief + view + the
 * strategist config), never its construction order or the wall clock. So: the SAME authored plan reloads to
 * the SAME sha (a cache hit is a byte-identity, not a re-author); ANY changed byte in the frozen plan yields a
 * NEW sha (no silent contamination — a stale/edited fixture can never masquerade as the frozen one). The
 * loader ({@link readPlanFixture}) re-derives the sha on read and refuses a fixture whose stored sha does not
 * match its content (fail-closed, RUNTIME §8).
 *
 * ## Scope (owner v1)
 * The seed fixture is authored from a KNOWN-VALID Plan document (the `@ spot` live-authoring grammar bug is
 * OUT OF SCOPE — see bd issues), so the fan-out is exercised independently of live authoring. A rehydrated
 * strategist replays only its OPEN turn; it captured no re-brief, so on a watcher escalation the fan runs in
 * `frozenStrategist` mode ({@link cascadeAgent}): the over-reach is REFUSED and the frozen plan is HELD ARMED
 * (fail-closed to a PASS) rather than torn down, so the fan measures watcher management against a live armed
 * plan instead of collapsing to $0 (docs/results/author-and-fan/, PR #8). This is confined to the recorded
 * fan — the LIVE cascade still disarms/re-briefs exactly as before, and the admission Gate is untouched.
 */

import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
import { dirname } from "node:path";

import {
  OPEN_WAKE_KEY,
  recordedAgent,
  fixedPlanAgent,
  BACKTEST_CONFIG,
  type Agent,
  type AgentConfig,
  type AgentTurn,
  type CapturedTurn,
  type CapturedTurns,
  type SupersedeAction,
  type WakeKey,
} from "../agent.ts";
import { runSimulateSession, type SimulateResult } from "../simulate.ts";
import { buildGrid, type Grid } from "../grid.ts";
import { cellConfigId } from "../config.ts";
import { cascadeAgent, cascadeConfigOf, attributeCascade, type Attribution, type CascadeStep } from "./cascade.ts";
import { sha256 } from "../../crypto/sha256.ts";
import { canonicalizeJson } from "../../canonical/json.ts";
import { assertGradedMandate } from "../../frame/types.ts";
import { assertBriefHashHonest } from "../../frame/brief.ts";
import type { Blotter } from "../../blotter/index.ts";
import type { BusEvent } from "../../bus/types.ts";
import type { Brief, Mandate } from "../../frame/types.ts";
import type { FillModelName } from "../sim.ts";
import type { TimeOfDay } from "../day.ts";

// ─────────────────────────────────────────────────────────────────────────────
// The frozen plan fixture — freeze-once (ADR-0032 §7)
// ─────────────────────────────────────────────────────────────────────────────

/** The DATA that is frozen and content-hashed — everything the fan-out replays, WITHOUT the sha itself.
 * `openTurn` is the strategist's captured OPEN turn ({@link CapturedTurns}[{@link OPEN_WAKE_KEY}] — the armed
 * Plan/journal); `config` is the strategist's advertised {@link AgentConfig} (the tier identity a rehydrated
 * {@link recordedAgent} carries); `mandate`/`brief`/`view` are the frame context authored alongside the Plan
 * (ADR-0026 — the hard Mandate, the soft Brief, the View lens). Absent optionals are dropped by the
 * canonicalizer, so a fixture that carries no Brief hashes identically whether the field is absent or omitted. */
export interface PlanFixtureContent {
  /** The strategist's captured OPEN turn — the armed Plan (a `supersede` document) + its journal. */
  readonly openTurn: AgentTurn;
  /** The strategist tier's advertised config — the identity a rehydrated {@link recordedAgent} replays under. */
  readonly config: AgentConfig;
  /** The hard {@link Mandate} authored for the frame (ADR-0026). */
  readonly mandate?: Mandate;
  /** The soft {@link Brief} authored for the frame (ADR-0026). */
  readonly brief?: Brief;
  /** The View lens document the strategist authored the plan under (a Kestrel `VIEW` statement), when forced. */
  readonly view?: string;
}

/** A frozen, content-addressed plan fixture: the {@link PlanFixtureContent} plus its stable
 * `planFixtureSha` (the ADR-0032 §7 `plan_fixture_sha`). A cache KEY — the same authored plan always mints
 * the same sha; any changed byte mints a new one. */
export interface PlanFixture extends PlanFixtureContent {
  /** `sha256` over the canonical JSON of the {@link PlanFixtureContent} (ADR-0032 §7 `plan_fixture_sha`). */
  readonly planFixtureSha: string;
}

/** The exact bytes {@link planFixtureSha} hashes — the ONE {@link canonicalizeJson canonical byte order}
 * (canonical/json: sorted keys, dropped `undefined`, identical to the receipt / Blotter canonicalizer) over
 * ONLY the frozen content (never the sha), so a fixture's bytes depend only on its DATA. */
function canonicalContent(content: PlanFixtureContent): string {
  return canonicalizeJson({ openTurn: content.openTurn, config: content.config, mandate: content.mandate, brief: content.brief, view: content.view });
}

/**
 * The ADR-0032 §7 `plan_fixture_sha` — `sha256` over the canonical JSON of the frozen {@link
 * PlanFixtureContent}. PURE and order-insensitive: the same authored plan (same OPEN turn + mandate + brief +
 * view + config) in ANY key order folds to the SAME 64-hex sha; ANY changed byte yields a different sha. No
 * wall clock, no RNG.
 */
export function planFixtureSha(content: PlanFixtureContent): string {
  return sha256(canonicalContent(content));
}

/**
 * Freeze an authored plan: bind the {@link PlanFixtureContent} to its {@link planFixtureSha}.
 *
 * ## Fail-closed at the HASH-COMMIT boundary (kestrel-voy9)
 * `plan_fixture_sha` is the content-address that binds "this performance came from this thesis" into
 * grade provenance (ADR-0032 §7), and the deterministic grade driver ({@link runSimulateSession})
 * never renders — so honesty is enforced HERE, where the mandate + brief COMMIT, not at a render seam
 * the grade never reaches:
 *   - {@link assertGradedMandate} — a frozen graded plan MUST declare a complete {@link Mandate}
 *     (objective + `1R=$X>0` + success criterion + bounded-risk rule); a missing/incomplete mandate
 *     can never mint a stable sha.
 *   - {@link assertBriefHashHonest} — a supplied Brief whose `hash` LIES about its `text` (a raw
 *     `{text,hash}` literal that bypassed {@link makeBrief}) can never mint a sha that binds the wrong
 *     thesis; the mismatch is refused before the content is hashed.
 * Pure; deterministic. Throws {@link KernelHonestyError} on either violation.
 */
export function capturePlanFixture(content: PlanFixtureContent): PlanFixture {
  assertGradedMandate(content.mandate);
  if (content.brief !== undefined) assertBriefHashHonest(content.brief);
  return { ...content, planFixtureSha: planFixtureSha(content) };
}

/** Additional frame context to fold into a fixture captured from a run (the {@link Mandate}/{@link Brief}/View
 * the run was authored under — a {@link SimulateResult} does not carry them back). */
export interface PlanFixtureFromRunOptions {
  readonly mandate?: Mandate;
  readonly brief?: Brief;
  readonly view?: string;
}

/**
 * Freeze the strategist's OPEN turn out of a completed {@link SimulateResult} — the freeze-once step after a
 * ONE-TIME authoring run. Reads {@link SimulateResult.captured}[{@link OPEN_WAKE_KEY}] (the armed Plan) and
 * {@link SimulateResult.config} (the strategist identity), folds in the supplied frame context, and
 * content-hashes. Fail-closed: a result with no captured OPEN turn throws (a strategist that stood down at
 * open authored no plan to freeze).
 */
export function planFixtureFromRun(res: SimulateResult, opts: PlanFixtureFromRunOptions = {}): PlanFixture {
  // The fixture freezes the TURN alone (never the capture's deliberation-cost wrapper, ADR-0040):
  // `planFixtureSha` is a content address of the authored plan, so a clocked authoring run and a
  // latency-blind one freeze byte-identical fixtures for the same authored document.
  const openTurn = res.captured.get(OPEN_WAKE_KEY)?.turn;
  if (openTurn === undefined) {
    throw new Error(`planFixtureFromRun: run captured no OPEN turn (key ${OPEN_WAKE_KEY}) — nothing to freeze`);
  }
  return capturePlanFixture({
    openTurn,
    config: res.config,
    ...(opts.mandate !== undefined ? { mandate: opts.mandate } : {}),
    ...(opts.brief !== undefined ? { brief: opts.brief } : {}),
    ...(opts.view !== undefined ? { view: opts.view } : {}),
  });
}

/** Pull the frozen Plan SOURCE document out of a fixture's OPEN turn (the `supersede` document), or `null`
 * when the OPEN turn armed nothing (a stand-down). The frozen-Plan counterfactual arm ({@link fanWatchers})
 * replays THIS document alone through {@link fixedPlanAgent}. */
export function fixturePlanDoc(fixture: PlanFixture): string | null {
  const sup = fixture.openTurn.actions.find((a): a is SupersedeAction => a.kind === "supersede");
  return sup?.document ?? null;
}

// ─────────────────────────────────────────────────────────────────────────────
// Rehydrate a fixture into a strategist tier — replay-many, ZERO model calls
// ─────────────────────────────────────────────────────────────────────────────

/**
 * Rehydrate a frozen fixture into a strategist {@link Agent} backed by the EXISTING {@link recordedAgent}: its
 * `open` re-emits the frozen OPEN turn byte-identically with NO provider call; its `decide` (only reached on a
 * watcher escalation) fails closed to a stand-down (the frozen plan captured no re-brief turn). Because this
 * tier IS a recording, {@link fanWatchers} runs the cascade in `frozenStrategist` mode so that fail-closed
 * stand-down HOLDS the frozen plan armed rather than disarming it (see {@link cascadeAgent}). This is the tier
 * every fan-out watcher runs against.
 */
export function recordedStrategistOf(fixture: PlanFixture): Agent {
  // COSTLESS by construction: the fixture froze the authored turn, not a deliberation cost, so the
  // rehydrated strategist replays only latency-blind (a clock-honest replay refuses at open, ADR-0040 §7).
  const turns: CapturedTurns = new Map<WakeKey, CapturedTurn>([[OPEN_WAKE_KEY, { turn: fixture.openTurn }]]);
  return recordedAgent(fixture.config, turns);
}

// ─────────────────────────────────────────────────────────────────────────────
// Persist / load — the fixtures dir (the cache)
// ─────────────────────────────────────────────────────────────────────────────

/** Serialize a fixture to pretty JSON (the on-disk cache form). Pure. */
export function serializePlanFixture(fixture: PlanFixture): string {
  return JSON.stringify(fixture, null, 2) + "\n";
}

/** Write a fixture to `path` (creating parent dirs) — persist to the fixtures dir. */
export function writePlanFixture(fixture: PlanFixture, path: string): void {
  mkdirSync(dirname(path), { recursive: true });
  writeFileSync(path, serializePlanFixture(fixture), "utf8");
}

/**
 * Load a fixture from `path` and RE-VERIFY its content hash. Fail-closed: a fixture whose stored
 * `planFixtureSha` does not match a fresh {@link planFixtureSha} of its content is REFUSED (a contaminated /
 * hand-edited fixture can never masquerade as the frozen plan). A clean reload returns byte-identical content
 * and the SAME sha — the cache-hit guarantee.
 *
 * BRIEF-HASH honesty on read (kestrel-voy9): the outer sha only proves the bytes are self-consistent, NOT
 * that a supplied `brief.hash` tells the truth about `brief.text` — a hand-forged fixture can carry a matching
 * outer sha over a LYING brief hash (it would content-address, and thus bind into grade provenance, the wrong
 * thesis). {@link assertBriefHashHonest} re-derives and REFUSES it before the fixture can be replayed.
 */
export function readPlanFixture(path: string): PlanFixture {
  const raw = JSON.parse(readFileSync(path, "utf8")) as PlanFixture;
  const { planFixtureSha: stored, ...content } = raw;
  const recomputed = planFixtureSha(content);
  if (recomputed !== stored) {
    throw new Error(
      `readPlanFixture: plan_fixture_sha mismatch — stored ${stored} != recomputed ${recomputed} ` +
        `(the fixture at ${path} is contaminated; refusing to replay a plan that is not the frozen one)`,
    );
  }
  if (raw.brief !== undefined) assertBriefHashHonest(raw.brief);
  return raw;
}

// ─────────────────────────────────────────────────────────────────────────────
// fanWatchers — one frozen plan → N watchers → N deterministic grid columns
// ─────────────────────────────────────────────────────────────────────────────

/** One watcher in the fan-out: its manage-only {@link Agent} (recorded / stub / prompted) and an optional
 * human label for the column (defaults to the watcher config's own `label`). */
export interface WatcherSpec {
  readonly watcher: Agent;
  /** Optional display label for this column; the grid ConfigId is derived from the cascade config regardless. */
  readonly label?: string;
}

/** The frozen frame every watcher runs against — the byte-identical tape + cadence + fill + mandate + brief
 * (ADR-0032 §7: "the identical frozen plan+frame"). Mirrors the `runSimulateSession` knobs the cascade tracer
 * uses, so the ONLY variable across columns is the watcher tier. */
export interface FanFrame {
  readonly events: readonly BusEvent[];
  readonly wakes?: readonly TimeOfDay[];
  readonly fillModel: FillModelName;
  readonly rUsd: number;
  readonly mandate?: Mandate;
  readonly brief?: Brief;
  /** The buy-and-hold / null baseline EV for the `strategist_alpha` leg; defaults to the frozen-plan EV
   * (⇒ `strategistAlpha = 0`, an honest null when no benchmark is supplied). */
  readonly benchmarkEv?: number;
  /**
   * Opt in to the PER-WAKE mark-to-model of held option legs (the theta cell — kestrel theta-cell seam b,
   * {@link import("../../fill/mark-to-model.ts").markHeldLegsPerWake}). `false`/absent (every existing
   * cell) ⇒ held options are marked only at the terminal intrinsic settle, exactly as today (pinned
   * floors unchanged, byte-identical). `true` (the FOMC-OPTIONS theta cell) ⇒ a held straddle's value
   * decays between wakes as theta bleeds, so standing down on a bleeding position is natively costly.
   */
  readonly markToModel?: boolean;
}

/** The per-column participation summary the fan surfaces so a run can MEASURE PASS-vs-act-vs-held rather
 * than INFER it from P&L (kestrel harness-observability, Bug 2). The four buckets PARTITION every invoked
 * ordinal (open + each wake) exactly once — a held/escalated turn is counted DISTINCTLY, never pooled into
 * the PASS bucket. Derived from the cascade routing {@link CascadeStep} log + the captured turns. */
export interface ColumnSummary {
  /** Non-escalated ordinals whose terminal turn EMITTED at least one action (`actions.length > 0`). */
  readonly actedTurns: number;
  /** Non-escalated ordinals whose terminal turn was a PASS (empty actions — a genuine stand-pat). */
  readonly passTurns: number;
  /** Ordinals where the watcher escalated and the frozen plan was HELD ARMED (over-reach refused). */
  readonly escalationHeldTurns: number;
  /** Ordinals where the watcher escalated and the strategist RE-BRIEFED (its turn crossed to the Bus). */
  readonly escalationRebriefTurns: number;
}

/** One fan-out column: the cascade run for one watcher against the frozen plan, keyed by its cell ConfigId. */
export interface WatcherColumn {
  /** The cell config-axis id ({@link cellConfigId}) of this cascade column — its distinct grid key. */
  readonly configId: string;
  /** The cascade config's label (`cascade(strategist»watcher)`) — legible column name. */
  readonly label: string;
  /** The graded Blotter of this watcher's cascade run. */
  readonly blotter: Blotter;
  /** The captured cascade tape (the byte-identical replay input for this column). */
  readonly captured: CapturedTurns;
  /** The three-level attribution over the frozen-plan counterfactual (ADR-0032 §6). */
  readonly attribution: Attribution;
  /** The cascade routing audit for this column — one {@link CascadeStep} per invoked ordinal (open + each
   * wake), including any escalation-held / escalation-rebrief events (kestrel harness-observability, Bug 2). */
  readonly steps: readonly CascadeStep[];
  /** The PASS/act/held participation summary derived from {@link steps} + {@link captured} (Bug 2). */
  readonly summary: ColumnSummary;
}

/** The result of a fan-out: the frozen plan's sha, the frozen-plan EV (the shared counterfactual baseline),
 * one {@link WatcherColumn} per watcher, and the aggregated comparison {@link Grid} (one cell per watcher). */
export interface FanResult {
  /** The frozen plan's `plan_fixture_sha` — the cache key every column shares. */
  readonly planFixtureSha: string;
  /** `EV(frozen strategist Plan, replayed pure-algo, no watcher)` — the armed Plan alone (the shared baseline). */
  readonly frozenPlanEv: number;
  readonly columns: readonly WatcherColumn[];
  /** The {@link buildGrid} of all cascade columns — N distinct cells (one per watcher ConfigId). */
  readonly grid: Grid;
}

/**
 * OPT-IN CLOCK-HONEST fan mode (ADR-0040 / clock-honest wakes §7, bd kestrel-yb75). Supplied to
 * {@link fanWatchers}, it makes EACH watcher column's cascade run CLOCKED: the real Simulate driver measures
 * the wall-clock deliberation cost of every Seat turn at its `now` seam and records `measuredMs` into that
 * column's {@link CapturedTurns} (the persisted replay corpus tracer-1 lifted, PR #223). ABSENT (the
 * default) ⇒ the fan runs LATENCY-BLIND, byte-identical to before this seam existed.
 *
 * The eligibility gate stays SOVEREIGN: this mode does not bypass the driver's own
 * {@link import("../simulate.ts").clockHonestIneligibility} refusal at open — an ineligible/undeclared
 * `dataRung` (minute/session/absent/off-lattice) still REFUSES the run loudly. And the shared frozen-plan
 * counterfactual ({@link FanResult.frozenPlanEv}) is UNTOUCHED: it has no Seat deliberating (pure-algo), so
 * it stays latency-blind and deterministic — only the watcher columns are clocked.
 */
export interface ClockHonestFan {
  /** The network/infra latency allowance added to EVERY measured Seat turn (integer ms ≥ 0). The driver
   * has NO default — a clock-honest run without it is refused at open (never a designed default); `0` is
   * legal and declares a colocated seat. A ConfigId axis (a clocked column never pools with an unclocked). */
  readonly latencyBufferMs: number;
  /** The tape's DECLARED observability rung — threaded to the driver, which VERIFIES it fail-closed
   * (`tick`/`second` eligible; `minute`/`session`/absent/off-lattice REFUSED at open). Declared, never
   * trusted here — the sovereign gate lives in the driver. */
  readonly dataRung: string;
  /** The injectable wall clock the driver measures with. Defaults to {@link monotonicNow}(`Date.now`).
   * Whatever is supplied is wrapped in {@link monotonicNow} so a broken clock FAILS CLOSED. */
  readonly now?: () => number;
}

/**
 * A CLOCK-HONEST MEASUREMENT DEFECT — broken timing (a dead/backward clock) or a capture that failed the
 * clock-honest replayability belt (PR #226 review, finding C). Distinct from a Seat/model failure so a
 * downstream attrition/coverage read can bucket "the measurement machinery failed" separately from "the
 * model failed" instead of regex-matching free-text `error` strings. Thrown by {@link monotonicNow} and by
 * the roster's fail-closed capture belt; `instanceof` survives propagation through the real driver. */
export class ClockHonestDefectError extends Error {}

/**
 * Wrap a wall clock so the clock-honest measurement path FAILS CLOSED on broken timing (bd kestrel-yb75).
 * A reading that is NON-FINITE (`NaN`/`Infinity` — a dead clock) or that STEPS BACKWARD (a non-monotonic
 * clock — an NTP slew) THROWS instead of yielding a bogus `measuredMs`: a backward step would round to `0`
 * and silently claim a colocated seat, and a `NaN` would serialize to `null` under a clock-honest label —
 * exactly the "never silently record measuredMs=0 or a fabricated cost" hazard §7 forbids. The throw
 * propagates out of the real driver so the column is marked CRASHED (never a clock-honest capture carrying
 * fabricated cost). A well-behaved clock (`Date.now` in normal operation; a constant injected clock, which
 * measures a legitimate `0`) is unaffected. Pure factory; the returned closure carries the last reading.
 */
export function monotonicNow(base: () => number): () => number {
  let last = Number.NEGATIVE_INFINITY;
  return () => {
    const v = base();
    if (!Number.isFinite(v)) {
      throw new ClockHonestDefectError(
        `clock-honest measurement: wall clock returned a non-finite reading ${JSON.stringify(v)} — a broken clock cannot ground a measured deliberation cost (fail closed, bd kestrel-yb75)`,
      );
    }
    if (v < last) {
      throw new ClockHonestDefectError(
        `clock-honest measurement: wall clock stepped BACKWARD (${v} < ${last}) — a non-monotonic clock would round to a bogus measuredMs=0 and silently claim a colocated seat (fail closed, bd kestrel-yb75)`,
      );
    }
    last = v;
    return v;
  };
}

/**
 * FAN OUT one frozen plan to N watchers (ADR-0032 §7). Rehydrates the fixture into a recorded strategist
 * (ZERO model calls), runs the frozen-Plan counterfactual ONCE (the shared `frozenPlanEv`), then for EACH
 * watcher builds `cascadeAgent({ strategist: recordedStrategist, watcher })` and runs it through the EXISTING
 * `runSimulateSession` against the byte-identical frozen frame. Collects grade + {@link attributeCascade
 * attribution} per watcher and aggregates all columns via {@link buildGrid} into N distinct grid cells (each
 * watcher label mints a distinct cascade ConfigId — the ADR-0013 controlled-division discipline). Deterministic
 * across the fan (no wall clock, no RNG in the driver): the same fixture + watchers yields byte-identical columns.
 */
/** Derive the {@link ColumnSummary} from a column's cascade routing {@link CascadeStep} log + its captured
 * turns. The four buckets PARTITION every invoked ordinal exactly once: an escalation-held / escalation-
 * rebrief step is counted in its OWN bucket (never pooled into PASS/act); a non-escalated step is classified
 * by whether its captured terminal turn EMITTED an action. Pure — no clock, no RNG. */
function summarizeColumn(steps: readonly CascadeStep[], captured: CapturedTurns): ColumnSummary {
  let actedTurns = 0;
  let passTurns = 0;
  let escalationHeldTurns = 0;
  let escalationRebriefTurns = 0;
  for (const step of steps) {
    if (step.kind === "escalation-hold-armed") {
      escalationHeldTurns++;
      continue;
    }
    if (step.kind === "escalation-rebrief") {
      escalationRebriefTurns++;
      continue;
    }
    const turn = captured.get(step.key)?.turn;
    if ((turn?.actions.length ?? 0) > 0) actedTurns++;
    else passTurns++;
  }
  return { actedTurns, passTurns, escalationHeldTurns, escalationRebriefTurns };
}

export async function fanWatchers(fixture: PlanFixture, watchers: readonly WatcherSpec[], frame: FanFrame, clock?: ClockHonestFan): Promise<FanResult> {
  const strategist = recordedStrategistOf(fixture);

  // The frozen-Plan counterfactual — the armed Plan alone (pure-algo, no watcher), run ONCE. The shared
  // baseline for every column's watcher_alpha (ADR-0032 §6). A fixture that armed nothing → NaN EV (honest).
  // The frozen frame knobs, spread WITHOUT explicit `undefined` (exactOptionalPropertyTypes discipline).
  const wakesOpt = frame.wakes !== undefined ? { wakes: frame.wakes } : {};
  const mandateOpt = frame.mandate !== undefined ? { mandate: frame.mandate } : {};
  const briefOpt = frame.brief !== undefined ? { brief: frame.brief } : {};
  // THE THETA GATE (theta-cell seam b): thread the frame's opt-in into the sim so held option legs are
  // marked-to-model per wake and the bleed is folded into `totals.floor` — the frozen-plan baseline AND
  // every watcher column grade on the same theta-graded floor. Absent (every cell but FOMC-OPTIONS) ⇒ no
  // spread ⇒ byte-identical to before the flag was wired (exactOptionalPropertyTypes discipline).
  const markToModelOpt = frame.markToModel === true ? { markToModel: true as const } : {};

  const planDoc = fixturePlanDoc(fixture);
  const frozen =
    planDoc !== null
      ? await runSimulateSession({ events: frame.events, agent: fixedPlanAgent(planDoc, BACKTEST_CONFIG), fillModel: frame.fillModel, rUsd: frame.rUsd, ...wakesOpt, ...markToModelOpt })
      : null;
  const frozenPlanEv = frozen?.blotter.totals.floor ?? NaN;
  const benchmark = frame.benchmarkEv ?? frozenPlanEv;

  const columns: WatcherColumn[] = [];
  const blotters: Blotter[] = [];
  for (const spec of watchers) {
    // `frozenStrategist: true` — this fan's strategist tier is a `recordedStrategistOf` recording with NO
    // captured re-brief. So a watcher escalation must NOT let the strategist's fail-closed stand-down
    // DISARM the frozen plan (which would collapse the whole fan to $0); the over-reach is refused and the
    // armed plan is HELD (fail-closed to a PASS). SCOPED to this recorded fan — the LIVE cascade never sets
    // it (see cascade.ts CascadeOptions.frozenStrategist); the admission Gate is untouched.
    // Wire a routing log sink through the cascade so each column can REPORT its escalation events (esp. the
    // frozenStrategist HELD_ARMED_TURN holds) instead of discarding them, and SURFACE a PASS/act/held summary
    // (kestrel harness-observability, Bug 2). Above the determinism line — audit evidence, never a graded input.
    const steps: CascadeStep[] = [];
    // CLOCK-HONEST (bd kestrel-yb75, opt-in): stamp the cascade config's `clockHonest`/`latencyBufferMs`
    // ConfigId axes (so a clocked column never pools with an unclocked one) and thread the DECLARED
    // `dataRung` + a MONOTONIC-GUARDED clock into the driver. The driver verifies eligibility fail-closed at
    // open (the sovereign gate) and mints `measuredMs` at its one `now` seam. A FRESH monotonic wrapper per
    // column (each column is its own measurement session). Absent ⇒ no config override, no dataRung/now
    // spread ⇒ byte-identical latency-blind behavior (exactOptionalPropertyTypes discipline).
    const clockedConfigOpt =
      clock !== undefined
        ? { config: { ...cascadeConfigOf(strategist.config, spec.watcher.config), clockHonest: true as const, latencyBufferMs: clock.latencyBufferMs } }
        : {};
    const clockOpt = clock !== undefined ? { dataRung: clock.dataRung, now: monotonicNow(clock.now ?? Date.now) } : {};
    const cascade = cascadeAgent({ strategist, watcher: spec.watcher, frozenStrategist: true, log: steps, ...clockedConfigOpt });
    const res = await runSimulateSession({
      events: frame.events,
      agent: cascade,
      fillModel: frame.fillModel,
      rUsd: frame.rUsd,
      ...wakesOpt,
      ...mandateOpt,
      ...briefOpt,
      ...markToModelOpt,
      ...clockOpt,
    });
    const attribution = attributeCascade({ benchmark, frozenPlan: frozenPlanEv, cascade: res.blotter.totals.floor });
    columns.push({
      configId: cellConfigId(res.config),
      label: spec.label ?? res.config.label,
      blotter: res.blotter,
      captured: res.captured,
      attribution,
      steps,
      summary: summarizeColumn(steps, res.captured),
    });
    blotters.push(res.blotter);
  }

  return { planFixtureSha: fixture.planFixtureSha, frozenPlanEv, columns, grid: buildGrid(blotters) };
}