// nano-workforce — the app-side engine-reset reconciliation surface (issue #622).
//
// When the Nano engine is reset, restored, or rolled to an incarnation whose key generator has
// rewound (Magikcraft/nano-bpm#1065), `app.db` keeps projecting ENGINE-BACKED inflight work that no
// longer exists on the engine: open user tasks pointing at dead instances, active runs keyed on a
// `process_key` the fresh engine has re-minted for something unrelated (the release-train run
// recorded `process_key=41`; the fresh engine re-minted 41 for an unrelated probe). Without a
// supported way to converge, the app silently trusts stale projections — orphaned human gates and
// key-collision identity confusion.
//
// `reconcile` is the first-class remedy. It runs ON STARTUP (main.ts) and ON DEMAND (the
// `reconcileEngineState` operator command), scoped NARROWLY to claimed inflight work:
//
//   • Detection — engine INCARNATION EPOCH (preferred over fragile per-key 404 probing). The engine
//     stamps a monotonic incarnation id at boot and exposes it on `/v2/topology`; the app persists
//     the last-seen value (`engine_incarnation`). An epoch REGRESSION (observed < recorded) — the
//     #1065 rewind signature — or its absence where one was recorded means "engine was reset/rewound
//     → reconcile", ONE cheap check instead of N per-instance probes.
//   • Convergence — for every NON-terminal, engine-backed app row (a nano.app.json instanceTracking
//     binding whose `statusField` is still in its `activeStatuses` set and whose `keyField` is
//     populated), drive the row to the defined `orphaned` terminal WITH PROVENANCE
//     (`reconcile_provenance`: the reason, the observed engine epoch, and the reconcile run id) —
//     instead of trusting a stale projection or silently dropping data.
//   • Guardrails — TERMINAL rows (done/failed/merged/abandoned/…) and append-only / non-engine-backed
//     surfaces (presence, audit, provenance) are NEVER mutated: reconcile only touches rows whose
//     status is in a binding's `activeStatuses`. Every pass is recorded in `reconcile_runs`.
//   • Idempotent — a second pass with a matching epoch is a no-op (nothing regressed, and every
//     already-orphaned row has left its `activeStatuses`, so it is not re-scanned). An UNREACHABLE
//     engine is a no-op too: reconcile NEVER orphans when it could not confirm a reset (a 401/5xx or
//     a network error yields `reachable:false`, not a false "engine missing").
//
// A second, complementary pass covers the "instance absent/unknown" gap (issue #630) the epoch signal
// alone cannot: an inflight run whose engine instance has VANISHED from the read model — its
// `keyField` (process_key) has NO `_urban_instance_state` row at all (engine clean-reset, cluster
// rebuild, or read-model pruning removed it), so the derived tracking edge has no `TERMINATED` row to
// match and the run freezes at its last worker-owned status, wedging Active forever with no
// reconciliation path (the observed pre-reset orphan `Magikcraft/nano-bpm#1051`, process_key 71506).
// `reconcileVanishedInstances` drives every such row — active, dispatched, its key absent from
// `_urban_instance_state`, and PAST A GRACE WINDOW (so a still-starting run not yet projected is
// spared) — to the same `orphaned` terminal, with a DISTINCT provenance reason so an operator can
// tell a vanished-instance orphan apart from an epoch-regression one. Because `_urban_instance_state`
// is an app-side projection that can lag / be pruned / be rebuilt while the instance is still ACTIVE
// on the engine, "no projection row" is NOT proof the instance vanished — so before folding, the pass
// CROSS-CHECKS ENGINE TRUTH (`/v2/process-instances/search`, issue #736): an instance the engine still
// reports ACTIVE (or whose truth cannot be established) is SPARED, closing the false-orphan class on
// deployments whose engine omits the incarnation epoch. `runEngineReconcile` runs BOTH passes, so
// startup and the operator command converge both failure modes in one call.
//
// The provenance is app-owned (not urban's `_urban_write_provenance`, which is a domain-free
// insert-join sidecar written only inside a job): reconcile runs at boot / over HTTP, outside any
// job, and needs to record the REASON + epoch + run id — which the app-owned `reconcile_provenance`
// table carries, and the existing `app.db` backup convention makes the whole mutation reversible.

import type { DataLayer, GatewayDataSource as DataSource, InstanceTracking } from "@nanobpm/urban";
import type { TopologyProbe } from "./enginePreflight.ts";
import { activeStatusesFor, baseStatusFieldFor, engineBackedBindings, keyFieldFor } from "./instanceTracking.ts";

/** The defined terminal state a reset-orphaned engine-backed row is driven to. Deliberately DISTINCT
 *  from a binding's natural terminal (`abandoned`/`failed`/…) so an operator can tell a row that was
 *  orphaned by an engine reset apart from one that drained normally. Not in any binding's
 *  `activeStatuses`, so an orphaned row is never re-scanned (idempotency) nor re-polled by the urban
 *  instance-tracking reconciler. */
export const ORPHANED_STATUS = "orphaned";

/** The provenance reason stamped on every orphaned transition: the engine was reset/rewound and the
 *  recorded incarnation epoch regressed (the #1065 signature). */
export const RECONCILE_ORPHAN_REASON = "engine-reset/epoch-regression";

/** The provenance reason stamped when a row is orphaned because its engine instance VANISHED from the
 *  read model — the run's `keyField` (process_key) has NO `_urban_instance_state` row at all, so the
 *  instance is absent/unknown in engine truth (engine clean-reset, cluster rebuild, or read-model
 *  pruning removed the instance-state row entirely — issue #630). Deliberately DISTINCT from
 *  {@link RECONCILE_ORPHAN_REASON} so an operator can tell an epoch-regression orphan apart from a
 *  vanished-instance orphan, even though both land on the same `orphaned` terminal. */
export const RECONCILE_VANISHED_REASON = "engine-instance/vanished";

/** The default grace window (ms) a dispatched-but-instance-less row is spared before it is considered
 *  vanished. A run dispatched moments ago (its `process_key` set) has not yet been polled into
 *  `_urban_instance_state` by the instanceTracking reconciler (`pollMs` 5s + engine search latency),
 *  so it transiently looks "vanished". This window (comfortably larger than a poll cycle) keeps a
 *  legitimately-still-starting run from being folded to terminal prematurely (issue #630 AC #2). */
export const DEFAULT_VANISHED_GRACE_MS = 5 * 60_000;

/** The framework's canonical per-instance engine-lifecycle projection table (urban's
 *  `_urban_instance_state`, keyed by `process_instance_key`). The vanished-instance reconcile joins
 *  each engine-backed row's `keyField` against it: a run whose key has NO row here has no backing
 *  instance in engine truth. `_urban_` prefixed (framework bookkeeping) so it is provisioned by the
 *  runtime, not our migrations — the reconcile guards on its existence before acting. */
const INSTANCE_STATE_TABLE = "_urban_instance_state";

/** The single-row epoch ledger + its append-only run/provenance sidecars (migration 092). */
const INCARNATION_TABLE = "engine_incarnation";
const RUNS_TABLE = "reconcile_runs";
const PROVENANCE_TABLE = "reconcile_provenance";
/** The conventional last-touched timestamp column stamped on every status transition; orphaning
 *  refreshes it too, but only on the tables that actually declare it (introspected per binding). */
const UPDATED_AT_COLUMN = "updated_at";

/** What a `/v2/topology` epoch probe observed. `reachable:false` means the engine could not be
 *  confirmed (network error, or a non-2xx like 401/5xx) — reconcile then does NOTHING, so a
 *  transient outage can never be mistaken for a reset and orphan live work. `reachable:true` with a
 *  null `epoch` means the engine answered but exposes no incarnation id (e.g. a stock Camunda 8
 *  gateway, or before Magikcraft/nano-bpm#1068 ships). That null is a no-op ONLY when no epoch was
 *  ever recorded; if a concrete epoch WAS recorded, a now-null observation reads as a regression
 *  ("the epoch disappeared" — the reset signature), so reconcile orphans inflight work. See the
 *  decision table on {@link reconcileEngineBackedWork}. */
export interface EngineEpochObservation {
  reachable: boolean;
  epoch: number | null;
}

/** Why a reconcile pass acted (or did not). */
export type ReconcileReason =
  | "epoch-regression"
  | "seed-epoch"
  | "no-op"
  | "engine-unreachable"
  | "instance-vanished";

/** One orphaned engine-backed row. */
export interface OrphanedRow {
  table: string;
  pk: string;
  key: string | null;
  fromStatus: string;
}

/** The outcome of one reconcile pass — the same shape the run row records and the operator command
 *  returns. */
export interface ReconcileResult {
  runId: string;
  reason: ReconcileReason;
  observedEpoch: number | null;
  recordedEpoch: number | null;
  orphanedCount: number;
  orphaned: OrphanedRow[];
}

export interface ReconcileLog {
  info(msg: string): void;
  warn(msg: string): void;
}

export interface ReconcileOptions {
  /** Injectable clock (defaults to `Date`), so tests are deterministic. */
  now?: () => Date;
  /** Injectable run id (defaults to a random UUID). */
  runId?: string;
  /** The data source name to reconcile (defaults to the DataLayer's default source). */
  sourceName?: string;
  log?: ReconcileLog;
}

/** Options for the vanished-instance reconcile pass ({@link reconcileVanishedInstances}). */
export interface VanishedReconcileOptions extends ReconcileOptions {
  /** How long (ms) a dispatched-but-instance-less row is spared before it is folded to terminal, so a
   *  still-starting run (not yet projected into `_urban_instance_state`) is not orphaned prematurely.
   *  Defaults to {@link DEFAULT_VANISHED_GRACE_MS}. */
  graceMs?: number;
  /** Cross-check against ENGINE TRUTH before orphaning a candidate row (issue #736). Given the row's
   *  `keyField` (process instance key), it reports whether the engine still considers the instance
   *  ACTIVE. An ACTIVE instance is NEVER orphaned — the app-side `_urban_instance_state` projection is
   *  merely lagging — and an instance whose truth could not be established (`null`) is spared too (we
   *  never orphan what we could not confirm dead). Only a row the engine positively confirms is gone
   *  (`false`) is folded. When omitted the pass falls back to projection-only behaviour (no live
   *  cross-check); production always wires one via {@link runEngineReconcile}. See
   *  {@link makeEngineActiveProbe}. */
  engineActive?: EngineActiveProbe;
}

/** Read the incarnation epoch out of a `/v2/topology` body — `nano.incarnation` (or its `epoch`
 *  alias), coerced from a number or a numeric string. Any other shape (absent, non-numeric) yields
 *  null: "the engine exposes no epoch". A null is a no-op ONLY when no epoch was previously recorded;
 *  when one WAS recorded, reconcile reads a now-null observation as a regression ("epoch disappeared"),
 *  not a no-op — see the decision table on {@link reconcileEngineBackedWork}. */
export function parseEngineEpoch(body: TopologyProbe | null | undefined): number | null {
  const raw = body?.nano?.incarnation ?? body?.nano?.epoch;
  if (raw == null) return null;
  const n = typeof raw === "number" ? raw : Number(raw);
  return Number.isFinite(n) ? n : null;
}

/** Probe `/v2/topology` for the engine incarnation epoch. Never throws: a network error or a non-2xx
 *  yields `reachable:false` (reconcile then does nothing), so an outage can never orphan live work. */
export async function probeEngineEpoch(
  restAddress: string,
  opts: { token?: string; fetchImpl?: typeof fetch; timeoutMs?: number } = {},
): Promise<EngineEpochObservation> {
  const fetchImpl = opts.fetchImpl ?? fetch;
  const url = `${restAddress.replace(/\/+$/, "")}/topology`;
  const headers: Record<string, string> = { accept: "application/json" };
  if (opts.token) headers.authorization = `Bearer ${opts.token}`;
  try {
    const res = await fetchImpl(url, {
      headers,
      signal: AbortSignal.timeout(opts.timeoutMs ?? 3000),
    });
    if (!res.ok) return { reachable: false, epoch: null };
    const body: TopologyProbe = await res.json();
    return { reachable: true, epoch: parseEngineEpoch(body) };
  } catch {
    return { reachable: false, epoch: null };
  }
}

/** A cross-check against ENGINE TRUTH for one process instance key, used to spare a live instance from
 *  the vanished-instance pass. Returns:
 *   • `true`  — the engine reports the instance ACTIVE. It is live; the app-side `_urban_instance_state`
 *               projection is merely lagging/pruned/rebuilding, so the row MUST NOT be orphaned.
 *   • `false` — the engine answered and the instance is NOT active: absent from the read model, or in a
 *               known terminal state ({@link ENGINE_TERMINAL_STATES}). It is genuinely gone in engine
 *               truth → orphan-eligible.
 *   • `null`  — engine truth could NOT be established (unreachable, non-2xx, malformed — including an
 *               item whose `state` is missing or outside {@link ENGINE_TERMINAL_STATES}). We never
 *               orphan a row we could not confirm dead, so the caller spares it (a transient outage
 *               must never fold live work). */
export type EngineActiveProbe = (processKey: string) => Promise<boolean | null>;

/** One item of a `/v2/process-instances/search` result, narrowed to what the engine-truth cross-check
 *  reads: the instance key (to match the row we probed for) and its lifecycle `state`. Keys are
 *  stringified defensively (the wire may send a JSON number or string); `state` is the engine's
 *  lifecycle enum (`ACTIVE`/`COMPLETED`/`CANCELED`/…). */
interface InstanceSearchStateItem {
  processInstanceKey?: string | number;
  state?: string;
}

/** The lifecycle states that POSITIVELY mean "this instance is no longer running" — urban's
 *  `ProcessInstanceState` terminals (`COMPLETED`/`TERMINATED`) plus the Camunda-8-parity v2 REST
 *  terminals (`CANCELED`/`FAILED`), since the probe reads that raw surface rather than the typed
 *  client. ONLY these may answer `false` (orphan-eligible): a `state` that is missing, empty, or
 *  outside this set is a partial/malformed read this app cannot interpret, so it degrades to `null`
 *  ("unknown" → the caller spares the row). Classifying an unrecognized state as "gone" would fold
 *  live work off a wire shape we misread — the exact failure mode the #736 cross-check exists to stop. */
export const ENGINE_TERMINAL_STATES: ReadonlySet<string> = new Set(["COMPLETED", "TERMINATED", "CANCELED", "FAILED"]);

/** Build an {@link EngineActiveProbe} that queries the engine's own `/v2/process-instances/search` for
 *  a single process instance key and reports whether the engine still considers it ACTIVE. This is the
 *  authoritative engine-truth check the vanished-instance pass consults before orphaning: an ACTIVE
 *  engine instance must NEVER be orphaned regardless of the app-side projection (issue #736), so a
 *  merlin-style deployment whose `_urban_instance_state` lags no longer false-orphans live work.
 *  Never throws — every transport/parse failure, and every `state` this app cannot interpret (missing,
 *  or outside {@link ENGINE_TERMINAL_STATES}), degrades to `null` ("unknown"), which the caller treats
 *  as "spare" (we never orphan what we could not confirm dead). */
export function makeEngineActiveProbe(
  engineRest: { restAddress: string; token?: string },
  opts: { fetchImpl?: typeof fetch; timeoutMs?: number } = {},
): EngineActiveProbe {
  const fetchImpl = opts.fetchImpl ?? fetch;
  const base = engineRest.restAddress.replace(/\/+$/, "");
  const headers: Record<string, string> = { accept: "application/json", "content-type": "application/json" };
  if (engineRest.token) headers.authorization = `Bearer ${engineRest.token}`;
  return async (processKey: string): Promise<boolean | null> => {
    try {
      const res = await fetchImpl(`${base}/process-instances/search`, {
        method: "POST",
        headers,
        body: JSON.stringify({ filter: { processInstanceKey: processKey }, page: { from: 0, limit: 10 } }),
        signal: AbortSignal.timeout(opts.timeoutMs ?? 3000),
      });
      if (!res.ok) return null;
      // biome-ignore lint/plugin: runtime/framework contract boundary for external data shape
      const body = (await res.json()) as { items?: InstanceSearchStateItem[] };
      const items = body.items ?? [];
      const match = items.find((it) => it.processInstanceKey != null && String(it.processInstanceKey) === processKey);
      // Engine answered but the instance is absent from the read model → genuinely gone (not active).
      if (!match) return false;
      const state = String(match.state ?? "").trim().toUpperCase();
      if (state === "ACTIVE") return true;
      // A KNOWN terminal state is a positive "gone". Anything else — a missing/empty `state`, or one
      // outside the enum this app can interpret — is a partial or malformed answer, NOT a confirmed
      // death, so it degrades to `null` and the caller spares the row.
      return ENGINE_TERMINAL_STATES.has(state) ? false : null;
    } catch {
      return null;
    }
  };
}

/** Double-quote a SQL identifier (table/column) so a manifest-declared name is safe to interpolate. */
function q(id: string): string {
  return `"${id.replace(/"/g, '""')}"`;
}

/** The schema of `table` we need to orphan a row: its primary-key column (the first `pk`-flagged
 *  column from `PRAGMA table_info`, or `rowid` when the table declares none — so provenance always
 *  records a stable row identity) and whether it carries an `updated_at` column to stamp. */
async function tableShape(src: DataSource, table: string): Promise<{ pkCol: string; hasUpdatedAt: boolean }> {
  const cols = await src.query<{ name: string; pk: number }>(`PRAGMA table_info(${q(table)})`);
  const pk = cols.find((c) => Number(c.pk) > 0);
  return { pkCol: pk?.name ?? "rowid", hasUpdatedAt: cols.some((c) => c.name === UPDATED_AT_COLUMN) };
}

/** The recorded last-seen epoch, or null when none was ever recorded (no row, or a null epoch). */
async function readRecordedEpoch(src: DataSource): Promise<number | null> {
  const rows = await src.query<{ epoch: number | null }>(
    `SELECT epoch FROM ${INCARNATION_TABLE} WHERE id = 1`,
  );
  const epoch = rows.length ? rows[0].epoch : null;
  return epoch == null ? null : Number(epoch);
}

/** Persist (seed or advance) the last-seen epoch. Only ever called with a concrete number, so a
 *  recorded epoch always means "an epoch was actually observed". */
async function persistEpoch(src: DataSource, epoch: number, at: string): Promise<void> {
  await src.exec(
    `INSERT INTO ${INCARNATION_TABLE} (id, epoch, observed_at) VALUES (1, ?, ?) ` +
      `ON CONFLICT(id) DO UPDATE SET epoch = excluded.epoch, observed_at = excluded.observed_at`,
    [epoch, at],
  );
}

/** The resolved schema + tracking selectors for one engine-backed binding, or `null` when the binding
 *  carries no `activeStatuses` selector (it cannot classify "in-flight", so it is skipped). */
interface BindingShape {
  table: string;
  active: readonly string[];
  statusField: string;
  keyField: string;
  pkCol: string;
  hasUpdatedAt: boolean;
}

/** The shape of a row selected for possible orphaning. */
interface OrphanCandidate {
  __pk: unknown;
  __key: unknown;
  __status: unknown;
  __updated?: unknown;
}

/** Resolve a binding's tracking selectors + physical schema, or `null` to skip a selector-less
 *  binding (activeStatusesFor would throw; we tolerate it here). */
async function resolveShape(src: DataSource, binding: InstanceTracking): Promise<BindingShape | null> {
  if (!binding.activeStatuses?.length) return null;
  const table = binding.table;
  const { pkCol, hasUpdatedAt } = await tableShape(src, table);
  return {
    table,
    active: activeStatusesFor(table),
    statusField: baseStatusFieldFor(table),
    keyField: keyFieldFor(table),
    pkCol,
    hasUpdatedAt,
  };
}

/** Drive ONE candidate row to `orphaned` with provenance, GUARDED: the UPDATE re-asserts the exact
 *  status read AND a populated key (plus any `extraGuardSql`, e.g. the still-vanished re-check), so a
 *  writer that flipped the row to a newer terminal status (or an instance that reappeared) between the
 *  SELECT and this UPDATE wins the race — we never clobber that history back to `orphaned`. Only a row
 *  we actually transitioned (`res.changed > 0`) gets provenance and is returned. `updated_at` is
 *  stamped (when the table has one) so the transition refreshes the row's timestamp like every other
 *  status transition in the codebase. Runs inside the caller's transaction. */
async function orphanRow(
  src: DataSource,
  shape: BindingShape,
  row: OrphanCandidate,
  reason: string,
  observedEpoch: number | null,
  runId: string,
  at: string,
  extraGuardSql = "",
): Promise<OrphanedRow | null> {
  const { table, pkCol, statusField, keyField, hasUpdatedAt } = shape;
  const pk = String(row.__pk);
  const key = row.__key == null ? null : String(row.__key);
  const fromStatus = String(row.__status);
  const res = await src.exec(
    `UPDATE ${q(table)} SET ${q(statusField)} = ?` +
      (hasUpdatedAt ? `, ${q(UPDATED_AT_COLUMN)} = ?` : "") +
      ` WHERE ${q(pkCol)} = ? AND ${q(statusField)} = ? AND ${q(keyField)} IS NOT NULL${extraGuardSql}`,
    hasUpdatedAt ? [ORPHANED_STATUS, at, row.__pk, fromStatus] : [ORPHANED_STATUS, row.__pk, fromStatus],
  );
  if (res.changed <= 0) return null;
  await src.exec(
    `INSERT INTO ${PROVENANCE_TABLE} ` +
      `(run_id, source_table, pk_value, key_value, from_status, to_status, reason, observed_epoch, at) ` +
      `VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    [runId, table, pk, key, fromStatus, ORPHANED_STATUS, reason, observedEpoch, at],
  );
  return { table, pk, key, fromStatus };
}

/** Orphan every NON-terminal, engine-backed row across all instanceTracking bindings, recording one
 *  `reconcile_provenance` row per transition. Runs inside the caller's transaction. */
async function orphanEngineBackedRows(
  src: DataSource,
  runId: string,
  observedEpoch: number | null,
  at: string,
): Promise<OrphanedRow[]> {
  const orphaned: OrphanedRow[] = [];
  for (const binding of engineBackedBindings()) {
    const shape = await resolveShape(src, binding);
    if (!shape) continue;
    const placeholders = shape.active.map(() => "?").join(", ");
    const rows = await src.query<OrphanCandidate>(
      `SELECT ${q(shape.pkCol)} AS __pk, ${q(shape.keyField)} AS __key, ${q(shape.statusField)} AS __status ` +
        `FROM ${q(shape.table)} WHERE ${q(shape.statusField)} IN (${placeholders}) AND ${q(shape.keyField)} IS NOT NULL`,
      [...shape.active],
    );
    for (const row of rows) {
      const o = await orphanRow(src, shape, row, RECONCILE_ORPHAN_REASON, observedEpoch, runId, at);
      if (o) orphaned.push(o);
    }
  }
  return orphaned;
}

/** Whether a row's `updated_at` is younger than the grace window — i.e. it was (re)dispatched too
 *  recently to have been projected into `_urban_instance_state` yet, so it must NOT be folded. A
 *  null/unparseable timestamp is treated as "within grace" (spared): when we cannot establish a row's
 *  age we must NOT orphan it — a nullable `updated_at` (e.g. `delivery_units.updated_at`,
 *  db/migrations/088_delivery_units.sql) would otherwise fold a live row. Erring toward sparing at
 *  worst leaves a genuinely-vanished ageless row for a later pass once it carries a usable timestamp;
 *  erring the other way wedges/destroys a live run, so we choose the safe default. */
function withinGrace(updated: unknown, nowMs: number, graceMs: number): boolean {
  if (updated == null) return true;
  const t = Date.parse(String(updated));
  if (!Number.isFinite(t)) return true;
  return nowMs - t < graceMs;
}

/** A vanished-instance candidate: the row selected for possible orphaning plus the binding shape that
 *  resolves its physical schema. Selected OUTSIDE any transaction, because the engine-truth
 *  cross-check that narrows these is network I/O (see {@link confirmVanishedGone}). */
interface VanishedCandidate {
  shape: BindingShape;
  row: OrphanCandidate;
}

/** SELECT every NON-terminal, engine-backed row whose `keyField` (process instance key) has NO
 *  `_urban_instance_state` row — the instance is absent/unknown in the app-side read model (vanished,
 *  issue #630) — and whose last transition is older than the grace window. READ-ONLY and network-free,
 *  so it is safe to run outside the orphaning transaction. */
async function selectVanishedCandidates(
  src: DataSource,
  nowMs: number,
  graceMs: number,
): Promise<VanishedCandidate[]> {
  const candidates: VanishedCandidate[] = [];
  for (const binding of engineBackedBindings()) {
    const shape = await resolveShape(src, binding);
    if (!shape) continue;
    const placeholders = shape.active.map(() => "?").join(", ");
    const updatedSel = shape.hasUpdatedAt ? `, ${q(UPDATED_AT_COLUMN)} AS __updated` : "";
    // Active, dispatched (key populated) rows whose engine instance key has NO matching
    // `_urban_instance_state` row — absent/unknown in engine truth.
    const rows = await src.query<OrphanCandidate>(
      `SELECT ${q(shape.pkCol)} AS __pk, ${q(shape.keyField)} AS __key, ${q(shape.statusField)} AS __status${updatedSel} ` +
        `FROM ${q(shape.table)} b WHERE ${q(shape.statusField)} IN (${placeholders}) AND ${q(shape.keyField)} IS NOT NULL ` +
        `AND NOT EXISTS (SELECT 1 FROM ${q(INSTANCE_STATE_TABLE)} s WHERE s.process_instance_key = b.${q(shape.keyField)})`,
      [...shape.active],
    );
    for (const row of rows) {
      if (shape.hasUpdatedAt && withinGrace(row.__updated, nowMs, graceMs)) continue;
      candidates.push({ shape, row });
    }
  }
  return candidates;
}

/** Narrow candidates to the ones ENGINE TRUTH positively confirms are gone (issue #736): the
 *  `_urban_instance_state` projection is an app-side read model that can lag / be pruned / be rebuilt
 *  while the instance is still ACTIVE on the engine, so "no projection row" is NOT "instance vanished".
 *  An ACTIVE instance (`true`) is spared, and one whose truth could not be established (`null` — engine
 *  unreachable, malformed answer, or a probe that THREW) is spared too: we never orphan a row we could
 *  not positively confirm is gone, and an injected probe's failure must never abort the pass. Only a
 *  `false` (engine confirms absent/terminated) survives. Runs OUTSIDE the DB transaction — this is
 *  network I/O, and awaiting it under an open write transaction would hold the SQLite lock for up to
 *  the probe's timeout PER ROW, stalling every other writer (including boot). Without a probe (omitted)
 *  the pass falls back to projection-only behaviour, so every candidate survives. */
async function confirmVanishedGone(
  candidates: VanishedCandidate[],
  engineActive?: EngineActiveProbe,
): Promise<VanishedCandidate[]> {
  if (!engineActive) return candidates;
  // Engine truth is PER INSTANCE, not per row, and one instance can back several tracked rows: the
  // `delivery_units` aggregate is a DB-trigger mirror of its legacy base row (db/migrations/089) and
  // carries the same `process_key`, so both are candidates for one vanished instance. Probe each
  // DISTINCT key once and apply that verdict to every row carrying it — the same answer, without
  // doubling engine calls that can each block for the probe's full timeout.
  const verdicts = new Map<string, boolean | null>();
  const gone: VanishedCandidate[] = [];
  for (const candidate of candidates) {
    const key = candidate.row.__key == null ? null : String(candidate.row.__key);
    // Defensive: the SELECT requires a populated key, so with no key there is nothing to cross-check
    // and the projection-only verdict stands.
    if (key == null) {
      gone.push(candidate);
      continue;
    }
    let verdict = verdicts.get(key);
    if (verdict === undefined) {
      try {
        verdict = await engineActive(key);
      } catch {
        // A probe that throws established nothing — treat it exactly like an unreachable engine.
        verdict = null;
      }
      verdicts.set(key, verdict);
    }
    if (verdict === false) gone.push(candidate);
  }
  return gone;
}

/** Fold each confirmed-gone candidate to the `orphaned` terminal, recording one
 *  `reconcile_provenance` row per transition (reason {@link RECONCILE_VANISHED_REASON}). WRITE-ONLY and
 *  network-free, so the caller's transaction stays short. Runs inside the caller's transaction. */
async function orphanVanishedCandidates(
  t: DataSource,
  candidates: VanishedCandidate[],
  runId: string,
  at: string,
): Promise<OrphanedRow[]> {
  const orphaned: OrphanedRow[] = [];
  for (const { shape, row } of candidates) {
    // Re-assert "still no instance-state row" in the guarded UPDATE (which also re-asserts the exact
    // status read), so a row that went terminal — or an instance that reappeared, the poller recording
    // it — between the out-of-transaction SELECT/probe and this UPDATE wins the race.
    const stillVanishedGuard =
      ` AND NOT EXISTS (SELECT 1 FROM ${q(INSTANCE_STATE_TABLE)} s ` +
      `WHERE s.process_instance_key = ${q(shape.table)}.${q(shape.keyField)})`;
    const o = await orphanRow(t, shape, row, RECONCILE_VANISHED_REASON, null, runId, at, stillVanishedGuard);
    if (o) orphaned.push(o);
  }
  return orphaned;
}

/** Whether the framework `_urban_instance_state` projection exists in this source. When it does NOT,
 *  the vanished-instance pass is a hard no-op: without the projection every dispatched row would look
 *  "vanished", so we must never orphan on its absence. */
async function instanceStateTableExists(src: DataSource): Promise<boolean> {
  const rows = await src.query<{ n: number }>(
    `SELECT COUNT(*) AS n FROM sqlite_master WHERE type = 'table' AND name = ?`,
    [INSTANCE_STATE_TABLE],
  );
  return rows.length > 0 && Number(rows[0].n) > 0;
}

/**
 * Reconcile the app's engine-backed projections against one epoch observation. Pure of I/O beyond the
 * data layer (the topology probe is {@link probeEngineEpoch}, injected as `observation`), so the
 * red/green test drives it with a seeded row + a regressed epoch directly.
 *
 * Decision table (engine reachable):
 *   • recorded != null AND (observed == null OR observed < recorded)  → REGRESSION: orphan inflight.
 *   • recorded == null AND observed != null                          → SEED: first epoch learned.
 *   • otherwise                                                       → NO-OP (incl. a matching epoch).
 * The epoch is persisted whenever a concrete one was observed (seed, advance, or the fresh
 * post-rewind incarnation), so the very next pass with that same epoch is a pure no-op.
 *
 * This surface is deliberately EPOCH-SCOPED: it only ever acts on the epoch signal. An engine that
 * is reachable but exposes NO epoch and for which none was ever recorded (observed == null AND
 * recorded == null) is, by contract, an intentional no-op — we have no reset signal to act on, and
 * we never orphan live work speculatively. Converging engine-backed rows against an engine with no
 * epoch support (e.g. via a per-instance existence probe) is out of contract for this surface.
 */
export async function reconcileEngineBackedWork(
  data: DataLayer,
  observation: EngineEpochObservation,
  opts: ReconcileOptions = {},
): Promise<ReconcileResult> {
  const src = data.open(opts.sourceName);
  const at = (opts.now?.() ?? new Date()).toISOString();
  const runId = opts.runId ?? crypto.randomUUID();

  // An unreachable engine is a hard no-op: we could not confirm a reset, so we NEVER orphan.
  if (!observation.reachable) {
    const recorded = await readRecordedEpoch(src);
    await recordRun(src, { runId, at, observedEpoch: null, recordedEpoch: recorded, reason: "engine-unreachable", orphanedCount: 0 });
    opts.log?.warn("reconcile: engine unreachable — skipped (no rows orphaned; live work left intact).");
    return { runId, reason: "engine-unreachable", observedEpoch: null, recordedEpoch: recorded, orphanedCount: 0, orphaned: [] };
  }

  const observedEpoch = observation.epoch;
  const recordedEpoch = await readRecordedEpoch(src);
  const regression = recordedEpoch != null && (observedEpoch == null || observedEpoch < recordedEpoch);

  const result = await src.tx(async (t) => {
    let orphaned: OrphanedRow[] = [];
    let reason: ReconcileReason;
    if (regression) {
      orphaned = await orphanEngineBackedRows(t, runId, observedEpoch, at);
      reason = "epoch-regression";
    } else if (recordedEpoch == null && observedEpoch != null) {
      reason = "seed-epoch";
    } else {
      reason = "no-op";
    }
    if (observedEpoch != null) await persistEpoch(t, observedEpoch, at);
    await recordRun(t, { runId, at, observedEpoch, recordedEpoch, reason, orphanedCount: orphaned.length });
    return { reason, orphaned };
  });

  if (result.reason === "epoch-regression") {
    opts.log?.warn(
      `reconcile: engine epoch regressed ${recordedEpoch} → ${observedEpoch} (reset/rewind) — ` +
        `orphaned ${result.orphaned.length} inflight row(s) [run ${runId}].`,
    );
  } else if (result.reason === "seed-epoch") {
    opts.log?.info(`reconcile: recorded engine epoch ${observedEpoch} (first observation) [run ${runId}].`);
  } else {
    opts.log?.info(`reconcile: engine epoch ${observedEpoch ?? "n/a"} unchanged — no-op [run ${runId}].`);
  }

  return { runId, reason: result.reason, observedEpoch, recordedEpoch, orphanedCount: result.orphaned.length, orphaned: result.orphaned };
}

/**
 * Reconcile engine-backed inflight work against the framework's canonical per-instance projection
 * (`_urban_instance_state`) — the "instance absent/unknown" gap (issue #630), DISTINCT from the
 * epoch-regression reset the {@link reconcileEngineBackedWork} pass handles.
 *
 * When an engine instance VANISHES from the read model — engine clean-reset, cluster rebuild, or
 * read-model pruning removes the `_urban_instance_state` row entirely — there is no `TERMINATED` row
 * for the derived tracking edge to match, so the run freezes at its last worker-owned status
 * (`escalated`/`awaiting_operator`) and wedges the Active list forever with no reconciliation path.
 * "The instance backing this run no longer exists in engine truth" is a terminal condition: this pass
 * drives every such row (active, dispatched, its `keyField` absent from `_urban_instance_state`, and
 * past the grace window) to the defined `orphaned` terminal WITH PROVENANCE.
 *
 * Safety:
 *   • GRACE WINDOW — a just-dispatched run has not yet been polled into `_urban_instance_state`; only
 *     rows whose last transition is older than `graceMs` are folded, so a still-starting run is never
 *     prematurely orphaned (AC #2).
 *   • PROJECTION-PRESENT — if `_urban_instance_state` does not exist (the runtime has not provisioned
 *     it), every dispatched row would look vanished, so the pass is a hard no-op.
 *   • GUARDED — the same status-re-assert as the epoch pass, plus a still-vanished re-check, so a
 *     concurrent terminal write or a reappearing instance wins the race.
 *   • ENGINE-TRUTH CROSS-CHECK (issue #736) — the `_urban_instance_state` projection is an app-side
 *     read model that can lag / be pruned / be rebuilt for an instance that is still ACTIVE on the
 *     engine, so "no projection row" is NOT "instance vanished". Before folding, the pass consults
 *     `opts.engineActive` (production wires one from the live engine via {@link makeEngineActiveProbe};
 *     {@link runEngineReconcile}): an instance the engine reports ACTIVE — or whose truth could not be
 *     established (engine unreachable, malformed answer, a probe that threw) — is SPARED. Only an
 *     instance the engine positively confirms is gone is orphaned. Without a cross-check (omitted) the
 *     pass falls back to projection-only.
 *   • SHORT TRANSACTION — the cross-check is network I/O, so the candidate SELECT and every probe run
 *     OUTSIDE `src.tx(...)`; the transaction covers only the guarded UPDATE + provenance writes. A
 *     slow or unreachable engine therefore cannot hold the SQLite write lock open (for up to the
 *     probe's timeout per candidate row), stalling other writers or boot. The guards make the split
 *     safe: a row that went terminal, or an instance that reappeared in the projection, between the
 *     out-of-transaction read and the in-transaction UPDATE wins the race and is not folded.
 */
export async function reconcileVanishedInstances(
  data: DataLayer,
  opts: VanishedReconcileOptions = {},
): Promise<ReconcileResult> {
  const src = data.open(opts.sourceName);
  const clock = opts.now?.() ?? new Date();
  const at = clock.toISOString();
  const nowMs = clock.getTime();
  const runId = opts.runId ?? crypto.randomUUID();
  const graceMs = opts.graceMs ?? DEFAULT_VANISHED_GRACE_MS;

  // Without the framework projection we cannot tell a vanished instance from a live one — every
  // dispatched row would look vanished. NO-OP rather than orphan live work.
  if (!(await instanceStateTableExists(src))) {
    await recordRun(src, { runId, at, observedEpoch: null, recordedEpoch: null, reason: "no-op", orphanedCount: 0 });
    opts.log?.info(`reconcile(vanished): instance-state projection absent — no-op [run ${runId}].`);
    return { runId, reason: "no-op", observedEpoch: null, recordedEpoch: null, orphanedCount: 0, orphaned: [] };
  }

  // READ + PROBE first, transaction second: the engine-truth cross-check is network I/O and must never
  // be awaited under an open write transaction (see SHORT TRANSACTION above).
  const candidates = await selectVanishedCandidates(src, nowMs, graceMs);
  const confirmedGone = await confirmVanishedGone(candidates, opts.engineActive);

  const orphaned = await src.tx(async (t) => {
    const rows = await orphanVanishedCandidates(t, confirmedGone, runId, at);
    const reason: ReconcileReason = rows.length > 0 ? "instance-vanished" : "no-op";
    await recordRun(t, { runId, at, observedEpoch: null, recordedEpoch: null, reason, orphanedCount: rows.length });
    return rows;
  });

  if (orphaned.length > 0) {
    opts.log?.warn(
      `reconcile(vanished): orphaned ${orphaned.length} inflight row(s) whose engine instance ` +
        `vanished from the read model [run ${runId}].`,
    );
  } else {
    opts.log?.info(`reconcile(vanished): no vanished instances — no-op [run ${runId}].`);
  }
  return {
    runId,
    reason: orphaned.length > 0 ? "instance-vanished" : "no-op",
    observedEpoch: null,
    recordedEpoch: null,
    orphanedCount: orphaned.length,
    orphaned,
  };
}

async function recordRun(
  src: DataSource,
  run: { runId: string; at: string; observedEpoch: number | null; recordedEpoch: number | null; reason: ReconcileReason; orphanedCount: number },
): Promise<void> {
  await src.exec(
    `INSERT INTO ${RUNS_TABLE} (run_id, started_at, observed_epoch, recorded_epoch, reason, orphaned_count) ` +
      `VALUES (?, ?, ?, ?, ?, ?)`,
    [run.runId, run.at, run.observedEpoch, run.recordedEpoch, run.reason, run.orphanedCount],
  );
}

/** Probe the engine's incarnation epoch, then reconcile — the wiring both startup (main.ts) and the
 *  `reconcileEngineState` operator command share, so the two paths can never diverge. Runs BOTH the
 *  epoch-regression pass (engine reset/rewind) and the vanished-instance pass (an inflight run whose
 *  instance is absent/unknown in `_urban_instance_state` — issue #630), returning ONE merged result.
 *  Reason precedence: `epoch-regression` always wins; otherwise, if the vanished pass orphaned any
 *  rows the reason is `instance-vanished` (even when the epoch pass reported a non-regression state
 *  such as `engine-unreachable`); otherwise the epoch pass's reason stands. `orphanedCount` /
 *  `orphaned` cover both passes. The two passes never double-fold a row: once the epoch pass orphans a
 *  row it leaves `activeStatuses`, so the vanished pass no longer selects it. */
export async function runEngineReconcile(
  data: DataLayer,
  engineRest: { restAddress: string; token?: string },
  opts: VanishedReconcileOptions & { fetchImpl?: typeof fetch } = {},
): Promise<ReconcileResult> {
  const observation = await probeEngineEpoch(engineRest.restAddress, {
    token: engineRest.token,
    fetchImpl: opts.fetchImpl,
  });
  const epoch = await reconcileEngineBackedWork(data, observation, opts);
  // A distinct run id so the vanished pass's `reconcile_runs`/provenance rows never collide with the
  // epoch pass's (run_id is a PRIMARY KEY). DERIVE it from the epoch pass's resolved run id — which is
  // also the merged result's `runId` — so it is `<runId>-vanished` on EVERY path, including the boot
  // path where `opts.runId` is omitted (a bare random UUID here would be non-correlatable to the
  // returned `runId`). Operators can always locate the vanished pass's provenance from the reported id.
  const vanished = await reconcileVanishedInstances(data, {
    ...opts,
    runId: `${epoch.runId}-vanished`,
    // Cross-check ENGINE TRUTH before orphaning any vanished-instance candidate (issue #736): an
    // instance the engine still reports ACTIVE is spared even when its `_urban_instance_state`
    // projection is absent (the merlin false-orphan: the projection lags, the instance is live). A
    // caller-supplied `engineActive` wins (tests inject a deterministic one); otherwise build one from
    // the same engine address/token the epoch probe used.
    engineActive: opts.engineActive ?? makeEngineActiveProbe(engineRest, { fetchImpl: opts.fetchImpl }),
  });

  const orphaned = [...epoch.orphaned, ...vanished.orphaned];
  const reason: ReconcileReason =
    epoch.reason === "epoch-regression"
      ? "epoch-regression"
      : vanished.orphanedCount > 0
        ? "instance-vanished"
        : epoch.reason;
  return {
    runId: epoch.runId,
    reason,
    observedEpoch: epoch.observedEpoch,
    recordedEpoch: epoch.recordedEpoch,
    orphanedCount: orphaned.length,
    orphaned,
  };
}