// ── The runner's raw SQL, against a real MariaDB ───────────────────────────
//
// EVENTS_PLAN.md Phase 2. `eventRunner.test.js` stubs the three tables and
// exercises everything the runner DECIDES. It cannot prove the statements whose
// whole correctness is a server contract, and on this codebase that gap has
// already cost something once: engagement's cooldown claim was green against its
// stub and always allowed the send against a real server, because the connector
// defaults `foundRows: true` and a no-op UPDATE reports 1 rather than 0.
//
// So the five statements that decide who owns what run here, for real:
//
//   • **`claimStart`** — the CAS `scheduled -> starting`. "Exactly one winner" is
//     `affectedRows = 1` for one caller and 0 for every other, and that is a
//     property of the SERVER's answer, not of the SQL's shape.
//   • **`claimTick`** — the same, for a run already in flight, and with no
//     owner-matches escape clause. A live lease must refuse its own holder, or
//     one `setInterval` that overran advances one run twice.
//   • **`transition`** — a guarded status move. The guard is the whole thing: a
//     run cancelled between the read and the write must not be transitioned.
//   • **`reclaimStale`** on steps — two statements in a fixed ORDER, give-up
//     before hand-back. Reversing them makes `MAX_ATTEMPTS` unreachable and the
//     row cycles forever (Engagement Phase 14's defect), and **neither statement
//     may touch `attempts`**.
//   • **`holdNext`** — `UPDATE ... ORDER BY seq LIMIT 1` with a guard, which is
//     both a MariaDB-specific syntax and a correctness claim: it must move the
//     next PENDING step and only ever push a due date later.
//
// **Phase 3 added four more**, and each of them is a control a staff member
// presses against a live game world:
//
//   * **`confirmParked` / `skipByHuman`** - both keyed on `status = 'running'
//     AND claim_expires_at IS NULL`. That pair, and only that pair, means "a cue
//     waiting on a human". If the clause let a LEASED step through, confirm would
//     race the process mid-dispatch on that row.
//   * **`cancelOpen`** - pending steps and parked cues, never a leased one.
//   * **`lastStartedSeq`** - a `MAX(seq) ... WHERE status <> 'pending'`, which is
//     what decides whether retry is offered. The first draft asked for the LOWEST
//     unsettled seq instead, which is a different step whenever a phase carried
//     on past an `on_failure: skip` failure.
//
// **Phase 4 added two reads**, and a read earns a place here when a stub cannot
// tell it is wrong:
//
//   * **`findSchedulable`** - the query the runner runs on EVERY tick to decide
//     what has a recurrence to expand. It joins a definition to its published
//     version and left-joins the series, and every stub of it in
//     `eventSchedule.test.js` is a hand-written object rather than that join. A
//     syntax error or a wrong join direction here is a runner that materialises
//     nothing, silently, for ever.
//   * **`repinScheduled`** - the UPDATE publish runs over already-materialised
//     occurrences. Its guard is the whole statement, and the two rows it must
//     NOT touch are a run that has started and a run that is already terminal.
//   * **`listInWindow`** - the calendar's real half, with a correlated subquery
//     for `waiting_steps` and a LEFT JOIN that must not drop a definition with no
//     series.
//
// **Phase 5 added the gate statements**, and the increment is the one on this
// list with the closest precedent: engagement's cooldown claim was green against
// its stub and always allowed the send against a real server, because the
// connector defaults `foundRows: true`. A gate's tally is the same shape of
// claim — a conditional UPDATE whose answer decides something — so it is proved
// here rather than believed.
//
//   * **`GATE_COUNT`** - `tally = tally + 1` with `CASE WHEN tally + 1 >=
//     needed` inside the same statement, guarded on `satisfied_at IS NULL`. Two
//     emits arriving together must each add one and exactly ONE of them must
//     cross the threshold; a read-then-write would let both see 2 of 3. **This
//     is the statement that failed here first**: MariaDB evaluates SET
//     assignments left to right with the values already assigned, so an
//     increment written before the CASE made the CASE read the new tally and
//     close a two-firing gate on the first firing. The ORDER of the SET list is
//     load-bearing, and only a real server says so.
//   * **`GATE_SATISFY`** - the same guard for the two reasons that are not a
//     firing: an `after` deadline passing, and a human forcing it. A force that
//     races the tick must lose harmlessly.
//   * **`GATE_OPEN_FOR_TRIGGER`** - the emit path's only query, and the one
//     index in this feature on a hot path. Its JOIN is what stops a gate
//     belonging to a cancelled run counting for ever.
//   * **`uq_evgate_phase`** - what makes `open` an INSERT IGNORE, so a process
//     that died between entering a phase and opening its gate opens no second
//     one on the next tick.
//
// **Phase 8 added the resource ledger**, and its unique key is the single most
// server-dependent thing in this feature. "Two events cannot hold a lease on one
// target" has to hold among LIVE rows only — last week's finished event must not
// keep this week's from leasing the same rate — and MariaDB has no partial index,
// so the encoding is a STORED generated column that goes NULL once the row is no
// longer ours. Whether multiple NULLs collide in a unique index is a property of
// the server and of nothing else, and TEAMS.md §2.5 already had to be corrected
// once on this exact shape: MariaDB refuses ON DELETE SET NULL on a foreign key
// whose column is a base column of a stored generated column (error 1901), which
// is why the expression reads `status` alone and `step_id` stays a SET NULL FK.
// Both halves of that are proved below rather than believed.
//
// Plus the two unique indexes that are load-bearing rather than tidy:
// `uq_evrun_occurrence` (which, not the claim, is what stops two runs of one
// occurrence existing) and `uq_evstep_slot` (which is what makes re-materialising
// a phase a no-op).
//
// **It SKIPS when there is no database**, deliberately: CI runs the suite with
// the pool pointed at a dead port, and a file that failed there would make every
// PR red for a reason unrelated to itself. Run it against this machine's
// container with:
//
//   DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=... DB_PASSWORD=... \
//     node --test test/eventRunnerSql.test.js
//
// It creates its tables in a throwaway database named after the process and
// drops it again, so it can never touch a real schema.

const { test, before, after, beforeEach } = require('node:test')
const assert = require('node:assert/strict')
const mariadb = require('mariadb')

// Trimmed to the columns these statements read or write. The ENUMs are verbatim,
// because "is `missed` a legal value" is one of the things being proved.
const SCHEMA = `
CREATE TABLE event_series (
  id       INT AUTO_INCREMENT PRIMARY KEY,
  name     VARCHAR(160) NOT NULL,
  slug     VARCHAR(160) NOT NULL,
  ordering INT NOT NULL DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_definitions (
  id            INT AUTO_INCREMENT PRIMARY KEY,
  title         VARCHAR(200) NOT NULL DEFAULT 'x',
  slug          VARCHAR(200) NOT NULL DEFAULT 'x',
  state         ENUM('draft','ready','archived') NOT NULL DEFAULT 'draft',
  current_version_id INT NULL,
  series_id     INT NULL,
  concurrency_key VARCHAR(190) NULL,
  timezone      VARCHAR(64) NOT NULL DEFAULT 'UTC',
  grace_seconds INT NOT NULL DEFAULT 900
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_versions (
  id            INT AUTO_INCREMENT PRIMARY KEY,
  definition_id INT NOT NULL,
  version       INT NOT NULL DEFAULT 1,
  spec          JSON NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_runs (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  definition_id INT      NOT NULL,
  version_id    INT      NOT NULL,
  scope         VARCHAR(190) NOT NULL DEFAULT '',
  status        ENUM('scheduled','starting','running','paused','ending',
                     'completed','cancelled','failed','missed')
                NOT NULL DEFAULT 'scheduled',
  health        ENUM('ok','degraded','stalled') NOT NULL DEFAULT 'ok',
  current_phase VARCHAR(64)  NULL,
  scheduled_for DATETIME NOT NULL,
  concurrency_key VARCHAR(190) NULL,
  started_at    DATETIME NULL,
  ended_at      DATETIME NULL,
  claimed_by    VARCHAR(64)  NULL,
  claim_expires_at DATETIME  NULL,
  last_error    VARCHAR(500) NULL,
  created_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uq_evrun_occurrence (definition_id, scope, scheduled_for),
  INDEX idx_evrun_due (status, scheduled_for),
  INDEX idx_evrun_concurrency (concurrency_key, status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_run_steps (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  run_id        BIGINT   NOT NULL,
  phase         VARCHAR(64)  NOT NULL,
  seq           INT      NOT NULL,
  action_id     VARCHAR(96)  NOT NULL,
  params        JSON     NULL,
  action_version INT     NOT NULL DEFAULT 1,
  status        ENUM('pending','running','done','failed','skipped','refused','cancelled')
                NOT NULL DEFAULT 'pending',
  due_at        DATETIME NULL,
  attempts      INT      NOT NULL DEFAULT 0,
  on_failure    VARCHAR(32)  NOT NULL DEFAULT 'skip',
  idempotency_key CHAR(40)   NOT NULL,
  claimed_by    VARCHAR(64)  NULL,
  claim_expires_at DATETIME  NULL,
  last_error    VARCHAR(500) NULL,
  started_at    DATETIME NULL,
  finished_at   DATETIME NULL,
  created_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uq_evstep_slot (run_id, phase, seq),
  INDEX idx_evstep_due (status, due_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_run_phase_gates (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  run_id        BIGINT   NOT NULL,
  phase         VARCHAR(64)  NOT NULL,
  kind          ENUM('after','on') NOT NULL,
  after_seconds INT          NULL,
  trigger_id    VARCHAR(96)  NULL,
  conditions    JSON         NULL,
  needed        INT      NOT NULL DEFAULT 1,
  tally         INT      NOT NULL DEFAULT 0,
  entered_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  due_at        DATETIME NULL,
  last_event    JSON     NULL,
  last_event_at DATETIME NULL,
  satisfied_at  DATETIME NULL,
  satisfied_by  VARCHAR(16)  NULL,
  forced_by     INT          NULL,
  created_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  CONSTRAINT fk_evgate_run FOREIGN KEY (run_id) REFERENCES event_runs(id) ON DELETE CASCADE,
  UNIQUE KEY uq_evgate_phase (run_id, phase),
  INDEX idx_evgate_open (trigger_id, satisfied_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_run_budget (
  id          BIGINT AUTO_INCREMENT PRIMARY KEY,
  run_id      BIGINT      NOT NULL,
  dimension   VARCHAR(96) NOT NULL,
  consumed    INT         NOT NULL DEFAULT 0,
  cap         INT         NULL,
  effective_from VARCHAR(96) NULL,
  created_at  DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at  DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  CONSTRAINT fk_evbud_run FOREIGN KEY (run_id) REFERENCES event_runs(id) ON DELETE CASCADE,
  UNIQUE KEY uq_evbud_dim (run_id, dimension)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_run_resources (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  run_id        BIGINT       NOT NULL,
  step_id       BIGINT       NULL,
  owner_module  VARCHAR(64)  NOT NULL,
  kind          VARCHAR(64)  NOT NULL,
  ref           VARCHAR(190) NOT NULL,
  payload       JSON         NULL,
  lease_until   DATETIME     NULL,
  status        ENUM('pending','confirmed','reverting','reverted','orphaned','drifted')
                NOT NULL DEFAULT 'pending',
  revert_attempts INT        NOT NULL DEFAULT 0,
  last_error    VARCHAR(500) NULL,
  member_key    VARCHAR(190) NULL,
  created_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  live_marker   TINYINT      AS (IF(status IN ('pending','confirmed','reverting'), 1, NULL)) STORED,
  CONSTRAINT fk_evres_run  FOREIGN KEY (run_id)  REFERENCES event_runs(id) ON DELETE CASCADE,
  CONSTRAINT fk_evres_step FOREIGN KEY (step_id) REFERENCES event_run_steps(id) ON DELETE SET NULL,
  UNIQUE KEY uq_evres_target (owner_module, kind, ref, live_marker),
  INDEX idx_evres_run (run_id, status),
  INDEX idx_evres_live (status, lease_until)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE event_action_settings (
  action_id   VARCHAR(96) NOT NULL PRIMARY KEY,
  enabled     TINYINT(1)  NOT NULL DEFAULT 0,
  caps        JSON        NULL,
  updated_by  INT         NULL,
  updated_at  DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
`

// The statements under test, verbatim from `eventRuns.db.js` and
// `eventRunSteps.db.js`. Duplicated rather than required, because requiring the
// modules would drag in `utils/db`'s pool, which the harness has already pointed
// at a dead port. The pool below leaves `foundRows` at the connector's default,
// exactly as `utils/db.js` does — pinning it here would make this file agree with
// the code by construction and prove nothing about the pool the server runs.
const CLAIM_START = `
UPDATE event_runs
   SET status = 'starting', claimed_by = ?, claim_expires_at = ?,
       started_at = COALESCE(started_at, NOW())
 WHERE id = ? AND status = 'scheduled'`

const CLAIM_TICK = `
UPDATE event_runs
   SET claimed_by = ?, claim_expires_at = ?
 WHERE id = ?
   AND status IN ('starting','running','ending')
   AND (claim_expires_at IS NULL OR claim_expires_at < ?)`

const TRANSITION = `
UPDATE event_runs SET status = ?, current_phase = ?
 WHERE id = ? AND status IN (?)`

const CLAIM_STEP = `
UPDATE event_run_steps
   SET status = 'running', attempts = attempts + 1, claimed_by = ?, claim_expires_at = ?,
       started_at = COALESCE(started_at, NOW())
 WHERE id = ? AND status = 'pending' AND (due_at IS NULL OR due_at <= ?)`

const STEP_GIVE_UP = `
UPDATE event_run_steps
   SET status = 'failed', last_error = 'gave up after repeated interruptions',
       finished_at = NOW(), claimed_by = NULL, claim_expires_at = NULL
 WHERE status = 'running'
   AND claim_expires_at IS NOT NULL AND claim_expires_at < ?
   AND attempts >= ?`

const STEP_RECLAIM = `
UPDATE event_run_steps
   SET status = 'pending', claimed_by = NULL, claim_expires_at = NULL
 WHERE status = 'running' AND claim_expires_at IS NOT NULL AND claim_expires_at < ?`

const HOLD_NEXT = `
UPDATE event_run_steps
   SET due_at = ?
 WHERE run_id = ? AND phase = ? AND seq > ? AND status = 'pending'
   AND (due_at IS NULL OR due_at < ?)
 ORDER BY seq LIMIT 1`

// Phase 3's four, verbatim from `eventRunSteps.db.js`.
const CONFIRM_PARKED = `
UPDATE event_run_steps
   SET status = 'done', finished_at = NOW(), claimed_by = NULL,
       last_error = ?
 WHERE id = ? AND status = 'running' AND claim_expires_at IS NULL`

const SKIP_BY_HUMAN = `
UPDATE event_run_steps
   SET status = 'skipped', finished_at = NOW(), claimed_by = NULL,
       last_error = ?
 WHERE id = ?
   AND (status = 'pending' OR (status = 'running' AND claim_expires_at IS NULL))`

const REQUEUE = `
UPDATE event_run_steps
   SET status = 'pending', attempts = 0, due_at = NULL, last_error = NULL,
       claimed_by = NULL, claim_expires_at = NULL, finished_at = NULL
 WHERE id = ? AND status = 'failed'`

const CANCEL_OPEN = `
UPDATE event_run_steps
   SET status = 'cancelled', finished_at = NOW(), claimed_by = NULL
 WHERE run_id = ?
   AND (status = 'pending' OR (status = 'running' AND claim_expires_at IS NULL))`

const LAST_STARTED_SEQ = `
SELECT MAX(seq) AS seq FROM event_run_steps
 WHERE run_id = ? AND phase = ? AND status <> 'pending'`

// Phase 5's four, verbatim from `eventPhaseGates.db.js`.
const GATE_OPEN = `
INSERT IGNORE INTO event_run_phase_gates
   (run_id, phase, kind, after_seconds, trigger_id, conditions, needed, entered_at, due_at)
 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`

const GATE_COUNT = `
UPDATE event_run_phase_gates
    SET satisfied_at = CASE WHEN tally + 1 >= needed THEN ? ELSE NULL END,
        satisfied_by = CASE WHEN tally + 1 >= needed THEN 'condition' ELSE NULL END,
        last_event = ?,
        last_event_at = ?,
        tally = tally + 1
  WHERE id = ? AND satisfied_at IS NULL`

const GATE_SATISFY = `
UPDATE event_run_phase_gates
    SET satisfied_at = ?, satisfied_by = ?, forced_by = ?
  WHERE id = ? AND satisfied_at IS NULL`

const GATE_OPEN_FOR_TRIGGER = `
SELECT g.* FROM event_run_phase_gates g
   JOIN event_runs r ON r.id = g.run_id
  WHERE g.trigger_id = ? AND g.satisfied_at IS NULL
    AND r.status IN ('running','paused')
    AND r.current_phase = g.phase
  LIMIT 200`

const MATERIALISE_RUN = `
INSERT IGNORE INTO event_runs (definition_id, version_id, scope, scheduled_for, concurrency_key)
VALUES (?, ?, ?, ?, ?)`

const MATERIALISE_STEP = `
INSERT IGNORE INTO event_run_steps (run_id, phase, seq, action_id, idempotency_key)
VALUES (?, ?, ?, ?, ?)`

const FIND_MISSED = `
SELECT r.id FROM event_runs r
  JOIN event_definitions d ON d.id = r.definition_id
 WHERE r.status = 'scheduled'
   AND r.scheduled_for + INTERVAL d.grace_seconds SECOND < ?`

// Verbatim `eventDefinitions.db#findSchedulable`.
const FIND_SCHEDULABLE = `
SELECT d.id, d.title, d.slug, d.timezone, d.grace_seconds, d.concurrency_key,
       d.current_version_id, d.series_id, v.spec AS version_spec,
       s.name AS series_name, s.slug AS series_slug
  FROM event_definitions d
  JOIN event_versions v ON v.id = d.current_version_id
  LEFT JOIN event_series s ON s.id = d.series_id
 WHERE d.state = 'ready'
 ORDER BY d.id`

// Verbatim `eventRuns.db#listInWindow`, with no optional filter applied.
const LIST_IN_WINDOW = `
SELECT r.*, d.title AS definition_title, d.slug AS definition_slug,
       d.series_id AS series_id, se.name AS series_name, se.slug AS series_slug,
       v.version AS version_number,
       (SELECT COUNT(*) FROM event_run_steps s
         WHERE s.run_id = r.id AND s.status = 'running' AND s.claim_expires_at IS NULL) AS waiting_steps
  FROM event_runs r
  JOIN event_definitions d ON d.id = r.definition_id
  JOIN event_versions    v ON v.id = r.version_id
  LEFT JOIN event_series se ON se.id = d.series_id
 WHERE r.scheduled_for >= ? AND r.scheduled_for < ?
 ORDER BY r.scheduled_for, r.id
 LIMIT 500`

// Verbatim `eventRuns.db#repinScheduled`.
const REPIN_SCHEDULED = `
UPDATE event_runs
   SET version_id = ?
 WHERE definition_id = ?
   AND status = 'scheduled'
   AND started_at IS NULL
   AND version_id <> ?`

const DB = `rg_events_test_${process.pid}`
let pool = null
let available = false

const poolOpts = () => ({
  host: process.env.DB_HOST || '127.0.0.1',
  port: Number(process.env.DB_PORT) || 3306,
  user: process.env.DB_USER || 'root',
  password: process.env.DB_PASSWORD || '',
})

before(async () => {
  const admin = mariadb.createPool({
    ...poolOpts(),
    connectionLimit: 1,
    connectTimeout: 2000,
    initializationTimeout: 2000,
    multipleStatements: true,
  })
  try {
    await admin.query(`CREATE DATABASE ${DB}`)
    available = true
  } catch {
    available = false
  } finally {
    await admin.end().catch(() => {})
  }
  if (!available) return

  pool = mariadb.createPool({
    ...poolOpts(),
    database: DB,
    connectionLimit: 3,
    multipleStatements: true,
    bigIntAsNumber: true,
    insertIdAsNumber: true,
  })
  await pool.query(SCHEMA)
})

after(async () => {
  if (pool) {
    await pool.query(`DROP DATABASE IF EXISTS ${DB}`).catch(() => {})
    await pool.end().catch(() => {})
  }
})

// Checked INSIDE each test, never as a `{ skip }` option: the option is evaluated
// when the file is read, which is before `before()` has had a chance to find out
// whether there is a database. Every test skipped unconditionally is what that
// mistake looks like, and it looks exactly like a passing suite.
const SKIP = 'no database reachable - set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD to run'
const needDb = (t) => {
  if (available) return false
  t.skip(SKIP)
  return true
}

const T0 = new Date('2026-09-02T12:00:00Z')
const later = (ms) => new Date(T0.getTime() + ms)
const rows = (r) => Number(r.affectedRows)

beforeEach(async () => {
  if (!available) return
  await pool.query('DELETE FROM event_run_phase_gates')
  await pool.query('DELETE FROM event_run_resources')
  await pool.query('DELETE FROM event_run_steps')
  await pool.query('DELETE FROM event_runs')
  await pool.query('DELETE FROM event_definitions')
  await pool.query('DELETE FROM event_versions')
  await pool.query('DELETE FROM event_series')
})

async function seedRun(over = {}) {
  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (?)', [
    over.graceSeconds ?? 900,
  ])
  const r = await pool.query(
    `INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, concurrency_key,
                             claimed_by, claim_expires_at)
     VALUES (?, 1, ?, ?, ?, ?, ?, ?)`,
    [
      def.insertId,
      over.scope ?? '',
      over.status ?? 'scheduled',
      over.scheduledFor ?? T0,
      over.concurrencyKey ?? null,
      over.claimedBy ?? null,
      over.claimExpiresAt ?? null,
    ],
  )
  return { runId: r.insertId, definitionId: def.insertId }
}

const seedStep = async (runId, over = {}) =>
  (
    await pool.query(
      `INSERT INTO event_run_steps (run_id, phase, seq, action_id, status, due_at, attempts,
                                    claimed_by, claim_expires_at, idempotency_key)
       VALUES (?, ?, ?, 'test.noop', ?, ?, ?, ?, ?, ?)`,
      [
        runId,
        over.phase ?? 'main',
        over.seq ?? 0,
        over.status ?? 'pending',
        over.dueAt ?? null,
        over.attempts ?? 0,
        over.claimedBy ?? null,
        over.claimExpiresAt ?? null,
        over.key ?? 'k'.repeat(40),
      ],
    )
  ).insertId

const stepById = async (id) => (await pool.query('SELECT * FROM event_run_steps WHERE id = ?', [id]))[0]
const runById = async (id) => (await pool.query('SELECT * FROM event_runs WHERE id = ?', [id]))[0]

// ── claimStart: exactly one winner ─────────────────────────────────────────

test('claimStart: the first caller wins and every other gets zero', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun()

  const first = await pool.query(CLAIM_START, ['host:1', later(60_000), runId])
  const second = await pool.query(CLAIM_START, ['host:2', later(60_000), runId])

  assert.equal(rows(first), 1, 'the winner is told 1')
  assert.equal(rows(second), 0, 'the loser is told 0, not 1 with foundRows')
  assert.equal((await runById(runId)).claimed_by, 'host:1')
  assert.equal((await runById(runId)).status, 'starting')
})

test('claimStart: started_at is stamped once and never moved', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun()
  await pool.query(CLAIM_START, ['host:1', later(60_000), runId])
  const first = (await runById(runId)).started_at

  await pool.query("UPDATE event_runs SET status = 'scheduled' WHERE id = ?", [runId])
  await pool.query(CLAIM_START, ['host:2', later(60_000), runId])

  assert.deepEqual((await runById(runId)).started_at, first, 'COALESCE keeps the original instant')
})

// ── claimTick: a live lease refuses even its own holder ────────────────────

test('claimTick: a live lease is not re-enterable by the process that took it', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running', claimedBy: 'host:1', claimExpiresAt: later(60_000) })

  const again = await pool.query(CLAIM_TICK, ['host:1', later(120_000), runId, T0])
  assert.equal(rows(again), 0, 'a tick that overran must not advance its own run twice')
})

test('claimTick: an expired lease is takeable, by anyone', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running', claimedBy: 'host:1', claimExpiresAt: later(-60_000) })

  const taken = await pool.query(CLAIM_TICK, ['host:2', later(60_000), runId, T0])
  assert.equal(rows(taken), 1)
  assert.equal((await runById(runId)).claimed_by, 'host:2')
})

test('claimTick: a paused run is never claimable', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'paused' })
  assert.equal(rows(await pool.query(CLAIM_TICK, ['host:1', later(60_000), runId, T0])), 0)
})

// ── transition: the guard is the whole point ───────────────────────────────

test('transition: a run cancelled underneath the tick is not transitioned', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  await pool.query("UPDATE event_runs SET status = 'cancelled' WHERE id = ?", [runId])

  const moved = await pool.query(TRANSITION, ['completed', 'main', runId, 'running'])
  assert.equal(rows(moved), 0)
  assert.equal((await runById(runId)).status, 'cancelled')
})

test('transition: running -> running is a guarded write, not a no-op', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })

  // This is how the runner advances `current_phase`, and `foundRows` is exactly
  // what makes it report 1 despite `status` not changing — which is the answer
  // the caller needs, because what it is checking is that the run is STILL
  // running, not that the status moved.
  const moved = await pool.query(TRANSITION, ['running', 'closing', runId, 'running'])
  assert.equal(rows(moved), 1)
  assert.equal((await runById(runId)).current_phase, 'closing')
})

// ── The step claim ─────────────────────────────────────────────────────────

test('the step claim: one winner, and attempts is incremented by the claim alone', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const stepId = await seedStep(runId)

  assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(60_000), stepId, T0])), 1)
  assert.equal(rows(await pool.query(CLAIM_STEP, ['host:2', later(60_000), stepId, T0])), 0)
  assert.equal((await stepById(stepId)).attempts, 1, 'one claim, one attempt')
})

test('the step claim: a step held behind a core.wait is not due', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const stepId = await seedStep(runId, { dueAt: later(300_000) })

  assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(60_000), stepId, T0])), 0)
  assert.equal(rows(await pool.query(CLAIM_STEP, ['host:1', later(360_000), stepId, later(301_000)])), 1)
})

// ── The reclaim: order, and what it must not touch ─────────────────────────

test('the reclaim hands a stale step back WITHOUT resetting attempts', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const stepId = await seedStep(runId, { status: 'running', attempts: 2, claimExpiresAt: later(-1000) })

  await pool.query(STEP_GIVE_UP, [T0, 3])
  await pool.query(STEP_RECLAIM, [T0])

  const step = await stepById(stepId)
  assert.equal(step.status, 'pending')
  assert.equal(step.attempts, 2, 'Engagement Phase 14: a reclaim that reset this made MAX_ATTEMPTS unreachable')
})

test('the reclaim gives up FIRST, so a spent step leaves running as failed', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const stepId = await seedStep(runId, { status: 'running', attempts: 3, claimExpiresAt: later(-1000) })

  const gaveUp = await pool.query(STEP_GIVE_UP, [T0, 3])
  const reclaimed = await pool.query(STEP_RECLAIM, [T0])

  assert.equal(rows(gaveUp), 1)
  assert.equal(rows(reclaimed), 0, 'reversing these two makes the row retry forever')
  assert.equal((await stepById(stepId)).status, 'failed')
  assert.equal((await stepById(stepId)).attempts, 3)
})

test('the reclaim leaves a PARKED step alone, however long it waits', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const stepId = await seedStep(runId, { status: 'running', attempts: 1, claimExpiresAt: null })

  await pool.query(STEP_GIVE_UP, [later(365 * 24 * 3600 * 1000), 3])
  await pool.query(STEP_RECLAIM, [later(365 * 24 * 3600 * 1000)])

  const step = await stepById(stepId)
  assert.equal(step.status, 'running', 'a NULL lease is a parked cue, not staleness')
  assert.equal(step.attempts, 1)
})

// ── holdNext ───────────────────────────────────────────────────────────────

test('holdNext moves the next PENDING step of the phase, and only one', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  await seedStep(runId, { seq: 0, status: 'done' })
  const second = await seedStep(runId, { seq: 1 })
  const third = await seedStep(runId, { seq: 2 })

  const moved = await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
  assert.equal(rows(moved), 1)
  assert.deepEqual((await stepById(second)).due_at, later(300_000))
  assert.equal((await stepById(third)).due_at, null, 'a wait holds the next step, not the rest of the phase')
})

test('holdNext never pulls a due date earlier, so a re-dispatch cannot double the wait', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  await seedStep(runId, { seq: 0, status: 'done' })
  const second = await seedStep(runId, { seq: 1, dueAt: later(600_000) })

  const moved = await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
  assert.equal(rows(moved), 0)
  assert.deepEqual((await stepById(second)).due_at, later(600_000))
})

test('holdNext skips a step that is already running', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  await seedStep(runId, { seq: 0, status: 'done' })
  await seedStep(runId, { seq: 1, status: 'running' })
  const third = await seedStep(runId, { seq: 2 })

  await pool.query(HOLD_NEXT, [later(300_000), runId, 'main', 0, later(300_000)])
  assert.deepEqual((await stepById(third)).due_at, later(300_000))
})

// ── The two unique indexes that carry the weight ───────────────────────────

test('uq_evrun_occurrence, not the claim, is what stops two runs of one occurrence', async (t) => {
  if (needDb(t)) return
  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')

  const a = await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])
  const b = await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])

  assert.equal(rows(a), 1)
  assert.equal(rows(b), 0, 'INSERT IGNORE answers honestly rather than raising a 1062')
  const all = await pool.query('SELECT COUNT(*) AS n FROM event_runs')
  assert.equal(Number(all[0].n), 1)
})

test("scope '' rather than NULL is what makes that index work at all", async (t) => {
  if (needDb(t)) return
  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')

  // The empty-string case collides, as it must. A NULL scope would NOT: multiple
  // NULLs do not collide in MariaDB, which would silently permit two runs of one
  // occurrence — the reason the column is NOT NULL DEFAULT ''.
  await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])
  assert.equal(rows(await pool.query(MATERIALISE_RUN, [def.insertId, 1, '', T0, null])), 0)

  // Two different scopes are two different occurrences, which is what lets a
  // worldwide event fan out across servers without colliding with itself.
  assert.equal(rows(await pool.query(MATERIALISE_RUN, [def.insertId, 1, 'atlantic', T0, null])), 1)
})

test('uq_evstep_slot makes re-materialising a phase a no-op', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })

  assert.equal(rows(await pool.query(MATERIALISE_STEP, [runId, 'main', 0, 'test.noop', 'a'.repeat(40)])), 1)
  assert.equal(rows(await pool.query(MATERIALISE_STEP, [runId, 'main', 0, 'test.noop', 'b'.repeat(40)])), 0)

  const [step] = await pool.query('SELECT idempotency_key FROM event_run_steps WHERE run_id = ?', [runId])
  assert.equal(step.idempotency_key, 'a'.repeat(40), 'a re-materialise cannot overwrite a key a dispatch already sent')
})

// ── The grace window is per definition ─────────────────────────────────────

test('findMissed compares against each definition’s own grace window', async (t) => {
  if (needDb(t)) return
  const tight = await seedRun({ graceSeconds: 600 })
  const generous = await seedRun({ graceSeconds: 3600, scope: 'b' })

  const missed = (await pool.query(FIND_MISSED, [later(30 * 60 * 1000)])).map((r) => Number(r.id))
  assert.deepEqual(missed, [tight.runId])
  assert.ok(!missed.includes(generous.runId), 'inside its own window a run starts late rather than being missed')
})

// -- Phase 3: the controls a human presses ----------------------------------

test('confirm resolves a parked cue and cannot touch a step being dispatched', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const parked = await seedStep(runId, { seq: 0, status: 'running', claimedBy: 'host:1', claimExpiresAt: null })
  const busy = await seedStep(runId, { seq: 1, status: 'running', claimedBy: 'host:1', claimExpiresAt: later(60_000), key: 'x'.repeat(40) })

  assert.equal(rows(await pool.query(CONFIRM_PARKED, ['gate opened', parked])), 1)
  assert.equal(rows(await pool.query(CONFIRM_PARKED, ['nope', busy])), 0, 'a live lease is a step somebody owns')

  assert.equal((await stepById(parked)).status, 'done')
  assert.equal((await stepById(parked)).last_error, 'gate opened')
  assert.equal((await stepById(busy)).status, 'running')
})

test('a confirm of an already-confirmed cue reports 0, not 1', async (t) => {
  if (needDb(t)) return
  // The engagement Phase 4a shape: a connector that defaults `foundRows: true`
  // reports 1 for an UPDATE that matched and changed nothing, and a control that
  // read that as success would tell a second staff member their press worked.
  const { runId } = await seedRun({ status: 'running' })
  const parked = await seedStep(runId, { status: 'running', claimExpiresAt: null })

  assert.equal(rows(await pool.query(CONFIRM_PARKED, [null, parked])), 1)
  assert.equal(rows(await pool.query(CONFIRM_PARKED, [null, parked])), 0)
})

test('skip takes a pending step and a parked cue, and refuses a leased one', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const pending = await seedStep(runId, { seq: 0, status: 'pending' })
  const parked = await seedStep(runId, { seq: 1, status: 'running', claimExpiresAt: null, key: 'y'.repeat(40) })
  const busy = await seedStep(runId, { seq: 2, status: 'running', claimExpiresAt: later(60_000), key: 'z'.repeat(40) })
  const failed = await seedStep(runId, { seq: 3, status: 'failed', key: 'w'.repeat(40) })

  assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, pending])), 1)
  assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, parked])), 1)
  assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, busy])), 0)
  assert.equal(rows(await pool.query(SKIP_BY_HUMAN, [null, failed])), 0, 'a failed step is terminal; resume carries the phase past it')
})

test('cancelOpen closes pending steps and parked cues, and leaves a leased one alone', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  const done = await seedStep(runId, { seq: 0, status: 'done' })
  const busy = await seedStep(runId, { seq: 1, status: 'running', claimExpiresAt: later(60_000), key: 'p'.repeat(40) })
  const parked = await seedStep(runId, { seq: 2, status: 'running', claimExpiresAt: null, key: 'q'.repeat(40) })
  const pending = await seedStep(runId, { seq: 3, status: 'pending', key: 'r'.repeat(40) })

  assert.equal(rows(await pool.query(CANCEL_OPEN, [runId])), 2)

  assert.equal((await stepById(done)).status, 'done')
  assert.equal((await stepById(busy)).status, 'running', 'nothing can recall a command already sent')
  assert.equal((await stepById(parked)).status, 'cancelled', 'a cancelled run must stop claiming to wait on somebody')
  assert.equal((await stepById(pending)).status, 'cancelled')
})

test('requeue only takes a failed step, and puts attempts back to zero', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'paused' })
  const failed = await seedStep(runId, { seq: 0, status: 'failed', attempts: 3 })
  const pending = await seedStep(runId, { seq: 1, status: 'pending', key: 's'.repeat(40) })

  assert.equal(rows(await pool.query(REQUEUE, [failed])), 1)
  assert.equal(rows(await pool.query(REQUEUE, [pending])), 0)

  const row = await stepById(failed)
  assert.equal(row.status, 'pending')
  assert.equal(Number(row.attempts), 0)
  assert.equal(row.due_at, null, 'a re-queued step is due now, not at the retry backoff it was left on')
})

test('lastStartedSeq names the furthest step of the phase, not the earliest unsettled one', async (t) => {
  if (needDb(t)) return
  // The defect this replaced: a phase that carried on past a failed step (an
  // `on_failure` of `skip`) and then paused at a later one. "The lowest seq that
  // is not settled" answers with the FIRST failure - a step the runner has long
  // since stepped over - and retry would re-queue a row behind its own cursor.
  const { runId } = await seedRun({ status: 'paused' })
  await seedStep(runId, { seq: 0, status: 'failed', key: 'a'.repeat(40) })
  await seedStep(runId, { seq: 1, status: 'done', key: 'b'.repeat(40) })
  await seedStep(runId, { seq: 2, status: 'failed', key: 'c'.repeat(40) })
  await seedStep(runId, { seq: 3, status: 'pending', key: 'd'.repeat(40) })

  const [row] = await pool.query(LAST_STARTED_SEQ, [runId, 'main'])
  assert.equal(Number(row.seq), 2)
})

test('lastStartedSeq is NULL for a phase nothing has touched', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedRun({ status: 'running' })
  await seedStep(runId, { seq: 0, status: 'pending' })

  const [row] = await pool.query(LAST_STARTED_SEQ, [runId, 'main'])
  assert.equal(row.seq, null, 'a null must read as "nothing to retry", not as seq 0')
})

test('a guarded transition refuses a run that was cancelled underneath it', async (t) => {
  if (needDb(t)) return
  // What the admin cancel looks like from the runner's side, mid-tick: the
  // guarded write returns 0 and the tick treats the run as taken rather than
  // advancing a run somebody has just stopped.
  const { runId } = await seedRun({ status: 'running' })
  await pool.query("UPDATE event_runs SET status = 'cancelled' WHERE id = ?", [runId])

  assert.equal(rows(await pool.query(TRANSITION, ['running', 'two', runId, 'running'])), 0)
  assert.equal((await runById(runId)).status, 'cancelled')
})


// ── Phase 4: the two reads ────────────────────────────────────────

/** A definition with a published version, and optionally a series. */
async function seedDefinition({ state = 'ready', spec = { schedule: { kind: 'manual' } }, series = null } = {}) {
  let seriesId = null
  if (series) {
    const s = await pool.query('INSERT INTO event_series (name, slug) VALUES (?, ?)', [series, series])
    seriesId = s.insertId
  }
  const d = await pool.query(
    'INSERT INTO event_definitions (state, series_id, timezone) VALUES (?, ?, ?)',
    [state, seriesId, 'Europe/Berlin'],
  )
  const v = await pool.query(
    'INSERT INTO event_versions (definition_id, version, spec) VALUES (?, 1, ?)',
    [d.insertId, JSON.stringify(spec)],
  )
  await pool.query('UPDATE event_definitions SET current_version_id = ? WHERE id = ?', [
    v.insertId,
    d.insertId,
  ])
  return { definitionId: d.insertId, versionId: v.insertId, seriesId }
}

test('findSchedulable returns ready definitions with their PUBLISHED spec, series or not', async (t) => {
  if (needDb(t)) return
  const withSeries = await seedDefinition({
    spec: { schedule: { kind: 'weekly', days: ['friday'], time: '20:00' } },
    series: 'royal-spy',
  })
  const withoutSeries = await seedDefinition({ spec: { schedule: { kind: 'manual' } } })

  const found = await pool.query(FIND_SCHEDULABLE)
  const ids = found.map((r) => r.id).sort((a, b) => a - b)
  assert.deepEqual(ids, [withSeries.definitionId, withoutSeries.definitionId].sort((a, b) => a - b))

  // The LEFT JOIN must not drop the definition that belongs to no arc — an
  // inner join here would make every event outside a series unschedulable, and
  // most events are outside one.
  const plain = found.find((r) => r.id === withoutSeries.definitionId)
  assert.equal(plain.series_name, null)

  const arced = found.find((r) => r.id === withSeries.definitionId)
  assert.equal(arced.series_name, 'royal-spy')
  assert.equal(arced.timezone, 'Europe/Berlin')

  // The spec really came back, and really came back parseable.
  const spec = typeof arced.version_spec === 'string' ? JSON.parse(arced.version_spec) : arced.version_spec
  assert.equal(spec.schedule.kind, 'weekly')
})

test('findSchedulable skips a draft, an archived one, and one with no published version', async (t) => {
  if (needDb(t)) return
  await seedDefinition({ state: 'draft' })
  await seedDefinition({ state: 'archived' })
  // `ready` with a dangling version pointer: the JOIN is what must drop it, and
  // a definition whose version row went missing must not become a runner crash.
  const orphan = await seedDefinition({ state: 'ready' })
  await pool.query('DELETE FROM event_versions WHERE id = ?', [orphan.versionId])

  assert.equal((await pool.query(FIND_SCHEDULABLE)).length, 0)
})

test('listInWindow is half-open on the window, and counts only PARKED steps as waiting', async (t) => {
  if (needDb(t)) return
  const def = await seedDefinition()
  const at = async (when) => {
    const r = await pool.query(
      'INSERT INTO event_runs (definition_id, version_id, scope, scheduled_for) VALUES (?, ?, ?, ?)',
      [def.definitionId, def.versionId, '', when],
    )
    return r.insertId
  }
  const before = await at(new Date('2026-09-01T00:00:00Z'))
  const onFrom = await at(new Date('2026-09-02T00:00:00Z'))
  const inside = await at(new Date('2026-09-05T00:00:00Z'))
  const onTo = await at(new Date('2026-09-09T00:00:00Z'))

  // `>= from AND < to` — the instant ON the upper bound belongs to the NEXT
  // window. A closed interval would draw the last day of one month and the first
  // of the next as the same occurrence twice.
  const found = await pool.query(LIST_IN_WINDOW, [
    new Date('2026-09-02T00:00:00Z'),
    new Date('2026-09-09T00:00:00Z'),
  ])
  assert.deepEqual(found.map((r) => r.id), [onFrom, inside])
  assert.ok(!found.some((r) => r.id === before || r.id === onTo))

  // A parked step is `running` with a NULL lease; a leased one is the runner
  // mid-dispatch and is not waiting on anybody.
  await seedStep(inside, { status: 'running', claimExpiresAt: null, key: 'a'.repeat(40) })
  await seedStep(inside, {
    seq: 1,
    status: 'running',
    claimedBy: 'host',
    claimExpiresAt: later(60_000),
    key: 'b'.repeat(40),
  })
  const again = await pool.query(LIST_IN_WINDOW, [
    new Date('2026-09-02T00:00:00Z'),
    new Date('2026-09-09T00:00:00Z'),
  ])
  assert.equal(Number(again.find((r) => r.id === inside).waiting_steps), 1)
})


test('repinScheduled moves the occurrences that have not begun, and only those', async (t) => {
  if (needDb(t)) return
  // The case: an editor fixes a typo on a weekly event on Wednesday. Two Fridays
  // are already materialised on v3, last Friday's run is finished, and one is in
  // flight right now.
  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
  const mk = async (status, versionId, startedAt, when) =>
    (
      await pool.query(
        `INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
         VALUES (?, ?, ?, ?, ?, ?)`,
        [def.insertId, versionId, `s${when}`, status, T0, startedAt],
      )
    ).insertId

  const ahead1 = await mk('scheduled', 3, null, 1)
  const ahead2 = await mk('scheduled', 3, null, 2)
  const running = await mk('running', 3, T0, 3)
  const done = await mk('completed', 3, T0, 4)
  // Already on the new version: excluded by `version_id <> ?`, so a second
  // publish of an unchanged definition is not a fleet of pointless writes.
  const already = await mk('scheduled', 4, null, 5)

  const moved = rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4]))
  assert.equal(moved, 2)

  const versionOf = async (id) =>
    Number((await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [id]))[0].version_id)
  assert.equal(await versionOf(ahead1), 4)
  assert.equal(await versionOf(ahead2), 4)
  // A run that has begun keeps the version it pinned, for ever: that pin is what
  // makes it explicable afterwards.
  assert.equal(await versionOf(running), 3)
  assert.equal(await versionOf(done), 3)
  assert.equal(await versionOf(already), 4)
})

test('a scheduled run whose started_at is somehow set is left alone', async (t) => {
  if (needDb(t)) return
  // Belt and braces on the guard: `status = 'scheduled'` and `started_at IS NULL`
  // are two conditions rather than one because a row that has both is the only
  // row that is provably untouched.
  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
  const r = await pool.query(
    `INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
     VALUES (?, 3, '', 'scheduled', ?, ?)`,
    [def.insertId, T0, T0],
  )
  assert.equal(rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4])), 0)
  const after = (await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [r.insertId]))[0]
  assert.equal(Number(after.version_id), 3)
})


// ── Phase 5: the gate statements ───────────────────────────────────────────

/** A run in `running` on one phase, plus an open gate for it. */
async function seedGate({ kind = 'on', needed = 1, trigger = 'uo.champ.boss_up', status = 'running', phase = 'boss' } = {}) {
  const { runId } = await seedRun({ status })
  await pool.query('UPDATE event_runs SET current_phase = ? WHERE id = ?', [phase, runId])
  const g = await pool.query(GATE_OPEN, [
    runId,
    phase,
    kind,
    kind === 'after' ? 1800 : null,
    kind === 'on' ? trigger : null,
    kind === 'on' ? JSON.stringify({ variable: 'region', cmp: 'eq', value: 'Yew' }) : null,
    needed,
    T0,
    kind === 'after' ? later(1800_000) : null,
  ])
  return { runId, gateId: Number(g.insertId) }
}

const gateById = async (id) => (await pool.query('SELECT * FROM event_run_phase_gates WHERE id = ?', [id]))[0]

test('two firings arriving together each count, and exactly one crosses the threshold', async (t) => {
  if (needDb(t)) return
  const { gateId } = await seedGate({ needed: 2 })

  // The whole reason the CASE is inside the UPDATE: a read-then-write would let
  // both of these see "1 of 2" and neither satisfy, or both satisfy and advance
  // one phase twice.
  const [a, b] = await Promise.all([
    pool.query(GATE_COUNT, [T0, null, T0, gateId]),
    pool.query(GATE_COUNT, [T0, null, T0, gateId]),
  ])
  assert.equal(rows(a) + rows(b), 2, 'both increments land')

  const gate = await gateById(gateId)
  assert.equal(Number(gate.tally), 2)
  assert.ok(gate.satisfied_at, 'and the second one closed it')
  assert.equal(gate.satisfied_by, 'condition')
})

test('the threshold is read BEFORE the increment, not after it', async (t) => {
  if (needDb(t)) return
  // The defect this file was written to catch, and did: MariaDB evaluates SET
  // assignments left to right with the values already assigned, so an increment
  // placed FIRST makes the CASE after it read `new + 1 >= needed` and close a
  // two-firing gate on the first firing. Every stub of this agrees with the
  // intent rather than with the server, which is exactly why it survived one.
  const { gateId } = await seedGate({ needed: 2 })
  await pool.query(GATE_COUNT, [T0, null, T0, gateId])
  const half = await gateById(gateId)
  assert.equal(Number(half.tally), 1)
  assert.equal(half.satisfied_at, null, 'one of two is not two')

  await pool.query(GATE_COUNT, [T0, null, T0, gateId])
  assert.ok((await gateById(gateId)).satisfied_at, 'and the second one is')
})

test('an increment against a closed gate is refused, not applied', async (t) => {
  if (needDb(t)) return
  const { gateId } = await seedGate({ needed: 1 })
  assert.equal(rows(await pool.query(GATE_COUNT, [T0, null, T0, gateId])), 1)

  // `WHERE satisfied_at IS NULL` — and 0 rather than 1, which is the whole
  // reason this file exists: `foundRows: true` would answer 1 for a no-op.
  assert.equal(rows(await pool.query(GATE_COUNT, [T0, null, T0, gateId])), 0)
  assert.equal(Number((await gateById(gateId)).tally), 1, 'the tally does not keep climbing after the phase moved on')
})

test('a force and a deadline race the same guard, and only one of them wins', async (t) => {
  if (needDb(t)) return
  const { gateId } = await seedGate({ kind: 'after' })
  const [forced, elapsed] = await Promise.all([
    pool.query(GATE_SATISFY, [T0, 'forced', 7, gateId]),
    pool.query(GATE_SATISFY, [T0, 'elapsed', null, gateId]),
  ])
  assert.equal(rows(forced) + rows(elapsed), 1, 'exactly one of the two writes')

  const gate = await gateById(gateId)
  assert.ok(['forced', 'elapsed'].includes(gate.satisfied_by))
  // Whichever won is the one in the row, and the log records that one — never
  // both.
  assert.equal(gate.satisfied_by === 'forced' ? Number(gate.forced_by) : gate.forced_by, gate.satisfied_by === 'forced' ? 7 : null)
})

test('opening a gate twice writes one row', async (t) => {
  if (needDb(t)) return
  const { runId } = await seedGate()
  // The INSERT IGNORE that makes recovery from a process which died between
  // entering a phase and opening its gate uneventful.
  const again = await pool.query(GATE_OPEN, [runId, 'boss', 'on', null, 'uo.champ.boss_up', null, 1, later(60_000), null])
  assert.equal(rows(again), 0)
  const all = await pool.query('SELECT * FROM event_run_phase_gates WHERE run_id = ?', [runId])
  assert.equal(all.length, 1)
  assert.equal(new Date(all[0].entered_at).getTime(), T0.getTime(), 'and the first entry time is the one that stands')
})

test('the emit path sees only gates whose run is live and on that phase', async (t) => {
  if (needDb(t)) return
  const live = await seedGate()
  const paused = await seedGate({ status: 'paused' })
  const cancelled = await seedGate({ status: 'scheduled' })
  await pool.query('UPDATE event_runs SET status = ? WHERE id = ?', ['cancelled', cancelled.runId])

  const moved = await seedGate()
  await pool.query('UPDATE event_runs SET current_phase = ? WHERE id = ?', ['cleanup', moved.runId])

  const closed = await seedGate()
  await pool.query(GATE_SATISFY, [T0, 'forced', null, closed.gateId])

  const found = (await pool.query(GATE_OPEN_FOR_TRIGGER, ['uo.champ.boss_up'])).map((g) => Number(g.id))
  assert.deepEqual(found.sort((a, b) => a - b), [live.gateId, paused.gateId].sort((a, b) => a - b))
  // `paused` counts: the world does not stop because an operator paused the
  // console, and discarding firings during a pause would make pause destructive.
  assert.ok(found.includes(paused.gateId))
})

test('a gate goes with its run', async (t) => {
  if (needDb(t)) return
  // ON DELETE CASCADE, which is what keeps an orphaned gate from outliving the
  // run whose phase it was about.
  const { runId, gateId } = await seedGate()
  await pool.query('DELETE FROM event_run_steps WHERE run_id = ?', [runId])
  await pool.query('DELETE FROM event_runs WHERE id = ?', [runId])
  assert.equal((await pool.query('SELECT * FROM event_run_phase_gates WHERE id = ?', [gateId])).length, 0)
})

// ── Phase 6: the cap's conditional increment ───────────────────────────────
//
// **The plan's own acceptance criterion**: *"two concurrent steps against one cap
// proving neither over-spends"*. It belongs here rather than beside the stub for
// the reason the gate's increment does — the guard lives in a WHERE that the
// server evaluates against the pre-update row, and a stub reproduces the reading
// rather than the server. Engagement's cooldown claim was green against its stub
// and always allowed the send, because the connector defaults `foundRows: true`
// and a no-op UPDATE reports 1: exactly the mistake this shape of statement
// invites, and the reason `spend()` reads `affectedRows` at all.
//
// The SET list has ONE assignment, and that is Phase 5's lesson applied rather
// than rediscovered: MariaDB evaluates SET assignments left to right with the
// values already assigned, so nothing in this statement may read `consumed`
// after writing it. The guard stays in the WHERE.

// **Requiring a shipping model into THIS file starts a second pool**, and it is
// the only file in the suite where that matters. Everywhere else the environment
// points at a dead port and `utils/db`'s pool never connects; here it points at a
// live server, so the pool holds open connections and the process never exits —
// 49 green tests and a file that hangs until the harness kills it. Every other
// event test file already closes it in an `after`; this one now has a reason to.
//
// **And that second pool has to be aimed at the throwaway database**, which it
// was not before Phase 8 noticed. `utils/db` builds its pool at REQUIRE time from
// `DB_NAME`, and its own `dotenv.config()` reads `server/.env` — so a model test
// run on a developer's machine was reaching that developer's real schema while
// the fixtures it was asserting against were being written next door. It passed
// only because the two tables happened to exist in both. `dotenv` does not
// overwrite a variable that already exists, so setting it here, before the
// require below, is what beats the file. This file drops its database in `after`,
// so aiming the model pool at it is also what keeps the whole run disposable.
process.env.DB_NAME = DB

const budgetDb = require('../src/model/events/eventRunBudget.db')
const resourcesDb = require('../src/model/events/eventRunResources.db')
const runsDb = require('../src/model/events/eventRuns.db')
const appDb = require('../src/utils/db')
after(() => appDb.close())

const seedBudget = async (over = {}) => {
  const { runId } = await seedRun({ status: 'running' })
  await pool.query(
    'INSERT INTO event_run_budget (run_id, dimension, consumed, cap, effective_from) VALUES (?, ?, ?, ?, ?)',
    // `'cap' in over`, not `over.cap ?? 30` -- `null` is a MEANINGFUL cap here
    // (uncapped) and `??` coalesces it straight back to 30, which made the
    // uncapped test seed a capped row and fail against the correct statement.
    [
      runId,
      over.dimension ?? 'uo.creatures',
      over.consumed ?? 0,
      'cap' in over ? over.cap : 30,
      over.from ?? 'uo.creature.spawn',
    ],
  )
  return runId
}

const consumedOf = async (runId, dimension = 'uo.creatures') =>
  Number(
    (
      await pool.query('SELECT consumed FROM event_run_budget WHERE run_id = ? AND dimension = ?', [
        runId,
        dimension,
      ])
    )[0].consumed,
  )

// The statement, lifted verbatim from `eventRunBudget.db.js`. Run through this
// file's pool rather than through the module, because the module builds its own
// pool from the environment while this pool points at the throwaway database.
const SPEND = `
  UPDATE event_run_budget
     SET consumed = consumed + ?
   WHERE run_id = ? AND dimension = ? AND (cap IS NULL OR consumed + ? <= cap)`

const spend = async (runId, amount, dimension = 'uo.creatures') =>
  Number((await pool.query(SPEND, [amount, runId, dimension, amount])).affectedRows) > 0

test('two steps spending one cap at once: the second is refused, not queued behind the first', async (t) => {
  if (needDb(t)) return
  // 28 of 30 spent, and two steps each wanting 5 arrive together. A
  // read-then-write would let both see 28 and both spend, ending at 38 of 30 —
  // the exact over-spend the conditional increment exists to make impossible.
  const runId = await seedBudget({ consumed: 28, cap: 30 })
  const [a, b] = await Promise.all([spend(runId, 5), spend(runId, 5)])
  assert.equal(a, false)
  assert.equal(b, false)
  assert.equal(await consumedOf(runId), 28)
})

test('two steps that BOTH fit both spend, and the total is exact', async (t) => {
  if (needDb(t)) return
  // The other half, and the one a too-strict guard would break: a cap is not a
  // lock. Three steps of 5 against 30 must all succeed and land on 15, or the
  // statement is refusing legal work.
  const runId = await seedBudget({ consumed: 0, cap: 30 })
  const results = await Promise.all([spend(runId, 5), spend(runId, 5), spend(runId, 5)])
  assert.deepEqual(results, [true, true, true])
  assert.equal(await consumedOf(runId), 15)
})

test('a spend that exactly reaches the cap is allowed; one over it is not', async (t) => {
  if (needDb(t)) return
  // `<=`, not `<`. A cap of 30 means thirty creatures are permitted, and an
  // off-by-one here is a deployment that can never use the last unit of anything
  // it configured.
  const runId = await seedBudget({ consumed: 25, cap: 30 })
  assert.equal(await spend(runId, 5), true)
  assert.equal(await consumedOf(runId), 30)
  assert.equal(await spend(runId, 1), false)
  assert.equal(await consumedOf(runId), 30)
})

test('a NULL cap is uncapped, and still counts', async (t) => {
  if (needDb(t)) return
  // The row exists so the meter has something to show; nothing bounds it. The
  // distinction matters because a MISSING row is a refusal — a step spending a
  // dimension its own run's version never priced — and the two must not collapse
  // into one behaviour.
  const runId = await seedBudget({ consumed: 0, cap: null })
  assert.equal(await spend(runId, 1000000), true)
  assert.equal(await consumedOf(runId), 1000000)
})

test('spending a dimension with no row is refused', async (t) => {
  if (needDb(t)) return
  const runId = await seedBudget()
  assert.equal(await spend(runId, 1, 'uo.bosses'), false)
})

test('seeding is INSERT IGNORE: a tick that overruns cannot reset a spent cap', async (t) => {
  if (needDb(t)) return
  // The idempotence `materialisePhase` and `gates.open` both have. Without it a
  // second seed would either error on the unique key or — worse, written as an
  // upsert — hand a run that has spent 28 of 30 a fresh 0.
  const { runId } = await seedRun({ status: 'running' })
  const SEED =
    'INSERT IGNORE INTO event_run_budget (run_id, dimension, consumed, cap, effective_from) VALUES (?, ?, 0, ?, ?)'
  await pool.query(SEED, [runId, 'uo.creatures', 30, 'uo.creature.spawn'])
  await spend(runId, 28)
  await pool.query(SEED, [runId, 'uo.creatures', 30, 'uo.creature.spawn'])
  assert.equal(await consumedOf(runId), 28)
  assert.equal((await pool.query('SELECT * FROM event_run_budget WHERE run_id = ?', [runId])).length, 1)
})

test('a refund floors at zero rather than going negative', async (t) => {
  if (needDb(t)) return
  // `GREATEST(consumed - ?, 0)`. A negative `consumed` would make every later cap
  // check lie in the permissive direction, permanently — a worse outcome than a
  // refund that is slightly too small.
  const runId = await seedBudget({ consumed: 3, cap: 30 })
  await pool.query(
    'UPDATE event_run_budget SET consumed = GREATEST(consumed - ?, 0) WHERE run_id = ? AND dimension = ?',
    [10, runId, 'uo.creatures'],
  )
  assert.equal(await consumedOf(runId), 0)
})

test('a budget goes with its run', async (t) => {
  if (needDb(t)) return
  const runId = await seedBudget()
  await pool.query('DELETE FROM event_runs WHERE id = ?', [runId])
  assert.equal((await pool.query('SELECT * FROM event_run_budget WHERE run_id = ?', [runId])).length, 0)
})

test('a version records that a dry run passed against it', async (t) => {
  if (needDb(t)) return
  // §K's gate is two columns on an immutable table, and the ALTER that adds them
  // has to actually apply — `verified_at` is what `runsModel.create` reads to
  // decide whether a scheduled occurrence may start at all, so a column that
  // silently did not exist would hold every schedule on the deployment.
  await pool.query('ALTER TABLE event_versions ADD COLUMN IF NOT EXISTS verified_at DATETIME NULL')
  await pool.query('ALTER TABLE event_versions ADD COLUMN IF NOT EXISTS verified_by INT NULL')
  const v = await pool.query('INSERT INTO event_versions (definition_id, spec) VALUES (1, ?)', ['{}'])
  const before = (await pool.query('SELECT verified_at FROM event_versions WHERE id = ?', [v.insertId]))[0]
  assert.equal(before.verified_at, null)
  const moved = await pool.query('UPDATE event_versions SET verified_at = ?, verified_by = ? WHERE id = ?', [
    T0,
    1,
    v.insertId,
  ])
  assert.equal(Number(moved.affectedRows), 1)
  const after = (
    await pool.query('SELECT verified_at, verified_by FROM event_versions WHERE id = ?', [v.insertId])
  )[0]
  assert.ok(after.verified_at instanceof Date)
  assert.equal(Number(after.verified_by), 1)
})

test('a non-positive spend never reaches the database', async (t) => {
  if (needDb(t)) return
  // Through the shipping module rather than a copy of its statement. An action
  // whose `cost()` answers 0 for its params is telling core it consumes nothing,
  // and pricing that as a query would be one round trip per step behind a fact
  // the caller already has.
  const runId = await seedBudget({ consumed: 0, cap: 10 })
  assert.equal(await budgetDb.spend(runId, 'uo.creatures', 0), true)
  assert.equal(await consumedOf(runId), 0)
})

// ── The resource ledger's unique key (Phase 8) ─────────────────────────────
//
// Every test here is about a property of the SERVER. A stub can enforce whatever
// rule its author had in mind; only MariaDB can say whether this encoding of
// "unique among live rows" actually is one.

const insertResource = async (runId, over = {}) =>
  (
    await pool.query(
      `INSERT INTO event_run_resources (run_id, step_id, owner_module, kind, ref, payload, lease_until, member_key, status)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [
        runId,
        over.stepId ?? null,
        over.owner ?? 'demo',
        over.kind ?? 'creature',
        over.ref ?? '0xA',
        over.payload ?? null,
        over.leaseUntil ?? null,
        over.memberKey ?? null,
        over.status ?? 'pending',
      ],
    )
  ).insertId

const dup = async (fn) => {
  try {
    await fn()
    return null
  } catch (err) {
    return err.code || String(err.errno)
  }
}

test('two runs cannot hold one target, and the refusal comes from the server', async (t) => {
  if (needDb(t)) return
  // §D: "the unique index is what stops two events leasing one target." Not a
  // read-then-insert — two runs entering the same tick would both pass the check
  // — so the whole conflict story is this one constraint answering.
  const a = await seedRun()
  const b = await seedRun()
  await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
  const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
  assert.equal(code, 'ER_DUP_ENTRY')
})

test('the key is released by the three statuses that mean it is no longer ours', async (t) => {
  if (needDb(t)) return
  // Amended 2026-09-03. §D says "among non-reverted rows", which was written
  // before the six statuses had their meanings; taken literally it makes
  // `drifted` and `orphaned` hold a target for ever, so one bad night would
  // disable a lease permanently with no control able to clear it. `drifted` means
  // somebody else has hold of the value and this run has let go; `orphaned` means
  // it vanished. Neither is a claim on the target.
  for (const status of ['reverted', 'drifted', 'orphaned']) {
    await pool.query('DELETE FROM event_run_resources')
    const a = await seedRun()
    const b = await seedRun()
    await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status })
    const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
    assert.equal(code, null, `a ${status} row must not hold the target`)
  }
})

test('the key is HELD by the three that mean core still believes it is ours', async (t) => {
  if (needDb(t)) return
  for (const status of ['pending', 'confirmed', 'reverting']) {
    await pool.query('DELETE FROM event_run_resources')
    const a = await seedRun()
    const b = await seedRun()
    await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status })
    const code = await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' }))
    assert.equal(code, 'ER_DUP_ENTRY', `a ${status} row must hold the target`)
  }
})

test('many released rows on one target coexist, which is the whole encoding', async (t) => {
  if (needDb(t)) return
  // The property the NULL depends on: multiple NULLs do not collide in a unique
  // index. A weekly event that leases the same rate every Saturday accumulates one
  // released row per week, and the fiftieth must not fail to insert.
  const a = await seedRun()
  for (let i = 0; i < 5; i++) {
    await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'reverted' })
  }
  const held = await pool.query(
    'SELECT COUNT(*) AS n FROM event_run_resources WHERE ref = ? AND live_marker IS NULL',
    ['demo.rate'],
  )
  assert.equal(Number(held[0].n), 5)
  // And a live one still goes on top of them.
  assert.equal(await dup(() => insertResource(a.runId, { kind: 'override', ref: 'demo.rate' })), null)
})

test('an UPDATE that releases a row frees the target at once', async (t) => {
  if (needDb(t)) return
  // The generated column is STORED, so this is really asking whether MariaDB
  // recomputes it on UPDATE and re-indexes. Cleanup depends on it entirely: the
  // moment a lease is restored, the next event may take it.
  const a = await seedRun()
  const b = await seedRun()
  const id = await insertResource(a.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
  assert.equal(await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' })), 'ER_DUP_ENTRY')
  await pool.query("UPDATE event_run_resources SET status = 'reverted' WHERE id = ?", [id])
  assert.equal(await dup(() => insertResource(b.runId, { kind: 'override', ref: 'demo.rate' })), null)
})

test('the key is per owner and per kind, not per ref', async (t) => {
  if (needDb(t)) return
  // `kind` and `ref` are module-opaque strings core stores verbatim, so two
  // modules using the same word for different things must not collide — and one
  // module's `creature:0xA` and `item:0xA` are two objects.
  const a = await seedRun()
  await insertResource(a.runId, { owner: 'demo', kind: 'creature', ref: '0xA', status: 'confirmed' })
  assert.equal(await dup(() => insertResource(a.runId, { owner: 'other', kind: 'creature', ref: '0xA' })), null)
  assert.equal(await dup(() => insertResource(a.runId, { owner: 'demo', kind: 'item', ref: '0xA' })), null)
  assert.equal(await dup(() => insertResource(a.runId, { owner: 'demo', kind: 'creature', ref: '0xB' })), null)
})

test('a deleted step leaves its resources behind, and a deleted run does not', async (t) => {
  if (needDb(t)) return
  // TEAMS.md §2.5's correction, held as a test: `step_id` is SET NULL and it only
  // works because the generated column reads `status` alone. If `step_id` were in
  // that expression MariaDB would refuse the constraint outright (error 1901), and
  // the migration would fail on a fresh install rather than here.
  //
  // The two directions are different on purpose. A record of what was changed in
  // the WORLD must outlive the row that scheduled it — `engagement_sends`'
  // argument — while a deleted RUN takes its ledger with it, because the ledger
  // exists to answer questions about a run.
  const a = await seedRun()
  const stepId = await seedStep(a.runId)
  const id = await insertResource(a.runId, { stepId, status: 'confirmed' })

  await pool.query('DELETE FROM event_run_steps WHERE id = ?', [stepId])
  const orphaned = (await pool.query('SELECT step_id, status FROM event_run_resources WHERE id = ?', [id]))[0]
  assert.equal(orphaned.step_id, null)
  assert.equal(orphaned.status, 'confirmed')

  await pool.query('DELETE FROM event_runs WHERE id = ?', [a.runId])
  const gone = await pool.query('SELECT id FROM event_run_resources WHERE id = ?', [id])
  assert.equal(gone.length, 0)
})

test('the ledger model reserves, confirms, reverts and refuses for real', async (t) => {
  if (needDb(t)) return
  // Through the shipping module rather than a copy of its statements, like the
  // budget tests above: `reserve` reads ER_DUP_ENTRY as a refusal and looks the
  // holder up to name it, and both halves of that are the connector's behaviour
  // rather than this file's.
  const a = await seedRun()
  const b = await seedRun()

  const first = await resourcesDb.reserve({ runId: a.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
  assert.equal(first.ok, true)
  await resourcesDb.confirm(first.id)

  const second = await resourcesDb.reserve({ runId: b.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
  assert.equal(second.ok, false)
  assert.equal(second.code, 'held')
  assert.equal(Number(second.holder.run_id), Number(a.runId))
  assert.equal(second.holder.status, 'confirmed')

  // And once it is given back the second run gets it.
  await resourcesDb.markReverted(first.id)
  const third = await resourcesDb.reserve({ runId: b.runId, owner: 'demo', kind: 'override', ref: 'demo.rate' })
  assert.equal(third.ok, true)
})

test('failRevert increments and NEVER resets, and only a human clears it', async (t) => {
  if (needDb(t)) return
  // Engagement Phase 14's rule at the statement level: `revert_attempts =
  // revert_attempts + 1` is written in one place, and the reset is a separate
  // statement with an actor behind it. A `SET revert_attempts = ?` anywhere in
  // the sweep would make the ceiling unreachable and the row cycle for ever.
  const a = await seedRun()
  const id = await resourcesDb
    .reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
    .then((r) => r.id)
  await resourcesDb.confirm(id)

  await resourcesDb.failRevert(id, 'the shard did not answer')
  await resourcesDb.failRevert(id, 'still nothing')
  let row = (await pool.query('SELECT * FROM event_run_resources WHERE id = ?', [id]))[0]
  assert.equal(Number(row.revert_attempts), 2)
  assert.equal(row.status, 'confirmed')
  assert.equal(row.last_error, 'still nothing')

  assert.equal(await resourcesDb.resetAttempts(a.runId), 1)
  row = (await pool.query('SELECT * FROM event_run_resources WHERE id = ?', [id]))[0]
  assert.equal(Number(row.revert_attempts), 0)
})

test('claimRevert is a compare-and-set, and reverting is not re-claimable', async (t) => {
  if (needDb(t)) return
  // The cleanup leg and the manual cleanup route can both be working one run at
  // once. `reverting` is deliberately not claimable — a row another pass is
  // mid-revert on is left alone, exactly as a step with a live claim is.
  const a = await seedRun()
  const id = await resourcesDb
    .reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
    .then((r) => r.id)
  await resourcesDb.confirm(id)

  assert.equal(await resourcesDb.claimRevert(id), true)
  assert.equal(await resourcesDb.claimRevert(id), false, 'a second pass must not take a row mid-revert')

  await resourcesDb.markReverted(id)
  assert.equal(await resourcesDb.claimRevert(id), false, 'and a reverted row is finished')
})

test('the placeholder is resolvable exactly once, and only if it is a placeholder', async (t) => {
  if (needDb(t)) return
  // The `kind = '@step'` guard in the statement, not in JavaScript. It is what
  // stops a bug elsewhere resolving a real resource — which would be core marking
  // a live creature as given back without asking anyone.
  const a = await seedRun()
  const placeholder = await resourcesDb
    .reserve({ runId: a.runId, owner: 'demo', kind: resourcesDb.STEP_KIND, ref: 'k'.repeat(40) })
    .then((r) => r.id)
  const real = await resourcesDb
    .reserve({ runId: a.runId, owner: 'demo', kind: 'creature', ref: '0xA' })
    .then((r) => r.id)

  assert.equal(await resourcesDb.resolvePlaceholder(placeholder), true)
  assert.equal(await resourcesDb.resolvePlaceholder(placeholder), false)
  assert.equal(await resourcesDb.resolvePlaceholder(real), false)
  const stillThere = (await pool.query('SELECT status FROM event_run_resources WHERE id = ?', [real]))[0]
  assert.equal(stillThere.status, 'pending')
})

test('the unresolved reads see the five statuses that still want something', async (t) => {
  if (needDb(t)) return
  // `cleanup_status` is derived from this count, so what it includes IS the
  // definition of "clean". A `drifted` row left out of it would let a run whose
  // lease somebody else took call itself complete.
  const a = await seedRun()
  for (const status of ['pending', 'confirmed', 'reverting', 'reverted', 'orphaned', 'drifted']) {
    await insertResource(a.runId, { ref: `ref-${status}`, status })
  }
  assert.equal(await resourcesDb.unresolvedCount(a.runId), 5)
  const rows = await resourcesDb.unresolvedForRun(a.runId)
  assert.deepEqual(
    rows.map((r) => r.status).sort(),
    ['confirmed', 'drifted', 'orphaned', 'pending', 'reverting'],
  )
  const counts = await resourcesDb.unresolvedCounts([a.runId])
  assert.equal(counts.get(a.runId) ?? counts.get(String(a.runId)), 5)
})

test('payload comes back as an object rather than as a string', async (t) => {
  if (needDb(t)) return
  // A lease's baseline lives in here and the cleanup sweep reads it out to pass
  // to `restore`. The connector hands JSON back as text, so a missing hydration
  // is a `restore(undefined)` — a lease put back to nothing, silently.
  const a = await seedRun()
  const id = await resourcesDb
    .reserve({
      runId: a.runId,
      owner: 'demo',
      kind: 'override',
      ref: 'demo.rate',
      payload: { baseline: 1, applied: 3 },
    })
    .then((r) => r.id)
  void id
  const [row] = await resourcesDb.forRun(a.runId)
  assert.deepEqual(row.payload, { baseline: 1, applied: 3 })
})

test('the cleanup scan finds a run that owes something, whatever its status column says', async (t) => {
  if (needDb(t)) return
  await pool.query('ALTER TABLE event_runs ADD COLUMN IF NOT EXISTS cleanup_status ' +
    "ENUM('not_required','pending','complete','incomplete') NOT NULL DEFAULT 'not_required'")

  // **`not_required` is in the scan because of a live-walk defect**, not for
  // tidiness. A run whose only resource was a LEASE never went through the
  // ledger's `markRunDirty` — `core.lease` reserves its own row — so its column
  // stayed `not_required` and the lease was never given back at all. A terminal
  // run with an unresolved row has work to do whatever any summary column says.
  const lease = await seedRun({ status: 'completed' })
  await insertResource(lease.runId, { kind: 'override', ref: 'demo.rate', status: 'confirmed' })
  const marked = await seedRun({ status: 'cancelled' })
  await insertResource(marked.runId, { ref: '0xB', status: 'pending' })
  await pool.query("UPDATE event_runs SET cleanup_status = 'pending' WHERE id = ?", [marked.runId])

  // Three that must NOT be selected, one per reason.
  const running = await seedRun({ status: 'running' })
  await insertResource(running.runId, { ref: '0xC', status: 'confirmed' })
  const done = await seedRun({ status: 'completed' })
  await insertResource(done.runId, { ref: '0xD', status: 'reverted' })
  const givenUp = await seedRun({ status: 'completed' })
  await insertResource(givenUp.runId, { ref: '0xE', status: 'confirmed' })
  await pool.query("UPDATE event_runs SET cleanup_status = 'incomplete' WHERE id = ?", [givenUp.runId])

  const found = (await resourcesDb.runsNeedingCleanup(10, 3)).map((r) => Number(r.id))
  assert.deepEqual(found.sort(), [Number(lease.runId), Number(marked.runId)].sort())
})

test('the scan stops selecting a run whose attempts are spent', async (t) => {
  if (needDb(t)) return
  // Without the bound in the join, a run whose rows are all spent would be
  // selected, worked over and found to have nothing to do on every tick for the
  // rest of its life.
  const a = await seedRun({ status: 'completed' })
  const id = await insertResource(a.runId, { status: 'confirmed' })
  await pool.query("UPDATE event_runs SET cleanup_status = 'pending' WHERE id = ?", [a.runId])
  assert.equal((await resourcesDb.runsNeedingCleanup(10, 3)).length, 1)

  await pool.query('UPDATE event_run_resources SET revert_attempts = 3 WHERE id = ?', [id])
  assert.equal((await resourcesDb.runsNeedingCleanup(10, 3)).length, 0)
})

test('setCleanupStatus is guarded, which is what stops a late row re-opening a swept run', async (t) => {
  if (needDb(t)) return
  const a = await seedRun({ status: 'completed' })
  assert.equal(await runsDb.setCleanupStatus(a.runId, 'complete'), true)
  assert.equal(await runsDb.setCleanupStatus(a.runId, 'pending', ['not_required']), false)
  assert.equal(await runsDb.setCleanupStatus(a.runId, 'pending'), true)
})