server/db/schema.sql

@@ -1686,3 +1686,164 @@ CREATE TABLE IF NOT EXISTS notification_channel_prefs (
 -- has no digest mode to be asked into instead.
 INSERT IGNORE INTO notification_channel_prefs (user_id, stream_id, channel, mode)
   SELECT user_id, stream_id, 'push', 'instant' FROM notification_subscriptions;
+
+-- ── The engagement engine (ENGAGEMENT.md §4.1, §4.2a, §4.5 — Phase 4a) ──────
+--
+-- Five tables and no delivery. A rule says "when this trigger fires, for these
+-- people, on these channels, no more often than this"; the outbox is the queue
+-- the grace window needs; the cooldown table is what makes "once per house" mean
+-- once per house; and the send log is the first answer this deployment has ever
+-- had to "did user X get the mail?".
+--
+-- Nothing here sends anything. Core seeds no rules and `enabled` defaults to 0,
+-- so on a real deployment these five tables stay empty until an operator turns a
+-- rule on from the screen Phase 4b builds.
+
+-- What an operator actually configures: trigger -> audience -> template -> timing.
+--
+-- `trigger_id` deliberately has NO foreign key and no existence check: a trigger
+-- is DECLARED IN CODE (§4.3), so the set of them is whatever registered on this
+-- boot. A rule naming a trigger no module currently registers is DORMANT — it is
+-- listed, it never fires, and it starts working again when the module comes back
+-- (§7.3). Deleting it on uninstall would silently destroy an operator's
+-- configuration on the strength of a module being temporarily absent.
+CREATE TABLE IF NOT EXISTS engagement_rules (
+  id              INT AUTO_INCREMENT PRIMARY KEY,
+  trigger_id      VARCHAR(96)  NOT NULL,
+  name            VARCHAR(160) NOT NULL,
+  -- OFF by default (§7.1 Q3). A rule arrives inert and an operator turns it on,
+  -- so no import, seed or restore can start mailing on its own.
+  enabled         TINYINT(1)   NOT NULL DEFAULT 0,
+  audience        VARCHAR(32)  NOT NULL DEFAULT 'owner',
+  audience_segment_id INT      NULL,
+  -- §7.1 Q3: the hard stop that makes operator-editable rules safe to choose over
+  -- code-registered ones. Counted in engagement_sends, enforced before the outbox
+  -- row is written, never overridable from the rule editor beyond this column.
+  max_sends_per_hour INT       NOT NULL DEFAULT 100,
+  channels        JSON         NOT NULL,
+  template_keys   JSON         NOT NULL,
+  conditions      JSON         NULL,
+  cooldown_seconds INT         NOT NULL DEFAULT 0,
+  delay_seconds   INT          NOT NULL DEFAULT 0,
+  cancel_on       JSON         NULL,
+  updated_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_engr_user FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL,
+  INDEX idx_engr_trigger (trigger_id, enabled)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- §5.1a: an operator-composed segment over module-declared audiences. Stored as a
+-- boolean tree of audience ids + params; `ceiling` is DERIVED at save time as the
+-- NARROWEST ceiling in the tree (ceilings.meetAll) and re-checked against the
+-- trigger's own ceiling, so composition can never widen. It is a column rather
+-- than a runtime computation so an audit can read what a rule was allowed to
+-- reach without re-resolving it — and so a module that has since changed its
+-- audience's ceiling cannot retroactively widen a saved segment.
+--
+-- `engagement_rules.audience_segment_id` above points here with NO foreign key,
+-- on purpose and for the same reason `trigger_id` has none: a rule whose segment
+-- has been deleted must go DORMANT, not silently fall back to its plain
+-- `audience` column. ON DELETE SET NULL would be exactly that silent fallback,
+-- and the fallback reaches a DIFFERENT set of people (§5.1a rule 4).
+CREATE TABLE IF NOT EXISTS engagement_audience_segments (
+  id          INT AUTO_INCREMENT PRIMARY KEY,
+  name        VARCHAR(160) NOT NULL,
+  expression  JSON         NOT NULL,
+  ceiling     VARCHAR(32)  NOT NULL,
+  updated_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_engseg_user FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- §4.1. NOT `settings`: cooldown state is high-cardinality (recipients x rules x
+-- subjects), written on every fire, and asked "is this one pair still cooling?".
+-- A JSON blob under one settings key would be a read-modify-write of the whole
+-- deployment's cooldown state per event, with a lost-update race between two
+-- concurrent triggers.
+--
+-- `subject_key` is what makes "one IDOC mail per player per day" the right rule
+-- instead of the wrong one: a player with four houses decaying should hear about
+-- all four, once each. Cooling per (rule, user) alone silently drops three.
+CREATE TABLE IF NOT EXISTS engagement_cooldowns (
+  rule_id      INT          NOT NULL,
+  user_id      INT          NOT NULL,
+  -- The SUBJECT the cooldown is about, opaque to core: a house serial, a vendor
+  -- id, ''. NOT NULL with a '' default, because this is a PRIMARY KEY column and
+  -- MariaDB would coerce a NULL one anyway. '' is "this rule cools per user, not
+  -- per subject".
+  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
+  last_fired_at DATETIME    NOT NULL,
+  fire_count   INT          NOT NULL DEFAULT 1,
+  PRIMARY KEY (rule_id, user_id, subject_key),
+  CONSTRAINT fk_engc_rule FOREIGN KEY (rule_id) REFERENCES engagement_rules(id) ON DELETE CASCADE,
+  CONSTRAINT fk_engc_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
+  -- So a prune worker can drop rows older than the longest configured cooldown.
+  -- Without it this table grows without bound, which is the failure mode
+  -- teamActivityPrune was written for.
+  INDEX idx_engc_sweep (last_fired_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- §4.2a. Modelled on announce_jobs / announce_job_legs. One row per
+-- (rule, user, channel) occurrence of an event.
+CREATE TABLE IF NOT EXISTS engagement_outbox (
+  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
+  rule_id      INT          NOT NULL,
+  trigger_id   VARCHAR(96)  NOT NULL,       -- denormalized; survives a rule edit
+  user_id      INT          NOT NULL,
+  channel      VARCHAR(32)  NOT NULL,       -- VARCHAR, never ENUM: the channel set is data
+  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
+  payload      JSON         NOT NULL,       -- the declared variables, snapshotted at emit
+  dedupe_key   VARCHAR(190) NULL,
+  status       ENUM('scheduled','sending','sent','failed','cancelled','suppressed') NOT NULL DEFAULT 'scheduled',
+  due_at       DATETIME     NOT NULL,
+  attempts     SMALLINT     NOT NULL DEFAULT 0,
+  last_error   TEXT         NULL,
+  created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  sent_at      DATETIME     NULL,
+  CONSTRAINT fk_engo_rule FOREIGN KEY (rule_id) REFERENCES engagement_rules(id) ON DELETE CASCADE,
+  CONSTRAINT fk_engo_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
+  -- **Scoped to the row's identity, and §4.2a's global `UNIQUE (dedupe_key)` is
+  -- a defect this phase found while building it.** A dedupe key names the EVENT
+  -- ("house 0x4001 entered IDOC"), and one event legitimately becomes many rows:
+  -- an audience of fifty users is fifty rows, a rule spanning email and in-app
+  -- doubles that, and two rules on one trigger double it again. Under a global
+  -- unique index the FIRST of those inserts wins and every other one is silently
+  -- ignored — ninety-nine recipients dropped by the mechanism meant to stop a
+  -- replayed event becoming a second mail. Scoping it to (rule, user, channel)
+  -- keeps exactly that guarantee and nothing more.
+  UNIQUE KEY uq_engo_dedupe (rule_id, user_id, channel, dedupe_key),
+  INDEX idx_engo_due (status, due_at),
+  -- What a RESOLVING event queries: a house repaired back to LikeNew cancels
+  -- every scheduled row for that (rule, user, house).
+  INDEX idx_engo_cancel (rule_id, user_id, subject_key, status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- G15: the per-message record. Today "did user X get the mail?" is unanswerable.
+--
+-- It is deliberately NOT a second address book: the address is stored as a
+-- sha256, which is enough to correlate a bounce (Phase 9) and useless as a
+-- mailing list. `user_id` is SET NULL rather than CASCADE so the log survives an
+-- account deletion — an audit of what this deployment sent must not be erasable
+-- by deleting the recipient.
+CREATE TABLE IF NOT EXISTS engagement_sends (
+  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
+  outbox_id    BIGINT       NULL,
+  rule_id      INT          NULL,
+  trigger_id   VARCHAR(96)  NOT NULL,
+  user_id      INT          NULL,
+  channel      VARCHAR(32)  NOT NULL,
+  transport    VARCHAR(32)  NULL,
+  address_hash CHAR(64)     NULL,
+  status       ENUM('sent','failed','suppressed','bounced','complained') NOT NULL,
+  detail       VARCHAR(500) NULL,
+  created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CONSTRAINT fk_engs_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL,
+  INDEX idx_engs_trigger (trigger_id, created_at),
+  INDEX idx_engs_user (user_id, created_at),
+  -- The per-rule hourly ceiling (§7.1 Q3) is counted here, so the count has to be
+  -- an index range scan rather than a table scan: it runs once per rule per event.
+  INDEX idx_engs_rule_window (rule_id, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

server/src/engagement/audiences.js

@@ -0,0 +1,138 @@
+// ── Resolving a rule's audience to recipients ──────────────────────────────
+//
+// ENGAGEMENT.md §5.1a / §4.5, Phase 4a. A rule names an audience two ways and
+// only ever one at a time: a **plain ceiling name** (`owner`, `staff`,
+// `subscribers`, `authenticated`, `everyone`) resolved from core's own tables, or
+// an **`audience_segment_id`** pointing at an operator-composed tree of
+// module-declared audiences (segments.js). This file turns either into user ids.
+//
+// **Three things it is careful about, all of them the same worry.** The set this
+// function returns is the set that gets mailed, so:
+//
+// 1. Every id is checked against `users.status = 'active'` - including the ones a
+//    MODULE's resolver produced, which core has no reason to trust with account
+//    status it does not know about.
+// 2. A dormant segment (its module uninstalled) resolves to EMPTY and says so.
+//    The caller must not send. Falling back to the rule's plain `audience`
+//    column would reach a different population than the one composed (§5.1a
+//    rule 4), which is the failure mode this whole design exists to avoid.
+// 3. `members` as a PLAIN audience resolves to nobody. It is the ceiling for
+//    "a module-declared list", and without a segment there is no list - core
+//    knows no game vocabulary and cannot guess which members were meant. A rule
+//    saved that way is inert and visible as such, rather than quietly falling
+//    back to something wider.
+
+const registries = require('../modules/registries')
+const channels = require('./channels')
+const segments = require('./segments')
+const segmentsDb = require('../model/engagement/engagementSegments.db')
+const recipients = require('../model/engagement/engagementRecipients.db')
+const ceilings = require('../modules/ceilings')
+const log = require('../utils/logger')('engagement')
+
+/**
+ * Which registered channels default to something other than 'off'?
+ *
+ * Read once per resolution rather than hardcoded, because it is the difference
+ * between "opted in" meaning a stored row and meaning the absence of one
+ * (§3.1, G9). All three of core's channels default 'off' today, so this is empty
+ * and `subscribers` is the simple query - but the answer lives in the registry.
+ */
+const defaultOnChannels = () => channels.all().filter((c) => c.defaultMode !== 'off').map((c) => c.id)
+
+/**
+ * Resolve one rule against one event.
+ *
+ * @returns {{ userIds: number[], ceiling: string|null, dormant: boolean, reason: string|null }}
+ *   `dormant` means "this rule cannot be resolved right now"; `reason` names why
+ *   for the log and, in Phase 4b, for the admin list's dormant badge.
+ */
+async function resolveForRule(rule, event) {
+  if (rule.audience_segment_id) {
+    const segment = await segmentsDb.getById(rule.audience_segment_id)
+    if (!segment) {
+      // The segment was deleted out from under the rule. `audience_segment_id`
+      // deliberately has no ON DELETE SET NULL (see schema.sql), because that
+      // would silently fall back to the rule's plain `audience` column and mail
+      // a different set of people.
+      return { userIds: [], ceiling: null, dormant: true, reason: 'audience segment no longer exists' }
+    }
+    const { dormant, userIds } = await segments.resolve(segment.expression)
+    if (dormant) {
+      return { userIds: [], ceiling: segment.ceiling, dormant: true, reason: 'audience segment is dormant' }
+    }
+    return {
+      userIds: await recipients.filterActive(userIds),
+      // The STORED ceiling, not one re-derived now: a module that has since
+      // widened its own audience's ceiling must not widen a segment that was
+      // saved under the old one.
+      ceiling: segment.ceiling,
+      dormant: false,
+      reason: null,
+    }
+  }
+
+  switch (rule.audience) {
+    case 'owner': {
+      if (!event.ownerUserId) {
+        // Not dormant: the rule is fine and this particular event simply has no
+        // owner to mail. A trigger that never carries one is an operator's
+        // mistake the rule editor should catch (Phase 4b), not a runtime error.
+        return { userIds: [], ceiling: 'owner', dormant: false, reason: 'event carries no ownerUserId' }
+      }
+      return {
+        userIds: await recipients.filterActive([event.ownerUserId]),
+        ceiling: 'owner',
+        dormant: false,
+        reason: null,
+      }
+    }
+    case 'staff':
+      return {
+        userIds: await recipients.staff(ceilings.STAFF_CEILING_ROLES),
+        ceiling: 'staff',
+        dormant: false,
+        reason: null,
+      }
+    case 'subscribers':
+      return {
+        userIds: await recipients.subscribers(event.triggerId, defaultOnChannels()),
+        ceiling: 'subscribers',
+        dormant: false,
+        reason: null,
+      }
+    case 'authenticated':
+    case 'everyone':
+      return { userIds: await recipients.active(), ceiling: rule.audience, dormant: false, reason: null }
+    case 'members':
+      return {
+        userIds: [],
+        ceiling: 'members',
+        dormant: false,
+        reason: 'a "members" audience needs a segment naming which list',
+      }
+    default:
+      // Fails closed on an audience name the lattice does not know - the same
+      // posture `ceilings.permits` takes, and for the same reason.
+      log.warn('rule names an unknown audience', { rule: rule.id, audience: rule.audience })
+      return { userIds: [], ceiling: null, dormant: true, reason: `unknown audience "${rule.audience}"` }
+  }
+}
+
+/**
+ * The G24 gate, re-run at SEND time and not only at save time.
+ *
+ * A rule's audience was checked against its trigger's ceiling when it was saved,
+ * so this can only fail when something changed underneath: a module upgraded and
+ * narrowed its trigger's ceiling, or a module was replaced by one declaring the
+ * same id more tightly. That is precisely the case where a stale rule would
+ * otherwise mail a population the current declaration forbids, which is what
+ * makes this the security boundary rather than a duplicate check.
+ */
+function permitted(triggerId, ceiling) {
+  const declaration = registries.eventTrigger(triggerId)
+  if (!declaration) return false
+  return ceilings.permits(declaration.ceiling, ceiling)
+}
+
+module.exports = { resolveForRule, permitted, defaultOnChannels }

server/src/engagement/conditions.js

@@ -0,0 +1,251 @@
+// ── Rule conditions — a predicate over a trigger's DECLARED variables ───────
+//
+// ENGAGEMENT.md §4.5, Phase 4a. `engagement_rules.conditions` is the half of a
+// rule that decides *whether* this particular firing is interesting: "only when
+// decayStatus is IDOC", "only for threads in this Team". Without it every rule is
+// all-or-nothing per trigger, and an operator's only way to narrow is to ask a
+// module author for a second trigger.
+//
+// **It is validated against the declaration, not against a payload.** A condition
+// naming a variable the trigger does not declare is refused at SAVE, with the
+// variable named, for the same reason §4.3 gives the template editor: a predicate
+// that silently reads `undefined` is a rule that silently never fires (or always
+// does), and the day you find out is the day the mail did not go.
+//
+// **The grammar is small and closed on purpose.** No arbitrary expressions, no
+// arithmetic, no regex. An operator composes and/or/not over comparisons of one
+// declared variable against a literal, and every operator here is one a rule
+// editor can render as a dropdown. Anything that needs more than this is asking
+// for a condition the module should have declared as a variable.
+//
+// Nothing in this file reaches the database or the network.
+
+const registries = require('../modules/registries')
+
+// Comparison operators, grouped by what they may be applied to. The grouping is
+// the whole of the type check: `gt` on a boolean and `startsWith` on an int are
+// both refused at save rather than quietly answering false forever.
+const OPERATORS = {
+  eq: { label: 'is', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 1 },
+  ne: { label: 'is not', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 1 },
+  in: { label: 'is one of', types: ['string', 'int', 'float', 'url'], arity: 'list' },
+  nin: { label: 'is none of', types: ['string', 'int', 'float', 'url'], arity: 'list' },
+  gt: { label: 'is greater than', types: ['int', 'float', 'datetime'], arity: 1 },
+  gte: { label: 'is at least', types: ['int', 'float', 'datetime'], arity: 1 },
+  lt: { label: 'is less than', types: ['int', 'float', 'datetime'], arity: 1 },
+  lte: { label: 'is at most', types: ['int', 'float', 'datetime'], arity: 1 },
+  contains: { label: 'contains', types: ['string', 'url'], arity: 1 },
+  startsWith: { label: 'starts with', types: ['string', 'url'], arity: 1 },
+  // The one operator that takes no value: "the emit carried this variable at
+  // all". It is the honest way to write a rule about an OPTIONAL variable, and
+  // without it `ne` would have to double as a presence test and get it wrong
+  // (an absent variable is not "not equal to X"; it is absent).
+  present: { label: 'is present', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 0 },
+  absent: { label: 'is absent', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 0 },
+}
+
+const BOOLEAN_OPS = ['and', 'or', 'not']
+
+// A list literal an operator may type. Bounded because it is stored in a JSON
+// column an admin can write, and an unbounded IN list is an unbounded predicate
+// evaluated on every event.
+const MAX_LIST = 50
+// Depth of the and/or/not tree. Three levels is more nesting than any rule
+// editor should offer; the bound is here so a hand-written JSON body cannot
+// recurse this evaluator into a stack overflow on the emit path.
+const MAX_DEPTH = 5
+
+const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v)
+
+/**
+ * Check one literal against the declared type of the variable it is compared to.
+ *
+ * `datetime` accepts anything `Date` parses and is normalised to an ISO string,
+ * which is what `engagementEmit.coerce` does to the payload side — so both sides
+ * of every comparison are the same representation of a moment, and a lexical
+ * `<` on two ISO strings is a chronological one.
+ */
+function checkLiteral(type, raw) {
+  switch (type) {
+    case 'string':
+    case 'url':
+      return typeof raw === 'string' ? { value: raw } : { error: 'expected a string' }
+    case 'int':
+      return Number.isInteger(raw) ? { value: raw } : { error: 'expected an integer' }
+    case 'float':
+      return typeof raw === 'number' && Number.isFinite(raw)
+        ? { value: raw }
+        : { error: 'expected a finite number' }
+    case 'boolean':
+      return typeof raw === 'boolean' ? { value: raw } : { error: 'expected a boolean' }
+    case 'datetime': {
+      const d = raw instanceof Date ? raw : new Date(raw)
+      if (Number.isNaN(d.getTime())) return { error: 'expected a date' }
+      return { value: d.toISOString() }
+    }
+    default:
+      return { error: `unsupported type "${type}"` }
+  }
+}
+
+/**
+ * Validate a condition tree against a trigger declaration.
+ *
+ * Returns `{ ok: true, conditions }` with a NEW normalised tree — literals
+ * coerced, unknown keys dropped — or `{ ok: false, errors }` listing every
+ * problem rather than the first, the posture `validatePayload` takes and for the
+ * same reason: an operator fixing one clause at a time is an operator making six
+ * round trips through a form.
+ *
+ * `null` and `undefined` are valid and mean "no conditions" — a rule that fires
+ * on every occurrence of its trigger, which is the common case.
+ */
+function validate(declaration, raw) {
+  const errors = []
+  const variables = new Map((declaration?.variables || []).map((v) => [v.name, v]))
+
+  function walk(node, depth, path) {
+    if (depth > MAX_DEPTH) {
+      errors.push(`${path}: nested deeper than ${MAX_DEPTH}`)
+      return null
+    }
+    if (!isPlainObject(node)) {
+      errors.push(`${path}: expected an object`)
+      return null
+    }
+
+    if (BOOLEAN_OPS.includes(node.op)) {
+      // `not` takes exactly one node; `and`/`or` take a list. Both are written
+      // as `nodes` so a client walks one shape.
+      const raws = Array.isArray(node.nodes) ? node.nodes : []
+      if (!raws.length) {
+        errors.push(`${path}: "${node.op}" has no nodes`)
+        return null
+      }
+      if (node.op === 'not' && raws.length !== 1) {
+        errors.push(`${path}: "not" takes exactly one node`)
+        return null
+      }
+      const nodes = raws.map((child, i) => walk(child, depth + 1, `${path}.nodes[${i}]`)).filter(Boolean)
+      return nodes.length === raws.length ? { op: node.op, nodes } : null
+    }
+
+    if (node.op !== undefined) {
+      errors.push(`${path}: unknown operator "${node.op}"`)
+      return null
+    }
+
+    // A leaf: { variable, cmp, value }.
+    const variable = variables.get(node.variable)
+    if (!variable) {
+      errors.push(`${path}: "${node.variable}" is not a variable of "${declaration?.id}"`)
+      return null
+    }
+    const operator = OPERATORS[node.cmp]
+    if (!operator) {
+      errors.push(`${path}: unknown comparison "${node.cmp}"`)
+      return null
+    }
+    if (!operator.types.includes(variable.type)) {
+      errors.push(`${path}: "${node.cmp}" cannot be applied to a ${variable.type}`)
+      return null
+    }
+
+    if (operator.arity === 0) return { variable: variable.name, cmp: node.cmp }
+
+    if (operator.arity === 'list') {
+      if (!Array.isArray(node.value) || !node.value.length) {
+        errors.push(`${path}: "${node.cmp}" needs a non-empty list`)
+        return null
+      }
+      if (node.value.length > MAX_LIST) {
+        errors.push(`${path}: "${node.cmp}" list is longer than ${MAX_LIST}`)
+        return null
+      }
+      const value = []
+      let bad = false
+      node.value.forEach((item, i) => {
+        const checked = checkLiteral(variable.type, item)
+        if (checked.error) {
+          errors.push(`${path}.value[${i}]: ${checked.error}`)
+          bad = true
+        } else value.push(checked.value)
+      })
+      return bad ? null : { variable: variable.name, cmp: node.cmp, value }
+    }
+
+    const checked = checkLiteral(variable.type, node.value)
+    if (checked.error) {
+      errors.push(`${path}: ${checked.error}`)
+      return null
+    }
+    return { variable: variable.name, cmp: node.cmp, value: checked.value }
+  }
+
+  if (raw === null || raw === undefined) return { ok: true, conditions: null }
+  const conditions = walk(raw, 0, 'conditions')
+  return errors.length ? { ok: false, errors } : { ok: true, conditions }
+}
+
+/** Compare one already-normalised leaf against a payload. */
+function evaluateLeaf(leaf, data) {
+  const present = Object.prototype.hasOwnProperty.call(data, leaf.variable)
+  const actual = data[leaf.variable]
+
+  if (leaf.cmp === 'present') return present
+  if (leaf.cmp === 'absent') return !present
+  // Every other comparison against an absent variable is FALSE, never true.
+  // `ne` is the one that tempts otherwise — "not equal to X" reads as satisfied
+  // by nothing at all — and treating it as true would make an optional variable's
+  // absence fire the rule.
+  if (!present) return false
+
+  switch (leaf.cmp) {
+    case 'eq': return actual === leaf.value
+    case 'ne': return actual !== leaf.value
+    case 'in': return leaf.value.includes(actual)
+    case 'nin': return !leaf.value.includes(actual)
+    case 'gt': return actual > leaf.value
+    case 'gte': return actual >= leaf.value
+    case 'lt': return actual < leaf.value
+    case 'lte': return actual <= leaf.value
+    case 'contains': return typeof actual === 'string' && actual.includes(leaf.value)
+    case 'startsWith': return typeof actual === 'string' && actual.startsWith(leaf.value)
+    default: return false
+  }
+}
+
+/**
+ * Does this event's payload satisfy the rule's conditions?
+ *
+ * `null` conditions are satisfied — a rule with no conditions fires on every
+ * occurrence. A tree this evaluator does not recognise answers **false**, which
+ * is the fail-closed direction: a stored condition that no longer parses (a rule
+ * saved against an older trigger version, say) must stop the mail rather than
+ * become "no conditions" and mail everyone.
+ */
+function evaluate(conditions, data = {}) {
+  if (conditions === null || conditions === undefined) return true
+  if (!isPlainObject(conditions)) return false
+
+  if (conditions.op === 'and') return (conditions.nodes || []).every((n) => evaluate(n, data))
+  if (conditions.op === 'or') return (conditions.nodes || []).some((n) => evaluate(n, data))
+  if (conditions.op === 'not') return !evaluate((conditions.nodes || [])[0], data)
+  if (conditions.op !== undefined) return false
+
+  return evaluateLeaf(conditions, data)
+}
+
+/**
+ * The operator vocabulary a rule editor renders, with the variable types each
+ * one applies to. Served with the rule surface in Phase 4b rather than hardcoded
+ * in the client, on the same argument the ceiling vocabulary is served with the
+ * trigger catalog: a second copy of a rule is a copy that drifts.
+ */
+const vocabulary = () =>
+  Object.entries(OPERATORS).map(([cmp, o]) => ({ cmp, label: o.label, types: o.types, arity: o.arity }))
+
+/** Convenience for a caller holding only a trigger id. */
+const validateFor = (triggerId, raw) => validate(registries.eventTrigger(triggerId), raw)
+
+module.exports = { validate, validateFor, evaluate, vocabulary, OPERATORS, MAX_LIST, MAX_DEPTH }

server/src/engagement/engine.js

@@ -0,0 +1,241 @@
+// ── The engagement engine ──────────────────────────────────────────────────
+//
+// ENGAGEMENT.md Phase 4a. `ctx.events.emit` validated a payload against a
+// declaration and stopped (Phase 2); this is what it now hands the validated
+// event to. The engine's whole job is to answer, for one event, **who gets told,
+// on what, and not too often** - and then to write that down as outbox rows.
+// It never delivers: `engagementWorker` drains the outbox, and what actually
+// carries a message arrives with the channels' `deliver` in Phases 6 and 7.
+//
+// **The order of the gates is the design, and each one is here because skipping
+// it is a way to mail the wrong people or too many of them:**
+//
+//   1. enabled rules for this trigger      - nothing is seeded, nothing is on by default
+//   2. conditions                          - is this particular firing interesting
+//   3. audience -> user ids                - core's tables, or a composed segment
+//   4. ceiling re-check (G24)              - re-run at SEND time, not only at save
+//   5. per-channel preference              - a user's own opt-in, effective mode
+//   6. per-rule hourly ceiling (§7.1 Q3)   - the hard stop that makes rules-as-data safe
+//   7. cooldown, per (rule, user, subject) - one statement, so two emits cannot race
+//   8. enqueue, deduped                    - a replayed event is one row, not two
+//
+// Steps 6 and 7 are in that order deliberately. The hourly ceiling is about the
+// RULE and is the thing that stops a mail storm; the cooldown is about one
+// recipient and one subject. Checking the cheap global bound before consuming a
+// per-recipient cooldown slot means a rule that has hit its ceiling does not also
+// silently burn everybody's cooldowns on sends that never happen.
+//
+// **Nothing here throws at its caller.** It is invoked from inside a game-event
+// handler by way of `ctx.events.emit`, and a database problem of core's must not
+// become a module's control flow (the same posture the emit validator takes).
+
+const rulesDb = require('../model/engagement/engagementRules.db')
+const outboxDb = require('../model/engagement/engagementOutbox.db')
+const cooldownsDb = require('../model/engagement/engagementCooldowns.db')
+const sendsDb = require('../model/engagement/engagementSends.db')
+const recipients = require('../model/engagement/engagementRecipients.db')
+const conditions = require('./conditions')
+const audiences = require('./audiences')
+const channels = require('./channels')
+const log = require('../utils/logger')('engagement')
+
+const HOUR_MS = 60 * 60 * 1000
+
+/**
+ * Which of a rule's channels are actually deliverable right now?
+ *
+ * A rule stores channel ids as data (`channels JSON`), so it can name one whose
+ * module has been removed since. An unregistered channel is dropped rather than
+ * failing the rule: the other channels of that rule are still correct, and a
+ * dropped one is visible in the log line below.
+ */
+const liveChannels = (rule) => (rule.channels || []).filter((c) => channels.has(c))
+
+/**
+ * Narrow a candidate set to the users whose EFFECTIVE mode for (id, channel) is
+ * not 'off'.
+ *
+ * Effective, not stored: a row exists only where a user has expressed something,
+ * and absence means the channel's `defaultMode` (§3.1). Reading the stored rows
+ * and applying the default here keeps that answer in the registry, which is the
+ * invariant Phase 3 established.
+ *
+ * A 'digest' preference is kept, not dropped. Digest delivery is Phase 6's, and
+ * an outbox row for it is still the right record of "this person should be told";
+ * what changes in Phase 6 is who drains it.
+ */
+async function subscribedTo(userIds, streamId, channel) {
+  if (!userIds.length) return []
+  const stored = await recipients.storedModes(userIds, streamId, channel)
+  const fallback = channels.defaultMode(channel)
+  return userIds.filter((id) => (stored.get(id) ?? fallback) !== 'off')
+}
+
+/**
+ * Run one rule against one event. Returns a small summary, for the log line and
+ * for tests; it is not read by the caller for control flow.
+ */
+async function applyRule(rule, event, now) {
+  const summary = { ruleId: rule.id, enqueued: 0, deduped: 0, cooled: 0, capped: 0, skipped: null }
+
+  if (!conditions.evaluate(rule.conditions, event.data)) {
+    summary.skipped = 'conditions'
+    return summary
+  }
+
+  const resolved = await audiences.resolveForRule(rule, event)
+  if (resolved.dormant) {
+    summary.skipped = resolved.reason || 'dormant'
+    return summary
+  }
+  if (!resolved.userIds.length) {
+    summary.skipped = resolved.reason || 'empty audience'
+    return summary
+  }
+
+  // G24, re-run at send time. A rule saved when its trigger permitted a wider
+  // audience must not keep reaching it after a module upgrade narrowed the
+  // declaration - and that is the only way this can fail, since the save path
+  // ran the same check.
+  if (!audiences.permitted(event.triggerId, resolved.ceiling)) {
+    log.warn('rule audience exceeds its trigger ceiling - refusing', {
+      rule: rule.id,
+      trigger: event.triggerId,
+      audience: resolved.ceiling,
+    })
+    summary.skipped = 'ceiling'
+    return summary
+  }
+
+  const live = liveChannels(rule)
+  if (!live.length) {
+    summary.skipped = 'no registered channel'
+    return summary
+  }
+
+  // The per-rule hourly ceiling (§7.1 Q3). Counted once for the whole event
+  // rather than per channel: an operator setting "100 an hour" means a hundred
+  // messages, not a hundred per channel per event.
+  const sentThisHour = await sendsDb.countSentSince(rule.id, new Date(now.getTime() - HOUR_MS))
+  let budget = Math.max(0, rule.max_sends_per_hour - sentThisHour)
+  if (budget === 0) {
+    log.warn('rule is at its hourly send ceiling', {
+      rule: rule.id,
+      trigger: event.triggerId,
+      ceiling: rule.max_sends_per_hour,
+    })
+    summary.skipped = 'hourly ceiling'
+    return summary
+  }
+
+  const subjectKey = (event.subject ?? '').toString().slice(0, 190)
+  const dueAt = new Date(now.getTime() + Math.max(0, rule.delay_seconds) * 1000)
+
+  for (const channel of live) {
+    const eligible = await subscribedTo(resolved.userIds, event.triggerId, channel)
+    for (const userId of eligible) {
+      if (budget <= 0) {
+        summary.capped += 1
+        continue
+      }
+      // One statement, guarded on the interval, so two concurrent emits cannot
+      // both pass a read-then-write check (§4.1).
+      const allowed = await cooldownsDb.claim(rule.id, userId, subjectKey, rule.cooldown_seconds, now)
+      if (!allowed) {
+        summary.cooled += 1
+        continue
+      }
+      const id = await outboxDb.enqueue({
+        rule_id: rule.id,
+        trigger_id: event.triggerId,
+        user_id: userId,
+        channel,
+        subject_key: subjectKey,
+        payload: event.data,
+        // Scoped per (rule, user, channel) by the unique index, so one event
+        // fanned out to fifty people is fifty rows carrying the same key.
+        dedupe_key: event.dedupeKey,
+        due_at: dueAt,
+      })
+      if (id === null) summary.deduped += 1
+      else {
+        summary.enqueued += 1
+        budget -= 1
+      }
+    }
+  }
+
+  return summary
+}
+
+/**
+ * Cancel pending rows that this event resolves (§4.2a).
+ *
+ * This is the actual point of `delay_seconds`: without cancellation a delay is
+ * just a late mail. A house repaired back to LikeNew fires a trigger that some
+ * rule names in its `cancel_on`, and every still-scheduled row for that
+ * (rule, subject) stops.
+ *
+ * When the resolving event names an owner, only that user's rows are cancelled;
+ * when it does not, every user queued about that subject is - which is the
+ * house-repaired case, where the event is about the house and not about any one
+ * of the people who were going to be told.
+ */
+async function applyCancellations(event, summary) {
+  const rules = await rulesDb.enabledCancelledBy(event.triggerId)
+  if (!rules.length) return
+  const subjectKey = (event.subject ?? '').toString().slice(0, 190)
+  for (const rule of rules) {
+    const n = await outboxDb.cancel(rule.id, subjectKey, event.ownerUserId || null)
+    if (n) {
+      summary.cancelled += n
+      log.info('cancelled scheduled sends', {
+        rule: rule.id,
+        by: event.triggerId,
+        subject: subjectKey,
+        rows: n,
+      })
+    }
+  }
+}
+
+/**
+ * Dispatch one validated event. Called by `engagementEmit.emit` after the payload
+ * has been checked against the declaration.
+ *
+ * @param {object} event the envelope `engagementEmit` built
+ * @returns {Promise<{ rules: number, enqueued: number, cancelled: number }>}
+ */
+async function dispatch(event, now = new Date()) {
+  const summary = { rules: 0, enqueued: 0, deduped: 0, cooled: 0, capped: 0, cancelled: 0 }
+  try {
+    const rules = await rulesDb.enabledForTrigger(event.triggerId)
+    summary.rules = rules.length
+
+    for (const rule of rules) {
+      const result = await applyRule(rule, event, now)
+      summary.enqueued += result.enqueued
+      summary.deduped += result.deduped
+      summary.cooled += result.cooled
+      summary.capped += result.capped
+    }
+
+    await applyCancellations(event, summary)
+
+    // Keys and counts, never values - the same rule the emit log line follows.
+    // A payload carries player names, house locations and forum excerpts, and a
+    // log that reproduces them is a second copy of exactly the content
+    // `engagement_sends` is careful to keep out of the database.
+    if (summary.rules || summary.cancelled) {
+      log.info('event dispatched', { trigger: event.triggerId, ...summary })
+    }
+  } catch (err) {
+    // A database problem of core's must not become the module's control flow at
+    // three in the morning. The emit already succeeded as a contract; what failed
+    // is delivery, and it is logged as core's failure.
+    log.error('dispatch failed', { trigger: event.triggerId, message: err.message })
+  }
+  return summary
+}
+
+module.exports = { dispatch, applyRule, applyCancellations, subscribedTo, liveChannels, HOUR_MS }

server/src/engagement/segments.js

@@ -0,0 +1,232 @@
+// ── Audience segments — operator composition over module-declared audiences ──
+//
+// ENGAGEMENT.md §5.1a, Phase 4a. A module declares named audiences over its own
+// data ("members of a Team", "the governors"); an operator combines them with
+// and/or/not into a saved segment; a rule points at the segment. This file is the
+// two halves of that: derive the segment's ceiling at save time, and resolve the
+// expression to user ids at send time.
+//
+// **Composition must NARROW, never widen** (§5.1a rule 3), and that is the whole
+// security content of this file. `A OR B` takes the TIGHTER of the two ceilings,
+// not the looser - a ceiling states what an expression is *allowed* to reach, not
+// what it will resolve to, so the direction of the boolean operator is
+// irrelevant. Union-widens is the intuitive implementation and it is the wrong
+// one; `ceilings.meetAll` is the arithmetic, settled in Phase 2, and this is its
+// first consumer.
+//
+// The second rule that shows up in both halves is **dormancy** (§5.1a rule 4).
+// An audience whose module has been uninstalled resolves to the EMPTY set and
+// flags itself, never to an error and never to some other set of people. A
+// segment containing one is dormant, and a rule using a dormant segment does not
+// send. Resolving the rest of the tree instead would mail a DIFFERENT population
+// than the one the operator composed.
+
+const registries = require('../modules/registries')
+const ceilings = require('../modules/ceilings')
+
+const BOOLEAN_OPS = ['and', 'or', 'not']
+// Same bounds and the same reason as conditions.js: this tree comes out of a JSON
+// column an admin can write, and it is walked on the emit path.
+const MAX_DEPTH = 5
+const MAX_NODES = 50
+
+const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v)
+const isNot = (node) => isPlainObject(node) && node.op === 'not'
+
+/** Check one audience's declared params against what the operator supplied. */
+function checkParams(declaration, raw, path, errors) {
+  const params = {}
+  const supplied = isPlainObject(raw) ? raw : {}
+  for (const p of declaration.params || []) {
+    const value = supplied[p.id]
+    if (value === undefined || value === null || value === '') {
+      if (p.required) errors.push(`${path}: "${p.id}" is required`)
+      continue
+    }
+    if (p.type === 'int') {
+      const n = Number(value)
+      if (!Number.isInteger(n)) {
+        errors.push(`${path}: "${p.id}" expected an integer`)
+        continue
+      }
+      params[p.id] = n
+    } else if (p.type === 'boolean') {
+      if (typeof value !== 'boolean') {
+        errors.push(`${path}: "${p.id}" expected a boolean`)
+        continue
+      }
+      params[p.id] = value
+    } else {
+      if (typeof value !== 'string') {
+        errors.push(`${path}: "${p.id}" expected a string`)
+        continue
+      }
+      params[p.id] = value
+    }
+  }
+  return params
+}
+
+/**
+ * Validate an expression and derive its ceiling in one walk.
+ *
+ * Returns `{ ok: true, expression, ceiling }` with a normalised tree, or
+ * `{ ok: false, errors }`.
+ *
+ * **`not` is legal only as a child of `and`**, and that restriction is what makes
+ * a complement mean something. A complement needs a universe, and the only
+ * universe available here that does not widen is the set its siblings already
+ * produced: `A AND NOT B` is "A, less B", which is exactly what an operator
+ * wants and cannot be composed into a broadcast. A bare `NOT B`, or `A OR NOT B`,
+ * would have to mean "everyone except..." - a way to build the whole deployment
+ * out of one narrow audience, which is the widening rule 3 forbids. Refusing it
+ * at save is better than a semantics nobody can predict from the screen.
+ *
+ * Two failure modes, and they are different:
+ *
+ * - a leaf naming an audience nobody registers is refused AT SAVE, because an
+ *   operator composing a segment out of a typo should hear about it now rather
+ *   than discovering a permanently-empty rule later. (A segment that was VALID
+ *   when saved and whose module has since gone is a different case - that is
+ *   dormancy, handled in `resolve`, and it is not refused.)
+ * - two incomparable ceilings have NO meet, so the composition is refused rather
+ *   than resolved to a guess. `staff AND owner` is not `owner`; it is a question
+ *   the lattice declines to answer, and picking a side would be a widening.
+ */
+function validate(raw) {
+  const errors = []
+  let nodes = 0
+
+  // `underAnd` is the only context in which a `not` is legal.
+  function walk(node, depth, path, underAnd) {
+    if (++nodes > MAX_NODES) {
+      errors.push(`${path}: expression has more than ${MAX_NODES} nodes`)
+      return null
+    }
+    if (depth > MAX_DEPTH) {
+      errors.push(`${path}: nested deeper than ${MAX_DEPTH}`)
+      return null
+    }
+    if (!isPlainObject(node)) {
+      errors.push(`${path}: expected an object`)
+      return null
+    }
+
+    if (node.op === 'not') {
+      if (!underAnd) {
+        errors.push(`${path}: "not" is only allowed inside an "and" - a complement needs a set to take it from`)
+        return null
+      }
+      const children = Array.isArray(node.nodes) ? node.nodes : []
+      if (children.length !== 1) {
+        errors.push(`${path}: "not" takes exactly one node`)
+        return null
+      }
+      const inner = walk(children[0], depth + 1, `${path}.nodes[0]`, false)
+      if (!inner) return null
+      // A `not` contributes NO ceiling. Excluding people cannot widen who the
+      // expression reaches, so folding the excluded audience's ceiling into the
+      // meet would refuse perfectly safe segments: `members AND NOT staff` would
+      // hit meet('members','staff') = null and be rejected, even though it
+      // reaches strictly fewer people than `members` alone.
+      return { node: { op: 'not', nodes: [inner.node] }, ceiling: null, complement: true }
+    }
+
+    if (node.op === 'and' || node.op === 'or') {
+      const children = Array.isArray(node.nodes) ? node.nodes : []
+      if (!children.length) {
+        errors.push(`${path}: "${node.op}" has no nodes`)
+        return null
+      }
+      const walked = children.map((c, i) => walk(c, depth + 1, `${path}.nodes[${i}]`, node.op === 'and'))
+      if (walked.some((w) => w === null)) return null
+      const positives = walked.filter((w) => !w.complement)
+      if (!positives.length) {
+        errors.push(`${path}: "${node.op}" has nothing but complements - there is no set to exclude from`)
+        return null
+      }
+      return {
+        node: { op: node.op, nodes: walked.map((w) => w.node) },
+        ceiling: ceilings.meetAll(positives.map((w) => w.ceiling)),
+      }
+    }
+
+    if (node.op !== undefined) {
+      errors.push(`${path}: unknown operator "${node.op}"`)
+      return null
+    }
+
+    const declaration = registries.audience(node.audienceId)
+    if (!declaration) {
+      errors.push(`${path}: no audience "${node.audienceId}" is registered`)
+      return null
+    }
+    const params = checkParams(declaration, node.params, path, errors)
+    return { node: { audienceId: declaration.id, params }, ceiling: declaration.ceiling }
+  }
+
+  if (!isPlainObject(raw)) return { ok: false, errors: ['expression: expected an object'] }
+  const walked = walk(raw, 0, 'expression', false)
+  if (errors.length || !walked) return { ok: false, errors: errors.length ? errors : ['expression: invalid'] }
+  if (!walked.ceiling) {
+    return {
+      ok: false,
+      errors: [
+        'expression: the audiences combined here have no common ceiling, so there is no bound this segment could be given',
+      ],
+    }
+  }
+  return { ok: true, expression: walked.node, ceiling: walked.ceiling }
+}
+
+/**
+ * Resolve a validated expression to a set of user ids.
+ *
+ * Returns `{ dormant, userIds }`. `dormant` is true the moment ANY leaf names an
+ * audience that is no longer registered, and when it is true the caller must not
+ * send: `userIds` is empty, because the tree it would have come from is not the
+ * tree the operator composed.
+ *
+ * `and` is the intersection of its positive children, less the union of its
+ * complements. `or` is the union of its children, which are all positive because
+ * `validate` refused any other shape.
+ */
+async function resolve(expression) {
+  let dormant = false
+
+  async function walk(node) {
+    if (!isPlainObject(node)) return new Set()
+
+    if (node.op === 'and' || node.op === 'or') {
+      const children = Array.isArray(node.nodes) ? node.nodes : []
+      const positives = children.filter((c) => !isNot(c))
+      const complements = children.filter(isNot)
+
+      let out = new Set()
+      for (let i = 0; i < positives.length; i += 1) {
+        const set = await walk(positives[i])
+        if (i === 0) out = set
+        else if (node.op === 'and') out = new Set([...out].filter((id) => set.has(id)))
+        else for (const id of set) out.add(id)
+      }
+      for (const c of complements) {
+        const excluded = await walk((c.nodes || [])[0])
+        out = new Set([...out].filter((id) => !excluded.has(id)))
+      }
+      return out
+    }
+
+    // A `not` reached directly (never produced by validate, but a stored row
+    // predates nothing and this must not throw): no universe, so no members.
+    if (node.op !== undefined) return new Set()
+
+    const { dormant: gone, userIds } = await registries.resolveAudience(node.audienceId, node.params || {})
+    if (gone) dormant = true
+    return new Set(userIds)
+  }
+
+  const set = await walk(expression)
+  return { dormant, userIds: dormant ? [] : [...set] }
+}
+
+module.exports = { validate, resolve, MAX_DEPTH, MAX_NODES }

server/src/model/engagement/engagementCooldowns.db.js

@@ -0,0 +1,78 @@
+const { query } = require('../../utils/db')
+
+/**
+ * Claim a fire for (rule, user, subject), or refuse it because the pair is still
+ * cooling. ENGAGEMENT.md §4.1.
+ *
+ * **Two statements, each of which is its own atomic decision** - and it is worth
+ * saying why it is not the single `INSERT ... ON DUPLICATE KEY UPDATE` §4.1
+ * describes, because that version was written, tested green against an in-memory
+ * stub, and disproved by the first run against a real MariaDB.
+ *
+ * The one-statement form reads its answer out of `affectedRows`, on the usual
+ * contract: 1 for an insert, 2 for an update that changed something, and 0 for a
+ * duplicate key whose update changed nothing - that 0 being "the guard failed, so
+ * this pair is still cooling". **The mariadb Node connector sets `foundRows: true`
+ * by default**, which makes `affectedRows` report rows MATCHED rather than rows
+ * CHANGED, and `utils/db.js` does not override it. Under that pool the no-op case
+ * returns 1, indistinguishable from a fresh insert: every cooldown would have
+ * passed, always, and nothing in a stubbed test could have noticed.
+ *
+ * So the guard moves into a WHERE clause, where a row either matches or does not
+ * and `foundRows` has nothing to fold together:
+ *
+ *   1. UPDATE the row, guarded on the interval. `affectedRows = 1` means this
+ *      caller moved it and owns the fire.
+ *   2. If that matched nothing, the row either does not exist yet or is still
+ *      cooling. `INSERT IGNORE` separates the two: 1 means we inserted the first
+ *      fire, 0 means the row was there and step 1 already said it is cooling.
+ *
+ * It is still race-free, and each race resolves the right way:
+ *   - two concurrent first fires: neither UPDATEs, both INSERT IGNORE, exactly
+ *     one gets 1 (the primary key decides). The loser is treated as cooling.
+ *   - two concurrent fires after expiry: the row is locked by the first UPDATE,
+ *     and the second re-evaluates its guard against the committed row - which now
+ *     holds `now`, so it fails and is refused.
+ *
+ * `cooldown_seconds = 0` always passes, which is the documented meaning of a rule
+ * with no cooldown: the guard becomes `last_fired_at <= now`, and it is.
+ */
+async function claim(ruleId, userId, subjectKey, cooldownSeconds, now = new Date()) {
+  const moved = await query(
+    `UPDATE engagement_cooldowns
+        SET last_fired_at = ?, fire_count = fire_count + 1
+      WHERE rule_id = ? AND user_id = ? AND subject_key = ?
+        AND last_fired_at <= ? - INTERVAL ? SECOND`,
+    [now, ruleId, userId, subjectKey, now, cooldownSeconds],
+  )
+  if (Number(moved?.affectedRows || 0) === 1) return true
+
+  const inserted = await query(
+    `INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, last_fired_at, fire_count)
+     VALUES (?, ?, ?, ?, 1)`,
+    [ruleId, userId, subjectKey, now],
+  )
+  return Number(inserted?.affectedRows || 0) === 1
+}
+
+const get = async (ruleId, userId, subjectKey) => {
+  const [row] = await query(
+    'SELECT * FROM engagement_cooldowns WHERE rule_id = ? AND user_id = ? AND subject_key = ?',
+    [ruleId, userId, subjectKey],
+  )
+  return row || null
+}
+
+/**
+ * Drop cooldown rows older than `olderThan`.
+ *
+ * `idx_engc_sweep (last_fired_at)` exists for this: the table is written on every
+ * fire and read once per fire, so without a prune it is the unbounded-growth
+ * failure `teamActivityPrune` was written for. A dropped row means the next fire
+ * is treated as a first fire, which is correct as long as the retention window is
+ * longer than the longest configured cooldown - the caller's job, not this one's.
+ */
+const prune = (olderThan) =>
+  query('DELETE FROM engagement_cooldowns WHERE last_fired_at < ?', [olderThan])
+
+module.exports = { claim, get, prune }

server/src/model/engagement/engagementOutbox.db.js

@@ -0,0 +1,158 @@
+const { query } = require('../../utils/db')
+const { parseJson } = require('./engagementRules.db')
+
+const hydrate = (row) => row && { ...row, payload: parseJson(row.payload, {}) }
+
+/**
+ * Enqueue one (rule, user, channel) row, idempotently.
+ *
+ * `INSERT IGNORE` rather than a plain INSERT, because `uq_engo_dedupe` is the
+ * replay guard (§4.2a): the sidecar feed is at-least-once and a reconnect
+ * backfills, so the same event arriving twice must produce one row and not two
+ * mails. IGNORE turns that into a silent no-op, which is what a replay should be.
+ *
+ * Returns the new id, or null when the row already existed. A null is a
+ * SUCCESSFUL duplicate, not a failure - the caller counts it as such.
+ *
+ * A NULL dedupe_key never collides (multiple NULLs are legal under a UNIQUE
+ * index), so an emit that carries no key always enqueues. That is the right
+ * default: dedupe is something the emitter opts into by naming a key, and core
+ * cannot invent one that means anything.
+ */
+async function enqueue(row) {
+  const result = await query(
+    `INSERT IGNORE INTO engagement_outbox
+       (rule_id, trigger_id, user_id, channel, subject_key, payload, dedupe_key, due_at)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+    [
+      row.rule_id,
+      row.trigger_id,
+      row.user_id,
+      row.channel,
+      row.subject_key || '',
+      JSON.stringify(row.payload || {}),
+      row.dedupe_key ?? null,
+      row.due_at,
+    ],
+  )
+  return Number(result?.affectedRows || 0) === 1 ? result.insertId : null
+}
+
+/**
+ * Rows that are due. `idx_engo_due (status, due_at)` is this query.
+ *
+ * It selects rather than claims - claiming is `claim()` below, one row at a
+ * time - so two instances sweeping at once both see the same candidates and then
+ * disagree, harmlessly, about which of them owns each.
+ */
+const findDue = async (now, limit = 100) =>
+  (
+    await query(
+      "SELECT * FROM engagement_outbox WHERE status = 'scheduled' AND due_at <= ? ORDER BY due_at, id LIMIT ?",
+      [now, limit],
+    )
+  ).map(hydrate)
+
+/**
+ * Take ownership of one due row: a compare-and-set from 'scheduled' to 'sending'.
+ *
+ * **This is §7.1 Q2's answer** (settled by the org lead 2026-08-29, over
+ * `SELECT ... FOR UPDATE SKIP LOCKED`). The winner is whoever the server reports
+ * `affectedRows = 1` to; every other sweeper gets 0 and moves on. No explicit
+ * transaction, no MariaDB version floor, and it uses a status the ENUM already
+ * carried for exactly this.
+ *
+ * What it makes safe is the OUTBOX and only the outbox. `announceWorker`,
+ * `teamDigestWorker`, `teamForumUploadSweep` and `teamActivityPrune` are all
+ * still written for a single instance, so this does not make the deployment
+ * multi-instance - it makes the one table that will carry mail ready for the day
+ * it is, which is cheap now and expensive after mail has doubled once.
+ */
+async function claim(id) {
+  const result = await query(
+    `UPDATE engagement_outbox
+        SET status = 'sending', attempts = attempts + 1
+      WHERE id = ? AND status = 'scheduled'`,
+    [id],
+  )
+  return Number(result?.affectedRows || 0) === 1
+}
+
+/**
+ * Release a claimed row back to 'scheduled' with a later `due_at` - a transient
+ * failure that should be retried. The mirror of announceJobs' backoff.
+ */
+const reschedule = (id, dueAt, error) =>
+  query(
+    "UPDATE engagement_outbox SET status = 'scheduled', due_at = ?, last_error = ? WHERE id = ? AND status = 'sending'",
+    [dueAt, error ? String(error).slice(0, 2000) : null, id],
+  )
+
+/** A terminal outcome: 'sent', 'failed' or 'suppressed'. */
+const finish = (id, status, error) =>
+  query(
+    `UPDATE engagement_outbox
+        SET status = ?, last_error = ?, sent_at = IF(? = 'sent', NOW(), sent_at)
+      WHERE id = ?`,
+    [status, error ? String(error).slice(0, 2000) : null, status, id],
+  )
+
+/**
+ * Cancel every still-scheduled row for a (rule, subject) - the point of the
+ * grace window (§4.2a). `userId` narrows it to one recipient when the resolving
+ * event names one; a resolving event with no owner cancels for everyone the
+ * original event was queued for, which is the house-repaired case.
+ *
+ * Only 'scheduled' rows are touched: a row already claimed into 'sending' is
+ * somebody's in-flight send and cancelling it would leave two workers writing
+ * one row's outcome.
+ */
+async function cancel(ruleId, subjectKey, userId = null) {
+  const params = [ruleId, subjectKey]
+  let sql = "UPDATE engagement_outbox SET status = 'cancelled' WHERE rule_id = ? AND subject_key = ? AND status = 'scheduled'"
+  if (userId !== null && userId !== undefined) {
+    sql += ' AND user_id = ?'
+    params.push(userId)
+  }
+  const result = await query(sql, params)
+  return Number(result?.affectedRows || 0)
+}
+
+/**
+ * Recover rows stranded in 'sending' by a crash between the claim and the
+ * outcome.
+ *
+ * Without this the CAS claim leaks: the claiming process died, no other sweeper
+ * will ever match `status = 'scheduled'`, and the row sits in 'sending' forever.
+ * `updated_at` is the clock (it is ON UPDATE CURRENT_TIMESTAMP, so the claim
+ * stamped it), and the window has to be comfortably longer than the slowest
+ * legitimate send or this reclaims rows that are merely slow.
+ */
+const reclaimStale = (before) =>
+  query(
+    "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?",
+    [before],
+  )
+
+const getById = async (id) => {
+  const [row] = await query('SELECT * FROM engagement_outbox WHERE id = ?', [id])
+  return hydrate(row)
+}
+
+/** Admin/read surfaces (Phase 4b) and tests. */
+const listForRule = async (ruleId, limit = 100) =>
+  (
+    await query('SELECT * FROM engagement_outbox WHERE rule_id = ? ORDER BY id DESC LIMIT ?', [ruleId, limit])
+  ).map(hydrate)
+
+module.exports = {
+  enqueue,
+  findDue,
+  claim,
+  reschedule,
+  finish,
+  cancel,
+  reclaimStale,
+  getById,
+  listForRule,
+}

server/src/model/engagement/engagementRecipients.db.js

@@ -0,0 +1,125 @@
+const { query } = require('../../utils/db')
+
+// A bound on every "resolve an audience" query. `authenticated` on a large
+// deployment is the whole user table, and the engine turns each id into an
+// outbox row - so the read that feeds it has to have a ceiling of its own. The
+// per-rule hourly cap (§7.1 Q3) is the operator-facing limit; this is the one
+// that keeps a single emit from loading a hundred thousand rows into memory.
+const MAX_AUDIENCE = 5000
+
+const ids = (rows) => rows.map((r) => Number(r.id)).filter((n) => Number.isInteger(n) && n > 0)
+
+const marks = (list) => list.map(() => '?').join(', ')
+
+/**
+ * Every active user. The `authenticated` audience - and `everyone`, which has no
+ * distinct meaning here: a signed-out visitor has no address, no device and no
+ * inbox, so the widest set the engine can actually deliver to is this one. The
+ * ceiling lattice still distinguishes them (a trigger ceilinged `everyone`
+ * permits an `authenticated` rule and not the reverse); only the resolution
+ * coincides.
+ *
+ * `status = 'active'` on every query in this file: a banned or disabled account
+ * is refused at login, and mailing it engagement content would be the one
+ * surface that did not get the message.
+ */
+const active = async (limit = MAX_AUDIENCE) =>
+  ids(await query("SELECT id FROM users WHERE status = 'active' ORDER BY id LIMIT ?", [limit]))
+
+/** The `staff` audience. Roles come from `ceilings.STAFF_CEILING_ROLES`. */
+const staff = async (roles, limit = MAX_AUDIENCE) => {
+  if (!roles.length) return []
+  return ids(
+    await query(
+      `SELECT id FROM users WHERE status = 'active' AND role IN (${marks(roles)}) ORDER BY id LIMIT ?`,
+      [...roles, limit],
+    ),
+  )
+}
+
+/**
+ * The `subscribers` audience: active users who have opted into this id on at
+ * least one channel.
+ *
+ * "Opted in" is the EFFECTIVE mode, not the stored one, and that is why this is
+ * not simply `WHERE mode <> 'off'`. A row exists only where a user said
+ * something; absence means the channel's `defaultMode` (§3.1). All three of
+ * core's channels default 'off' today, so the second half of the WHERE matches
+ * nobody - but writing it means the day a channel ships with a non-off default,
+ * this audience is already right rather than silently excluding everyone who
+ * never opened the preferences screen.
+ *
+ * `defaultOnChannels` is the caller's list of channels whose defaultMode is not
+ * 'off'; it comes from the channel registry, so the default lives in exactly one
+ * place here too.
+ */
+const subscribers = async (streamId, defaultOnChannels = [], limit = MAX_AUDIENCE) => {
+  const optedIn = `EXISTS (
+      SELECT 1 FROM notification_channel_prefs p
+       WHERE p.user_id = u.id AND p.stream_id = ? AND p.mode <> 'off')`
+
+  if (!defaultOnChannels.length) {
+    return ids(
+      await query(
+        `SELECT u.id FROM users u WHERE u.status = 'active' AND ${optedIn} ORDER BY u.id LIMIT ?`,
+        [streamId, limit],
+      ),
+    )
+  }
+
+  // "At least one default-on channel has no row for this user" - counted rather
+  // than NOT EXISTS, because NOT EXISTS would mean "none of them has a row".
+  const defaulted = `(
+      SELECT COUNT(*) FROM notification_channel_prefs p2
+       WHERE p2.user_id = u.id AND p2.stream_id = ? AND p2.channel IN (${marks(defaultOnChannels)})
+    ) < ?`
+
+  return ids(
+    await query(
+      `SELECT u.id FROM users u
+        WHERE u.status = 'active' AND (${optedIn} OR ${defaulted})
+        ORDER BY u.id LIMIT ?`,
+      [streamId, streamId, ...defaultOnChannels, defaultOnChannels.length, limit],
+    ),
+  )
+}
+
+/**
+ * Narrow a set of user ids to the active ones.
+ *
+ * Every audience that does NOT come from a query in this file goes through here:
+ * `owner` is a single id off the event envelope, and a module-declared audience
+ * (§5.1a) is a list of ids a module's own resolver produced. Neither has any
+ * notion of account status, and a module must not be able to mail a banned
+ * account by returning its id.
+ */
+const filterActive = async (userIds) => {
+  const wanted = [...new Set(userIds.map(Number).filter((n) => Number.isInteger(n) && n > 0))]
+  if (!wanted.length) return []
+  const capped = wanted.slice(0, MAX_AUDIENCE)
+  return ids(
+    await query(
+      `SELECT id FROM users WHERE status = 'active' AND id IN (${marks(capped)}) ORDER BY id`,
+      capped,
+    ),
+  )
+}
+
+/**
+ * The stored mode for one (id, channel) across a set of users, as a Map.
+ *
+ * The caller applies the channel's `defaultMode` to anyone missing from the map,
+ * which keeps the defaulting in the one place §3.1 put it. Returning stored rows
+ * rather than a decision is what makes that possible.
+ */
+const storedModes = async (userIds, streamId, channel) => {
+  if (!userIds.length) return new Map()
+  const rows = await query(
+    `SELECT user_id, mode FROM notification_channel_prefs
+      WHERE stream_id = ? AND channel = ? AND user_id IN (${marks(userIds)})`,
+    [streamId, channel, ...userIds],
+  )
+  return new Map(rows.map((r) => [Number(r.user_id), r.mode]))
+}
+
+module.exports = { active, staff, subscribers, filterActive, storedModes, MAX_AUDIENCE }

server/src/model/engagement/engagementRules.db.js

@@ -0,0 +1,127 @@
+const { query } = require('../../utils/db')
+
+// JSON columns come back from the driver already parsed on some MariaDB/driver
+// combinations and as a string on others (it depends on whether the column is a
+// real JSON type or the LONGTEXT + CHECK alias MariaDB implements it as). Every
+// read below goes through this, so no caller has to know which it got.
+function parseJson(value, fallback) {
+  if (value === null || value === undefined) return fallback
+  if (typeof value !== 'string') return value
+  try {
+    return JSON.parse(value)
+  } catch {
+    return fallback
+  }
+}
+
+const hydrate = (row) =>
+  row && {
+    ...row,
+    enabled: Boolean(row.enabled),
+    channels: parseJson(row.channels, []),
+    template_keys: parseJson(row.template_keys, {}),
+    conditions: parseJson(row.conditions, null),
+    cancel_on: parseJson(row.cancel_on, []),
+  }
+
+const list = async () =>
+  (await query('SELECT * FROM engagement_rules ORDER BY trigger_id, name, id')).map(hydrate)
+
+const getById = async (id) => {
+  const [row] = await query('SELECT * FROM engagement_rules WHERE id = ?', [id])
+  return hydrate(row)
+}
+
+/**
+ * Every ENABLED rule for one trigger. The engine's hot path: one indexed read
+ * per emit, and `idx_engr_trigger (trigger_id, enabled)` is exactly this query.
+ */
+const enabledForTrigger = async (triggerId) =>
+  (await query('SELECT * FROM engagement_rules WHERE trigger_id = ? AND enabled = 1', [triggerId])).map(hydrate)
+
+/**
+ * Every enabled rule that names `triggerId` in its `cancel_on`.
+ *
+ * A JSON_CONTAINS rather than a scan: `cancel_on` is a small array on a small
+ * table, but this runs on EVERY emit — including the overwhelming majority that
+ * cancel nothing — so it must not be a full table read of the rule set.
+ */
+const enabledCancelledBy = async (triggerId) =>
+  (
+    await query(
+      "SELECT * FROM engagement_rules WHERE enabled = 1 AND cancel_on IS NOT NULL AND JSON_CONTAINS(cancel_on, JSON_QUOTE(?))",
+      [triggerId],
+    )
+  ).map(hydrate)
+
+const insert = async (rule) => {
+  const result = await query(
+    `INSERT INTO engagement_rules
+       (trigger_id, name, enabled, audience, audience_segment_id, max_sends_per_hour,
+        channels, template_keys, conditions, cooldown_seconds, delay_seconds, cancel_on, updated_by)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+    [
+      rule.trigger_id,
+      rule.name,
+      rule.enabled ? 1 : 0,
+      rule.audience,
+      rule.audience_segment_id,
+      rule.max_sends_per_hour,
+      JSON.stringify(rule.channels),
+      JSON.stringify(rule.template_keys),
+      rule.conditions === null ? null : JSON.stringify(rule.conditions),
+      rule.cooldown_seconds,
+      rule.delay_seconds,
+      JSON.stringify(rule.cancel_on || []),
+      rule.updated_by,
+    ],
+  )
+  return result.insertId
+}
+
+const update = (id, rule) =>
+  query(
+    `UPDATE engagement_rules
+        SET name = ?, enabled = ?, audience = ?, audience_segment_id = ?, max_sends_per_hour = ?,
+            channels = ?, template_keys = ?, conditions = ?, cooldown_seconds = ?,
+            delay_seconds = ?, cancel_on = ?, updated_by = ?
+      WHERE id = ?`,
+    [
+      rule.name,
+      rule.enabled ? 1 : 0,
+      rule.audience,
+      rule.audience_segment_id,
+      rule.max_sends_per_hour,
+      JSON.stringify(rule.channels),
+      JSON.stringify(rule.template_keys),
+      rule.conditions === null ? null : JSON.stringify(rule.conditions),
+      rule.cooldown_seconds,
+      rule.delay_seconds,
+      JSON.stringify(rule.cancel_on || []),
+      rule.updated_by,
+      id,
+    ],
+  )
+
+const remove = (id) => query('DELETE FROM engagement_rules WHERE id = ?', [id])
+
+/** Does any rule still point at this segment? The check before a segment delete. */
+const countUsingSegment = async (segmentId) => {
+  const [row] = await query(
+    'SELECT COUNT(*) AS n FROM engagement_rules WHERE audience_segment_id = ?',
+    [segmentId],
+  )
+  return Number(row?.n || 0)
+}
+
+module.exports = {
+  list,
+  getById,
+  enabledForTrigger,
+  enabledCancelledBy,
+  insert,
+  update,
+  remove,
+  countUsingSegment,
+  parseJson,
+}

server/src/model/engagement/engagementRules.model.js

@@ -0,0 +1,225 @@
+// ── Engagement rules — the save path ───────────────────────────────────────
+//
+// ENGAGEMENT.md §4.5 / §7.1 Q3, Phase 4a. A rule is **operator-editable data**,
+// not code, and that was a deliberate choice with a condition attached: it is
+// safe to choose only because `enabled` defaults to 0 and every rule carries a
+// hard per-hour send ceiling. Both of those live in this file's validation, not
+// in the screen that calls it - Phase 4b builds a form over this, and a rule that
+// arrives by any other route (a restore, a fixture, a future import) gets the
+// same answer.
+//
+// **Every check here is a boundary, not a convenience.** The rule editor will
+// re-implement some of them for the sake of a good error message, and that
+// second copy is expected to drift - so this one is the one that decides.
+//
+// The check with teeth is the ceiling (G24): an operator may narrow a rule's
+// audience as much as they like and may never widen it past what the trigger
+// declared. `ceilings.permits` is that arithmetic, `segments.validate` derives
+// it for a composed audience, and the engine re-runs the same check at SEND
+// time in case a module upgrade narrowed the declaration underneath a saved rule.
+
+const db = require('./engagementRules.db')
+const segmentsDb = require('./engagementSegments.db')
+const registries = require('../../modules/registries')
+const ceilings = require('../../modules/ceilings')
+const channels = require('../../engagement/channels')
+const conditions = require('../../engagement/conditions')
+
+// A day. Longer than this and "cooldown" is really "send once", which a rule
+// expresses by being disabled rather than by a decade-long interval.
+const MAX_COOLDOWN_SECONDS = 86_400
+// The grace window (§4.2a). A delay longer than a day outlives the thing it is
+// about - and, more practically, a queue row that sits for a week is a row whose
+// payload no longer describes the world.
+const MAX_DELAY_SECONDS = 86_400
+// The upper bound on the operator-set hourly ceiling. It is not "unlimited by
+// another name": the number exists so that a misconfiguration is a bad hour
+// rather than an unbounded one, and a ceiling nobody can raise past a bound is
+// what makes rules-as-data safe (§7.1 Q3).
+const MAX_SENDS_PER_HOUR = 10_000
+
+const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v)
+
+/**
+ * Validate a rule against the registries and the lattice.
+ *
+ * Returns `{ ok: true, rule }` with a normalised row ready for insert/update, or
+ * `{ ok: false, errors }` listing every problem.
+ *
+ * `triggerId` may name a trigger nobody currently registers ONLY on an update of
+ * an existing rule - a dormant rule must stay editable (its module can come
+ * back), and refusing to save it would make an uninstall destructive after the
+ * fact. A NEW rule must name a live trigger, because there is nothing to
+ * preserve and a typo should be caught now.
+ */
+async function validate(input, { existing = null } = {}) {
+  const errors = []
+  const raw = isPlainObject(input) ? input : {}
+
+  const triggerId = typeof raw.triggerId === 'string' ? raw.triggerId : existing?.trigger_id
+  const declaration = triggerId ? registries.eventTrigger(triggerId) : null
+  if (!triggerId) errors.push('triggerId is required')
+  else if (!declaration && !existing) errors.push(`no trigger "${triggerId}" is registered`)
+
+  const name = typeof raw.name === 'string' ? raw.name.trim() : ''
+  if (!name) errors.push('name is required')
+  else if (name.length > 160) errors.push('name is longer than 160 characters')
+
+  // Channels are stored as data and checked against the registry, so a rule
+  // cannot name a sink that does not exist. Phase 4b's form offers the registered
+  // set; this is what makes that an affordance rather than the rule.
+  const wanted = Array.isArray(raw.channels) ? [...new Set(raw.channels)] : []
+  if (!wanted.length) errors.push('at least one channel is required')
+  for (const c of wanted) if (!channels.has(c)) errors.push(`no channel "${c}" is registered`)
+
+  // `template_keys` is { channel: templateKey }. Phase 5 owns templates, so the
+  // KEYS are checked for shape and not for existence - a rule may legitimately
+  // name a template that has not been authored yet, and Phase 5's editor is where
+  // that becomes resolvable.
+  const templateKeys = {}
+  if (raw.templateKeys !== undefined && !isPlainObject(raw.templateKeys)) {
+    errors.push('templateKeys must be an object of { channel: templateKey }')
+  } else {
+    for (const [channel, key] of Object.entries(raw.templateKeys || {})) {
+      if (!wanted.includes(channel)) {
+        errors.push(`templateKeys names "${channel}", which is not one of this rule's channels`)
+        continue
+      }
+      if (typeof key !== 'string' || !/^[a-z0-9][a-z0-9-]{0,63}$/.test(key)) {
+        errors.push(`templateKeys.${channel} is not a valid template key`)
+        continue
+      }
+      templateKeys[channel] = key
+    }
+  }
+
+  const numbers = [
+    ['cooldownSeconds', 'cooldown_seconds', MAX_COOLDOWN_SECONDS, 0],
+    ['delaySeconds', 'delay_seconds', MAX_DELAY_SECONDS, 0],
+    ['maxSendsPerHour', 'max_sends_per_hour', MAX_SENDS_PER_HOUR, 1],
+  ]
+  const scalars = {}
+  for (const [key, column, max, min] of numbers) {
+    const supplied = raw[key]
+    const fallback = existing ? existing[column] : column === 'max_sends_per_hour' ? 100 : 0
+    const value = supplied === undefined || supplied === null ? fallback : Number(supplied)
+    if (!Number.isInteger(value) || value < min || value > max) {
+      errors.push(`${key} must be an integer between ${min} and ${max}`)
+    } else scalars[column] = value
+  }
+
+  // `cancel_on` names trigger ids, and they are NOT checked for registration for
+  // the dormancy reason (§7.3): a resolving event whose module is temporarily
+  // absent should stop cancelling, not make the rule unsaveable.
+  const cancelOn = Array.isArray(raw.cancelOn) ? [...new Set(raw.cancelOn.filter((t) => typeof t === 'string'))] : []
+  if (cancelOn.length && !scalars.delay_seconds) {
+    // Not an error - it is a rule that will never cancel anything, because there
+    // is no window in which to do it. Worth saying out loud rather than silently
+    // accepting a setting that cannot take effect.
+    errors.push('cancelOn has no effect without a delaySeconds grace window')
+  }
+
+  const checked = conditions.validate(declaration, raw.conditions === undefined ? existing?.conditions : raw.conditions)
+  if (!checked.ok) errors.push(...checked.errors)
+
+  // ── The audience, and the one check that is a security boundary ──────────
+  let audience = typeof raw.audience === 'string' ? raw.audience : existing?.audience || declaration?.audience
+  let segmentId = raw.audienceSegmentId === undefined ? existing?.audience_segment_id ?? null : raw.audienceSegmentId
+  segmentId = segmentId === null || segmentId === '' ? null : Number(segmentId)
+
+  let effectiveCeiling = null
+  if (segmentId !== null) {
+    if (!Number.isInteger(segmentId)) errors.push('audienceSegmentId must be an integer')
+    else {
+      const segment = await segmentsDb.getById(segmentId)
+      if (!segment) errors.push(`no audience segment ${segmentId} exists`)
+      else {
+        // The segment's STORED ceiling, derived when it was saved by
+        // `segments.validate` from the narrowest audience it contains. A rule
+        // pointing at a segment takes that as its reach; the `audience` column
+        // is retained for display and is not what the engine resolves.
+        effectiveCeiling = segment.ceiling
+        audience = segment.ceiling
+      }
+    }
+  } else if (!ceilings.isCeiling(audience)) {
+    errors.push(`audience must be one of ${ceilings.CEILINGS.join(', ')}`)
+  } else {
+    effectiveCeiling = audience
+  }
+
+  if (declaration && effectiveCeiling && !ceilings.permits(declaration.ceiling, effectiveCeiling)) {
+    errors.push(
+      `audience "${effectiveCeiling}" is wider than trigger "${triggerId}" permits (ceiling "${declaration.ceiling}")`,
+    )
+  }
+
+  if (errors.length) return { ok: false, errors }
+
+  return {
+    ok: true,
+    rule: {
+      trigger_id: triggerId,
+      name,
+      enabled: raw.enabled === undefined ? Boolean(existing?.enabled) : Boolean(raw.enabled),
+      audience,
+      audience_segment_id: segmentId,
+      max_sends_per_hour: scalars.max_sends_per_hour,
+      channels: wanted,
+      template_keys: templateKeys,
+      conditions: checked.conditions,
+      cooldown_seconds: scalars.cooldown_seconds,
+      delay_seconds: scalars.delay_seconds,
+      cancel_on: cancelOn,
+      updated_by: Number.isInteger(raw.updatedBy) ? raw.updatedBy : null,
+    },
+  }
+}
+
+async function create(input) {
+  const checked = await validate(input)
+  if (!checked.ok) return checked
+  const id = await db.insert(checked.rule)
+  return { ok: true, rule: await db.getById(id) }
+}
+
+async function update(id, input) {
+  const existing = await db.getById(id)
+  if (!existing) return { ok: false, errors: [`no rule ${id} exists`], notFound: true }
+  const checked = await validate(input, { existing })
+  if (!checked.ok) return checked
+  await db.update(id, checked.rule)
+  return { ok: true, rule: await db.getById(id) }
+}
+
+/**
+ * List every rule, each annotated with whether it can currently fire.
+ *
+ * Dormancy is computed rather than stored (§7.3): a rule whose trigger or
+ * segment is not registered right now is listed, flagged, and left alone. The
+ * alternative - deleting or disabling it on uninstall - destroys an operator's
+ * configuration on the strength of a module being temporarily absent.
+ */
+async function listAnnotated() {
+  const rows = await db.list()
+  const segments = new Map((await segmentsDb.list()).map((s) => [s.id, s]))
+  return rows.map((rule) => {
+    const reasons = []
+    if (!registries.eventTrigger(rule.trigger_id)) reasons.push(`trigger "${rule.trigger_id}" is not registered`)
+    if (rule.audience_segment_id && !segments.has(rule.audience_segment_id)) {
+      reasons.push('its audience segment no longer exists')
+    }
+    for (const c of rule.channels || []) if (!channels.has(c)) reasons.push(`channel "${c}" is not registered`)
+    return { ...rule, dormant: reasons.length > 0, dormantReasons: reasons }
+  })
+}
+
+module.exports = {
+  validate,
+  create,
+  update,
+  listAnnotated,
+  MAX_COOLDOWN_SECONDS,
+  MAX_DELAY_SECONDS,
+  MAX_SENDS_PER_HOUR,
+}

server/src/model/engagement/engagementSegments.db.js

@@ -0,0 +1,37 @@
+const { query } = require('../../utils/db')
+const { parseJson } = require('./engagementRules.db')
+
+const hydrate = (row) => row && { ...row, expression: parseJson(row.expression, null) }
+
+const list = async () =>
+  (await query('SELECT * FROM engagement_audience_segments ORDER BY name, id')).map(hydrate)
+
+const getById = async (id) => {
+  const [row] = await query('SELECT * FROM engagement_audience_segments WHERE id = ?', [id])
+  return hydrate(row)
+}
+
+/**
+ * `ceiling` is written by the caller from `segments.deriveCeiling`, never taken
+ * from an operator. It is a stored column rather than a runtime computation so
+ * an audit can read what a rule was ALLOWED to reach without re-resolving it,
+ * and so a module that later widens its own audience's ceiling cannot
+ * retroactively widen a segment that was saved under the old one.
+ */
+const insert = async (segment) => {
+  const result = await query(
+    'INSERT INTO engagement_audience_segments (name, expression, ceiling, updated_by) VALUES (?, ?, ?, ?)',
+    [segment.name, JSON.stringify(segment.expression), segment.ceiling, segment.updated_by ?? null],
+  )
+  return result.insertId
+}
+
+const update = (id, segment) =>
+  query(
+    'UPDATE engagement_audience_segments SET name = ?, expression = ?, ceiling = ?, updated_by = ? WHERE id = ?',
+    [segment.name, JSON.stringify(segment.expression), segment.ceiling, segment.updated_by ?? null, id],
+  )
+
+const remove = (id) => query('DELETE FROM engagement_audience_segments WHERE id = ?', [id])
+
+module.exports = { list, getById, insert, update, remove }

server/src/model/engagement/engagementSegments.model.js

@@ -0,0 +1,87 @@
+// ── Audience segments — the save path ──────────────────────────────────────
+//
+// ENGAGEMENT.md §5.1a, Phase 4a. The thin model over `segments.js`: it validates,
+// derives the ceiling, and writes. The composition UI is Phase 4b's; this is what
+// it will call, and what any other route in must go through.
+//
+// The `ceiling` column is never taken from the caller. It is derived from the
+// expression by `segments.validate` as the narrowest ceiling in the tree, and
+// stored so an audit can read what a rule was ALLOWED to reach without
+// re-resolving it.
+
+const db = require('./engagementSegments.db')
+const rulesDb = require('./engagementRules.db')
+const registries = require('../../modules/registries')
+const segments = require('../../engagement/segments')
+
+async function save(input, { id = null } = {}) {
+  const errors = []
+  const name = typeof input?.name === 'string' ? input.name.trim() : ''
+  if (!name) errors.push('name is required')
+  else if (name.length > 160) errors.push('name is longer than 160 characters')
+
+  const checked = segments.validate(input?.expression)
+  if (!checked.ok) errors.push(...checked.errors)
+  if (errors.length) return { ok: false, errors }
+
+  const row = {
+    name,
+    expression: checked.expression,
+    ceiling: checked.ceiling,
+    updated_by: Number.isInteger(input?.updatedBy) ? input.updatedBy : null,
+  }
+
+  if (id) {
+    const existing = await db.getById(id)
+    if (!existing) return { ok: false, errors: [`no segment ${id} exists`], notFound: true }
+    await db.update(id, row)
+    return { ok: true, segment: await db.getById(id) }
+  }
+  const newId = await db.insert(row)
+  return { ok: true, segment: await db.getById(newId) }
+}
+
+/**
+ * Delete a segment, refusing while a rule still points at it.
+ *
+ * There is deliberately no foreign key doing this (schema.sql): the database
+ * options are CASCADE, which would delete an operator's rules, and SET NULL,
+ * which would silently fall the rule back to its plain `audience` column and mail
+ * a DIFFERENT set of people. Refusing here, with the count, is the third option
+ * and the only safe one.
+ */
+async function remove(id) {
+  const inUse = await rulesDb.countUsingSegment(id)
+  if (inUse > 0) {
+    return {
+      ok: false,
+      inUse,
+      errors: [`${inUse} rule${inUse === 1 ? '' : 's'} still use this segment`],
+    }
+  }
+  await db.remove(id)
+  return { ok: true }
+}
+
+/**
+ * Every segment, each annotated with whether it can currently resolve.
+ *
+ * A segment naming an audience whose module has been uninstalled is DORMANT, not
+ * broken: it is listed, it resolves to nobody, and it starts working again when
+ * the module comes back (§5.1a rule 4).
+ */
+async function listAnnotated() {
+  const rows = await db.list()
+  return rows.map((segment) => {
+    const missing = []
+    const walk = (node) => {
+      if (!node || typeof node !== 'object') return
+      if (node.op) (node.nodes || []).forEach(walk)
+      else if (!registries.audience(node.audienceId)) missing.push(node.audienceId)
+    }
+    walk(segment.expression)
+    return { ...segment, dormant: missing.length > 0, missingAudiences: [...new Set(missing)] }
+  })
+}
+
+module.exports = { save, remove, listAnnotated }

server/src/model/engagement/engagementSends.db.js

@@ -0,0 +1,73 @@
+const { query } = require('../../utils/db')
+
+/**
+ * Record one attempt's outcome. G15: "did user X get the mail?" has never been
+ * answerable on this deployment, and this row is the answer.
+ *
+ * `address_hash` is a sha256 the CALLER computes, never an address. The log has
+ * to correlate a bounce back to a recipient (Phase 9) and it must not become a
+ * second address book, and a hash does the first without the second.
+ */
+const record = async (entry) => {
+  const result = await query(
+    `INSERT INTO engagement_sends
+       (outbox_id, rule_id, trigger_id, user_id, channel, transport, address_hash, status, detail)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+    [
+      entry.outbox_id ?? null,
+      entry.rule_id ?? null,
+      entry.trigger_id,
+      entry.user_id ?? null,
+      entry.channel,
+      entry.transport ?? null,
+      entry.address_hash ?? null,
+      entry.status,
+      entry.detail ? String(entry.detail).slice(0, 500) : null,
+    ],
+  )
+  return result.insertId
+}
+
+/**
+ * How many sends this rule has made in the last hour - the count the per-rule
+ * ceiling (§7.1 Q3) is enforced against.
+ *
+ * It counts 'sent' only. A refusal that never left the building (`suppressed`)
+ * and an attempt that failed are not sends, and counting them would let a broken
+ * transport silently consume a rule's whole hourly budget and mute it.
+ *
+ * `idx_engs_rule_window (rule_id, created_at)` exists for this: it runs once per
+ * rule per event, so it has to be an index range scan.
+ */
+const countSentSince = async (ruleId, since) => {
+  const [row] = await query(
+    "SELECT COUNT(*) AS n FROM engagement_sends WHERE rule_id = ? AND status = 'sent' AND created_at >= ?",
+    [ruleId, since],
+  )
+  return Number(row?.n || 0)
+}
+
+/** The admin send log (Phase 4b/5), newest first. */
+const list = ({ triggerId = null, userId = null, ruleId = null, limit = 100, offset = 0 } = {}) => {
+  const where = []
+  const params = []
+  if (triggerId) {
+    where.push('trigger_id = ?')
+    params.push(triggerId)
+  }
+  if (userId) {
+    where.push('user_id = ?')
+    params.push(userId)
+  }
+  if (ruleId) {
+    where.push('rule_id = ?')
+    params.push(ruleId)
+  }
+  const clause = where.length ? `WHERE ${where.join(' AND ')}` : ''
+  return query(
+    `SELECT * FROM engagement_sends ${clause} ORDER BY id DESC LIMIT ? OFFSET ?`,
+    [...params, limit, offset],
+  )
+}
+
+module.exports = { record, countSentSince, list }

server/src/server.js

@@ -12,6 +12,7 @@ const announceWorker = require('./utils/announceWorker')
 const teamActivityPrune = require('./utils/teamActivityPrune')
 const teamForumUploadSweep = require('./utils/teamForumUploadSweep')
 const teamDigestWorker = require('./utils/teamDigestWorker')
+const engagementWorker = require('./utils/engagementWorker')
 const { ensureSchema, close } = require('./utils/db')
 const { seedDefaults, createInitialAdminFromEnv } = require('../db/seed')
 const settings = require('./model/settings/settings.model')
@@ -160,6 +161,10 @@ async function start() {
   teamForumUploadSweep.start()
   teamDigestWorker.start()
 
+  // Drain the engagement outbox (ENGAGEMENT.md §4.2a). No-op until an operator
+  // enables a rule: core seeds none and `enabled` defaults to 0.
+  engagementWorker.start()
+
   setupShutdown(server, internalServer)
 }
 
@@ -180,6 +185,7 @@ function setupShutdown(server, internalServer) {
     teamActivityPrune.stop() // stop the Team activity retention timer
     teamForumUploadSweep.stop() // stop the forum upload sweep
     teamDigestWorker.stop() // stop the Team forum digest timer
+    engagementWorker.stop() // stop the engagement outbox worker
     server.close(() => log.info('http server closed'))
     if (internalServer) internalServer.close(() => log.info('internal http server closed'))
     try {

server/src/utils/engagementEmit.js

@@ -1,16 +1,17 @@
 // ── ctx.events.emit — the validating half of the engagement seam ────────────
 //
-// ENGAGEMENT.md §4.3 and §5.2, Phase 2. A registrant fires a declared event with
-// a payload; this checks the payload against the declaration and stops there.
-// **There is no delivery in this phase** — no rules, no cooldowns, no outbox, no
-// mail. Phase 4 replaces the log line at the bottom with the engine call, and
-// every validation rule below is already the one it will need.
+// ENGAGEMENT.md §4.3 and §5.2. A registrant fires a declared event with a
+// payload; this checks the payload against the declaration and, since Phase 4a,
+// hands the validated event to the engine.
 //
-// Landing the contract a phase before the engine is deliberate, and it is the
+// Landing the contract a phase before the engine was deliberate, and it is the
 // same argument registerCore() has always made: a seam whose first real exercise
-// is the thing that depends on it is a seam that has already drifted. Phase 6
-// migrates the Team mail onto this, and it should be migrating onto a validator
-// that has been running against core's own five triggers since Phase 2.
+// is the thing that depends on it is a seam that has already drifted. Every
+// validation rule below was written in Phase 2 for a caller that did not exist
+// yet, and the engine needed none of them changed.
+//
+// **The engine call is deliberately not awaited** — see `emit` below. Phase 6
+// migrates the Team mail onto this.
 //
 // **Two postures, one switch.** A malformed emit THROWS in development and is
 // DROPPED AND LOGGED in production, which is `ctx.teams.activity.push`'s posture
@@ -20,6 +21,7 @@
 // silently loses a variable is a template that silently renders `undefined`.
 
 const registries = require('../modules/registries')
+const engine = require('../engagement/engine')
 const createLogger = require('./logger')
 
 const log = createLogger('engagement')
@@ -203,8 +205,6 @@ function emit(owner, triggerId, envelope = {}) {
     data: payload.data,
   }
 
-  // Phase 2 ends here: validated, recorded, and deliberately undelivered.
-  //
   // The values are NOT logged. A payload carries player names, house locations
   // and forum excerpts, and an event log that reproduces them is a second copy
   // of exactly the content §4.5 was careful to keep out of `engagement_sends`
@@ -217,6 +217,19 @@ function emit(owner, triggerId, envelope = {}) {
     variables: Object.keys(event.data),
   })
 
+  // **Not awaited, and this is the point of the whole seam.** `emit` is called
+  // from inside a game-event handler; the caller's job is to say the event
+  // happened, and it must not be made to wait on rule lookups, audience
+  // resolution and a dozen inserts to find out whether it is allowed to carry on.
+  // That is the same reason the C# side's `Emit()` enqueues and returns rather
+  // than touching the socket from the Core thread. `dispatch` catches everything
+  // internally and never rejects, and the `.catch` is the belt to that braces.
+  //
+  // The consequence a test has to know about: `emit` returns before the outbox
+  // rows exist. `engine.dispatch(event)` is exported for a caller that needs to
+  // await the delivery decision, and the tests use it directly.
+  engine.dispatch(event).catch((err) => log.error('dispatch rejected', { trigger: triggerId, message: err.message }))
+
   return { ok: true, event }
 }
 

server/src/utils/engagementWorker.js

@@ -0,0 +1,168 @@
+// ── Engagement outbox worker ───────────────────────────────────────────────
+//
+// ENGAGEMENT.md §4.2a, Phase 4a. Every ENGAGEMENT_POLL_MS it sweeps
+// `engagement_outbox` for rows whose `due_at` has passed, claims each one, hands
+// it to its channel, and records the outcome in `engagement_sends`. Same
+// setInterval + unref + stop() shape as `announceWorker` and the three Team
+// sweepers, wired into server.js start/shutdown beside them.
+//
+// **Claiming is a compare-and-set, not a lock** (§7.1 Q2, settled by the org lead
+// 2026-08-29 over `SELECT ... FOR UPDATE SKIP LOCKED`): an
+// `UPDATE ... SET status='sending' WHERE id=? AND status='scheduled'`, and the
+// instance the server reports `affectedRows = 1` to owns the row. No transaction
+// to hold open, no MariaDB version floor, and it uses a status the ENUM already
+// carried for exactly this. What it makes safe is the outbox; the four existing
+// workers are still single-instance, so this does not by itself make the
+// deployment multi-instance.
+//
+// **Nothing is delivered in this phase, and that is visible rather than
+// pretended.** A channel's `deliver` arrives with email in Phase 6 and the in-app
+// inbox in Phase 7; until then `channels.get(id)` has no such function, the row
+// finishes as `failed` and the send log says why in as many words. The
+// alternatives were both worse: recording 'sent' would be a lie in the one table
+// whose whole purpose is answering "did they get it", and leaving the row
+// scheduled would mean an IDOC warning queued today arriving three weeks later
+// on the deploy that first shipped a mailer.
+//
+// In practice this path is unreachable on a real deployment for now: core seeds
+// no rules and `enabled` defaults to 0, so the outbox stays empty until an
+// operator turns a rule on from the screen Phase 4b builds.
+
+const outboxDb = require('../model/engagement/engagementOutbox.db')
+const sendsDb = require('../model/engagement/engagementSends.db')
+const channels = require('../engagement/channels')
+const log = require('./logger')('engagement-worker')
+
+const POLL_MS = Number(process.env.ENGAGEMENT_POLL_MS) || 30_000
+// How many rows one sweep will look at. A bound rather than a target: the sweep
+// runs again in POLL_MS, and an unbounded batch is how a backlog turns one tick
+// into a stall.
+const BATCH = Number(process.env.ENGAGEMENT_BATCH) || 100
+
+// A transient failure is retried with a flat backoff, and then given up on.
+// Flat rather than exponential because `due_at` is also the grace window's clock
+// and a doubling backoff would push a delayed message arbitrarily far past the
+// moment it was about.
+const MAX_ATTEMPTS = 5
+const RETRY_MS = 5 * 60 * 1000
+
+// A row claimed into 'sending' by a process that then died is invisible to every
+// other sweeper - `status='scheduled'` will never match it again. This window is
+// how long a claim may look alive before it is taken back; it has to be
+// comfortably longer than the slowest legitimate send or a slow one gets sent
+// twice.
+const STALE_MS = 15 * 60 * 1000
+
+/**
+ * Deliver one claimed row.
+ *
+ * @returns {{ outcome: 'sent'|'retry'|'terminal', detail?: string, transport?: string }}
+ */
+async function deliver(row) {
+  const channel = channels.get(row.channel)
+  if (!channel) {
+    // The channel's module was removed between enqueue and now. Terminal: there
+    // is nothing to retry towards, and leaving the row scheduled would make it
+    // sweep forever.
+    return { outcome: 'terminal', detail: `channel "${row.channel}" is no longer registered` }
+  }
+  if (typeof channel.deliver !== 'function') {
+    return { outcome: 'terminal', detail: `channel "${row.channel}" has no delivery implementation yet` }
+  }
+  try {
+    const result = await channel.deliver(row)
+    if (result && result.ok) return { outcome: 'sent', transport: result.transport, detail: result.detail }
+    if (result && result.retry) return { outcome: 'retry', detail: result.detail || 'transient failure' }
+    return { outcome: 'terminal', detail: (result && result.detail) || 'delivery refused' }
+  } catch (err) {
+    // A channel shouldn't throw, but if one does it is a transient failure
+    // rather than a crashed tick - announceWorker's posture with its legs.
+    log.error('channel deliver threw', { outbox: row.id, channel: row.channel, message: err.message })
+    return { outcome: 'retry', detail: err.message }
+  }
+}
+
+/**
+ * Claim, deliver, record. One row, start to finish.
+ *
+ * `deliverFn` is injectable so a test can drive the retry/give-up path without a
+ * channel that fails on demand - the alternative is registering a fake channel,
+ * which would make the registry, not this function, the thing under test.
+ */
+async function processRow(row, now = new Date(), deliverFn = deliver) {
+  if (!(await outboxDb.claim(row.id))) return null // another sweeper got there first
+
+  const result = await deliverFn(row)
+
+  if (result.outcome === 'retry' && row.attempts + 1 < MAX_ATTEMPTS) {
+    await outboxDb.reschedule(row.id, new Date(now.getTime() + RETRY_MS), result.detail)
+    return 'retry'
+  }
+
+  const status = result.outcome === 'sent' ? 'sent' : 'failed'
+  await outboxDb.finish(row.id, status, status === 'failed' ? result.detail : null)
+  // The send log is written for every terminal outcome, not only success. G15's
+  // question is "did user X get the mail?", and "no, and here is why" is an
+  // answer that table has to be able to give.
+  await sendsDb.record({
+    outbox_id: row.id,
+    rule_id: row.rule_id,
+    trigger_id: row.trigger_id,
+    user_id: row.user_id,
+    channel: row.channel,
+    transport: result.transport ?? null,
+    status,
+    detail: result.detail ?? null,
+  })
+  return status
+}
+
+async function tick(now = new Date()) {
+  try {
+    await outboxDb.reclaimStale(new Date(now.getTime() - STALE_MS))
+  } catch (err) {
+    log.error('failed to reclaim stale rows', { message: err.message })
+  }
+
+  let due
+  try {
+    due = await outboxDb.findDue(now, BATCH)
+  } catch (err) {
+    log.error('failed to load due rows', { message: err.message })
+    return
+  }
+  if (!due || !due.length) return
+
+  const counts = { sent: 0, failed: 0, retry: 0, taken: 0 }
+  for (const row of due) {
+    try {
+      const outcome = await processRow(row, now)
+      if (outcome === null) counts.taken += 1
+      else counts[outcome] += 1
+    } catch (err) {
+      log.error('row failed', { outbox: row.id, message: err.message })
+    }
+  }
+  log.info('outbox swept', counts)
+}
+
+let timer = null
+
+function start() {
+  if (timer) return timer
+  timer = setInterval(() => {
+    tick().catch((err) => log.error('engagement tick failed', { message: err.message }))
+  }, POLL_MS)
+  if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown)
+  log.info('engagement outbox worker started', { pollMs: POLL_MS, batch: BATCH })
+  return timer
+}
+
+function stop() {
+  if (timer) {
+    clearInterval(timer)
+    timer = null
+  }
+}
+
+module.exports = { start, stop, tick, processRow, deliver, POLL_MS, MAX_ATTEMPTS, RETRY_MS, STALE_MS }

server/test/engagementEngine.test.js

@@ -0,0 +1,908 @@
+// ── The engagement engine (ENGAGEMENT.md Phase 4a) ─────────────────────────
+//
+// The phase's acceptance criteria, one test apiece:
+//
+//   • a trigger fired twice inside `cooldown_seconds` for the same
+//     (rule, user, subject) sends once
+//   • the same trigger for a DIFFERENT subject sends again — the multi-house
+//     case §4.1 names, which is the one a per-user cooldown gets wrong
+//   • a scheduled row is cancelled by a `cancel_on` trigger and never sends
+//   • a restart mid-window still sends exactly once
+//   • a duplicate `dedupe_key` is a successful no-op
+//
+// …plus the two properties that are security boundaries rather than behaviour:
+// the G24 ceiling is re-checked at SEND time and not only at save, and a composed
+// segment takes the NARROWEST ceiling in its tree.
+//
+// **The five tables are stubbed at the `.db` layer** and the engine's own logic
+// runs for real against them, the shape `notificationChannelPrefs.test.js` uses.
+// The one place that is not enough is the raw SQL whose correctness IS a server
+// contract - the cooldown claim, the outbox compare-and-set, and the scoped
+// dedupe key. Those run against a real MariaDB in `engagementEngineSql.test.js`,
+// which skips when there is none, and the first time it ran it disproved the
+// cooldown statement this file's stub had been agreeing with.
+//
+// Point the DB at a closed port before requiring anything: the registries reach
+// utils/discordAnnounce, which builds the pool at require time.
+process.env.DB_HOST = '127.0.0.1'
+process.env.DB_PORT = '59999'
+
+const { test, beforeEach, afterEach, after } = require('node:test')
+const assert = require('node:assert/strict')
+
+const registries = require('../src/modules/registries')
+const channels = require('../src/engagement/channels')
+const engine = require('../src/engagement/engine')
+const conditions = require('../src/engagement/conditions')
+const segments = require('../src/engagement/segments')
+const worker = require('../src/utils/engagementWorker')
+const rules = require('../src/model/engagement/engagementRules.model')
+const rulesDb = require('../src/model/engagement/engagementRules.db')
+const outboxDb = require('../src/model/engagement/engagementOutbox.db')
+const cooldownsDb = require('../src/model/engagement/engagementCooldowns.db')
+const sendsDb = require('../src/model/engagement/engagementSends.db')
+const segmentsDb = require('../src/model/engagement/engagementSegments.db')
+const recipients = require('../src/model/engagement/engagementRecipients.db')
+const db = require('../src/utils/db')
+
+after(() => db.close())
+
+const T0 = new Date('2026-08-29T12:00:00Z')
+const later = (ms) => new Date(T0.getTime() + ms)
+
+// ── In-memory stand-ins for the five tables ────────────────────────────────
+
+let store
+const originals = {}
+
+function snapshotOriginals() {
+  for (const [name, mod] of [
+    ['rulesDb', rulesDb], ['outboxDb', outboxDb], ['cooldownsDb', cooldownsDb],
+    ['sendsDb', sendsDb], ['segmentsDb', segmentsDb], ['recipients', recipients],
+  ]) {
+    originals[name] = { mod, fns: { ...mod } }
+  }
+}
+snapshotOriginals()
+
+function restoreOriginals() {
+  for (const { mod, fns } of Object.values(originals)) Object.assign(mod, fns)
+}
+
+function installStubs() {
+  store = {
+    rules: new Map(),
+    segments: new Map(),
+    cooldowns: new Map(),
+    outbox: new Map(),
+    sends: [],
+    users: new Map(), // id -> { id, role, status }
+    prefs: new Map(), // "<user> <id> <channel>" -> mode
+    nextOutboxId: 1,
+  }
+
+  rulesDb.enabledForTrigger = async (triggerId) =>
+    [...store.rules.values()].filter((r) => r.enabled && r.trigger_id === triggerId)
+  rulesDb.enabledCancelledBy = async (triggerId) =>
+    [...store.rules.values()].filter((r) => r.enabled && (r.cancel_on || []).includes(triggerId))
+  rulesDb.getById = async (id) => store.rules.get(id) || null
+  rulesDb.list = async () => [...store.rules.values()]
+  rulesDb.countUsingSegment = async (segmentId) =>
+    [...store.rules.values()].filter((r) => r.audience_segment_id === segmentId).length
+
+  segmentsDb.getById = async (id) => store.segments.get(id) || null
+  segmentsDb.list = async () => [...store.segments.values()]
+
+  // The two statements' semantics, reproduced: a guarded UPDATE that matches
+  // claims the fire; otherwise an INSERT IGNORE claims a first fire; otherwise
+  // the pair is still cooling. `engagementEngineSql.test.js` is what proves the
+  // SQL itself - a stub can only ever agree with whoever wrote it, and in this
+  // case the first version of both was wrong together.
+  cooldownsDb.claim = async (ruleId, userId, subjectKey, cooldownSeconds, now) => {
+    const key = `${ruleId}|${userId}|${subjectKey}`
+    const row = store.cooldowns.get(key)
+    if (!row) {
+      store.cooldowns.set(key, { last_fired_at: now, fire_count: 1 })
+      return true
+    }
+    if (row.last_fired_at.getTime() <= now.getTime() - cooldownSeconds * 1000) {
+      row.fire_count += 1
+      row.last_fired_at = now
+      return true
+    }
+    return false
+  }
+
+  outboxDb.enqueue = async (row) => {
+    if (row.dedupe_key) {
+      const clash = [...store.outbox.values()].find(
+        (r) =>
+          r.dedupe_key === row.dedupe_key &&
+          r.rule_id === row.rule_id &&
+          r.user_id === row.user_id &&
+          r.channel === row.channel,
+      )
+      if (clash) return null
+    }
+    const id = store.nextOutboxId++
+    store.outbox.set(id, { id, status: 'scheduled', attempts: 0, subject_key: '', ...row })
+    return id
+  }
+  // Copies, not the live objects: a SQL SELECT hands back a snapshot, and
+  // `processRow` reads `row.attempts` as the value BEFORE its own claim
+  // incremented it. Returning references here made the retry budget off by one
+  // in the stub only, which is exactly the class of thing a stub must not invent.
+  outboxDb.findDue = async (now, limit = 100) =>
+    [...store.outbox.values()]
+      .filter((r) => r.status === 'scheduled' && r.due_at <= now)
+      .sort((a, b) => a.due_at - b.due_at || a.id - b.id)
+      .slice(0, limit)
+      .map((r) => ({ ...r }))
+  outboxDb.claim = async (id) => {
+    const row = store.outbox.get(id)
+    if (!row || row.status !== 'scheduled') return false
+    row.status = 'sending'
+    row.attempts += 1
+    return true
+  }
+  outboxDb.reschedule = async (id, dueAt, error) => {
+    const row = store.outbox.get(id)
+    if (row && row.status === 'sending') Object.assign(row, { status: 'scheduled', due_at: dueAt, last_error: error })
+  }
+  outboxDb.finish = async (id, status, error) => {
+    const row = store.outbox.get(id)
+    if (row) Object.assign(row, { status, last_error: error })
+  }
+  outboxDb.cancel = async (ruleId, subjectKey, userId = null) => {
+    let n = 0
+    for (const row of store.outbox.values()) {
+      if (row.rule_id !== ruleId || row.subject_key !== subjectKey || row.status !== 'scheduled') continue
+      if (userId !== null && userId !== undefined && row.user_id !== userId) continue
+      row.status = 'cancelled'
+      n += 1
+    }
+    return n
+  }
+  outboxDb.reclaimStale = async () => {}
+  outboxDb.getById = async (id) => (store.outbox.has(id) ? { ...store.outbox.get(id) } : null)
+
+  sendsDb.record = async (entry) => {
+    store.sends.push({ id: store.sends.length + 1, created_at: T0, ...entry })
+    return store.sends.length
+  }
+  sendsDb.countSentSince = async (ruleId, since) =>
+    store.sends.filter((s) => s.rule_id === ruleId && s.status === 'sent' && s.created_at >= since).length
+
+  const activeIds = () => [...store.users.values()].filter((u) => u.status === 'active').map((u) => u.id)
+  recipients.active = async () => activeIds()
+  recipients.staff = async (roles) =>
+    [...store.users.values()].filter((u) => u.status === 'active' && roles.includes(u.role)).map((u) => u.id)
+  recipients.subscribers = async (streamId, defaultOn = []) =>
+    activeIds().filter((id) => {
+      const rows = [...store.prefs.entries()].filter(([k]) => k.startsWith(`${id} ${streamId} `))
+      if (rows.some(([, mode]) => mode !== 'off')) return true
+      const named = new Set(rows.map(([k]) => k.split(' ')[2]))
+      return defaultOn.some((c) => !named.has(c))
+    })
+  recipients.filterActive = async (ids) =>
+    [...new Set(ids)].filter((id) => store.users.get(id)?.status === 'active')
+  recipients.storedModes = async (userIds, streamId, channel) =>
+    new Map(
+      userIds
+        .filter((id) => store.prefs.has(`${id} ${streamId} ${channel}`))
+        .map((id) => [id, store.prefs.get(`${id} ${streamId} ${channel}`)]),
+    )
+}
+
+// ── Fixtures ───────────────────────────────────────────────────────────────
+
+const addUser = (id, over = {}) => store.users.set(id, { id, role: 'player', status: 'active', ...over })
+const optIn = (userId, streamId, channel, mode = 'instant') =>
+  store.prefs.set(`${userId} ${streamId} ${channel}`, mode)
+
+let nextRuleId = 1
+function addRule(over = {}) {
+  const id = nextRuleId++
+  const rule = {
+    id,
+    trigger_id: 'uo.house.idoc_warning',
+    name: `rule ${id}`,
+    enabled: true,
+    audience: 'owner',
+    audience_segment_id: null,
+    max_sends_per_hour: 100,
+    channels: ['email'],
+    template_keys: {},
+    conditions: null,
+    cooldown_seconds: 0,
+    delay_seconds: 0,
+    cancel_on: [],
+    ...over,
+  }
+  store.rules.set(id, rule)
+  return rule
+}
+
+/** A validated event envelope, the shape `engagementEmit.emit` builds. */
+const event = (over = {}) => ({
+  triggerId: 'uo.house.idoc_warning',
+  owner: 'uo',
+  version: 1,
+  subject: 'house-4001',
+  ownerUserId: 10,
+  dedupeKey: null,
+  occurredAt: T0.toISOString(),
+  data: { house: 'The Silver Anvil', decayStatus: 'IDOC' },
+  ...over,
+})
+
+/** Register a batch, the way the loader's second pass commits one. */
+function register(owner, fn) {
+  const api = registries.stage(owner)
+  fn(api)
+  registries.apply(api.staged)
+}
+
+const IDOC_TRIGGER = {
+  id: 'uo.house.idoc_warning',
+  label: 'House approaching collapse',
+  ceiling: 'owner',
+  audience: 'owner',
+  subjectKey: 'house',
+  variables: [
+    { name: 'house', type: 'string', required: true, example: 'The Silver Anvil' },
+    { name: 'decayStatus', type: 'string', required: false, example: 'IDOC' },
+  ],
+}
+
+function registerUoTrigger(over = {}) {
+  register('uo', (api) => api.registerEventTriggers([{ ...IDOC_TRIGGER, ...over }]))
+}
+
+const outboxRows = (filter = () => true) => [...store.outbox.values()].filter(filter)
+const scheduled = () => outboxRows((r) => r.status === 'scheduled')
+
+// Core's channels register through `coreChannels`, the way app.js does.
+// Requiring `channels` alone gets the empty map — that is the design, and the
+// engine dropping every rule because no channel is registered is what a
+// boot-order regression would look like.
+function registerChannels() {
+  channels._reset()
+  delete require.cache[require.resolve('../src/engagement/coreChannels')]
+  // eslint-disable-next-line global-require
+  require('../src/engagement/coreChannels')
+}
+
+beforeEach(() => {
+  registries._reset()
+  registerChannels()
+  installStubs()
+  nextRuleId = 1
+  registerUoTrigger()
+  addUser(10)
+  // Every channel is opt-IN (§7.1 Q1, and channels.js `defaultMode: 'off'`), so
+  // a fixture that wants mail to happen has to say so. The test below that turns
+  // this off again is the one asserting exactly that.
+  optIn(10, 'uo.house.idoc_warning', 'email')
+})
+
+afterEach(() => {
+  registries._reset()
+  restoreOriginals()
+})
+
+// ── Acceptance: cooldowns ──────────────────────────────────────────────────
+
+test('a trigger fired twice inside cooldown_seconds for the same (rule, user, subject) sends once', async () => {
+  addRule({ cooldown_seconds: 3600 })
+
+  await engine.dispatch(event(), T0)
+  await engine.dispatch(event(), later(60_000))
+
+  assert.equal(outboxRows().length, 1)
+})
+
+test('the same trigger for a DIFFERENT subject sends again — the multi-house case (§4.1)', async () => {
+  // The rule §4.1 warns about is "one IDOC mail per player per day": a player
+  // with four houses decaying should hear about all four, once each. Cooling on
+  // (rule, user) alone silently drops three of them, and this is the test that
+  // would fail if `subject_key` were ever dropped from the primary key.
+  addRule({ cooldown_seconds: 86_400 })
+
+  await engine.dispatch(event({ subject: 'house-4001' }), T0)
+  await engine.dispatch(event({ subject: 'house-4002' }), later(1000))
+  await engine.dispatch(event({ subject: 'house-4003' }), later(2000))
+  // …and the first house again, still inside the day.
+  await engine.dispatch(event({ subject: 'house-4001' }), later(3000))
+
+  assert.deepEqual(outboxRows().map((r) => r.subject_key).sort(), ['house-4001', 'house-4002', 'house-4003'])
+})
+
+test('a cooldown that has expired lets the same subject through again', async () => {
+  addRule({ cooldown_seconds: 60 })
+
+  await engine.dispatch(event(), T0)
+  await engine.dispatch(event(), later(61_000))
+
+  assert.equal(outboxRows().length, 2)
+})
+
+test('two rules on one trigger each get their own cooldown', async () => {
+  addRule({ cooldown_seconds: 3600 })
+  addRule({ cooldown_seconds: 3600 })
+
+  await engine.dispatch(event(), T0)
+
+  assert.equal(outboxRows().length, 2)
+})
+
+// ── Acceptance: dedupe ─────────────────────────────────────────────────────
+
+test('a duplicate dedupe_key is a successful no-op, not a second row and not an error', async () => {
+  addRule()
+
+  const first = await engine.dispatch(event({ dedupeKey: 'idoc:4001:2026-08-29' }), T0)
+  const replay = await engine.dispatch(event({ dedupeKey: 'idoc:4001:2026-08-29' }), later(1000))
+
+  assert.equal(first.enqueued, 1)
+  assert.equal(replay.enqueued, 0)
+  assert.equal(replay.deduped, 1)
+  assert.equal(outboxRows().length, 1)
+})
+
+test('one dedupe_key fans out to every recipient — the key is scoped, not global', async () => {
+  // §4.2a's `UNIQUE (dedupe_key)` was a defect: a dedupe key names the EVENT, and
+  // one event legitimately becomes one row per (rule, user, channel). A global
+  // unique index would have let the FIRST recipient's row in and silently dropped
+  // everyone else's, which is the opposite of what dedupe is for.
+  addUser(11)
+  addUser(12)
+  for (const id of [10, 11, 12]) {
+    optIn(id, 'uo.house.idoc_warning', 'email')
+    optIn(id, 'uo.house.idoc_warning', 'inapp')
+  }
+  registries._reset()
+  registerUoTrigger({ ceiling: 'subscribers', audience: 'subscribers' })
+  addRule({ audience: 'subscribers', channels: ['email', 'inapp'] })
+
+  const result = await engine.dispatch(event({ dedupeKey: 'idoc:4001' }), T0)
+
+  // three users x two channels
+  assert.equal(result.enqueued, 6)
+  assert.equal(new Set(outboxRows().map((r) => r.dedupe_key)).size, 1)
+})
+
+// ── Acceptance: the grace window and cancellation ──────────────────────────
+
+test('a scheduled row is cancelled by a cancel_on trigger and never sends', async () => {
+  register('uo', (api) =>
+    api.registerEventTriggers([
+      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', subjectKey: 'house' },
+    ]),
+  )
+  const rule = addRule({ delay_seconds: 1800, cancel_on: ['uo.house.repaired'] })
+
+  await engine.dispatch(event(), T0)
+  assert.equal(scheduled().length, 1)
+
+  const cancelled = await engine.dispatch(
+    event({ triggerId: 'uo.house.repaired', subject: 'house-4001' }),
+    later(60_000),
+  )
+  assert.equal(cancelled.cancelled, 1)
+
+  // The window has passed; the worker finds nothing to do.
+  await worker.tick(later(1_900_000))
+  assert.equal(store.outbox.get(1).status, 'cancelled')
+  assert.equal(store.sends.length, 0)
+  assert.equal(rule.id, 1)
+})
+
+test('a resolving event with no owner cancels every recipient queued about that subject', async () => {
+  register('uo', (api) =>
+    api.registerEventTriggers([
+      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', ceiling: 'authenticated', audience: 'authenticated' },
+    ]),
+  )
+  addUser(11)
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  register('uo', (api) =>
+    api.registerEventTriggers([
+      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', ceiling: 'authenticated', audience: 'authenticated' },
+    ]),
+  )
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  optIn(11, 'uo.house.idoc_warning', 'email')
+  addRule({ audience: 'authenticated', delay_seconds: 600, cancel_on: ['uo.house.repaired'] })
+
+  await engine.dispatch(event(), T0)
+  assert.equal(scheduled().length, 2)
+
+  await engine.dispatch(
+    event({ triggerId: 'uo.house.repaired', subject: 'house-4001', ownerUserId: null }),
+    later(1000),
+  )
+  assert.equal(scheduled().length, 0)
+})
+
+test('cancellation leaves an in-flight row alone', async () => {
+  register('uo', (api) =>
+    api.registerEventTriggers([{ ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired' }]),
+  )
+  addRule({ delay_seconds: 600, cancel_on: ['uo.house.repaired'] })
+  await engine.dispatch(event(), T0)
+
+  // A worker has claimed it: cancelling now would leave two writers on one row.
+  await outboxDb.claim(1)
+  const result = await engine.dispatch(event({ triggerId: 'uo.house.repaired' }), later(1000))
+
+  assert.equal(result.cancelled, 0)
+  assert.equal(store.outbox.get(1).status, 'sending')
+})
+
+// ── Acceptance: exactly once across a restart ──────────────────────────────
+
+test('a restart mid-window still sends exactly once', async () => {
+  addRule({ delay_seconds: 600 })
+  await engine.dispatch(event(), T0)
+
+  // "Restart" is the engine losing its process between enqueue and due_at. The
+  // outbox is the durable half, so the only question is whether the sweep after
+  // the restart double-delivers — and the CAS claim is what says it cannot.
+  await worker.tick(later(500_000)) // not yet due
+  assert.equal(store.sends.length, 0)
+
+  await worker.tick(later(700_000))
+  await worker.tick(later(700_001)) // a second instance, or the next tick
+  assert.equal(store.sends.length, 1)
+})
+
+test('two sweepers racing one due row: exactly one claim wins', async () => {
+  addRule()
+  await engine.dispatch(event(), T0)
+
+  // Both see the same candidate — findDue does not claim — and then disagree
+  // harmlessly about which of them owns it. §7.1 Q2's answer, as a test.
+  const [a] = await outboxDb.findDue(later(1000))
+  const [b] = await outboxDb.findDue(later(1000))
+  assert.equal(a.id, b.id)
+
+  assert.equal(await outboxDb.claim(a.id), true)
+  assert.equal(await outboxDb.claim(b.id), false)
+})
+
+// ── The send log ───────────────────────────────────────────────────────────
+
+test('a row whose channel has no deliver() finishes failed, and the send log says why', async () => {
+  // Phase 4a delivers nothing: `deliver` arrives with email in Phase 6 and the
+  // inbox in Phase 7. Recording 'sent' would be a lie in the one table whose
+  // purpose is answering "did they get it".
+  addRule()
+  await engine.dispatch(event(), T0)
+  await worker.tick(later(1000))
+
+  assert.equal(store.outbox.get(1).status, 'failed')
+  assert.equal(store.sends.length, 1)
+  assert.equal(store.sends[0].status, 'failed')
+  assert.match(store.sends[0].detail, /no delivery implementation/)
+  assert.equal(store.sends[0].user_id, 10)
+})
+
+test('a transient failure is retried, and then given up on', async () => {
+  addRule()
+  await engine.dispatch(event(), T0)
+
+  let attempts = 0
+  const failing = async () => {
+    attempts += 1
+    return { outcome: 'retry', detail: 'smtp timeout' }
+  }
+
+  let at = later(1000)
+  for (let i = 0; i < worker.MAX_ATTEMPTS + 2; i += 1) {
+    // eslint-disable-next-line no-await-in-loop
+    const [row] = await outboxDb.findDue(at)
+    if (!row) break
+    // eslint-disable-next-line no-await-in-loop
+    await worker.processRow(row, at, failing)
+    at = new Date(at.getTime() + worker.RETRY_MS + 1000)
+  }
+
+  // Tried MAX_ATTEMPTS times and then stopped, rather than retrying forever.
+  assert.equal(attempts, worker.MAX_ATTEMPTS)
+  assert.equal(store.outbox.get(1).status, 'failed')
+  assert.equal(store.sends.length, 1)
+  assert.match(store.sends[0].detail, /smtp timeout/)
+})
+
+// ── Preferences ────────────────────────────────────────────────────────────
+
+test("a user whose mode is 'off' for the channel is not enqueued", async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  addUser(11)
+  optIn(10, 'uo.house.idoc_warning', 'email', 'instant')
+  optIn(11, 'uo.house.idoc_warning', 'email', 'off')
+  addRule({ audience: 'authenticated' })
+
+  await engine.dispatch(event(), T0)
+
+  assert.deepEqual(outboxRows().map((r) => r.user_id), [10])
+})
+
+test('absence means the CHANNEL default, and all three of core default off', async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  store.prefs.clear()
+  addRule({ audience: 'authenticated' })
+
+  // Nobody has expressed anything, and email defaults 'off' (§3.1) — so an
+  // `authenticated` rule reaches nobody until people opt in. That is opt-IN
+  // working, not the engine failing.
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.enqueued, 0)
+  assert.equal(channels.defaultMode('email'), 'off')
+})
+
+test("a 'digest' preference still enqueues — batching is the drain's job, not the enqueue's", async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  optIn(10, 'uo.house.idoc_warning', 'email', 'digest')
+  addRule({ audience: 'authenticated' })
+
+  await engine.dispatch(event(), T0)
+  assert.equal(outboxRows().length, 1)
+})
+
+// ── The hourly ceiling (§7.1 Q3) ───────────────────────────────────────────
+
+test('a rule stops at its hourly send ceiling', async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  for (let i = 20; i < 30; i += 1) {
+    addUser(i)
+    optIn(i, 'uo.house.idoc_warning', 'email')
+  }
+  const rule = addRule({ audience: 'authenticated', max_sends_per_hour: 4 })
+
+  const result = await engine.dispatch(event(), T0)
+
+  // Eleven eligible recipients (the ten here plus the fixture's user 10), and a
+  // ceiling of four: four rows, and the rest are counted and dropped rather than
+  // queued for later - a rule at its ceiling is a rule an operator has to fix.
+  assert.equal(result.enqueued, 4)
+  assert.equal(result.enqueued + result.capped, 11)
+  assert.equal(rule.max_sends_per_hour, 4)
+})
+
+test('the hourly ceiling counts sends, not attempts', async () => {
+  // A broken transport must not silently consume a rule's whole budget and mute
+  // it: only rows the log records as 'sent' count against the ceiling.
+  const rule = addRule({ max_sends_per_hour: 2 })
+  store.sends.push({ rule_id: rule.id, status: 'failed', created_at: T0 })
+  store.sends.push({ rule_id: rule.id, status: 'suppressed', created_at: T0 })
+
+  const result = await engine.dispatch(event(), later(1000))
+  assert.equal(result.enqueued, 1)
+})
+
+// ── Ceilings: the security boundary, both halves ───────────────────────────
+
+test('a rule may not be SAVED with an audience wider than its trigger permits', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning', // ceiling: owner
+    name: 'IDOC warning',
+    channels: ['email'],
+    audience: 'authenticated',
+  })
+
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /wider than trigger/)
+})
+
+test('the ceiling is re-checked at SEND time, so a module narrowing its declaration stops a saved rule', async () => {
+  // The only way this can fail is the case it exists for: the rule was saved
+  // when the trigger permitted `authenticated`, and a module upgrade has since
+  // narrowed the declaration to `owner`. A save-time check alone would keep
+  // mailing the wider set forever.
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  addUser(11)
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  optIn(11, 'uo.house.idoc_warning', 'email')
+  addRule({ audience: 'authenticated' })
+
+  const before = await engine.dispatch(event(), T0)
+  assert.equal(before.enqueued, 2)
+
+  registries._reset()
+  registerUoTrigger({ ceiling: 'owner', audience: 'owner' }) // the upgrade
+
+  const after2 = await engine.dispatch(event({ subject: 'house-9' }), later(1000))
+  assert.equal(after2.enqueued, 0)
+})
+
+test('a rule for an unregistered trigger is dormant, not deleted and not an error', async () => {
+  const rule = addRule({ trigger_id: 'uo.gone.away' })
+  const listed = await rules.listAnnotated()
+  const found = listed.find((r) => r.id === rule.id)
+
+  assert.equal(found.dormant, true)
+  assert.match(found.dormantReasons.join(' '), /not registered/)
+})
+
+// ── Segments (§5.1a) ───────────────────────────────────────────────────────
+
+function registerAudiences() {
+  register('uo', (api) =>
+    api.registerAudiences([
+      { id: 'uo.team.members', label: 'Team members', ceiling: 'members', params: [{ id: 'teamId', type: 'int', required: true }], resolve: async ({ teamId }) => (teamId === 1 ? [10, 11] : [12]) },
+      { id: 'uo.governors', label: 'Governors', ceiling: 'members', resolve: async () => [11, 12] },
+      { id: 'uo.watchers', label: 'Watchers', ceiling: 'authenticated', resolve: async () => [10, 13] },
+      { id: 'uo.flagged', label: 'Flagged accounts', ceiling: 'staff', resolve: async () => [10] },
+    ]),
+  )
+}
+
+test('OR takes the TIGHTER ceiling — union-widens is the wrong implementation', async () => {
+  registerAudiences()
+  const checked = segments.validate({
+    op: 'or',
+    nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.watchers' }],
+  })
+
+  assert.equal(checked.ok, true)
+  // members is below authenticated, so the meet is members — NOT authenticated,
+  // which is what a "widest wins" reading would have given.
+  assert.equal(checked.ceiling, 'members')
+})
+
+test('two incomparable ceilings are refused rather than resolved to a guess', async () => {
+  registerAudiences()
+  const checked = segments.validate({
+    op: 'and',
+    nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.flagged' }],
+  })
+
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /no common ceiling/)
+})
+
+test('NOT does not constrain the ceiling — excluding people cannot widen', async () => {
+  registerAudiences()
+  // `members AND NOT staff` reaches strictly fewer people than `members`. If the
+  // complement's ceiling were folded into the meet, meet('members','staff') is
+  // null and this safe segment would be refused.
+  const checked = segments.validate({
+    op: 'and',
+    nodes: [{ audienceId: 'uo.governors' }, { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }],
+  })
+
+  assert.equal(checked.ok, true)
+  assert.equal(checked.ceiling, 'members')
+})
+
+test('NOT outside an AND is refused — a complement needs a set to take it from', async () => {
+  registerAudiences()
+  for (const expression of [
+    { op: 'not', nodes: [{ audienceId: 'uo.governors' }] },
+    { op: 'or', nodes: [{ audienceId: 'uo.governors' }, { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }] },
+  ]) {
+    const checked = segments.validate(expression)
+    assert.equal(checked.ok, false)
+    assert.match(checked.errors.join(' '), /only allowed inside an "and"/)
+  }
+})
+
+test('a segment resolves through the module resolvers, and AND NOT subtracts', async () => {
+  registerAudiences()
+  const checked = segments.validate({
+    op: 'and',
+    nodes: [
+      { audienceId: 'uo.team.members', params: { teamId: 1 } }, // [10, 11]
+      { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }, // [10]
+    ],
+  })
+  const resolved = await segments.resolve(checked.expression)
+
+  assert.equal(resolved.dormant, false)
+  assert.deepEqual(resolved.userIds, [11])
+})
+
+test('a segment whose module is uninstalled is DORMANT and sends to nobody', async () => {
+  registerAudiences()
+  const checked = segments.validate({ op: 'or', nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.watchers' }] })
+  store.segments.set(1, { id: 1, name: 'staff-ish', expression: checked.expression, ceiling: 'members' })
+  addUser(11)
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  optIn(11, 'uo.house.idoc_warning', 'email')
+  registries._reset()
+  registerUoTrigger({ ceiling: 'members', audience: 'members' })
+  addRule({ audience: 'members', audience_segment_id: 1 })
+
+  // The module is gone: `resolveAudience` answers dormant + empty, and the rule
+  // must NOT fall back to anything. Reaching a different population than the one
+  // composed is the failure §5.1a rule 4 forbids.
+  const result = await engine.dispatch(event(), T0)
+
+  assert.equal(result.enqueued, 0)
+  assert.equal(outboxRows().length, 0)
+})
+
+test('a rule pointing at a deleted segment is dormant, never a fallback to its plain audience', async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  addRule({ audience: 'authenticated', audience_segment_id: 99 }) // no such segment
+
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.enqueued, 0)
+})
+
+test("a plain 'members' audience with no segment reaches nobody", async () => {
+  registries._reset()
+  registerUoTrigger({ ceiling: 'members', audience: 'members' })
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  addRule({ audience: 'members' })
+
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.enqueued, 0)
+})
+
+// ── Conditions ─────────────────────────────────────────────────────────────
+
+test('a condition narrows which firings are interesting', async () => {
+  addRule({
+    conditions: { variable: 'decayStatus', cmp: 'in', value: ['Greatly damaged', 'IDOC'] },
+  })
+
+  await engine.dispatch(event({ data: { house: 'A', decayStatus: 'IDOC' } }), T0)
+  await engine.dispatch(event({ subject: 'house-2', data: { house: 'B', decayStatus: 'LikeNew' } }), later(1000))
+
+  assert.equal(outboxRows().length, 1)
+})
+
+test('a condition naming a variable the trigger does not declare is refused at save, with the name', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning',
+    name: 'typo',
+    channels: ['email'],
+    audience: 'owner',
+    conditions: { variable: 'decaystatus', cmp: 'eq', value: 'IDOC' },
+  })
+
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /"decaystatus" is not a variable/)
+})
+
+test('an absent variable makes every comparison false — including "is not"', async () => {
+  // `ne` is the one that tempts otherwise: "not equal to IDOC" reads as satisfied
+  // by nothing at all, and treating it that way would fire the rule on every
+  // event that omits an optional variable.
+  const c = { variable: 'decayStatus', cmp: 'ne', value: 'IDOC' }
+  assert.equal(conditions.evaluate(c, { house: 'A' }), false)
+  assert.equal(conditions.evaluate(c, { house: 'A', decayStatus: 'LikeNew' }), true)
+  assert.equal(conditions.evaluate({ variable: 'decayStatus', cmp: 'absent' }, { house: 'A' }), true)
+})
+
+test('a condition tree that no longer parses fails CLOSED', async () => {
+  // A stored condition that stops making sense must stop the mail, not decay
+  // into "no conditions" and reach everyone the rule could ever reach.
+  assert.equal(conditions.evaluate({ op: 'xor', nodes: [] }, {}), false)
+  assert.equal(conditions.evaluate('nonsense', {}), false)
+  assert.equal(conditions.evaluate(null, {}), true)
+})
+
+test('and / or / not compose', () => {
+  const data = { house: 'The Silver Anvil', decayStatus: 'IDOC' }
+  assert.equal(
+    conditions.evaluate(
+      { op: 'and', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'IDOC' }, { variable: 'house', cmp: 'contains', value: 'Silver' }] },
+      data,
+    ),
+    true,
+  )
+  assert.equal(
+    conditions.evaluate({ op: 'not', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'IDOC' }] }, data),
+    false,
+  )
+  assert.equal(
+    conditions.evaluate(
+      { op: 'or', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'LikeNew' }, { variable: 'house', cmp: 'startsWith', value: 'The' }] },
+      data,
+    ),
+    true,
+  )
+})
+
+test('an operator cannot be applied to a type it does not fit', () => {
+  const declaration = registries.eventTrigger('uo.house.idoc_warning')
+  const checked = conditions.validate(declaration, { variable: 'house', cmp: 'gt', value: 'x' })
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /cannot be applied to a string/)
+})
+
+// ── Rule validation, the rest ──────────────────────────────────────────────
+
+test('a new rule is created disabled unless it says otherwise (§7.1 Q3)', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning',
+    name: 'IDOC warning',
+    channels: ['email'],
+    audience: 'owner',
+  })
+  assert.equal(checked.ok, true)
+  assert.equal(checked.rule.enabled, false)
+  assert.equal(checked.rule.max_sends_per_hour, 100)
+})
+
+test('a rule naming an unregistered channel is refused', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning',
+    name: 'IDOC warning',
+    channels: ['carrier-pigeon'],
+    audience: 'owner',
+  })
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /no channel "carrier-pigeon"/)
+})
+
+test('the hourly ceiling has a hard upper bound an operator cannot type past', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning',
+    name: 'IDOC warning',
+    channels: ['email'],
+    audience: 'owner',
+    maxSendsPerHour: 10_000_000,
+  })
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /maxSendsPerHour/)
+})
+
+test('cancelOn without a delay is refused — there is no window to cancel in', async () => {
+  const checked = await rules.validate({
+    triggerId: 'uo.house.idoc_warning',
+    name: 'IDOC warning',
+    channels: ['email'],
+    audience: 'owner',
+    cancelOn: ['uo.house.repaired'],
+  })
+  assert.equal(checked.ok, false)
+  assert.match(checked.errors.join(' '), /no effect without a delaySeconds/)
+})
+
+test('a rule whose channel was removed is dormant but still editable', async () => {
+  const rule = addRule({ channels: ['email', 'carrier-pigeon'] })
+  const listed = await rules.listAnnotated()
+  assert.equal(listed.find((r) => r.id === rule.id).dormant, true)
+
+  // …and only the live channel is used when it fires.
+  optIn(10, 'uo.house.idoc_warning', 'email')
+  await engine.dispatch(event(), T0)
+  assert.deepEqual([...new Set(outboxRows().map((r) => r.channel))], ['email'])
+})
+
+// ── The dispatch contract ──────────────────────────────────────────────────
+
+test('dispatch never throws at its caller, even when the database is gone', async () => {
+  addRule()
+  rulesDb.enabledForTrigger = async () => {
+    throw new Error('connection lost')
+  }
+
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.enqueued, 0)
+})
+
+test('an event nobody has written a rule for is a no-op', async () => {
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.rules, 0)
+  assert.equal(outboxRows().length, 0)
+})
+
+test('a disabled rule does not fire', async () => {
+  addRule({ enabled: false })
+  const result = await engine.dispatch(event(), T0)
+  assert.equal(result.rules, 0)
+})

server/test/engagementEngineSql.test.js

@@ -0,0 +1,283 @@
+// ── The engine's raw SQL, against a real MariaDB ───────────────────────────
+//
+// ENGAGEMENT.md Phase 4a. `engagementEngine.test.js` stubs the five tables and
+// exercises the engine's logic against in-memory stand-ins, which is the right
+// shape for everything the engine DECIDES. It cannot prove the three statements
+// whose whole correctness is a server contract:
+//
+//   • the cooldown claim's answer is read out of `affectedRows`, and what that
+//     number MEANS depends on the pool's `foundRows` setting. This file is what
+//     found that: §4.1's single `INSERT ... ON DUPLICATE KEY UPDATE` was written,
+//     was green against the 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. A cooldown that never cools.
+//   • the outbox claim is a compare-and-set (§7.1 Q2), and "exactly one winner"
+//     is `affectedRows = 1` for one caller and 0 for the other.
+//   • `uq_engo_dedupe` is scoped to (rule, user, channel, dedupe_key), so one
+//     event's key fans out to every recipient instead of admitting the first.
+//
+// A stub that reproduces those from the same reading of the manual proves the
+// reading, not the server. So this file talks to a real database.
+//
+// **It SKIPS when there is none**, and that is deliberate rather than lax: CI
+// runs the suite without a database (the harness points the pool 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=3307 DB_USER=... DB_PASSWORD=... DB_NAME=... \
+//     node --test test/engagementEngineSql.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 } = require('node:test')
+const assert = require('node:assert/strict')
+const mariadb = require('mariadb')
+
+const SCHEMA = `
+CREATE TABLE engagement_cooldowns (
+  rule_id      INT          NOT NULL,
+  user_id      INT          NOT NULL,
+  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
+  last_fired_at DATETIME    NOT NULL,
+  fire_count   INT          NOT NULL DEFAULT 1,
+  PRIMARY KEY (rule_id, user_id, subject_key)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+CREATE TABLE engagement_outbox (
+  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
+  rule_id      INT          NOT NULL,
+  trigger_id   VARCHAR(96)  NOT NULL,
+  user_id      INT          NOT NULL,
+  channel      VARCHAR(32)  NOT NULL,
+  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
+  payload      JSON         NOT NULL,
+  dedupe_key   VARCHAR(190) NULL,
+  status       ENUM('scheduled','sending','sent','failed','cancelled','suppressed') NOT NULL DEFAULT 'scheduled',
+  due_at       DATETIME     NOT NULL,
+  attempts     SMALLINT     NOT NULL DEFAULT 0,
+  last_error   TEXT         NULL,
+  created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  sent_at      DATETIME     NULL,
+  UNIQUE KEY uq_engo_dedupe (rule_id, user_id, channel, dedupe_key),
+  INDEX idx_engo_due (status, due_at),
+  INDEX idx_engo_cancel (rule_id, user_id, subject_key, status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+`
+
+const DB = `rg_engage_test_${process.pid}`
+let pool = null
+let available = false
+
+// The statements under test, verbatim from the two `.db` files. They are
+// duplicated here 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 to `false` here would make this file agree with
+// the code by construction and prove nothing about the pool the server runs.
+const CLAIM_COOLDOWN_UPDATE = `
+UPDATE engagement_cooldowns
+   SET last_fired_at = ?, fire_count = fire_count + 1
+ WHERE rule_id = ? AND user_id = ? AND subject_key = ?
+   AND last_fired_at <= ? - INTERVAL ? SECOND`
+
+const CLAIM_COOLDOWN_INSERT = `
+INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, last_fired_at, fire_count)
+VALUES (?, ?, ?, ?, 1)`
+
+const CLAIM_OUTBOX = `
+UPDATE engagement_outbox SET status = 'sending', attempts = attempts + 1
+ WHERE id = ? AND status = 'scheduled'`
+
+const ENQUEUE = `
+INSERT IGNORE INTO engagement_outbox
+  (rule_id, trigger_id, user_id, channel, subject_key, payload, dedupe_key, due_at)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+
+before(async () => {
+  const admin = mariadb.createPool({
+    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 || '',
+    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({
+    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 || '',
+    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-08-29T12:00:00Z')
+const later = (ms) => new Date(T0.getTime() + ms)
+
+const claimCooldown = async (ruleId, userId, subject, seconds, now) => {
+  const moved = await pool.query(CLAIM_COOLDOWN_UPDATE, [now, ruleId, userId, subject, now, seconds])
+  if (Number(moved.affectedRows) === 1) return true
+  const inserted = await pool.query(CLAIM_COOLDOWN_INSERT, [ruleId, userId, subject, now])
+  return Number(inserted.affectedRows) === 1
+}
+
+// ── The cooldown claim ─────────────────────────────────────────────────────
+
+test('the cooldown claim: first fire inserts and is allowed', async (t) => {
+  if (needDb(t)) return
+  assert.equal(await claimCooldown(1, 10, 'h1', 3600, T0), true)
+})
+
+test('the cooldown claim: a second fire inside the window is REFUSED', async (t) => {
+  if (needDb(t)) return
+  await claimCooldown(2, 10, 'h1', 3600, T0)
+  // affectedRows = 0: a duplicate key whose update changed nothing.
+  assert.equal(await claimCooldown(2, 10, 'h1', 3600, later(60_000)), false)
+})
+
+test('the cooldown claim: a fire after the window is allowed, and counts', async (t) => {
+  if (needDb(t)) return
+  await claimCooldown(3, 10, 'h1', 60, T0)
+  assert.equal(await claimCooldown(3, 10, 'h1', 60, later(61_000)), true)
+  const [row] = await pool.query('SELECT fire_count FROM engagement_cooldowns WHERE rule_id = 3')
+  assert.equal(Number(row.fire_count), 2)
+})
+
+test('the cooldown claim: the refusal survives foundRows — the bug this file caught', async (t) => {
+  if (needDb(t)) return
+  // The regression, named. `foundRows: true` (the connector's default, and what
+  // `utils/db.js` gets) makes `affectedRows` count MATCHED rows, so the
+  // ON DUPLICATE KEY UPDATE form's "0 means still cooling" reading returns 1 and
+  // every send is allowed. Guarding in a WHERE clause is what makes the number
+  // mean one thing.
+  const seed = 'INSERT INTO engagement_cooldowns VALUES (7, 10, "h1", ?, 1)'
+  await pool.query(seed, [T0])
+  const noop = await pool.query(`${seed} ON DUPLICATE KEY UPDATE fire_count = fire_count`, [T0])
+  assert.equal(Number(noop.affectedRows), 1, 'a no-op ODKU reports 1 under foundRows, not 0')
+
+  // …and the shipped claim still refuses.
+  assert.equal(await claimCooldown(7, 10, 'h1', 3600, later(60_000)), false)
+})
+
+test('the cooldown claim: repeated expiries keep counting', async (t) => {
+  if (needDb(t)) return
+  // `fire_count` is moved by the same guarded UPDATE that moves `last_fired_at`,
+  // so a claim that succeeded and a claim that counted can never disagree.
+  await claimCooldown(4, 10, 'h1', 10, T0)
+  for (let i = 1; i <= 3; i += 1) await claimCooldown(4, 10, 'h1', 10, later(i * 11_000))
+  const [row] = await pool.query('SELECT fire_count FROM engagement_cooldowns WHERE rule_id = 4')
+  assert.equal(Number(row.fire_count), 4)
+})
+
+test('the cooldown claim: a different subject is a different row', async (t) => {
+  if (needDb(t)) return
+  assert.equal(await claimCooldown(5, 10, 'house-1', 86_400, T0), true)
+  assert.equal(await claimCooldown(5, 10, 'house-2', 86_400, later(1000)), true)
+  assert.equal(await claimCooldown(5, 10, 'house-1', 86_400, later(2000)), false)
+})
+
+test('the cooldown claim: cooldown_seconds = 0 always passes', async (t) => {
+  if (needDb(t)) return
+  assert.equal(await claimCooldown(6, 10, '', 0, T0), true)
+  assert.equal(await claimCooldown(6, 10, '', 0, later(1)), true)
+})
+
+// ── The outbox claim and the dedupe key ────────────────────────────────────
+
+const enqueue = async (over = {}) => {
+  const row = {
+    rule_id: 1, trigger_id: 'uo.house.idoc_warning', user_id: 10, channel: 'email',
+    subject_key: 'h1', dedupe_key: null, due_at: T0, ...over,
+  }
+  const r = await pool.query(ENQUEUE, [
+    row.rule_id, row.trigger_id, row.user_id, row.channel, row.subject_key,
+    JSON.stringify({ house: 'A' }), row.dedupe_key, row.due_at,
+  ])
+  return Number(r.affectedRows) === 1 ? Number(r.insertId) : null
+}
+
+test('the outbox claim: exactly one of two callers wins (§7.1 Q2)', async (t) => {
+  if (needDb(t)) return
+  const id = await enqueue({ rule_id: 20 })
+  const a = await pool.query(CLAIM_OUTBOX, [id])
+  const b = await pool.query(CLAIM_OUTBOX, [id])
+  assert.equal(Number(a.affectedRows), 1)
+  assert.equal(Number(b.affectedRows), 0)
+  const [row] = await pool.query('SELECT status, attempts FROM engagement_outbox WHERE id = ?', [id])
+  assert.equal(row.status, 'sending')
+  assert.equal(Number(row.attempts), 1)
+})
+
+test('the outbox claim under real concurrency: one winner, however many racers', async (t) => {
+  if (needDb(t)) return
+  const id = await enqueue({ rule_id: 21 })
+  // Fired at once on separate pooled connections, so the server - not the
+  // JavaScript event loop's ordering - is what serialises them.
+  const results = await Promise.all([1, 2, 3, 4, 5].map(() => pool.query(CLAIM_OUTBOX, [id])))
+  assert.equal(results.filter((r) => Number(r.affectedRows) === 1).length, 1)
+})
+
+test('a replayed event with the same dedupe key is IGNOREd, not duplicated', async (t) => {
+  if (needDb(t)) return
+  const first = await enqueue({ rule_id: 30, dedupe_key: 'idoc:4001' })
+  const replay = await enqueue({ rule_id: 30, dedupe_key: 'idoc:4001' })
+  assert.ok(first)
+  assert.equal(replay, null)
+})
+
+test('ONE dedupe key fans out to every recipient — the scoped unique key', async (t) => {
+  if (needDb(t)) return
+  // §4.2a's global `UNIQUE (dedupe_key)` would have admitted the first of these
+  // and silently ignored the other five: fifty recipients would have become one.
+  const ids = []
+  for (const user of [10, 11, 12]) {
+    for (const channel of ['email', 'inapp']) {
+      ids.push(await enqueue({ rule_id: 31, user_id: user, channel, dedupe_key: 'idoc:4001' }))
+    }
+  }
+  assert.equal(ids.filter(Boolean).length, 6)
+})
+
+test('a NULL dedupe key never collides — many NULLs are legal under a UNIQUE index', async (t) => {
+  if (needDb(t)) return
+  const a = await enqueue({ rule_id: 32, dedupe_key: null })
+  const b = await enqueue({ rule_id: 32, dedupe_key: null })
+  assert.ok(a && b && a !== b)
+})