feat(engagement): the rules engine, cooldowns and outbox (engagement Phase 4a)

Phase 4 of docs/website/ENGAGEMENT.md, split 4a/4b at the org lead's direction.
This is 4a: the engine, server only, with no HTTP surface at all. A fired trigger
now produces outbox rows and send-log entries; Admin - Engagement - Rules and the
segment composition UI are 4b.

Five tables (rules, audience segments, cooldowns, outbox, sends), the sweep
worker, audience resolution, condition evaluation, the grace window and its
cancellation, and the save-path validation 4b's form will call. engagementEmit's
Phase 2 log line becomes the engine call.

Two settled questions this phase was blocked on:

  Q2 (multi-instance) - neither SKIP LOCKED nor documented single-instance: the
  outbox claims each row with a compare-and-set into the 'sending' state the ENUM
  already carried. It makes the outbox safe for two instances, not the deployment.

  Q4 (admin surface) - its own top-level nav group, built in 4b.

Two defects in the plan's own section 4, both found by building it:

  The global UNIQUE(dedupe_key) was data loss. A dedupe key names the EVENT, and
  one event is one row per (rule, user, channel) - so a fifty-person audience
  would have had one row admitted and forty-nine silently ignored. Scoped.

  Section 4.1's single INSERT ... ON DUPLICATE KEY UPDATE cooldown claim always
  passes against this codebase's pool: the mariadb connector defaults
  foundRows:true, so a no-op update reports affectedRows 1 rather than 0. It is
  two statements now, with the interval guard in a WHERE clause.

The second defect is why there is a second test file. The stubbed suite was green
against the broken claim, because a stub can only agree with whoever wrote it;
engagementEngineSql.test.js runs the raw statements against a real MariaDB and
skips when there is none.

Verification: 43 new tests green in engagementEngine.test.js, 12 more against
MariaDB 11.8, and the whole path exercised end to end against a live database -
per-subject cooldowns, conditions, the CAS claim, the send log's honest failure
detail, and dormancy on uninstall. The three pre-existing Windows-only CRLF
failures in the generated-artifact tests are unchanged from clean edge.

Co-Authored-By: Claude <noreply@anthropic.com>

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 }