// Read-model coverage for the plan-family (Epic) projections — the wave rollups, the slice-PR
// delivery counts, and the per-row delivery/bucket/ack signals — now authored via Urban's ADR-0065
// declare-once primitives (`defineRollup`, app/planRollups.ts; `defineReadModel` + key-correlated
// rollup lookups, app/planReadModel.ts). Issue #493, the plan-family twin of app/featureReadModel.test.ts.
//
// 059/060/061 hand-authored the three GROUP-BY aggregates as SQL VIEWs AND a second time in the runtime
// TS; 074/080 hand-authored the per-row `delivery`/`list_bucket`/`ack_open` signals as SQL CASEs AND a
// TS oracle (`deriveDelivery`/`deriveEpicBucket`/`epicIsAcknowledgeable`, app/delivery.ts) — kept in
// lockstep by bespoke parity tests (app/plansReadModel.test.ts, app/planWaveSummary.test.ts,
// app/delivery.test.ts, the ADR-0065 drift surface #2). Migrations 082/083 supersede them: every
// rollup VIEW is emitted from its ONE `defineRollup`, and every derived read-model column from the ONE
// `defineReadModel` — both of which ALSO drive the runtime TS (`reduce`/`fnFor`, behind app/delivery.ts).
// This suite retires those hand-written parity tests in favour of the framework parity guard, and
// guards FOUR things:
//
//   1. DRIFT GUARD — migration 082 embeds each rollup's VIEW DDL VERBATIM from `rollup.viewDdl()`, and
//      migration 083 embeds each derived column VERBATIM from `planReadModel.sqlSelectFor(...)`, so the
//      checked-in VIEWs cannot drift from the declarations.
//   2. FRAMEWORK PARITY GUARD — `assertRollupParity` / `assertReadModelParity` prove the SQL and TS
//      lowerings each declaration compiles to agree (the role the retired hand-written tests played).
//   3. END-TO-END BEHAVIOUR on the REAL migration VIEWs (059→083 applied to an in-memory DB): the full
//      status × slice-PR × acknowledgement matrix vs the app/delivery.ts adapters, the hand-authored
//      display strings (`delivery_label`/`wave_label`, and 059's `bar` glyph), and the reconciler
//      `derived_status`-bypass.
//   4. PAGE BINDINGS — the operator pages bind the derived VIEWs (`plan_read_model`, `plan_wave_summary`,
//      `plan_wave_tasks`), never the raw `plans` table.
import { readFileSync } from "node:fs";
import { DatabaseSync } from "node:sqlite";
import { test } from "node:test";
import { fileURLToPath } from "node:url";
import { assertReadModelParity, assertRollupParity, type ParityDb, type ParitySample, type RollupInputs } from "@nanobpm/urban";
import type { AppApi } from "@nanobpm/urban";
import { assert, assertEquals } from "#test-assert";
import { acknowledgeVia } from "./acknowledge.ts";
import { deriveDelivery, deriveEpicBucket, epicIsAcknowledgeable } from "./delivery.ts";
import { planReadModel, PLAN_READ_MODEL_BASE_ALIAS, PLAN_READ_MODEL_DERIVED } from "./planReadModel.ts";
import { PLAN_ROLLUPS, planDeliveryCounts, planWaveCounts, planWaveProgress } from "./planRollups.ts";
import { noopLog } from "../test/log.ts";

const MIG = (name: string) => readFileSync(fileURLToPath(new URL(`../db/migrations/${name}`, import.meta.url)), "utf8");
const PAGE = (name: string) => JSON.parse(readFileSync(fileURLToPath(new URL(`../pages/${name}`, import.meta.url)), "utf8"));

const ROLLUPS_MIGRATION = "082_plan_rollups_declare_once.sql";
const READ_MODEL_MIGRATION = "097_plan_read_model_terminal_dismiss.sql";
// Passes `acknowledged_at` through the plan_read_model VIEW (issue #654) — the column the shared
// `acknowledgeVia` helper reads to tell an already-acknowledged epic (idempotent 200) from a live one
// (409). Supersedes 097's VIEW body verbatim except for that one added base pass-through.
const READ_MODEL_ACK_PASSTHROUGH_MIGRATION = "100_plan_read_model_pass_acknowledged_at.sql";
// The forward chain whose net effect the end-to-end tests exercise: the original hand-authored VIEWs
// (059/060/061/074/080), the declare-once supersessions (082/083), then the terminal-dismiss
// supersession (097), then the `acknowledged_at` pass-through (100). Mirrors the runtime migrator.
const MIGRATION_CHAIN = [
  "059_plan_wave_summary.sql",
  "060_plan_wave_rollup.sql",
  "061_plan_delivery_rollup.sql",
  "074_plan_read_model_derive_bucket.sql",
  "080_plan_read_model_derive_terminal.sql",
  ROLLUPS_MIGRATION,
  "083_plan_read_model_declare_once.sql",
  "084_plan_wave_tasks_effective_status.sql",
  READ_MODEL_MIGRATION,
  READ_MODEL_ACK_PASSTHROUGH_MIGRATION,
];

// The base `plans` / `plan_tasks` / `pull_requests` shapes the VIEWs read, plus a stand-in for the
// managed `plans__tracking` derived VIEW (ADR-0065) the read model reads its terminal-folded
// `derived_status` off. `derived_status_override` lets a test model the reconciler's derive edge (a
// terminated instance ⇒ `abandoned` while base `status` stays frozen). The vestigial stored
// `list_bucket`/`ack_open` columns (#439) are present so a test can seed STALE values and prove the
// VIEW ignores them.
function viewDb(): DatabaseSync {
  const db = new DatabaseSync(":memory:");
  db.exec(
    `CREATE TABLE plans (
       plan_key TEXT PRIMARY KEY, repo TEXT, issue_number INTEGER, issue_url TEXT, title TEXT,
       status TEXT, task_count INTEGER, process_key TEXT, outcome TEXT, created_at TEXT, updated_at TEXT,
       epic_phase TEXT, base_branch TEXT, wait_gate_label TEXT, bound_artifacts TEXT, promotion_pr TEXT,
       promotion_state TEXT, acknowledged_at TEXT, list_bucket TEXT, ack_open INTEGER,
       derived_status_override TEXT);
     CREATE TABLE plan_tasks (
       id INTEGER PRIMARY KEY, plan_key TEXT, task_index INTEGER, task_id TEXT, title TEXT, prompt TEXT,
       status TEXT, pr_key TEXT, summary TEXT, created_at TEXT, updated_at TEXT, wave INTEGER,
       open_question TEXT, answer TEXT, draft_pr_key TEXT, corr_key TEXT);
     CREATE TABLE pull_requests (pr_key TEXT PRIMARY KEY, url TEXT, status TEXT, process_key TEXT,
       derived_status_override TEXT);`,
  );
  // Stand-in for the managed `plans__tracking` / `pull_requests__tracking` VIEWs urban provisions at
  // mount: each re-exports `<base>.*` plus the terminal-folded `derived_status`. A test seeds
  // `derived_status_override` to model the reconciler's derive edge (a terminated instance ⇒ `abandoned`
  // while base `status` stays frozen); absent, it falls through to the base `status`. The plan-family
  // count rollups join `pull_requests__tracking.derived_status` (ADR-0065 derive-only), the SAME column
  // the canonical runtime reads (service.ts `prsTracking`), so the VIEW counts track a cancelled slice.
  db.exec(
    `CREATE VIEW plans__tracking AS
       SELECT p.*, COALESCE(p.derived_status_override, p.status) AS derived_status FROM plans p;
     CREATE VIEW pull_requests__tracking AS
       SELECT p.*, COALESCE(p.derived_status_override, p.status) AS derived_status FROM pull_requests p;`,
  );
  for (const m of MIGRATION_CHAIN) db.exec(MIG(m));
  return db;
}

interface SamplePlan {
  status: string;
  acknowledged_at?: string | null;
  derived_status_override?: string | null;
  /** Deliberately-stale STORED projection columns (a row the gateway last projected in another status).
   *  The VIEW must ignore these and re-derive. */
  stored?: Partial<Record<"list_bucket" | "ack_open", string | number>>;
}

let taskId = 0;
function addPlan(db: DatabaseSync, plan_key: string, plan: SamplePlan): void {
  const s = plan.stored ?? {};
  db.prepare(
    `INSERT INTO plans (plan_key, repo, issue_number, issue_url, title, status, acknowledged_at,
        derived_status_override, list_bucket, ack_open, created_at, updated_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
  ).run(
    plan_key,
    "o/r",
    1,
    `https://gh/${plan_key}`,
    `Epic ${plan_key}`,
    plan.status,
    plan.acknowledged_at ?? null,
    plan.derived_status_override ?? null,
    s.list_bucket ?? null,
    s.ack_open ?? null,
    "2026-01-01T00:00:00Z",
    "2026-01-01T00:00:00Z",
  );
}

/** Add one slice task, optionally with a PR. `prStatus === undefined` ⇒ no PR row (an un-opened slice);
 *  `prStatus === "missing"` ⇒ a `pr_key` with NO `pull_requests` row (the poller's dangling-PR sentinel).
 *  `prDerivedOverride` seeds the PR's `derived_status_override` (a DERIVE-ONLY terminated PR whose base
 *  `status` stays frozen but whose tracking `derived_status` recomputes, e.g. `abandoned`). */
function addTask(db: DatabaseSync, plan_key: string, opts: { status?: string; wave?: number | null; prStatus?: string; prDerivedOverride?: string | null }): void {
  const id = taskId++;
  const prKey = opts.prStatus === undefined ? null : `pr${id}`;
  db.prepare(
    "INSERT INTO plan_tasks (id, plan_key, task_index, task_id, status, pr_key, wave) VALUES (?, ?, ?, ?, ?, ?, ?)",
  ).run(id, plan_key, id, `t${id}`, opts.status ?? "opened", prKey, opts.wave ?? null);
  if (prKey !== null && opts.prStatus !== "missing") {
    db.prepare("INSERT INTO pull_requests (pr_key, url, status, process_key, derived_status_override) VALUES (?, ?, ?, ?, ?)").run(
      prKey,
      `https://gh/${prKey}`,
      opts.prStatus,
      `P${id}`,
      opts.prDerivedOverride ?? null,
    );
  }
}

function readModel(db: DatabaseSync, plan_key: string): Record<string, unknown> {
  return db.prepare("SELECT * FROM plan_read_model WHERE plan_key = ?").get(plan_key) as Record<string, unknown>;
}

// A `ParityDb` over node:sqlite's `DatabaseSync` for the framework parity guards (which need positional
// `exec`/`all`/`run`, whereas `DatabaseSync` exposes query methods on prepared statements).
function parityDb(db: DatabaseSync): ParityDb {
  return {
    exec: (sql) => db.exec(sql),
    all: <T = Record<string, unknown>>(sql: string, params: unknown[] = []) => db.prepare(sql).all(...(params as never[])) as T[],
    run: (sql, params: unknown[] = []) => {
      const r = db.prepare(sql).run(...(params as never[]));
      return { changes: Number(r.changes), lastInsertRowid: r.lastInsertRowid };
    },
  };
}

test("DRIFT GUARD: migration 082 embeds each rollup's VIEW DDL VERBATIM from rollup.viewDdl() (the VIEWs cannot drift from defineRollup)", () => {
  const sql = MIG(ROLLUPS_MIGRATION);
  for (const rollup of PLAN_ROLLUPS) {
    assert(
      sql.includes(rollup.viewDdl()),
      `migration ${ROLLUPS_MIGRATION} no longer embeds the declaration's VIEW DDL for rollup "${rollup.decl.name}" — ` +
        `regenerate it from app/planRollups.ts (or add a new superseding migration). Expected to contain:\n${rollup.viewDdl()}`,
    );
    // Each superseded VIEW is DROP+CREATEd (059/060/061 are immutable — this supersedes their bodies).
    assert(new RegExp(`DROP VIEW IF EXISTS ${rollup.decl.name};`).test(sql), `082 must DROP the superseded "${rollup.decl.name}" first`);
  }
});

test("DRIFT GUARD: migration 097 embeds each derived column VERBATIM from planReadModel.sqlSelectFor (the VIEW cannot drift from the declaration)", () => {
  const sql = MIG(READ_MODEL_MIGRATION);
  for (const col of PLAN_READ_MODEL_DERIVED) {
    const emitted = planReadModel.sqlSelectFor(col, { baseAlias: PLAN_READ_MODEL_BASE_ALIAS });
    assert(
      sql.includes(`${emitted} AS ${col}`),
      `migration ${READ_MODEL_MIGRATION} no longer embeds the declaration's SQL for "${col}" — regenerate it ` +
        `from app/planReadModel.ts (or add a new superseding migration). Expected to contain:\n  ${emitted} AS ${col}`,
    );
  }
  // DROP+CREATE that supersedes 083's plan_read_model VIEW body (the terminal-dismiss #641 arm), keeping
  // every base column an aliased pass-through so the static pages↔schema contract guard still sees them.
  assert(/DROP VIEW IF EXISTS plan_read_model;/.test(sql), "097 must DROP the superseded plan_read_model first");
  assert(/CREATE VIEW plan_read_model AS/.test(sql), "097 must (re)create plan_read_model");
  for (const base of ["plan_key", "repo", "issue_number", "title", "process_key", "epic_phase", "promotion_pr", "promotion_state"]) {
    assert(sql.includes(`pl.${base} AS ${base}`), `097 must pass base column "${base}" through the VIEW`);
  }
  // The hand-authored display strings (D3 — no TS twin) live in this VIEW over the derived columns.
  assert(sql.includes("AS delivery_label"), "097 must carry the hand-authored delivery_label display column");
  assert(sql.includes("AS wave_label"), "097 must carry the hand-authored wave_label display column");
  // The FROM/JOIN relation names are DERIVED from the declaration (baseTable + each lookup's rollup name
  // + join keys), not hand-hardcoded — so renaming `baseTable` or a rollup `.name` (which would make 082
  // create a different-named VIEW) breaks this guard instead of silently leaving 097 pointing at a
  // stale/missing relation.
  const alias = PLAN_READ_MODEL_BASE_ALIAS;
  assert(sql.includes(`FROM ${planReadModel.decl.baseTable} ${alias}`), `097's FROM must be the declaration's baseTable "${planReadModel.decl.baseTable}" (aliased ${alias})`);
  for (const lk of planReadModel.decl.lookups) {
    const rollupName = lk.rollup.decl.name;
    const on = lk.on.map((k) => `${alias}.${k.base} = ${lk.as}.${k.rollup}`).join(" AND ");
    const join = `LEFT JOIN ${rollupName} ${lk.as} ON ${on}`;
    assert(sql.includes(join), `097 must LEFT JOIN the declaration's "${rollupName}" lookup exactly as "${join}"`);
  }
});

// A minimal `AppApi` over node:sqlite so the shared `acknowledgeVia` helper (app/acknowledge.ts) can be
// driven against the REAL migration VIEW + base table — not a mock that could paper over a missing VIEW
// column. `data.table(name, key)` reads/writes `name` keyed on `key`, exactly as urban's does.
function sqliteApp(db: DatabaseSync): AppApi {
  return {
    data: {
      table: (name: string, key: string) => ({
        get: async (id: unknown) => db.prepare(`SELECT * FROM ${name} WHERE ${key} = ?`).get(id) ?? undefined,
        update: async (id: unknown, patch: Record<string, unknown>) => {
          const cols = Object.keys(patch);
          const set = cols.map((c) => `${c} = ?`).join(", ");
          const r = db.prepare(`UPDATE ${name} SET ${set} WHERE ${key} = ?`).run(...(cols.map((c) => patch[c]) as never[]), id as never);
          return Number(r.changes);
        },
      }),
    },
    log: noopLog(),
  } as unknown as AppApi;
}

const EPIC_ACK_SURFACE = {
  view: planReadModel.decl.name,
  baseTable: "plans",
  keyColumn: "plan_key",
  label: "epic",
  notDismissableError: "not terminal",
} as const;

test("DRIFT GUARD: migration 100 supersedes the plan_read_model VIEW, adding the acknowledged_at pass-through while re-embedding every derived column VERBATIM from planReadModel.sqlSelectFor", () => {
  const sql = MIG(READ_MODEL_ACK_PASSTHROUGH_MIGRATION);
  assert(/DROP VIEW IF EXISTS plan_read_model;/.test(sql), "100 must DROP the superseded plan_read_model first");
  assert(/CREATE VIEW plan_read_model AS/.test(sql), "100 must (re)create plan_read_model");
  // The one reason this migration exists: expose `acknowledged_at` as a projected column (the shared
  // acknowledgeVia helper reads it to distinguish an already-acknowledged epic from a live one).
  assert(sql.includes("pl.acknowledged_at AS acknowledged_at"), "100 must PROJECT the acknowledged_at pass-through");
  // Every derived column stays byte-identical to the declaration — no drift is smuggled in.
  for (const col of PLAN_READ_MODEL_DERIVED) {
    const emitted = planReadModel.sqlSelectFor(col, { baseAlias: PLAN_READ_MODEL_BASE_ALIAS });
    assert(
      sql.includes(`${emitted} AS ${col}`),
      `migration ${READ_MODEL_ACK_PASSTHROUGH_MIGRATION} no longer embeds the declaration's SQL for "${col}" — regenerate it ` +
        `from app/planReadModel.ts (or add a new superseding migration). Expected to contain:\n  ${emitted} AS ${col}`,
    );
  }
  for (const base of ["plan_key", "repo", "issue_number", "title", "process_key", "epic_phase", "promotion_pr", "promotion_state"]) {
    assert(sql.includes(`pl.${base} AS ${base}`), `100 must keep base column "${base}" as a pass-through`);
  }
});

test("issue #654: the plan_read_model VIEW PROJECTS acknowledged_at so acknowledgeVia can tell an already-acknowledged epic (idempotent 200) from a live one (409)", async () => {
  const db = viewDb();
  const stamp = "2026-02-02T00:00:00Z";
  // A `done` epic already dismissed (its `acknowledged_at` stamped): the VIEW folds it to History with
  // the Dismiss affordance closed (`ack_open=0`).
  addPlan(db, "o/r#1", { status: "done", acknowledged_at: stamp });
  const row = readModel(db, "o/r#1");
  // The bug (Copilot review, PR #655): `plan_read_model` only READ `acknowledged_at` inside its CASE
  // bodies and never PROJECTED it — unlike the PR / feature / delivery-graph read models — so the helper
  // saw `undefined` and could not tell "already acknowledged" from "still live". Assert the projection.
  assert("acknowledged_at" in row, "plan_read_model must PROJECT acknowledged_at (the helper depends on it)");
  assertEquals(row.acknowledged_at, stamp);
  assertEquals(row.ack_open, 0, "an acknowledged terminal epic has ack_open=0");

  // End-to-end: re-dismissing an already-acknowledged epic through the REAL VIEW is an idempotent 200,
  // NOT a spurious 409 — and it does not re-stamp.
  const res = await acknowledgeVia(sqliteApp(db), EPIC_ACK_SURFACE, "o/r#1");
  assertEquals(res.status, 200, "re-dismissing an already-acknowledged epic is an idempotent 200");
  assertEquals(res.body.ok, true);
  assertEquals(readModel(db, "o/r#1").acknowledged_at, stamp, "idempotent no-op must NOT re-stamp");
});

test("FRAMEWORK PARITY GUARD: each plan-family rollup's VIEW and TS reduce agree (assertRollupParity)", () => {
  // Sample leaf rows spanning: multi-wave plans, every task/PR status, dangling pr_key (no PR row),
  // un-levelized (NULL wave) tasks, and taskless plans — the predicates the rollups turn on.
  const sampleSets: RollupInputs[] = [
    {
      plan_tasks: [
        { plan_key: "a", pr_key: "a0", wave: 0, status: "opened" },
        { plan_key: "a", pr_key: "a1", wave: 0, status: "opened" },
        { plan_key: "a", pr_key: "a2", wave: 1, status: "escalated" },
        { plan_key: "a", pr_key: null, wave: 1, status: "blocked" },
        { plan_key: "a", pr_key: "a3", wave: null, status: "pending" },
        { plan_key: "b", pr_key: "b0", wave: 0, status: "skipped" },
        { plan_key: "b", pr_key: "bMissing", wave: 0, status: "opened" },
      ],
      pull_requests__tracking: [
        { pr_key: "a0", derived_status: "merged" },
        { pr_key: "a1", derived_status: "converging" },
        { pr_key: "a2", derived_status: "merged" },
        { pr_key: "a3", derived_status: "waiting_review" },
        { pr_key: "b0", derived_status: "abandoned" },
      ],
    },
    { plan_tasks: [], pull_requests__tracking: [] },
    {
      plan_tasks: [
        { plan_key: "c", pr_key: "c0", wave: 0, status: "opened" },
        { plan_key: "c", pr_key: "c1", wave: 2, status: "opened" },
      ],
      pull_requests__tracking: [
        { pr_key: "c0", derived_status: "merged" },
        { pr_key: "c1", derived_status: "converged" },
      ],
    },
  ];
  for (const rollup of PLAN_ROLLUPS) {
    const db = new DatabaseSync(":memory:");
    assertRollupParity(rollup, parityDb(db), sampleSets);
    db.close();
  }
});

test("FRAMEWORK PARITY GUARD: planReadModel's SQL and TS lowerings agree over the status × counts × ack matrix (assertReadModelParity)", () => {
  const samples: ParitySample[] = [];
  for (const status of ["planning", "dispatched", "done", "failed", "abandoned"]) {
    for (const derived_status of [status, "abandoned"]) {
      for (const acknowledged_at of [null, "2026-02-02T00:00:00Z"]) {
        for (const dc of [
          { prs_opened: 0, prs_merged: 0, prs_in_flight: 0 },
          { prs_opened: 3, prs_merged: 1, prs_in_flight: 2 },
          { prs_opened: 3, prs_merged: 3, prs_in_flight: 0 },
          { prs_opened: 2, prs_merged: 1, prs_in_flight: 0 },
        ]) {
          for (const wp of [[], [{ plan_key: "self", wave_count: 5, current_wave: 2 }]]) {
            samples.push({
              baseRow: { plan_key: "self", status, derived_status, acknowledged_at },
              lookups: { dc: [{ plan_key: "self", ...dc }], wp },
            });
          }
        }
      }
    }
  }
  const db = new DatabaseSync(":memory:");
  assertReadModelParity(planReadModel, parityDb(db), samples, { sql: { baseAlias: PLAN_READ_MODEL_BASE_ALIAS } });
  db.close();
});

test("the migration 083 VIEW derives delivery / list_bucket / ack_open EXACTLY like the app/delivery.ts adapters, over the status × slice-PR × ack matrix", () => {
  const db = viewDb();
  const inFlightPr = "waiting_review";
  // Slice-PR shapes exercising every delivery arm: none, all merged (landed), one in-flight
  // (converging), all terminal-not-merged (resolved-null), and a dangling pr_key (in-flight).
  const prSets: Record<string, string[]> = {
    none: [],
    landed: ["merged", "merged"],
    converging: ["merged", inFlightPr],
    resolved: ["merged", "abandoned"],
    convergedOnly: ["merged", "converged"],
    dangling: ["merged", "missing"],
  };
  const cases: Array<{ key: string; status: string; ackAt: string | null; override: string | null; prStatuses: string[] }> = [];
  let i = 0;
  for (const status of ["planning", "dispatched", "done", "failed", "abandoned"]) {
    for (const [shape, prStatuses] of Object.entries(prSets)) {
      for (const ackAt of [null, "2026-02-02T00:00:00Z"]) {
        for (const override of [null, "abandoned"]) {
          const key = `o/r#${i++}`;
          cases.push({ key, status, ackAt, override, prStatuses });
          addPlan(db, key, { status, acknowledged_at: ackAt, derived_status_override: override });
          prStatuses.forEach((ps, w) => addTask(db, key, { status: "opened", wave: w, prStatus: ps }));
          void shape;
        }
      }
    }
  }
  for (const { key, status, ackAt, override, prStatuses } of cases) {
    const row = readModel(db, key);
    const effectiveStatus = override ?? status;
    // `delivery` reads the BASE status (`done` is terminal, so base/effective agree on the gate).
    const expected = deriveDelivery(status, prStatuses);
    assertEquals(row.delivery, expected.delivery, `${key} (status=${status}): delivery`);
    assertEquals(row.delivery_label, expected.label, `${key} (status=${status}): delivery_label`);
    // `list_bucket`/`ack_open` classify on the terminal-folded effective status.
    assertEquals(row.list_bucket, deriveEpicBucket(effectiveStatus, expected.delivery, ackAt), `${key} (status=${effectiveStatus}): list_bucket`);
    const expectedAck = epicIsAcknowledgeable(effectiveStatus, expected.delivery) && ackAt === null ? 1 : 0;
    assertEquals(row.ack_open, expectedAck, `${key} (status=${effectiveStatus}): ack_open`);
  }
});

test("the migration 083 VIEW projects the wave frontier + 1-based wave_label from plan_wave_progress (taskless plan ⇒ all NULL)", () => {
  const db = viewDb();
  // 3 waves; wave 0 fully merged, wave 1 in-flight (the frontier), wave 2 pending ⇒ current_wave = 1.
  addPlan(db, "o/r#w", { status: "done" });
  addTask(db, "o/r#w", { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, "o/r#w", { status: "opened", wave: 1, prStatus: "waiting_review" });
  addTask(db, "o/r#w", { status: "opened", wave: 2, prStatus: "waiting_review" });
  const w = readModel(db, "o/r#w");
  assertEquals(w.wave_count, 3);
  assertEquals(w.current_wave, 1);
  assertEquals(w.wave_label, "2/3");

  // A settled plan (every wave merged) pins current_wave to the last index ⇒ "N/N".
  addPlan(db, "o/r#done", { status: "done" });
  addTask(db, "o/r#done", { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, "o/r#done", { status: "opened", wave: 1, prStatus: "merged" });
  assertEquals(readModel(db, "o/r#done").wave_label, "2/2");

  // A taskless plan has no rollup row ⇒ the LEFT JOIN reads NULL through every wave/delivery column.
  addPlan(db, "o/r#empty", { status: "done" });
  const empty = readModel(db, "o/r#empty");
  assertEquals({ wave_label: empty.wave_label, wave_count: empty.wave_count, delivery: empty.delivery }, { wave_label: null, wave_count: null, delivery: null });
});

test("the migration 083 VIEW IGNORES stale STORED list_bucket / ack_open columns — it reads only the derived signals", () => {
  const db = viewDb();
  // A settled+acknowledged epic whose STORED bucket lies (frozen while it was live). The VIEW re-derives.
  addPlan(db, "o/r#stale", { status: "done", acknowledged_at: "2026-02-02T00:00:00Z", stored: { list_bucket: "active", ack_open: 1 } });
  addTask(db, "o/r#stale", { status: "opened", wave: 0, prStatus: "merged" });
  const row = readModel(db, "o/r#stale");
  assertEquals(row.list_bucket, "history", "an acknowledged landed epic is History regardless of the stale stored value");
  assertEquals(row.ack_open, 0, "already acknowledged ⇒ no open Dismiss");
});

test("RED/GREEN #503 (+#641): a DERIVE-ONLY terminated epic (base status frozen 'dispatched', derived_status='abandoned') is classified off derived_status — Active+dismissable until acknowledged, then History", () => {
  // ADR-0065: cancel/terminate is DERIVE-ONLY — `plans__tracking.derived_status` recomputes `abandoned`
  // on READ while the base `plans.status` stays frozen at its last transient. The bucket classifies off
  // `derived_status`, so a terminated epic is handled on engine truth with no poller pass. Under #641
  // (uniform acknowledge-to-dismiss) a terminated epic now STAYS Active with a Dismiss affordance until
  // an operator ticks it off — mirroring features/PRs/DGs — rather than dropping straight to History.
  const db = viewDb();
  addPlan(db, "o/r#term", { status: "dispatched", stored: { list_bucket: "active" } });
  assertEquals(readModel(db, "o/r#term").list_bucket, "active", "precondition: a live dispatched epic is Active");

  db.prepare("UPDATE plans SET derived_status_override = 'abandoned' WHERE plan_key = ?").run("o/r#term");
  const row = readModel(db, "o/r#term");
  assertEquals(row.list_bucket, "active", "a derive-only terminated (unacknowledged) epic stays Active until dismissed (#641)");
  assertEquals(row.ack_open, 1, "…and carries the Dismiss affordance");
  assertEquals(row.list_bucket, deriveEpicBucket("abandoned", row.delivery === "converging" ? "converging" : null, null), "list_bucket tracks derived_status via the VIEW");

  // Acknowledging it (the operator tick-off) settles it to History — the derived_status-driven, no-
  // worker-write resolution the #503 phantom fix guaranteed, now gated on an explicit dismiss.
  db.prepare("UPDATE plans SET acknowledged_at = '2026-02-02T00:00:00Z' WHERE plan_key = ?").run("o/r#term");
  const acked = readModel(db, "o/r#term");
  assertEquals(acked.list_bucket, "history", "a dismissed terminated epic is History (classified off derived_status, no poller pass)");
  assertEquals(acked.ack_open, 0, "…and its Dismiss affordance is retracted");
});

test("REGRESSION (Copilot #493): a DERIVE-ONLY terminated slice PR (base status frozen 'converging', derived_status='abandoned') is counted RESOLVED — the VIEW joins pull_requests__tracking.derived_status", () => {
  // ADR-0065 derive-only: cancelling a PR instance recomputes `pull_requests__tracking.derived_status`
  // to `abandoned` on READ while the base `pull_requests.status` stays frozen at its last transient
  // (`converging`). The runtime (`service.ts` `derivePlanDelivery`) reads `prsTracking(...).derived_status`,
  // so the `plan_delivery_counts` rollup MUST join the SAME derived column — otherwise the VIEW would
  // read the frozen `converging` base, count the slice `prs_in_flight`, and WEDGE the epic at
  // `delivery='converging'` forever after its PR was cancelled. This asserts the rollup resolves it.
  const db = viewDb();
  addPlan(db, "o/r#cancel", { status: "done" });
  addTask(db, "o/r#cancel", { status: "opened", wave: 0, prStatus: "merged" });
  // A slice whose base PR row is frozen 'converging' but derived (terminal-folded) to 'abandoned'.
  addTask(db, "o/r#cancel", { status: "opened", wave: 1, prStatus: "converging", prDerivedOverride: "abandoned" });
  const row = readModel(db, "o/r#cancel");
  // Both slice PRs are terminal (one merged, one derived-abandoned) ⇒ nothing in flight, not all merged.
  assertEquals(row.delivery, null, "a cancelled slice is resolved-not-landed, not a wedged 'converging'");
  // The pre-derive/raw-status VIEW would have read 'converging' → prs_in_flight=1 → delivery='converging'.
  assert(row.delivery !== "converging", "the derived_status join closes the raw-status wedge (Copilot #493)");

  // A NON-overridden corpus stays byte-identical: derived_status falls through to base status, so a
  // genuinely in-flight 'converging' slice still reads converging (the fix only moves overridden rows).
  addPlan(db, "o/r#live", { status: "done" });
  addTask(db, "o/r#live", { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, "o/r#live", { status: "opened", wave: 1, prStatus: "converging" });
  assertEquals(readModel(db, "o/r#live").delivery, "converging", "a live (non-overridden) converging slice is unchanged");
});

test("the migration 082 plan_wave_summary VIEW still pre-formats the `bar` glyph over the (now framework-emitted) plan_wave_counts", () => {
  const db = viewDb();
  const plan = "o/r#bar";
  addPlan(db, plan, { status: "done" });
  // Wave 0 — 5 tasks: 3 merged, 1 converging (in-flight), 1 blocked (no PR).
  addTask(db, plan, { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, plan, { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, plan, { status: "opened", wave: 0, prStatus: "merged" });
  addTask(db, plan, { status: "opened", wave: 0, prStatus: "converging" });
  addTask(db, plan, { status: "blocked", wave: 0 });
  // Wave 1 — an escalated slice (its PR still escalated) and a skipped slice.
  addTask(db, plan, { status: "escalated", wave: 1, prStatus: "escalated" });
  addTask(db, plan, { status: "skipped", wave: 1 });

  const rows = db.prepare("SELECT wave, total, merged, in_flight, blocked, escalated, skipped, bar FROM plan_wave_summary WHERE plan_key = ? ORDER BY wave").all(plan) as Array<Record<string, unknown>>;
  assertEquals(rows.length, 2);
  assertEquals(rows[0].bar, "▓▓▓░░ 3/5 merged · 1 in-flight · 1 blocked");
  assertEquals(rows[1].bar, "░░ 0/2 merged · 1 escalated · 1 skipped");

  // A merged PR wins over the task's own status (the PR-merged predicate), and un-levelized (NULL wave)
  // tasks are excluded from every wave row.
  addPlan(db, "o/r#bar2", { status: "done" });
  addTask(db, "o/r#bar2", { status: "escalated", wave: 0, prStatus: "merged" });
  addTask(db, "o/r#bar2", { status: "pending", wave: null });
  const r2 = db.prepare("SELECT wave, total, merged, escalated FROM plan_wave_summary WHERE plan_key = ?").all("o/r#bar2") as Array<Record<string, unknown>>;
  assertEquals(r2.length, 1);
  assertEquals({ ...r2[0] }, { wave: 0, total: 1, merged: 1, escalated: 0 });
});

test("plan_wave_tasks carries each task's PR url + process_key link targets (unchanged display VIEW)", () => {
  const db = viewDb();
  addPlan(db, "o/r#lt", { status: "done" });
  addTask(db, "o/r#lt", { status: "opened", wave: 0, prStatus: "converging" });
  addTask(db, "o/r#lt", { status: "blocked", wave: 0 }); // no PR → null link targets
  const rows = db.prepare("SELECT pr_key, pr_url, process_key FROM plan_wave_tasks WHERE plan_key = ? ORDER BY task_index").all("o/r#lt") as Array<Record<string, unknown>>;
  assert(rows[0].pr_url != null && rows[0].process_key != null, "an opened slice links to its PR url + process instance");
  assertEquals({ pr_url: rows[1].pr_url, process_key: rows[1].process_key }, { pr_url: null, process_key: null });
});

test("plan_wave_tasks derives status=merged from the PR (never strands a landed slice at 'opened') — matching the summary bar", () => {
  // The reported defect (#530): a slice whose PR converged + merged kept reading Status "opened" in
  // the wave-state grid, because nothing writes `plan_tasks.status='merged'` on merge and the VIEW
  // exposed the raw task status. The fix DERIVES the displayed status the SAME way the count VIEWs
  // bucket `merged` — `pull_requests__tracking.derived_status = 'merged'` overrides the raw status —
  // so the per-task grid and the per-wave summary bar agree.
  const db = viewDb();
  addPlan(db, "o/r#eff", { status: "done" });
  addTask(db, "o/r#eff", { status: "opened", wave: 0, prStatus: "merged" }); // landed slice, task row frozen at "opened"
  addTask(db, "o/r#eff", { status: "opened", wave: 0, prStatus: "converging" }); // still converging → stays "opened"
  addTask(db, "o/r#eff", { status: "blocked", wave: 0 }); // no PR → raw status untouched
  // A DERIVE-ONLY merged edge (base PR status frozen, tracking recomputes to `merged`) also reads merged.
  addTask(db, "o/r#eff", { status: "opened", wave: 1, prStatus: "converging", prDerivedOverride: "merged" });
  const byWaveIdx = db
    .prepare("SELECT wave, task_index, status FROM plan_wave_tasks WHERE plan_key = ? ORDER BY task_index")
    .all("o/r#eff") as Array<Record<string, unknown>>;
  assertEquals(byWaveIdx.map((r) => r.status), ["merged", "opened", "blocked", "merged"]);
  // The count VIEW and the per-task grid now agree on the merged tally for wave 0 (3 slices, 1 merged).
  const c = db.prepare("SELECT merged, total FROM plan_wave_counts WHERE plan_key = ? AND wave = 0").get("o/r#eff") as Record<string, unknown>;
  assertEquals({ merged: Number(c.merged), total: Number(c.total) }, { merged: 1, total: 3 });
});

test("the operator pages bind the derived plan-family VIEWs (never the raw plans table)", () => {
  // Overview + Epic index + Epic detail all read the composite `plan_read_model`; the epic-detail
  // per-wave summary reads `plan_wave_summary` (the bar), and the wave-state grid `plan_wave_tasks`.
  const overview = PAGE("overview.page.json");
  const epics = (overview.nodes ?? []).find((n: { id: string }) => n.id === "overview-epics");
  assert(epics, "overview must keep the epics grid");
  assertEquals(epics.props.data.table, "plan_read_model");

  const epicIndex = PAGE("epic.page.json");
  const epicPlans = (epicIndex.nodes ?? []).find((n: { props?: { data?: { table?: string } } }) => n.props?.data?.table === "plan_read_model");
  assert(epicPlans, "the Epics index grid must read the derived plan_read_model VIEW");

  const detail = PAGE("epic-detail.page.json");
  const byId = (id: string) => (detail.nodes ?? []).find((n: { id: string }) => n.id === id);
  assertEquals(byId("wave-banner").props.data.table, "plan_read_model");
  // The primary epic detail grid must also read the derived read model, not the raw `plans` table — a
  // regression pointing it at `plans` would otherwise pass this guard (suppressed advisory test:400).
  assertEquals(byId("epic-plan").props.data.table, "plan_read_model");
  assertEquals(byId("wave-summary").props.data.table, "plan_wave_summary");
  const summaryCols: string[] = byId("wave-summary").props.columns.map((c: { field: string }) => c.field);
  for (const f of ["wave", "bar", "merged", "in_flight", "blocked", "escalated", "skipped", "total"]) {
    assert(summaryCols.includes(f), `the summary grid shows ${f}`);
  }
  assertEquals(byId("wave-state").props.data.table, "plan_wave_tasks");
});

test("the app/delivery.ts adapters route through the framework backends (planDeliveryCounts.reduce + planReadModel.evaluate) — the wave rollups compose", () => {
  // A guard that the declared rollups are wired as the adapters' engine: the two count rollups fold the
  // same slice-PR partition, and the composed wave-progress rollup reads the wave-counts rollup.
  assertEquals(planWaveProgress.sourceRelations.slice().sort(), ["plan_tasks", "pull_requests__tracking"]);
  assertEquals(planWaveCounts.groupBy, ["plan_key", "wave"]);
  assertEquals(planDeliveryCounts.groupBy, ["plan_key"]);
  // The adapters produce the framework-folded counts (deriveDelivery is the reduce()+evaluate() façade).
  const r = deriveDelivery("done", ["merged", "waiting_review", "missing"]);
  assertEquals({ prsOpened: r.prsOpened, prsMerged: r.prsMerged, prsInFlight: r.prsInFlight }, { prsOpened: 3, prsMerged: 1, prsInFlight: 2 });
  assertEquals(r.delivery, "converging");
  assertEquals(r.label, "1/3 slices merged, 2 converging");
});