// ── The event runner ───────────────────────────────────────────────────────
//
// EVENTS.md §E, and Phase 2 of EVENTS_PLAN.md. The eighth poller: same
// `setInterval` + `unref()` + `stop()` shape as `announceWorker`, the three Team
// sweepers and `engagementWorker`, wired into `server.js` beside them. In the
// **website** process rather than the bot, which cannot load module code.
//
// Its tick does four things, in this order:
//
//   1. **reclaim** — release leases whose holder died, never touching `attempts`
//   2. **materialise** — sweep occurrences past their grace window into `missed`
//   3. **advance** — claim each due run and move it through its phases
//   4. **prune** — the `event_run_log` retention sweep, on its own long clock
//
// **What a phase advances on, as of Phase 5.** Every step terminal, and — if the
// phase authored one — its GATE open as well. The gate is an ADDITIONAL
// condition and never a replacement: a phase whose steps are still running is
// not advanced by a boss that spawned early. `{ after: '30m' }` is closed by
// this file when its deadline passes; `{ on: '<trigger>', count: n }` is closed
// by the EMIT PATH, because a firing between two ticks is not observable from
// either of them. See `events/gates.js` for that division.
//
// **What "materialise" means, and why it is two halves.** Phase 4 completed it.
// The first half EXPANDS: every `ready` definition's recurrence is computed in
// its own IANA zone and every occurrence inside a fourteen-day horizon becomes a
// real `scheduled` row (`INSERT IGNORE` against the occurrence key, so the tick
// that already made one makes nothing). The second half SWEEPS: a run whose
// instant passed while the process was down does not start late and silently —
// it becomes `missed`, which is terminal and which a human can see (§L).
//
// **The two halves need each other, and the horizon is why.** Expansion looks
// forward from `now - grace` only, so an occurrence nobody ever materialised is
// never invented retroactively — waking up after three days down must not
// manufacture three days of history that no operator could have seen or
// cancelled. It does not have to: because rows exist a fortnight ahead of their
// instant, an outage that spans an occurrence finds the row already there, and
// the sweep marks it `missed` honestly. The horizon is what makes the missed
// sweep mean anything for a recurring event.
//
// **Two properties this file must not lose**, both already paid for once on this
// codebase:
//
// - **A reclaim never resets `attempts`.** Engagement Phase 14's defect: a sweep
//   that returned every stale row to its start state made the attempt ceiling
//   unreachable, so the row cycled forever, never terminal, and therefore never
//   eligible for any retention sweep.
// - **The unique index, not the claim, is what prevents a double run.** The claim
//   decides *who* advances an occurrence; `uq_evrun_occurrence` is what stops two
//   of them existing. Neither substitutes for the other.
//
// §N4 settled this deployment as single-instance, so the `--scale app=2` rig the
// engagement workstream used is deliberately not built. Every claim path is here
// exactly as §E specifies anyway, because the multi-instance case is not what
// they are for on a single container: the CAS is what protects a tick that
// overran into the next one, and the lease and its reclaim are what recover a
// step whose process died mid-dispatch. Both happen with one app.

const os = require('os')

const runsDb = require('../model/events/eventRuns.db')
const stepsDb = require('../model/events/eventRunSteps.db')
const logDb = require('../model/events/eventRunLog.db')
const versionsDb = require('../model/events/eventVersions.db')
const definitionsDb = require('../model/events/eventDefinitions.db')
const runsModel = require('../model/events/eventRuns.model')
const gatesDb = require('../model/events/eventPhaseGates.db')
const recurrence = require('../events/recurrence')
const gates = require('../events/gates')
const spec = require('../events/spec')
const registries = require('../modules/registries')
const { dispatchStep } = require('../events/dispatch')
const authorize = require('../events/authorize')
const log = require('./logger')('event-runner')

const POLL_MS = Number(process.env.EVENT_POLL_MS) || 15_000

// How many runs one sweep looks at, and how many steps it will drain from one
// run. Bounds rather than targets: the tick runs again in POLL_MS, and an
// unbounded batch is how a backlog turns one tick into a stall. The step bound
// also caps how long one run can hold the tick, which is what keeps a
// forty-step phase from starving every other run on the deployment.
const RUN_BATCH = Number(process.env.EVENT_RUN_BATCH) || 50
const STEPS_PER_TICK = Number(process.env.EVENT_STEPS_PER_TICK) || 25

// How far ahead a recurrence is turned into real rows (org lead, 2026-09-02).
// Fourteen days is a fortnight of occurrences an operator can see, cancel and
// reschedule ONE AT A TIME, which a projection is not — and it is short enough
// that a definition edited today affects almost everything still ahead of it.
// Beyond it the calendar projects rather than materialises, so a monthly event
// is still visible three weeks out without a row nobody will honour.
const HORIZON_DAYS = Number(process.env.EVENT_MATERIALISE_AHEAD_DAYS) || 14

// A bound on one definition's expansion in one tick, not a target. A daily
// schedule over a fortnight is fourteen; this is what stops a hand-written spec
// turning one tick into a thousand inserts.
const MAX_OCCURRENCES_PER_DEFINITION = 100

// A step's retries (org lead, 2026-09-02). §L specifies `retry(n) -> skip |
// pause | abort_run` and names no `n`; it lives here, as one number an operator
// can change, rather than in a column no authoring surface would ever show.
//
// Flat backoff rather than exponential, for the reason the outbox's is flat:
// `due_at` is also the event's own clock, and a doubling backoff pushes a step
// arbitrarily far past the moment the event was about.
const MAX_ATTEMPTS = Number(process.env.EVENT_STEP_MAX_ATTEMPTS) || 3
const RETRY_MS = Number(process.env.EVENT_STEP_RETRY_MS) || 60_000

// How long a claim may look alive before the reclaim takes it back. The run
// lease has to outlast a whole tick's work on one run; a step's is computed from
// its own action's `budgetMs` (see `leaseFor`), because a registry that lets an
// action declare an hour would otherwise have its steps reclaimed and
// re-dispatched fifty-nine minutes before they answered.
const RUN_LEASE_MS = Number(process.env.EVENT_RUN_LEASE_MS) || 15 * 60 * 1000
const STEP_LEASE_MARGIN_MS = 60_000

// The log retention horizon, and how often the sweep runs. It is folded into
// this tick rather than given a ninth timer because it shares the tick's only
// real dependency — a database — and a sweep that runs four times a day does not
// need an interval of its own. Only TERMINAL runs are ever eligible, which is the
// rule Engagement Phase 14 arrived at.
const LOG_RETENTION_DAYS = Number(process.env.EVENT_LOG_RETENTION_DAYS) || 90
const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000

// Who this process is, for `claimed_by`. Host and pid, so a stranded claim in the
// table names the thing that stranded it.
const OWNER = `${os.hostname()}:${process.pid}`.slice(0, 64)

const RUN_TERMINAL = runsDb.TERMINAL

/** The lease a step's dispatch gets: its action's own budget, plus a margin. */
function leaseFor(step, now) {
  const action = registries.eventAction(step.action_id)
  const budget = action?.budgetMs || 10_000
  return new Date(now.getTime() + budget + STEP_LEASE_MARGIN_MS)
}

// ── The disposition of a step that has run out of road ─────────────────────
//
// **All three dispositions write the step `failed`.** `on_failure` says what
// happens to the RUN, not what happened to the step, and a step that was
// attempted three times and never worked is `failed` under every one of them.
// `skipped` is reserved for a step a human skipped from the run console (Phase
// 3) — a status that meant both "nobody ran this" and "this failed and we moved
// on" would make the run console's summary line unreadable.
async function applyFailure(run, step, error, { status = 'failed', kind = 'step.status', extra = null } = {}) {
  // **`refused` shares this whole function with `failed`, and that is decision 3
  // of Phase 6** (org lead, 2026-09-03): a cap breach or a disabled action takes
  // the same disposition a terminal failure takes, so a `change` step's default
  // `pause` stops the run where it stands and an operator raises the cap, edits,
  // and resumes. What differs is the two words on the record — the STATUS the
  // step ends in and the KIND the log line carries — because "nothing here is
  // broken, this deployment does not permit that" is a different sentence from
  // "the shard did not answer", and an operator reading a stopped run at 2am
  // needs to tell them apart at a glance.
  await stepsDb.finish(step.id, status, error)
  await logDb.write({
    runId: run.id,
    stepId: step.id,
    kind,
    phase: step.phase,
    detail: {
      to: status,
      action: step.action_id,
      attempts: step.attempts + 1,
      onFailure: step.on_failure,
      error,
      ...(extra || {}),
    },
  })

  // A run that lost a step is degraded whatever happens next. Health is not
  // status (§E): a run can be genuinely running and degraded at once, and the
  // admin surface needs to say so without claiming the run stopped.
  if (await runsDb.setHealth(run.id, 'degraded')) {
    await logDb.write({ runId: run.id, kind: 'run.health', detail: { to: 'degraded', because: step.action_id } })
  }

  if (step.on_failure === 'abort_run') {
    // §L: the disposition for `irreversible`. Nothing further is dispatched, and
    // the pending steps are cancelled rather than left looking due forever.
    const cancelled = await stepsDb.cancelPending(run.id)
    await runsDb.transition(run.id, ['starting', 'running', 'ending'], 'failed', { error })
    await logDb.write({
      runId: run.id,
      kind: 'run.status',
      phase: step.phase,
      detail: { to: 'failed', because: step.action_id, cancelledSteps: cancelled },
    })
    return 'stop'
  }

  if (step.on_failure === 'pause') {
    // The default for `change`, and the right one when the world is half-altered:
    // stop advancing and wait for a human. `paused` is excluded from `findDue`,
    // so nothing here picks it up again — Phase 3's resume control is the only
    // thing that moves it.
    await runsDb.transition(run.id, ['starting', 'running'], 'paused', { error })
    await logDb.write({
      runId: run.id,
      kind: 'run.status',
      phase: step.phase,
      detail: { to: 'paused', because: step.action_id },
    })
    return 'stop'
  }

  // 'skip': the run carries on, degraded, and the failure is on the record.
  return 'continue'
}

/**
 * Claim one step, dispatch it, and record what came back.
 *
 * Answers `'continue'` (the phase may proceed), `'stop'` (it may not, for now or
 * ever) or `'taken'` (somebody else claimed it first).
 */
async function drainStep(run, step, now, carry = {}) {
  if (!(await stepsDb.claim(step.id, OWNER, leaseFor(step, now), now))) return 'taken'

  // ── The permission check, and it is the LAST thing before the dispatch ──
  //
  // §K's four layers behind one function (Phase 6). It sits after the claim, not
  // before it: the cap is held by a conditional UPDATE and two ticks that both
  // priced a step before either claimed it would both spend. It sits before the
  // dispatch because a refusal means the action does not happen — nothing is
  // sent, nothing is created, and the step never reaches the module at all.
  //
  // **`user` is null here, and that is the design rather than an omission.** The
  // role was checked when a human published the version and again when a human
  // or the scheduler started the run; a run in flight is deliberately not
  // re-gated against its starter's current role, because demoting an admin at
  // midnight should not silently strand every event they started. Cancel is the
  // control for a run that should stop.
  //
  // **A retry does not pay twice.** The spend happens on the first attempt only.
  // A retry re-dispatches the same idempotent operation against the same key, and
  // charging a cap for a flaky socket would exhaust a deployment's allowance
  // through unreliability rather than through effect. The corollary is that a
  // step which spent and then failed for good keeps its spend: the attempt may
  // have half-run, and a refund would be core asserting that it did not.
  const action = registries.eventAction(step.action_id)
  if (action) {
    const verdict = await authorize.mayInvoke({
      action,
      params: step.params || {},
      run,
      spend: step.attempts === 0,
    })
    if (!verdict.ok) {
      return applyFailure(run, step, verdict.reason, {
        status: 'refused',
        kind: 'step.refused',
        extra: {
          code: verdict.code,
          dimension: verdict.dimension,
          requested: verdict.requested,
          cap: verdict.cap,
          consumed: verdict.consumed,
        },
      })
    }
  }

  const result = await dispatchStep(step, { run })

  if (result.actionVersionDrift) {
    await logDb.write({
      runId: run.id,
      stepId: step.id,
      kind: 'step.status',
      phase: step.phase,
      detail: { action: step.action_id, versionDrift: result.actionVersionDrift },
    })
  }

  if (result.outcome === 'parked') {
    // The GM cue. The step stays `running` with a NULL lease: genuinely in
    // flight, nothing holding it, so the stale reclaim passes it by and a cue
    // posted on Friday is still waiting on Monday. Phase 3's confirm control is
    // what ends it.
    await stepsDb.park(step.id, null)
    await logDb.write({
      runId: run.id,
      stepId: step.id,
      kind: 'step.parked',
      phase: step.phase,
      detail: { action: step.action_id, params: step.params },
    })
    return 'stop'
  }

  if (result.outcome === 'done') {
    await stepsDb.finish(step.id, 'done', null)
    if (result.holdSeconds > 0) {
      // `core.wait`, and any module action that answers `holdFor`. The pause is
      // the NEXT step's `due_at` and it is set here, by the runner, because a
      // `perform()` that slept would hold its claim for the duration.
      const until = new Date(now.getTime() + result.holdSeconds * 1000)
      if (!(await stepsDb.holdNext(run.id, step.phase, step.seq, until))) {
        // **Nothing after it in this phase**, which is the case a wait written as
        // the last step of a phase produces. Dropping the hold here would make
        // "announce, wait five minutes, then the next phase" start the next phase
        // at once — a wait that silently meant nothing. The later phase's steps do
        // not exist yet, so the instant is carried out to `advanceRun` and applied
        // when they are materialised.
        carry.holdUntil = until
      }
    }
    await logDb.write({
      runId: run.id,
      stepId: step.id,
      kind: 'step.status',
      phase: step.phase,
      detail: { to: 'done', action: step.action_id, holdSeconds: result.holdSeconds || 0 },
    })
    return 'continue'
  }

  if (result.outcome === 'retry' && step.attempts + 1 < MAX_ATTEMPTS) {
    await stepsDb.reschedule(step.id, new Date(now.getTime() + RETRY_MS), result.error)
    await logDb.write({
      runId: run.id,
      stepId: step.id,
      kind: 'step.retry',
      phase: step.phase,
      detail: { action: step.action_id, attempt: step.attempts + 1, of: MAX_ATTEMPTS, error: result.error },
    })
    // Degraded from the FIRST retry, not from the eventual failure. An event
    // whose announcements are landing on the second attempt is having trouble
    // now, and that is when an operator wants to know.
    if (await runsDb.setHealth(run.id, 'degraded')) {
      await logDb.write({ runId: run.id, kind: 'run.health', detail: { to: 'degraded', because: step.action_id } })
    }
    return 'stop'
  }

  // Terminal, or transient with the attempts spent. Same disposition either way:
  // §L's `retry(n) -> ...` has arrived at the arrow.
  return applyFailure(run, step, result.error)
}

/**
 * Open a phase's advance gate, if it authored one.
 *
 * Called at phase entry, immediately after `materialisePhase` and BEFORE the
 * transition that makes the phase current — which is the safe order rather than
 * the tidy one. A process that died between the transition and this call would
 * leave a phase whose gate does not exist, and a missing gate does not hold a
 * phase: it advances on its steps alone, silently ignoring the condition its
 * author wrote. Opening first risks only a row for a phase this tick did not win,
 * which the winner's INSERT IGNORE then finds already correct.
 */
async function openGate(runId, phase, now) {
  const advance = phase?.advance
  if (!advance) return false

  const created =
    advance.after !== undefined
      ? await gatesDb.open({
          runId,
          phase: phase.key,
          kind: 'after',
          // Re-derived from the authored string rather than stored beside it:
          // `events/spec.js` owns what `'2h'` means, and this is the one caller
          // that needs the number.
          afterSeconds: spec.parseAfter(advance.after),
          now,
        })
      : await gatesDb.open({
          runId,
          phase: phase.key,
          kind: 'on',
          triggerId: advance.on,
          conditions: advance.where ?? null,
          needed: advance.count || 1,
          now,
        })

  if (created) {
    await logDb.write({
      runId,
      kind: 'phase.gate',
      phase: phase.key,
      detail:
        advance.after !== undefined
          ? { kind: 'after', after: advance.after }
          : { kind: 'on', trigger: advance.on, needed: advance.count || 1, where: gates.phrase(advance.where ?? null) },
    })
  }
  return created
}

/**
 * A run whose phase has waited past `EVENT_PHASE_STALL_MS` is `stalled`.
 *
 * §E's third health value, and the first thing in this system ever to write it.
 * It is VISIBILITY and not a timeout: nothing advances, nothing fails, and a
 * human decides — which is the org lead's answer of 2026-09-02 and the reason
 * there is no authored deadline in the spec. What it must not be is quiet,
 * because a held run also holds its concurrency key, so every later occurrence
 * of the same definition goes `missed` behind it.
 *
 * `setHealth` only ever escalates, so this cannot undo a `degraded` a retry
 * earned, and the `run.health` line is written once because `setHealth` answers
 * whether it changed anything.
 */
async function noteStall(run, gate, now) {
  const described = gates.describe(gate, now)
  if (!described?.stalled) return false
  if (!(await runsDb.setHealth(run.id, 'stalled'))) return false
  await logDb.write({
    runId: run.id,
    kind: 'run.health',
    phase: gate.phase,
    detail: {
      to: 'stalled',
      because: `waiting ${described.elapsedSeconds}s on ${gate.trigger_id}`,
      seen: gate.tally,
      needed: gate.needed,
    },
  })
  return true
}

/**
 * Advance one claimed run as far as it will go this tick.
 *
 * The loop is bounded by `STEPS_PER_TICK` and exits on the first thing it cannot
 * get past — a parked step, a step whose `due_at` is in the future, a step
 * somebody else holds, or a phase that is not finished.
 */
async function advanceRun(run, now) {
  const version = await versionsDb.getById(run.version_id)
  const phases = version?.spec?.phases
  if (!Array.isArray(phases) || !phases.length) {
    // The pinned version is unreadable. `version_id`'s foreign key RESTRICTs
    // precisely so this cannot be a deleted row, so it is corruption rather than
    // an ordinary race — terminal, named, and not retried.
    await runsDb.transition(run.id, ['scheduled', 'starting', 'running', 'ending'], 'failed', {
      error: 'the pinned version has no phases',
    })
    await logDb.write({ runId: run.id, kind: 'run.status', detail: { to: 'failed', because: 'pinned version has no phases' } })
    return 'failed'
  }

  let phaseKey = run.current_phase

  if (run.status === 'starting') {
    // Entering the first phase. `materialisePhase` is INSERT IGNORE against
    // `uq_evstep_slot`, so doing it again over the rows Phase 1's `create()`
    // already wrote is a no-op — which is what makes recovery from a process that
    // died between the claim and here uneventful.
    const first = phases[0]
    await stepsDb.materialisePhase(run.id, first.key, first.steps || [])
    await openGate(run.id, first, now)
    if (!(await runsDb.transition(run.id, 'starting', 'running', { phase: first.key }))) return 'taken'
    phaseKey = first.key
    await logDb.write({ runId: run.id, kind: 'run.status', phase: first.key, detail: { from: 'starting', to: 'running' } })
  }

  if (run.status === 'ending') {
    // A run that reached the wind-down and then lost its process. Phase 8 puts
    // cleanup here; until then `ending` is a state a run passes through rather
    // than one it does work in, and completing it is the whole recovery.
    await runsDb.transition(run.id, 'ending', 'completed')
    await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'ending', to: 'completed' } })
    return 'completed'
  }

  // A hold a `core.wait` could not place because nothing followed it in its own
  // phase. It crosses the phase boundary with the run rather than being dropped.
  const carry = {}

  for (let n = 0; n < STEPS_PER_TICK; n++) {
    // Re-read the run's status between steps, not just at the top of the tick.
    // This loop drains up to STEPS_PER_TICK steps from one run, and Phase 3 put
    // a pause and a cancel in a human's hand: without this, pausing a run in the
    // middle of a batch would answer by dispatching another two dozen steps,
    // which is not a pause. One indexed column read per step, against a control
    // whose entire value is that it takes effect at once.
    if (n > 0 && (await runsDb.statusOf(run.id)) !== 'running') return 'stopped'

    const phaseIndex = phases.findIndex((p) => p.key === phaseKey)
    if (phaseIndex < 0) {
      await runsDb.transition(run.id, ['running'], 'failed', { error: `phase "${phaseKey}" is not in the pinned version` })
      await logDb.write({ runId: run.id, kind: 'run.status', detail: { to: 'failed', because: `unknown phase "${phaseKey}"` } })
      return 'failed'
    }

    const step = await stepsDb.nextOpenStep(run.id, phaseKey)

    if (step && step.status === 'running') return 'in-flight' // parked, or somebody's dispatch
    if (step && step.due_at && new Date(step.due_at) > now) return 'waiting' // behind a core.wait

    if (step) {
      const outcome = await drainStep({ ...run, current_phase: phaseKey }, step, now, carry)
      if (outcome === 'continue') continue
      return outcome === 'taken' ? 'taken' : 'stopped'
    }

    // Every step of this phase is terminal — which is the whole of the advance
    // test for a phase with no gate, and half of it for a phase with one.
    const { open, gate } = await gates.check(run.id, phaseKey, now)
    if (!open) {
      await noteStall(run, gate, now)
      // **Nothing is logged per tick here, deliberately.** The gate row IS the
      // state — the tally, the deadline, the last related event — and the run
      // console reads it directly. A `phase.waiting` line every fifteen seconds
      // would bury `condition.evaluated`, which is the line that actually says
      // something happened.
      return 'waiting'
    }
    // **`phase.advanced` is written by whoever made the DECISION, and a forced
    // gate's decision was not this tick's.** The emit path closes an `on` gate
    // and logs only `condition.evaluated`, so the line for that one is the
    // runner's; `gates.check` closes an `after` gate here, so that one is too.
    // A human closing it through the advance control already wrote the line,
    // with the actor and the reason — things this tick does not have — and a
    // second line from here made the console show the phase advancing twice,
    // the less informative one last. Found in the live walk.
    if (gate && gate.satisfied_by !== 'forced') {
      await logDb.write({
        runId: run.id,
        kind: 'phase.advanced',
        phase: phaseKey,
        detail: {
          because: gate.satisfied_by,
          waitedSeconds: Math.max(0, Math.round((new Date(gate.satisfied_at) - new Date(gate.entered_at)) / 1000)),
          ...(gate.kind === 'on' ? { trigger: gate.trigger_id, seen: gate.tally, needed: gate.needed } : { after: gate.after_seconds }),
        },
      })
    }

    await logDb.write({ runId: run.id, kind: 'phase.completed', phase: phaseKey, detail: { index: phaseIndex } })

    const next = phases[phaseIndex + 1]
    if (!next) {
      // §E's `ending` exists for the reason `sending` does in the outbox — it is
      // what a claim sets — so the run passes through it even though Phase 2 has
      // no cleanup to do there. Phase 8 is what gives it work.
      if (!(await runsDb.transition(run.id, 'running', 'ending'))) return 'taken'
      await logDb.write({ runId: run.id, kind: 'run.status', phase: phaseKey, detail: { from: 'running', to: 'ending' } })
      await runsDb.transition(run.id, 'ending', 'completed')
      await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'ending', to: 'completed' } })
      return 'completed'
    }

    await stepsDb.materialisePhase(run.id, next.key, next.steps || [])
    await openGate(run.id, next, now)
    if (carry.holdUntil) {
      // `seq > -1` is the first step of the phase just created. Applied after
      // materialisation because that is the first moment there is a row to hold.
      await stepsDb.holdNext(run.id, next.key, -1, carry.holdUntil)
      carry.holdUntil = null
    }
    // `running -> running` is not a no-op: it is a guarded write of
    // `current_phase` that fails if the run stopped being `running` underneath
    // this tick, which is what a cancel from the admin surface looks like.
    if (!(await runsDb.transition(run.id, 'running', 'running', { phase: next.key }))) return 'taken'
    phaseKey = next.key
    await logDb.write({ runId: run.id, kind: 'phase.entered', phase: next.key, detail: { steps: (next.steps || []).length } })
  }

  return 'bounded' // more to do; the next tick picks it up
}

/** Claim one due run, work it, and hand the lease back if it is still in flight. */
async function processRun(run, now = new Date()) {
  if (run.status === 'scheduled') {
    const holder = await runsDb.concurrencyHolder(run.concurrency_key, run.id)
    if (holder) {
      // The org lead's answer for a held key (2026-09-02): hold at `scheduled`
      // and let the grace window decide. Nothing is destroyed, nothing starts
      // silently late, and if the holder outlasts the window the missed sweep
      // makes this run terminal and visible.
      //
      // Logged only when the reason CHANGES. A blocked run is re-examined every
      // tick, and a line per tick for the length of a grace window would bury the
      // one line that matters under a thousand identical ones.
      const message = `held: run ${holder.id} has concurrency key "${run.concurrency_key}"`
      if (run.last_error !== message) {
        await runsDb.transition(run.id, 'scheduled', 'scheduled', { error: message })
        await logDb.write({
          runId: run.id,
          kind: 'run.blocked',
          detail: { concurrencyKey: run.concurrency_key, heldBy: holder.id, holderStatus: holder.status },
        })
      }
      return 'blocked'
    }

    if (!(await runsDb.claimStart(run.id, OWNER, new Date(now.getTime() + RUN_LEASE_MS)))) return 'taken'
    await logDb.write({ runId: run.id, kind: 'run.status', detail: { from: 'scheduled', to: 'starting' } })
    run = { ...run, status: 'starting' }
  } else if (!(await runsDb.claimTick(run.id, OWNER, new Date(now.getTime() + RUN_LEASE_MS), now))) {
    return 'taken'
  }

  try {
    const outcome = await advanceRun(run, now)
    // A run that is not finished must not keep its lease: it would be
    // unadvanceable until the lease expired, which would turn every `core.wait`
    // into `max(wait, RUN_LEASE_MS)`. A terminal transition already cleared it.
    if (!['completed', 'failed'].includes(outcome)) await runsDb.releaseClaim(run.id, OWNER)
    return outcome
  } catch (err) {
    await runsDb.releaseClaim(run.id, OWNER)
    throw err
  }
}

/** Occurrences that passed their own grace window while nothing was running (§L). */
/**
 * Turn every `ready` definition's recurrence into rows inside the horizon.
 *
 * The first half of the materialise leg (§E). Answers how many occurrences were
 * newly created, which is zero on almost every tick — the horizon moves fifteen
 * seconds at a time, so a weekly event creates one row a week and answers
 * `created: false` for the same fourteen occurrences in between.
 *
 * **The window starts at `now - grace`, not at `now`.** An occurrence whose
 * instant just passed is still startable inside the definition's own grace
 * window, and that is exactly the case of a definition published four minutes
 * before its first occurrence. An occurrence older than that is not materialised
 * at all rather than materialised-then-swept: a row nobody could ever have seen
 * is not history, and writing one would put a `missed` event on the calendar for
 * a date on which this deployment had no such event.
 *
 * **Expansion is per definition and one failure does not stop the sweep.** A
 * spec written directly into the database with a shape the validator would have
 * refused is a bad row, not a bad tick.
 */
async function expandSchedules(now) {
  const definitions = await definitionsDb.findSchedulable()
  let created = 0
  const heldUnverified = new Set()

  for (const definition of definitions) {
    const schedule = definition.version_spec?.schedule
    if (!schedule || schedule.kind === 'manual') continue

    const from = now.getTime() - Number(definition.grace_seconds || 0) * 1000
    const to = now.getTime() + HORIZON_DAYS * recurrence.DAY_MS

    let occurrences
    try {
      occurrences = recurrence.occurrencesBetween(schedule, definition.timezone || 'UTC', from, to, {
        limit: MAX_OCCURRENCES_PER_DEFINITION,
      })
    } catch (err) {
      log.error('could not expand a schedule', {
        definition: definition.id,
        timezone: definition.timezone,
        message: err.message,
      })
      continue
    }

    for (const occurrence of occurrences) {
      try {
        // `runsModel.create` rather than a second insert path: it re-checks that
        // the definition is still `ready` and the version still has phases, and
        // it materialises the first phase's steps with their idempotency keys.
        // Nobody is watching a scheduled occurrence, so it needs those checks
        // more than a hand-started one does.
        const result = await runsModel.create(
          definition.id,
          // Scope is empty, deliberately (org lead, 2026-09-02). A fan-out across
          // named scopes needs a registry of what a scope IS, which no phase owns
          // yet; inventing one here would be a contract the modules were never
          // asked about. An admin's own start route still takes any scope.
          { scope: '', scheduledFor: occurrence.at, source: 'schedule' },
          null,
        )
        if (!result.ok) {
          if (result.code === 'unverified') {
            // §K's gate, and it must not be silent. There is no run row to hang
            // a diagnostic line on — that is the point, nothing was created — so
            // it is said once per definition per tick rather than once per
            // occurrence, and the admin surface says it where the author is
            // looking: a `ready` definition carries `versionVerified: false` and
            // the editor shows the one button that clears it.
            if (!heldUnverified.has(definition.id)) {
              heldUnverified.add(definition.id)
              log.warn('scheduled occurrences held: the published version has never been verified', {
                definition: definition.id,
                title: definition.title,
                version: definition.current_version_id,
              })
            }
            break
          }
          continue
        }
        if (!result.created) continue
        created += 1
        if (occurrence.adjusted) {
          // Why the clock reads oddly, recorded where an operator will look for
          // it rather than left to be rediscovered at 3am on the last Sunday in
          // October.
          await logDb.write({
            runId: result.run.id,
            kind: 'run.created',
            detail: {
              dstAdjusted: occurrence.adjusted,
              shiftMinutes: occurrence.shiftMinutes,
              timezone: definition.timezone,
              scheduledFor: occurrence.at,
            },
          })
        }
      } catch (err) {
        log.error('could not materialise an occurrence', {
          definition: definition.id,
          at: occurrence.at,
          message: err.message,
        })
      }
    }
  }

  if (created) log.info('occurrences materialised', { created, horizonDays: HORIZON_DAYS })
  return created
}

async function sweepMissed(now) {
  const missed = await runsDb.findMissed(now)
  let n = 0
  for (const run of missed) {
    if (await runsDb.transition(run.id, 'scheduled', 'missed', { error: 'the grace window passed' })) {
      await stepsDb.cancelPending(run.id)
      await logDb.write({
        runId: run.id,
        kind: 'run.status',
        detail: { to: 'missed', scheduledFor: run.scheduled_for },
      })
      n += 1
    }
  }
  return n
}

let lastPruneAt = 0

async function prune(now) {
  if (now.getTime() - lastPruneAt < PRUNE_EVERY_MS) return 0
  lastPruneAt = now.getTime()
  const before = new Date(now.getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000)
  const deleted = await logDb.pruneTerminal(before)
  if (deleted) log.info('event run log pruned', { deleted, before, retentionDays: LOG_RETENTION_DAYS })
  return deleted
}

async function tick(now = new Date()) {
  try {
    await runsDb.reclaimStale(now)
    await stepsDb.reclaimStale(now, MAX_ATTEMPTS)
  } catch (err) {
    log.error('failed to reclaim stale claims', { message: err.message })
  }

  try {
    await expandSchedules(now)
  } catch (err) {
    log.error('schedule expansion failed', { message: err.message })
  }

  try {
    await sweepMissed(now)
  } catch (err) {
    log.error('missed sweep failed', { message: err.message })
  }

  let due
  try {
    due = await runsDb.findDue(now, RUN_BATCH)
  } catch (err) {
    log.error('failed to load due runs', { message: err.message })
    return
  }

  const counts = {}
  for (const run of due || []) {
    try {
      const outcome = await processRun(run, now)
      counts[outcome] = (counts[outcome] || 0) + 1
    } catch (err) {
      log.error('run failed', { run: run.id, message: err.message })
    }
  }
  if (due && due.length) log.info('event runs swept', { due: due.length, ...counts })

  try {
    await prune(now)
  } catch (err) {
    log.error('log prune failed', { message: err.message })
  }
}

let timer = null
// Guards against this process running two ticks over the same runs at once.
// `setInterval` fires whether or not the last callback returned, and the CAS
// alone does not cover it now that a live lease is not re-enterable by its own
// owner — a second tick would simply find every run claimed and do nothing
// useful, one query at a time, for as long as the first one ran.
let ticking = false

function start() {
  if (timer) return timer
  timer = setInterval(() => {
    if (ticking) {
      log.warn('event tick still running; skipping this interval')
      return
    }
    ticking = true
    tick()
      .catch((err) => log.error('event tick failed', { message: err.message }))
      .finally(() => {
        ticking = false
      })
  }, POLL_MS)
  if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown)
  log.info('event runner started', { pollMs: POLL_MS, owner: OWNER, maxAttempts: MAX_ATTEMPTS })
  return timer
}

function stop() {
  if (timer) {
    clearInterval(timer)
    timer = null
  }
}

module.exports = {
  start,
  stop,
  tick,
  processRun,
  advanceRun,
  drainStep,
  sweepMissed,
  expandSchedules,
  prune,
  OWNER,
  POLL_MS,
  MAX_ATTEMPTS,
  RETRY_MS,
  HORIZON_DAYS,
  RUN_LEASE_MS,
  LOG_RETENTION_DAYS,
  RUN_TERMINAL,
}