/**
 * One tick: the periodic pass that sequences every other pass in this
 * directory.
 *
 * Nothing here decides anything on its own. `tick` reads the config afresh at
 * its own boundary, builds the one `Deps` the pass will use, and then calls the
 * modules beside it in the order their side effects require — supervision before
 * dispatch so a wedged orchestrator is known about before work is claimed,
 * settlement sweeps after dispatch so a run that just ended is swept on the same
 * pass, grooming last so a queue refilled by promotion is picked up by the next
 * tick rather than mid-pass. That ordering is the only thing this module owns,
 * and keeping it a call sequence rather than a set of registered hooks is what
 * makes it readable in one screen.
 *
 * The three probes that ride along — install surfaces, upgrade verification,
 * the DB snapshot cadence — live here rather than with their subjects because
 * they have no subject beyond "once in a while", and their schedule is the
 * tick's business.
 */
import { existsSync } from "node:fs";
import { admitCandidates, startOfToday, type AdmissionHold } from "../admission.ts";
import { dbBackupDirFor, findProject, loadConfig, resolveCaps, stateDir } from "../config.ts";
import { evaluateDecisionConditions, probeNpmVersion, probeRateLimitReset } from "../decisions.ts";
import { dbSnapshotDue, dbSnapshotMarkerKey, localDayKey } from "../digest-schedule.ts";
import { reconcileOrphanedPause, runDoctor } from "../doctor.ts";
import { fleetLayers, herdrPaneOmpStarts, resolveHerdrSession } from "../fleet.ts";
import { projectLabels } from "../label-projection.ts";
import { healthCheck } from "../lifecycle.ts";
import { errText, log, safeEscalate } from "../log.ts";
import { DEFAULT_MINING_WINDOW_MS, mineSignals, type MinedSignal } from "../mining.ts";
import { reconcileOrchestratorDown } from "../orchestrator-down.ts";
import { clearPauseIfUnchanged, isPaused, pauseInstance, pauseInstanceAt, pausedPath, setPaused } from "../pause.ts";
import { enqueueAvailableHeldNotices } from "../reports.ts";
import { effectiveLabels, isEligible, route } from "../routing.ts";
import { adoptSalvagedPrs, classifyAndRecover, reconcileGroomingClosures, reconcileStaleLabels, settlePushedGreen } from "../settlement.ts";
import { ACTIVE_STATES, DB_SNAPSHOT_RETENTION, dbPath, pruneDbSnapshots, snapshotDb } from "../store.ts";
import { DEFAULT_REPORT_POLICY, type ConductorConfig, type IssueSnapshot, type ReportingPolicy, type Store } from "../types.ts";
import { appendJournal, launchTransientUnit, readUpgradeJournal } from "../upgrade-journal.ts";
import { processStartTimeMs, runCommand, verifyPendingUpgrade, type UpgradeVerifyDeps } from "../upgrade-verify.ts";
import { inspectSurfaces } from "../upgrade.ts";
import { writeAdmissionAck } from "./ack.ts";
import { dispatchAdmissions, summarizeDispatch, summarizeHeldPass, type WorkerPool } from "./admission-pass.ts";
import { NO_ISSUE, PACKAGE_SRC_DIR, UNROUTABLE_TEXT, type Deps } from "./deps.ts";
import { handleIssue } from "./dispatch.ts";
import { consumeDrain, markDrained } from "./drain.ts";
import { dispatchToSpecGrooming } from "./groom-pass.ts";
import { INTEGRITY_SAMPLE, checkIntegrity, markPaged, packageManifest } from "./integrity.ts";
import { reconcilePanes } from "./panes.ts";
import { applyAdjudicationDispositions, dispatchReviewAdjudications, dispatchReviewRevisions } from "./review.ts";
import { cleanupRetainedRuns, watchBaseHealth, watchMergedBase } from "./settle-pass.ts";
import { wakeOrchestratorForMetConditions, watchOrchestrator, watchWorkerProgress } from "./supervision.ts";

/**
 * Record what this host has installed, once per dispatch pass (#919).
 *
 * Here rather than at render time, and this is the whole design of the slice:
 * the read costs three subprocesses, and calling it from the tick took the
 * tick's own suite from 8.4s to 83.4s while spawning three children every
 * fifteen minutes to answer a question that changes only when someone installs
 * something. The dispatch pass is already async and already spawns `gh`, so one
 * read per pass is free by comparison, and every cheap surface then reads a row.
 *
 * Advisory throughout: a probe that throws (no `herdr` on PATH, a `$PATH`
 * without `omp`) leaves the previous observation in place and logs. A stale
 * observation is still the truth about the last time anyone could look, and
 * losing a dispatch pass over a version string would be absurd.
 */
export async function recordInstallSurfaces(d: Deps): Promise<void> {
  if (d.probeInstallSurfaces === undefined) return;
  try {
    const surfaces = await d.probeInstallSurfaces();
    // The telegram peer's versions ride along on the same pass (#961). Read
    // here rather than in `status` because `status` runs every tick and this
    // touches the npm registry: one periodic read, rendered from the store as
    // often as anyone looks. A read that throws leaves the fields absent, which
    // renders as unverified — never as agreement.
    const telegram =
      d.probeTelegramFreshness === undefined ? undefined : await d.probeTelegramFreshness();
    const tgInstalled = telegram?.surfaces.installed;
    const tgDaemon = telegram?.surfaces.daemon;
    const tgPublished = telegram?.surfaces.published;
    d.store.recordInstallSurfaces({
      at: Date.now(),
      cliVersion: surfaces.cliVersion,
      ...(surfaces.ompVersion === undefined ? {} : { ompVersion: surfaces.ompVersion }),
      ...(surfaces.herdrSource === undefined ? {} : { herdrSource: surfaces.herdrSource }),
      ...(tgInstalled?.kind === "version" ? { telegramInstalled: tgInstalled.version } : {}),
      ...(tgDaemon?.kind === "version" ? { telegramDaemon: tgDaemon.version } : {}),
      ...(tgPublished?.kind === "version" ? { telegramPublished: tgPublished.version } : {}),
    });
  } catch (err) {
    log(`install-surface read skipped: ${errText(err)}`);
  }
}

// ==================================================== hourly signal mining
// (Phase 4.) The fleet already records every fact worth grooming — settlement
// flags that name a weakened test, failure classes that keep recurring — and
// then leaves them in `runs`, where only somebody reading stats would ever meet
// them. Mining turns them into intake items, which is the one queue grooming
// already reads.
//
// Three properties make this safe to run unattended:
//
//  - **Never auto-queued.** A mined item is `pending` intake, exactly like an
//    idea an operator typed. It reaches a worker only after grooming specs it
//    and the ready gate passes it. Mining proposes; it does not decide.
//  - **Idempotent by construction.** Each signal carries a stable `source`, and
//    Lane A derives the intake id from `project + source` with INSERT OR IGNORE.
//    An unresolved signal therefore does not multiply hourly — the tenth pass
//    over the same weakened test files nothing.
//  - **Bounded to what the store already holds.** `mineSignals` is pure and
//    takes pre-read rows; no network, no new table, no second clock.

// The window is `mining.ts`'s own `DEFAULT_MINING_WINDOW_MS`, imported rather
// than restated: the miner's occurrence thresholds are calibrated against it, so
// a second copy here would be a second answer to "what counts as recent".

/**
 * Run the mining pass at most once per wall-clock hour (Phase 4).
 *
 * The cadence rides the five-minute tick rather than adding a timer: a second
 * scheduler for a pass whose whole job is reading rows it already has would be
 * one more thing to reason about at shutdown. The hour is a bucket key in the
 * `notifications` ledger — the same once-per-key mechanism the daily db snapshot
 * uses — which makes it durable across restarts, where an in-memory `lastRun`
 * would let a flapping daemon mine on every boot.
 *
 * The key is claimed BEFORE the work, so the bound is "one attempt per hour".
 * That is the right way round: a miner that throws would otherwise retry every
 * five minutes and log every time, and losing one hour costs nothing at all —
 * the signals are derived from stored rows, so next hour sees the same ones.
 */
export function mineIntakeSignals(d: Pick<Deps, "project" | "store">, now = Date.now()): void {
  // Hour bucket, not a rolling 60 minutes: a pass at :59 and one at :01 are two
  // buckets and both run. That is the dbSnapshot cadence's own shape, and it is
  // harmless here precisely because filing is idempotent.
  const key = `signal-mining:${d.project.name}:${new Date(now).toISOString().slice(0, 13)}`;
  if (d.store.wasNotified(key)) return;
  d.store.markNotified(key);

  const since = now - DEFAULT_MINING_WINDOW_MS;
  let signals: readonly MinedSignal[];
  try {
    signals = mineSignals({
      project: d.project.name,
      now,
      runs: d.store.statsRuns(d.project.name, since),
      windowMs: DEFAULT_MINING_WINDOW_MS,
    });
  } catch (err) {
    log(`signal mining skipped: ${errText(err)}`);
    return;
  }
  if (signals.length === 0) return;

  const filed: string[] = [];
  let known = 0;
  let dismissed = 0;
  for (const signal of signals) {
    try {
      const item = d.store.recordIntake({
        project: d.project.name,
        text: signal.text,
        at: now,
        source: signal.source,
      });
      // The store returns the row that is NOW in the table, so a `createdAt`
      // older than this call proves the signal was already filed — the ordinary
      // case for anything not yet groomed, and it must read as "already known"
      // rather than as a tenth duplicate.
      if (item.createdAt === now) filed.push(signal.source);
      // A signal the operator dismissed stays dismissed: INSERT OR IGNORE cannot
      // resurrect it, and counting it as merely "known" would hide the fact that
      // the fleet keeps producing a signal somebody has already judged
      // uninteresting — which is itself worth seeing in the journal.
      else if (item.state !== "pending") dismissed += 1;
      else known += 1;
    } catch (err) {
      log(`intake for ${signal.source} not filed: ${errText(err)}`);
    }
  }
  log(
    `signal mining: ${filed.length} filed, ${known} already pending, ${dismissed} previously dismissed ` +
      `(${signals.length} signal(s) over ${Math.round(DEFAULT_MINING_WINDOW_MS / 86_400_000)}d)` +
      (filed.length === 0 ? "" : ` — ${filed.join(", ")}`),
  );
}

/**
 * The first-tick verification for a fleet-initiated upgrade (#486).
 *
 * The detached install unit journals its progress and stops, leaving dispatch
 * paused; whichever daemon comes back after the restart runs this on its
 * first tick — above every gate that a pause or an integrity mismatch could
 * close, because the install's own pause is the thing under verification.
 * A day with no journal costs one file read; only a pending request pays for
 * the version/health/ticks/pane/doctor checks and the outbox rows.
 */
export function upgradeVerifyDepsFor(d: Pick<Deps, "project" | "store">): UpgradeVerifyDeps {
  return {
    projectName: d.project.name,
    journal: (entry) => appendJournal(stateDir(), entry),
    run: runCommand,
    layers: (project) => fleetLayers(project),
    health: healthCheck,
    doctor: (name) => runDoctor(name),
    surfaces: () => inspectSurfaces({ run: runCommand, log, env: process.env }),
    enqueue: (draft) => {
      try {
        d.store.enqueueReport(draft);
      } catch (err) {
        log(`upgrade verification report could not be queued: ${errText(err)}`);
      }
    },
    pause: pauseInstance,
    resume: (project) => setPaused(false, undefined, project),
    holdGlobal: () => {
      if (pauseInstance() === undefined) {
        setPaused(true, { source: "upgrade-recovery", reason: "upgrade recovery verification required" });
      }
    },
    releaseGlobalRecovery: (expectedSince) => {
      const pause = pauseInstance();
      if (
        pause?.source === "upgrade-recovery" &&
        expectedSince !== undefined &&
        pause.since === expectedSince
      ) {
        setPaused(false);
      }
    },
    launchRollback: async (version) => {
      const launched = await launchTransientUnit(
        (argv) => runCommand(argv[0]!, argv.slice(1)),
        process.env,
        { kind: "rollback", version },
      );
      return launched.ok === true ? { ok: true, unit: launched.unit } : { ok: false, stderr: launched.stderr };
    },
    herdrSession: resolveHerdrSession(process.env),
    // The durable pane probe for the post-restart verifier (#832): every omp
    // process the fleet pane currently claims by herdr, resolved to its start
    // time — the evidence an external (pane-owned) orchestrator actually
    // restarted after the install began, through the same shared probe as the
    // in-process upgrade.
    probePaneOmp: (session) => herdrPaneOmpStarts(runCommand, session, processStartTimeMs),
    log,
    now: () => Date.now(),
  };
}

export async function verifyPendingUpgradeTick(
  d: Deps,
  run = verifyPendingUpgrade,
): Promise<void> {
  // Cheap pre-test: no journal, no pending request, nothing to touch.
  if (readUpgradeJournal(stateDir()).length === 0) return;
  const verifyDeps = upgradeVerifyDepsFor(d);
  try {
    const outcome = await run(verifyDeps);
    if (
      outcome.handled !== "none" &&
      outcome.handled !== "already-closed" &&
      outcome.handled !== "recovery-owed"
    ) {
      log(
        `upgrade journal handled: ${outcome.handled}` +
          `${outcome.version === undefined ? "" : ` for omp-conductor@${outcome.version}`}` +
          `${outcome.detail === undefined ? "" : ` — ${outcome.detail}`}`,
      );
    }
  } catch (err) {
    // A torn journal must not take the tick down: the operator still reads it
    // through `status`, and the next tick retries the same pass.
    log(`upgrade verification pass failed: ${errText(err)}`);
  }
}

/**
 * The daemon's conductor.db snapshot cadence (#289) — the ledger's only
 * durable copy, taken on the digest-aligned {@link dbSnapshotDue} window:
 * once per local day, at/after the digest's configured `at` when it has one.
 *
 * Host-global, like the store itself: one snapshot per day for every project,
 * so whichever project's daemon wins the day first publishes it and the
 * others no-op on the same durable marker. The marker is written through the
 * store's existing notification ledger (a one-row idempotence guard, the same
 * primitive escalations use not to act twice) *after* the snapshot is
 * published — a crash between publish and mark re-snapshots the next tick
 * instead of skipping the day, and a restored store that predates today's
 * marker takes a fresh snapshot on the next tick.
 *
 * Returns true when a snapshot was published. A store that does not exist
 * yet is a silent no-op: there is nothing to back up, and `doctor`'s
 * `db-backup` probe already treats that as a pass.
 */
export function runDbSnapshotCadence(args: {
  digestPolicy: ReportingPolicy["digest"];
  store: Pick<Store, "wasNotified" | "markNotified">;
  source: string;
  backupDir: string;
  now?: number;
  keep?: number;
  log?: (line: string) => void;
}): boolean {
  const { digestPolicy, store, source, backupDir } = args;
  // No store yet (first boot before any run persisted) → nothing to back up;
  // `doctor`'s `db-backup` probe already treats that as a pass.
  if (!existsSync(source)) return false;
  const now = args.now ?? Date.now();
  const today = localDayKey(now, digestPolicy.timezone);
  const alreadyToday = store.wasNotified(dbSnapshotMarkerKey(today));
  if (!dbSnapshotDue({ digest: digestPolicy }, alreadyToday ? today : undefined, now)) {
    return false;
  }
  const published = snapshotDb(source, backupDir, now);
  (args.log ?? log)(`conductor.db snapshot published: ${published}`);
  store.markNotified(dbSnapshotMarkerKey(today));
  try {
    const removed = pruneDbSnapshots(backupDir, args.keep ?? DB_SNAPSHOT_RETENTION);
    if (removed > 0) {
      (args.log ?? log)(`db snapshot retention pruned ${removed} file(s) beyond the retained ${args.keep ?? DB_SNAPSHOT_RETENTION}`);
    }
  } catch (err) {
    // The snapshot is already published and marked; an over-bound directory
    // costs disk until the next due day retries the prune, never the backup.
    (args.log ?? log)(`db snapshot retention prune failed: ${errText(err)}`);
  }
  return true;
}

export async function tick(d: Deps, workers?: WorkerPool): Promise<void> {
  // A config edit takes effect on the next tick, not the next daemon restart
  // (#170). Re-resolve the project and its caps at the tick boundary so a tick
  // and every run it admits see one consistent snapshot; a failed read keeps
  // the boot values rather than wedging the tick, and the next tick tries
  // again. The same reloaded config feeds the db-snapshot backup dir: when it
  // is unreadable the cadence falls back to the state-root default, matching
  // `doctor`'s probe of an absent field.
  let reloadedConfig: ConductorConfig | undefined;
  try {
    const cfg = loadConfig();
    reloadedConfig = cfg;
    const fresh = findProject(cfg, d.project.name);
    const freshCaps = resolveCaps(fresh, cfg.defaults);
    if (JSON.stringify(fresh) !== JSON.stringify(d.project) || JSON.stringify(freshCaps) !== JSON.stringify(d.caps)) {
      log(`config reloaded: project ${d.project.name} changed on disk — applying from this tick`);
    }
    d.project = fresh;
    d.caps = freshCaps;
    d.host = cfg.host;
    d.deliveryPolicyValid = true;
  } catch (err) {
    log(`config reload failed (${errText(err)}) — retaining boot values but blocking autonomous delivery`);
    d.deliveryPolicyValid = false;
  }

  // The fleet-installs-itself transaction (#486): the detached install unit
  // paused dispatch and stopped; the first tick after the restart verifies
  // against its journal and closes or rolls back. Above the pause gate on
  // purpose — the install's own pause is exactly the state being verified.
  try {
    await (d.upgradeVerifier ?? verifyPendingUpgradeTick)(d);
  } catch (err) {
    log(`upgrade journal pass failed: ${errText(err)}`);
  }

  // Availability-held notices are already durable. Once the freshly reloaded
  // policy opens (or newly allows their category), atomically hand a bounded
  // batch to the report outbox. Pauses do not suppress delivery.
  if (d.deliveryPolicyValid !== false) {
    try {
      const catchUp = enqueueAvailableHeldNotices(d.project, d.store, Date.now());
      if (catchUp !== undefined && !catchUp.deduped) {
        log(`availability catch-up ${catchUp.report.id} queued for ${d.project.name}`);
      }
    } catch (err) {
      // The daily digest may have claimed the same rows from another process
      // between selection and association. Either way the ledger still owns them.
      log(`availability catch-up handoff deferred (${errText(err)}) — retrying next tick`);
    }
  }

  // Before the pause check, deliberately. This one is not about dispatch: the
  // orchestrator is a different process, and it can be wedged while this fleet
  // is paused — which is exactly the state the reference fleet was in when the
  // failure happened. A pause silences claiming, not the operator's right to
  // know their supervising session stopped reading its queue.
  await watchOrchestrator(d);
  // The worker-side sibling, and above the pause gate with it: a stalled run
  // holds its slot on a parked fleet exactly as hard as on a busy one, and the
  // #1086 incident was an upgrade drain blocked for 26 minutes by one silent
  // session. Paused workers are skipped inside the pass — their silence is
  // the pause working (#938) — so this gate costs nothing there.
  try {
    await watchWorkerProgress(d);
  } catch (err) {
    log(`worker-progress sweep failed: ${errText(err)}`);
  }
  // The down incident is reconciled the same place and for the same reason: a
  // session that has actually died is as much the operator's concern as one
  // that is wedged, and restarting it is the daemon's restart either way. This
  // is what turns a crashed orchestrator into one page ("down since <t>") plus
  // a diverting count, instead of a warning only in daemon.log.
  await reconcileOrchestratorDown({
    project: d.project,
    store: d.store,
    orchestrator: d.orchestrator,
    escalate: (event) => d.escalate(event),
    log,
  });
  // The orphaned-pause watch rides the same above-the-pause-gate band (#1113):
  // a sentinel whose writer has exited is exactly the condition a paused
  // fleet cannot surface through its own gates below, and an `upgrade` pause
  // can never be cleared by any future event. The notification ledger dedupes
  // the page to once per sentinel instance; resuming stays a human decision.
  try {
    await reconcileOrphanedPause({
      project: d.project.name,
      escalate: (event) => d.escalate(event),
      log,
    });
  } catch (err) {
    log(`orphaned-pause watch failed: ${errText(err)}`);
  }

  // Beside the two orchestrator watches, and above the pause gate, for the same
  // reason they are: mining reads rows this store already holds and files intake
  // for a human to groom. It claims nothing, so a paused fleet has no cause to
  // stop noticing that a test was weakened three times this week. Its own hour
  // bucket bounds it; this call site just supplies the clock.
  try {
    mineIntakeSignals(d);
  } catch (err) {
    log(`signal mining pass failed: ${errText(err)}`);
  }

  // Settlement is maintenance, not dispatch. Run it before every gate that can
  // stop claiming — pause, integrity, spend, and capacity — so status converges
  // while the fleet is parked or workers are still active. Resident workers run
  // through the pool without blocking this five-minute tick. The return value
  // is what this pass settled; a held pass records it so the paused state reads
  // as work done, not just a clock that moved (#497).
  let settled = await settlePushedGreen(d);
  try {
    await watchMergedBase(d);
  } catch (err) {
    log(`base-branch check sweep failed: ${errText(err)}`);
  }
  try {
    await watchBaseHealth(d);
  } catch (err) {
    log(`current base-health sweep failed: ${errText(err)}`);
  }
  try {
    await adoptSalvagedPrs(d);
  } catch (err) {
    log(`salvaged PR adoption sweep failed: ${errText(err)}`);
  }

  // The conductor.db snapshot cadence (#289) is durability, not dispatch:
  // the verb ledger, the decision rows and the run history have no other
  // copy, so they are snapshotted once per day on the digest-aligned window.
  // Above the pause gate on purpose — a parked fleet still accumulates ledger
  // rows and still deserves a backup — and a failure costs the day's snapshot
  // placeholder, never the tick: failed cadence steps are logged and retried
  // by the five-minute loop, exactly like the sweeps above it.
  try {
    runDbSnapshotCadence({
      digestPolicy: (d.project.reporting ?? DEFAULT_REPORT_POLICY).digest,
      store: d.store,
      source: dbPath(),
      backupDir: dbBackupDirFor(reloadedConfig),
      log,
    });
  } catch (err) {
    log(`db snapshot cadence failed: ${errText(err)}`);
  }

  // Immediately after settlement and before any routing, so a class is on the
  // row before the next dispatch decision reads its budgets (#132). Above the
  // pause gate deliberately: classification and label reconciliation are
  // maintenance, and the four phantom `agent:failed` issues this exists to clear
  // are exactly as misleading on a parked fleet as on a busy one.
  //
  // Each is guarded on its own: a tracker that fails mid-classification must not
  // stop the label reconcile, and neither may stop the tick.
  try {
    settled += await classifyAndRecover({
      project: d.project,
      caps: d.caps,
      tracker: d.tracker,
      store: d.store,
      escalate: (e) => d.escalate(e),
      isPaused,
      setPaused,
    });
  } catch (err) {
    log(`classification sweep failed: ${errText(err)}`);
  }
  try {
    await reconcileStaleLabels(d);
  } catch (err) {
    log(`label reconcile failed: ${errText(err)}`);
  }
  try {
    // Same phase, same reason as the label reconcile above: a durable row the
    // tracker has moved past (#964). A separate call because that one iterates
    // labelled issues, and a promotable issue carries no label.
    await reconcileGroomingClosures(d);
  } catch (err) {
    log(`grooming closure reconcile failed: ${errText(err)}`);
  }

  // Drain the label projection outbox (#201). The maintenance phases above may
  // have enqueued ops (settlement releases, recovery requeues, reconciles);
  // each due op is applied now — or deferred with backoff for the next tick —
  // before the queue is read, so dispatch sees labels converging on what the
  // store decided. A refusing tracker defers ops instead of taking the tick
  // down.
  try {
    await projectLabels(d.store, d.tracker, d.project);
  } catch (err) {
    log(`label projection failed: ${errText(err)}`);
  }

  // Ledger maintenance, above the pause gate for the same reason the stall watch
  // is: a paused fleet still owes its operator the questions it asked, and a
  // condition that came true while dispatch was parked is exactly the thing the
  // orchestrator has to see promptly (#329).
  //
  // Expiry is synchronous (one UPDATE); condition evaluation is fire-and-forget
  // because it shells out to `gh` and `npm`, and a registry that hangs must cost
  // one unevaluated condition rather than the tick. A false→true transition
  // writes the same immediate-tick poke recover uses so the heartbeat does not
  // wait a full interval; repeated sweeps while the condition stays true never
  // re-enter `met` (store mark is idempotent).
  for (const expired of d.store.expireDueDecisions(d.project.name, Date.now())) {
    log(`decision ${expired.id} expired unanswered after seven days: ${expired.question}`);
  }
  void evaluateDecisionConditions(
    d.store,
    d.project.name,
    d.tracker,
    { npm: probeNpmVersion, rateLimit: probeRateLimitReset },
    Date.now,
  )
    .then((met) => {
      wakeOrchestratorForMetConditions(d.project.name, met);
    })
    .catch((err: unknown) => {
      log(`decision condition pass failed: ${errText(err)}`);
    });

  // A spend-cap pause is self-expiring (#780). The spend gate below persists a
  // pause once today's spend reaches the cap, but once that latch is on disk
  // every later pass returns at the pause gate and never reaches the spend
  // check again — so the fleet stayed stopped after the rolling window reset
  // until an operator ran `resume`. The same measurement that closed the gate
  // reopens it: clear a per-project spend-cap pause when the current
  // rolling-window spend is below the cap (or no cap is configured at all —
  // the gate that justified the pause is gone). Two guards keep the clear from
  // ever overriding another stop: only the exact instance that was read is
  // removed (a hold that replaced the sentinel meanwhile is untouched, review
  // #780), and only the per-project sentinel is ever considered — the legacy
  // global sentinel, which may carry an operator/integrity/setup hold, is
  // never removed, so the pause gate below still observes it.
  const spendPausePath = pausedPath(d.project.name);
  const spendPause = pauseInstanceAt(spendPausePath);
  if (spendPause?.source === "spend-cap") {
    const spent = d.store.spendSince(d.project.name, startOfToday());
    if (
      (d.caps.dailySpendUsd === null || spent < d.caps.dailySpendUsd) &&
      clearPauseIfUnchanged(spendPausePath, spendPause)
    ) {
      log(
        `spend-cap pause cleared: $${spent.toFixed(2)} ` +
          (d.caps.dailySpendUsd === null
            ? "(no daily cap configured)"
            : `below the $${d.caps.dailySpendUsd.toFixed(2)} daily cap`) +
          " — dispatch resumes",
      );
    }
  }

  // A paused fleet claims nothing. Checked first so pausing takes effect on the
  // next tick without signalling the process. But the pass still ran, and the
  // operator has to be able to see it: record it as a held pass — the work the
  // maintenance sweeps settled above this gate, and nothing admitted below it —
  // so `last dispatch` keeps moving while the fleet is deliberately parked. A
  // frozen clock is then a stall report, not a hold (#497).
  if (isPaused(d.project.name)) {
    // The held pass is the daemon-side admission acknowledgement: the daemon
    // has reached its admission boundary under the fence and claims nothing.
    // Written durably so the setup barrier can prove the fence was observed
    // (#651 review #3) — a pause cannot acknowledge itself.
    try {
      writeAdmissionAck(d.project.name);
    } catch (err) {
      log(`admission acknowledgement write failed: ${errText(err)}`);
    }
    d.store.recordDispatch(d.project.name, summarizeHeldPass(settled));
    return;
  }

  // The project drain is the self-expiring sibling of the pause fence (#484):
  // the same admission boundary — settlement above it, nothing claimed below —
  // but the intent is durable (it survives orchestrator loss) and bounded (the
  // record carries an absolute deadline, so a crash can never strand
  // admission). Reaching an empty active set SUCCEEDS the drain; it does not
  // end it (#1078): the release window the drain exists to create opens at
  // that moment and holds claims down until the deadline — exactly the quiet
  // interval a tick-driven orchestrator needs to cut a release, which the old
  // eight-second auto-clear destroyed before any tick could use it. Four
  // shapes, four behaviours:
  //   - fresh drain with live runs → a held pass, exactly like a pause;
  //   - fresh drain whose active set just reached zero → the opening is marked
  //     on the record once and logged once, and this pass admits nothing;
  //   - fresh drain already marked drained → the same held pass, silent;
  //   - malformed record → fails closed for THIS pass (it might be a fresh
  //     fence we cannot read), and the same consume removed it, so it can
  //     never become an unbounded permanent drain.
  const drain = consumeDrain(d.project.name);
  if (drain.kind === "active") {
    // Completion is the ACTIVE set, not the live-worker set: pushed-pending
    // and pushed-green PRs still make the `runs-settled` release gate fail, so
    // a window declared open over a remaining run would admit work on top of
    // a batch the releases still see as unfinished (#776 review #2).
    if (d.store.activeRuns(d.project.name).length === 0 && drain.drain.drainedAt === undefined) {
      markDrained(d.project.name);
      log(
        `project drain: fleet drained — release window open until ${drain.drain.expiresAt}` +
          " (claims stay paused; admission resumes at the deadline or on drain cancel)",
      );
    }
    d.store.recordDispatch(d.project.name, summarizeHeldPass(settled));
    return;
  } else if (drain.kind === "error") {
    log(`project drain record invalid (${drain.problem}) — removed; this pass admits nothing`);
    d.store.recordDispatch(d.project.name, summarizeHeldPass(settled));
    return;
  }

  const { project, caps, store } = d;

  // "Nobody patches the running conductor" is a hard boundary in both briefs —
  // which makes it prompt text, and prompt text is a request. This is the half
  // that does not negotiate: the package that dispatched the last worker must
  // still be the package on disk, or nothing else this tick does is
  // attributable. A legitimate deploy never trips it, because installing a new
  // build and restarting the unit re-records the baseline from the new files;
  // only an edit *underneath* a live daemon diverges from it.
  //
  // Below the pause gate on purpose, unlike the stall watch above. The property
  // being defended is that no work is dispatched under a package the operator
  // did not install — and a paused fleet dispatches nothing, so nothing needs
  // attributing yet. Tampering during a pause is not missed, only deferred: the
  // baseline is boot's, so the first tick after `resume` compares against it and
  // pauses again before claiming anything. Checking above the gate instead would
  // page on every legitimate build an operator deploys into a parked fleet,
  // which is exactly when they deploy them.
  const integrity = checkIntegrity(d.integrity, packageManifest());
  if (integrity.pause) {
    const shown = integrity.diff.slice(0, INTEGRITY_SAMPLE);
    log(
      `ERROR: the installed conductor changed under this daemon — ${integrity.diff.length} file(s) differ ` +
        `(${shown.join(", ")}${integrity.diff.length > shown.length ? ", …" : ""}) — pausing`,
    );
    setPaused(
      true,
      { source: "integrity", reason: "installed package changed under the daemon" },
      project.name,
    );
    if (integrity.page) {
      const delivered = await safeEscalate(d, {
        tier: 2,
        category: "fleet-stopped",
        project: project.name,
        issue: NO_ISSUE,
        // Dated for the same reason the spend cap is: the dedup key is the
        // summary, and a second tamper months later must not be swallowed as a
        // repeat of the first.
        summary:
          `Installed conductor changed under a running daemon on ${new Date().toISOString().slice(0, 10)}: ` +
          `${integrity.diff.length} file(s) differ (first: ${integrity.diff[0]}) — ${project.name} is paused`,
        detail: [
          `Package root: ${PACKAGE_SRC_DIR}`,
          ...shown,
          ...(integrity.diff.length > shown.length ? [`… and ${integrity.diff.length - shown.length} more`] : []),
          "",
          "If you deployed a new build, restart the daemon — the restart re-records the baseline.",
          "If you did not, the host edited itself while it was dispatching work: treat every run since",
          "the last known-good restart as unattributable before resuming.",
          "`omp-conductor resume` alone will not hold — the next tick re-pauses while the files differ.",
        ].join("\n"),
      });
      markPaged(d.integrity, delivered);
    }
    return;
  }

  // Review revisions are dispatch, not admission (#677): the orchestrator's
  // verb recorded them durably and the run row already occupies its issue, so
  // they bypass the ready queue and the claim path entirely — this pass wakes
  // them (bounded by the same worker slots) BEFORE the queue's capacity gate
  // reads live runs, so a claimed revision counts against capacity exactly like
  // the worker it is about to spawn.
  try {
    await dispatchReviewRevisions(d, workers);
  } catch (err) {
    log(`review revision dispatch failed: ${errText(err)}`);
  }

  // Adjudications are dispatch too (#932), and their own pass: an adjudication
  // occupies no worker slot and no issue — it opens a fresh read-only session
  // about a PR whose run has already settled — so it is neither bounded by the
  // worker cap nor allowed to consume a review round. Placed after revisions so
  // a PR with a live revision is never adjudicated in the same tick it is being
  // corrected in; the head re-read inside the pass is the durable guard.
  try {
    await dispatchReviewAdjudications(d, workers);
  } catch (err) {
    log(`adjudication dispatch failed: ${errText(err)}`);
  }

  // And what the verdicts imply (#876). Its own pass, after dispatch, because a
  // disposition acts on a SETTLED adjudication: running it before dispatch would
  // simply be a tick late, and running it inside dispatch would tie a tracker
  // mutation to the launch that produced the verdict.
  try {
    await applyAdjudicationDispositions(d);
  } catch (err) {
    log(`adjudication dispositions failed: ${errText(err)}`);
  }

  // route() filters the queue through isEligible() itself, so anything already
  // carrying a state label is gone before it gets here.
  const ready = await d.tracker.listReady();
  await cleanupRetainedRuns(
    d,
    new Set(ready.map((issue) => issue.number)),
    d.cleanup ?? { next: 0 },
  );
  // Label-projection overlay (#201): an issue whose outbox ops have not
  // reached GitHub yet is judged on what its labels *will* be. A pending
  // state-label removal stops a stale GitHub label from blocking redispatch,
  // and a pending queue-label removal drops the issue out of eligibility even
  // though the label is still physically present. isEligible stays pure; the
  // overlay happens here.
  const effective = ready.map((issue) => {
    const pending = store.pendingLabelOpsFor(project.name, issue.number);
    return pending.length === 0 ? issue : { ...issue, labels: effectiveLabels(issue.labels, pending) };
  });
  // Post-unblock stale-list revalidation (#891). `listReady` is a
  // label-FILTERED search, and GitHub's search index is eventually consistent:
  // measured 2026-08-22T09:18Z, an `unblock` cleared `agent:failed` and
  // restored the queue label, `gh issue view` returned the clean label set
  // immediately, and the very next pass still saw the stale labels, held #889
  // as `stale-lifecycle` and left the fleet at 0/5 with four spare slots. The
  // next scheduled pass usually recovers — this is throughput, not correctness
  // — but an operator-driven unblock that cannot refill the fleet has not done
  // what it says.
  //
  // So a lifecycle label read from the LIST is a suspicion, not a verdict: the
  // exact per-issue read decides. Bounded to candidates that would otherwise be
  // dropped for a lifecycle label — never the whole queue — and further to those
  // whose newest run is not active (a genuinely in-flight issue needs no
  // revalidation; its own row is the authority) and that are not operator-parked
  // (a park is a decision, not a stale label).
  //
  // An unreadable exact read keeps the stale labels, so the issue stays out of
  // this pass, and is held as `issue-state-lookup-error` rather than
  // `stale-lifecycle`: "the tracker could not say" is not evidence of a
  // residual label, and it must not summon Duty 1 to reconcile a label nobody
  // has read.
  const stateLabelSet = new Set(Object.values(project.stateLabels));
  const suspect = effective.filter(
    (issue) =>
      !isEligible(issue, project) &&
      issue.labels.some((l) => stateLabelSet.has(l)) &&
      !issue.labels.includes(project.stateLabels.backlog) &&
      !ACTIVE_STATES.includes(store.latestRun(project.name, issue.number)?.state ?? "merged"),
  );
  const revalidated = new Map<number, readonly string[]>();
  const unreadable = new Set<number>();
  for (const issue of suspect) {
    let snapshot: IssueSnapshot | undefined;
    try {
      snapshot = await d.tracker.issueSnapshot(issue.number);
    } catch {
      snapshot = undefined;
    }
    if (snapshot === undefined) {
      unreadable.add(issue.number);
      log(`#${issue.number} lifecycle labels could not be revalidated — holding this pass`);
      continue;
    }
    // A closed issue is nobody's candidate; leave the list labels alone and let
    // the ordinary gates speak.
    if (snapshot.state === "closed") continue;
    if (snapshot.labels.join("\u0000") === issue.labels.join("\u0000")) continue;
    revalidated.set(issue.number, snapshot.labels);
    log(`#${issue.number} lifecycle labels revalidated: the queue list was stale`);
  }
  const verified =
    revalidated.size === 0
      ? effective
      : effective.map((issue) => {
          const live = revalidated.get(issue.number);
          return live === undefined ? issue : { ...issue, labels: [...live] };
        });
  const { routed, unroutable } = route(verified, project);
  // route() drops a candidate for two reasons, only one of which is a claim
  // question. A lifecycle state label (agent:in-progress/blocked/failed)
  // marks a run-owned issue: it is genuinely in flight while its newest run
  // is live or settling, and a terminal newest run — or no run at all — means
  // the label is residual, the footprint of a settled run nobody cleared,
  // which is Duty 1 reconciliation work, not occupied capacity. A missing
  // queue label instead marks an issue the outbox is withdrawing (a pending
  // queue-label removal, e.g. releaseQueueLabel after a merge the PR did not
  // close), which belongs in none of the three populations. So `claimed` is
  // defined from actual ownership over genuinely lifecycle-labelled candidates
  // rather than as the residual of routing (#228, #611).
  // Every population below reads the VERIFIED list, so a candidate the exact
  // read cleared is not simultaneously admitted and counted as held (#891).
  const dropped = verified.filter(
    (issue) => !isEligible(issue, project) && issue.labels.some((l) => stateLabelSet.has(l)),
  );
  // The operator's park label is not a stale lifecycle label: the queue query
  // still returns a parked-and-queued issue, route() drops it as ineligible,
  // and the claim gate holds it as `issue-parked` when the park lands mid-pass
  // (#734). Counting it as stale-lifecycle below would summon the orchestrator
  // to reconcile a deliberate decision. Parked is its own population, derived
  // from the same isEligible read the gate uses, so the number status renders
  // cannot disagree with what admission would hold (#507).
  const parkedCandidates = verified.filter(
    (issue) => !isEligible(issue, project) && issue.labels.includes(project.stateLabels.backlog),
  );
  const parkedNumbers = new Set(parkedCandidates.map((issue) => issue.number));
  let claimed = 0;
  let parked = 0;
  const lifecycleHolds: AdmissionHold[] = [];
  for (const issue of dropped) {
    const newest = store.latestRun(project.name, issue.number);
    if (newest !== undefined && ACTIVE_STATES.includes(newest.state)) {
      claimed += 1;
    } else if (parkedNumbers.has(issue.number)) {
      // A park never kills a live run; a parked issue with no run is inventory
      // the operator is deliberately holding, not reconciliation work.
      parked += 1;
    } else if (unreadable.has(issue.number)) {
      // Fail closed, and say which read failed: a tracker that could not answer
      // is not evidence of a residual label (#891).
      lifecycleHolds.push({
        issue: issue.number,
        reason: "issue-state-lookup-error",
        detail: "lifecycle labels could not be revalidated against the exact issue read",
      });
    } else {
      lifecycleHolds.push({ issue: issue.number, reason: "stale-lifecycle" });
    }
  }
  const routingHolds: AdmissionHold[] = [
    ...unroutable.map(
      (u): AdmissionHold => ({ issue: u.issue.number, reason: `unroutable:${u.reason}` }),
    ),
    ...lifecycleHolds,
  ];
  const recordDispatch = (admitted: number, holds: readonly AdmissionHold[]): void => {
    store.recordDispatch(
      project.name,
      summarizeDispatch(ready.length, routed.length, claimed, admitted, holds, Date.now(), settled, parked),
    );
  };

  // An issue nobody can route never reaches a worker: guessing the target repo
  // is exactly the kind of improvisation this system exists to prevent. The
  // summary is stable so a queue left unfixed pages once, not every tick.
  for (const u of unroutable) {
    await safeEscalate(d, {
      tier: 1,
      project: project.name,
      issue: u.issue.number,
      summary: `#${u.issue.number} cannot be routed: ${UNROUTABLE_TEXT[u.reason]}`,
      detail: [
        u.issue.title,
        u.issue.url,
        `Repo labels seen: ${u.labels.length > 0 ? u.labels.join(", ") : "(none)"}`,
        `Configured repos: ${Object.keys(project.routing.repos).join(", ") || "(none)"}`,
        `Fix: put exactly one \`${project.routing.labelPrefix}<repo>\` label on the issue.`,
      ].join("\n"),
    });
  }

  const since = startOfToday();

  // Spend is the one cap that stops the fleet instead of merely deferring work.
  // A loop that is burning money has to halt itself; waiting for a human to
  // notice tomorrow is how a runaway becomes expensive. `null` means the
  // operator opted out — turns and wall-clock still brake every run (#46).
  const spent = store.spendSince(project.name, since);
  if (caps.dailySpendUsd !== null && spent >= caps.dailySpendUsd) {
    setPaused(
      true,
      { source: "spend-cap", reason: `daily spend reached $${caps.dailySpendUsd}` },
      project.name,
    );
    await safeEscalate(d, {
      tier: 2,
      category: "fleet-stopped",
      project: project.name,
      issue: NO_ISSUE,
      // Dated so the same cap pages again tomorrow, but only once per day.
      summary: `Daily spend cap reached on ${new Date().toISOString().slice(0, 10)} — ${project.name} is paused`,
      detail: [
        `Spent $${spent.toFixed(2)} of the $${caps.dailySpendUsd.toFixed(2)} daily cap.`,
        "Dispatch stays paused until today's spend falls below the cap, then resumes",
        "automatically on the next pass — `omp-conductor resume` only speeds that up.",
      ].join("\n"),
    });
    recordDispatch(0, [
      ...routingHolds,
      ...routed.map((r) => ({ issue: r.issue.number, reason: "daily-spend-cap" as const })),
    ]);
    return;
  }

  // Grooming, from here on, is the daemon's own duty (#1041). Its place in the
  // tick is the argument for it: BELOW the daily spend cap, because a batch of
  // scouts is real money and a fleet that has stopped spending must stop
  // spending on grooming too; ABOVE the capacity gate, because a full fleet
  // with a draining queue is precisely when the next specs have to be written —
  // the queue those workers will pull from next is empty right now, and waiting
  // for a free slot to notice is how the loop stalls between batches.
  //
  // It reads this pass's own routable count, so the drought and the response
  // happen in one pass instead of one apiece, and it never blocks: the launches
  // ride the worker pool exactly like adjudications.
  try {
    await dispatchToSpecGrooming(d, routed.length, workers);
  } catch (err) {
    log(`grooming pass failed: ${errText(err)}`);
  }

  // Two different questions, deliberately two queries. Capacity counts worker
  // *processes*, so a green PR awaiting a human merge must not consume a slot —
  // two of those would otherwise stop the fleet. That same PR's *issue* must
  // still be occupied, which is what `admitCandidates`' busy set is for.
  const live = store.liveRuns(project.name);
  const slots = caps.maxConcurrentWorkers - live.length;
  if (slots <= 0) {
    log(`at capacity: ${live.length}/${caps.maxConcurrentWorkers} workers`);
    recordDispatch(0, [
      ...routingHolds,
      ...routed.map((r) => ({ issue: r.issue.number, reason: "capacity" as const })),
    ]);
    return;
  }

  const pass = await admitCandidates(d, routed, slots);

  // The stop fence (#374): the run loop only re-reads `stopping` between
  // whole ticks, so a pass already in flight when SIGTERM/SIGINT landed must
  // re-check the drain signal here — after candidate admission, before any
  // claim or launch — or the shutdown path admits exactly the work it is
  // about to wait for and then loses to the stop timeout. The candidates keep
  // their queue labels; the next daemon start re-dispatches them.
  if (d.drain?.draining === true) {
    recordDispatch(0, [
      ...routingHolds,
      ...pass.holds,
      ...pass.admitted.map((a) => ({ issue: a.r.issue.number, reason: "shutting-down" as const })),
    ]);
    return;
  }
  recordDispatch(pass.admitted.length, [...routingHolds, ...pass.holds]);
  await recordInstallSurfaces(d);
  reconcilePanes(d, project.name, log);

  if (pass.admitted.length === 0) return;

  log(`dispatching ${pass.admitted.map((a) => `#${a.r.issue.number}`).join(" ")}`);
  await dispatchAdmissions(
    pass.admitted,
    (a) => handleIssue(d, a.r, a.attempt, a.lane, a.model),
    workers,
  );

  // Post-admission flush: freshly claimed runs enqueued their in-progress
  // label inside `handleIssue`; applying it now means the guard label lands on
  // GitHub within the same tick on the healthy path, not five minutes later
  // (#201).
  try {
    await projectLabels(store, d.tracker, project);
  } catch (err) {
    log(`label projection failed: ${errText(err)}`);
  }
}