feat(events): conditions, phase advancement and the diagnosis panel (Phase 5)
A phase used to advance on one fact - every step terminal. It can now also carry
an advance CONDITION: `{ after: '30m' }` or `{ on: '<triggerId>', where:
<conditions>, count: n }`, reusing `engagement/conditions.js` unchanged. The
phase's real deliverable is the diagnosis panel: "why didn't phase 3 start?"
answered in the condition builder's own words, with the tally, the elapsed time
and the last related firing whether or not it counted.
`POST /admin/events/runs/:runId/advance` arrives beside it. It has been absent
since Phase 3 for want of a meaning; a phase with a gate can wait on a boss that
will never spawn, and that is the one state "force it anyway" names.
One new table, `event_run_phase_gates`. The emit path writes the tally at the
moment a firing happens - a gate waiting on three spawns counts things that
occur between two ticks, and a tally held in a process's memory is one a restart
silently zeroes - and the runner's tick reads it.
A gate that never opens is HELD, with no automatic advance and no authored
timeout (org lead, 2026-09-02). What the engine owes instead is visibility:
`EVENT_PHASE_STALL_MS` takes the run's health to `stalled`, and `setHealth` is
now escalation-only so a later retry cannot demote it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T6t8mrAWhZU5vnyYgZTMtL
This commit is contained in:
244
server/src/events/gates.js
Normal file
244
server/src/events/gates.js
Normal file
@@ -0,0 +1,244 @@
|
||||
// ── Phase advance gates: the emit-path observer, and the words for the panel ─
|
||||
//
|
||||
// EVENTS.md §E and § Observability, and Phase 5 of EVENTS_PLAN.md. Two things
|
||||
// live here because they are two halves of one claim — that an operator can
|
||||
// answer *"why didn't phase 3 start?"* without reading a server log:
|
||||
//
|
||||
// `observe(event)` — the trigger stream's other subscriber. Beside
|
||||
// `engine.dispatch`, on the same seam, with the same
|
||||
// fire-and-forget posture.
|
||||
// `describe(gate)` — the same gate rendered in the condition builder's own
|
||||
// words, which is what the diagnosis panel shows.
|
||||
//
|
||||
// **Why the counting happens here and not on the runner's tick.** A gate that
|
||||
// waits for three boss spawns is counting things that happen *between* ticks. A
|
||||
// poller cannot count them: fifteen seconds after the third spawn there is
|
||||
// nothing left to observe, and a tally kept in a process's memory is a tally a
|
||||
// restart silently sets back to zero — with the phase then waiting for three
|
||||
// more of something that already happened. So the emit path writes, and the tick
|
||||
// reads. That division is the whole design of this file.
|
||||
//
|
||||
// **The cost of that, and its bound.** Every game event of every trigger some
|
||||
// run is waiting on costs one indexed lookup, and the common answer is zero
|
||||
// rows. Only when a gate is open does anything else happen, and then it is one
|
||||
// UPDATE per open gate — bounded by how many runs can be waiting on one trigger
|
||||
// at once, which is bounded by how many runs exist.
|
||||
//
|
||||
// **The words are rendered here, on the server, not in the client.** The panel's
|
||||
// entire value is that it reads the way the condition builder reads — `gte` as
|
||||
// *"is at least"*, `present` as *"is present"* — and those labels are defined in
|
||||
// `engagement/conditions.js`. A renderer in the browser would be a second
|
||||
// implementation of a grammar the server owns, and the first operator to meet a
|
||||
// clause it spelled differently would be the operator diagnosing a stalled run
|
||||
// at two in the morning.
|
||||
|
||||
const gatesDb = require('../model/events/eventPhaseGates.db')
|
||||
const runsDb = require('../model/events/eventRuns.db')
|
||||
const logDb = require('../model/events/eventRunLog.db')
|
||||
const conditions = require('../engagement/conditions')
|
||||
const log = require('../utils/logger')('event-gates')
|
||||
|
||||
// How long an `on` gate may wait before the run is called `stalled` (§E's third
|
||||
// health value, which nothing had ever written before this phase). It is a
|
||||
// VISIBILITY threshold and not a timeout: nothing advances, nothing fails, and
|
||||
// the operator decides. An hour is long enough that a champion spawn nobody has
|
||||
// killed yet is not an alarm, and short enough that a run which will wait for
|
||||
// ever is on the screen inside one shift.
|
||||
//
|
||||
// It does not apply to an `after` gate. A phase waiting out six hours it was
|
||||
// authored to wait is not stalled, it is working, and health that said otherwise
|
||||
// would train an operator to ignore it.
|
||||
const STALL_MS = Number(process.env.EVENT_PHASE_STALL_MS) || 60 * 60 * 1000
|
||||
|
||||
/** Every variable a condition tree names, in the order it names them. */
|
||||
function variablesIn(node, out = []) {
|
||||
if (!node || typeof node !== 'object') return out
|
||||
if (Array.isArray(node.nodes)) {
|
||||
node.nodes.forEach((child) => variablesIn(child, out))
|
||||
return out
|
||||
}
|
||||
if (node.variable && !out.includes(node.variable)) out.push(node.variable)
|
||||
return out
|
||||
}
|
||||
|
||||
const quote = (v) => (typeof v === 'string' ? `"${v}"` : String(v))
|
||||
|
||||
/**
|
||||
* One condition tree as a sentence, using the grammar's own operator labels.
|
||||
*
|
||||
* `null` answers null rather than "always": the caller renders *"on any
|
||||
* `uo.champ.boss_up`"* for a gate with no predicate, and a phrase saying
|
||||
* "everything is true" would be a clause an operator has to read past.
|
||||
*/
|
||||
function phrase(node) {
|
||||
if (!node || typeof node !== 'object') return null
|
||||
if (node.op === 'not') {
|
||||
const inner = phrase((node.nodes || [])[0])
|
||||
return inner ? `not (${inner})` : null
|
||||
}
|
||||
if (node.op === 'and' || node.op === 'or') {
|
||||
const parts = (node.nodes || []).map(phrase).filter(Boolean)
|
||||
if (!parts.length) return null
|
||||
// Parenthesised only where it changes the reading. A flat `and` of three
|
||||
// comparisons is a sentence; the same three wrapped in brackets is a
|
||||
// diagnosis an operator has to parse rather than read.
|
||||
const joined = parts.map((p) => (p.includes(' or ') || p.includes(' and ') ? `(${p})` : p))
|
||||
return joined.join(node.op === 'and' ? ' and ' : ' or ')
|
||||
}
|
||||
const operator = conditions.OPERATORS[node.cmp]
|
||||
if (!operator) return null
|
||||
if (operator.arity === 0) return `${node.variable} ${operator.label}`
|
||||
if (operator.arity === 'list') return `${node.variable} ${operator.label} ${node.value.map(quote).join(', ')}`
|
||||
return `${node.variable} ${operator.label} ${quote(node.value)}`
|
||||
}
|
||||
|
||||
/**
|
||||
* The shape the run console renders — a gate as an operator reads it.
|
||||
*
|
||||
* Derived rather than stored, every field of it. `stalled` in particular is a
|
||||
* comparison against the clock and not a column: a threshold that had been
|
||||
* written into a row at entry could not be changed by an operator raising
|
||||
* `EVENT_PHASE_STALL_MS`, and one written at the moment of stalling would be a
|
||||
* fourth writer on a row two already share.
|
||||
*/
|
||||
function describe(gate, now = new Date()) {
|
||||
if (!gate) return null
|
||||
const since = new Date(gate.entered_at)
|
||||
const satisfied = Boolean(gate.satisfied_at)
|
||||
// **A satisfied gate's clock stops when it was satisfied**, not at read time.
|
||||
// Live it answers "how long has this phase been waiting"; afterwards it
|
||||
// answers "how long did it wait", and those are the same number only while it
|
||||
// is still waiting. The walk caught it disagreeing with `phase.advanced`'s
|
||||
// own `waitedSeconds` by the age of the screen — 139s beside a logged 121.
|
||||
const until = satisfied ? new Date(gate.satisfied_at) : now
|
||||
const elapsedSeconds = Math.max(0, Math.round((until.getTime() - since.getTime()) / 1000))
|
||||
|
||||
return {
|
||||
phase: gate.phase,
|
||||
kind: gate.kind,
|
||||
satisfied,
|
||||
satisfiedAt: gate.satisfied_at || null,
|
||||
satisfiedBy: gate.satisfied_by || null,
|
||||
since: gate.entered_at,
|
||||
elapsedSeconds,
|
||||
// An `after` gate is never stalled; an `on` gate is stalled once it has
|
||||
// waited past the threshold and not before.
|
||||
stalled: !satisfied && gate.kind === 'on' && now.getTime() - since.getTime() >= STALL_MS,
|
||||
...(gate.kind === 'after'
|
||||
? { after: gate.after_seconds, dueAt: gate.due_at }
|
||||
: {
|
||||
waitingOn: gate.trigger_id,
|
||||
where: phrase(gate.conditions),
|
||||
seen: gate.tally,
|
||||
needed: gate.needed,
|
||||
lastEvent: gate.last_event,
|
||||
lastEventAt: gate.last_event_at || null,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The trigger stream's second subscriber.
|
||||
*
|
||||
* Called from `engagementEmit.emit` beside `engine.dispatch` and, like it, never
|
||||
* awaited and never allowed to reject. An emit is a module saying something
|
||||
* happened in the game; whether some event run cared is core's business, and a
|
||||
* failure of core's must not become the module's control flow.
|
||||
*
|
||||
* **Every firing is logged, matched or not** (§ Observability: "trigger
|
||||
* evaluations that did and did not satisfy a condition"). The near miss is the
|
||||
* more valuable of the two on the night: *"the boss did spawn, in Britain"* and
|
||||
* *"no boss has spawned"* are different answers, and without this line they look
|
||||
* identical on the screen.
|
||||
*/
|
||||
async function observe(event) {
|
||||
const summary = { gates: 0, counted: 0, satisfied: 0 }
|
||||
try {
|
||||
const open = await gatesDb.openForTrigger(event.triggerId)
|
||||
summary.gates = open.length
|
||||
if (!open.length) return summary
|
||||
|
||||
const now = new Date()
|
||||
for (const gate of open) {
|
||||
const matched = conditions.evaluate(gate.conditions, event.data || {})
|
||||
|
||||
// ONLY the variables the condition names, never the payload. This row is
|
||||
// read back onto an admin screen, and a copy of a whole game event's data
|
||||
// is a second copy of exactly the content `engagement_sends` is careful
|
||||
// not to keep. The named variables are also the useful ones: they are the
|
||||
// reason it did or did not count.
|
||||
const named = variablesIn(gate.conditions)
|
||||
const lastEvent = {
|
||||
trigger: event.triggerId,
|
||||
at: event.occurredAt,
|
||||
subject: event.subject ?? null,
|
||||
matched,
|
||||
variables: Object.fromEntries(
|
||||
named.filter((n) => event.data && n in event.data).map((n) => [n, event.data[n]]),
|
||||
),
|
||||
}
|
||||
|
||||
const result = matched
|
||||
? await gatesDb.count(gate.id, { lastEvent, now })
|
||||
: { counted: false, satisfied: false, tally: gate.tally, near: await gatesDb.noteNearMiss(gate.id, { lastEvent, now }) }
|
||||
|
||||
if (matched && result.counted) summary.counted += 1
|
||||
if (result.satisfied) summary.satisfied += 1
|
||||
|
||||
await logDb.write({
|
||||
runId: gate.run_id,
|
||||
kind: 'condition.evaluated',
|
||||
phase: gate.phase,
|
||||
detail: {
|
||||
trigger: event.triggerId,
|
||||
matched,
|
||||
// The tally the DATABASE holds after the write, not the one this
|
||||
// process predicted — two emits arriving together each read the same
|
||||
// stale number, and only one of them is right about what it became.
|
||||
seen: result.tally ?? gate.tally,
|
||||
needed: gate.needed,
|
||||
satisfied: result.satisfied,
|
||||
variables: lastEvent.variables,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
if (summary.counted || summary.satisfied) {
|
||||
log.info('phase gate advanced', { trigger: event.triggerId, ...summary })
|
||||
}
|
||||
} catch (err) {
|
||||
log.error('gate observation failed', { trigger: event.triggerId, message: err.message })
|
||||
}
|
||||
return summary
|
||||
}
|
||||
|
||||
/**
|
||||
* Is this phase's gate open — and if it is not, why not?
|
||||
*
|
||||
* The runner's question, and the one place an `after` gate is closed: its
|
||||
* deadline passing is not an event anything emits, so the tick that finds it
|
||||
* past `due_at` is what records it. Doing that here rather than in the runner
|
||||
* keeps `satisfied_by` a fact one file writes.
|
||||
*
|
||||
* Answers `{ open, gate }`. `open: true` with a null gate is a phase with no
|
||||
* advance condition at all — every phase before this one, and most after it.
|
||||
*/
|
||||
async function check(runId, phase, now = new Date()) {
|
||||
const gate = await gatesDb.forPhase(runId, phase)
|
||||
if (!gate) return { open: true, gate: null }
|
||||
if (gate.satisfied_at) return { open: true, gate }
|
||||
|
||||
if (gate.kind === 'after' && gate.due_at && new Date(gate.due_at) <= now) {
|
||||
if (await gatesDb.satisfy(gate.id, 'elapsed', { now })) {
|
||||
return { open: true, gate: await gatesDb.byId(gate.id) }
|
||||
}
|
||||
// Somebody else closed it between the read and the write — a force, or the
|
||||
// tick that overran into this one. Either way it is open, and whichever
|
||||
// reason won is the one in the row.
|
||||
return { open: true, gate: await gatesDb.byId(gate.id) }
|
||||
}
|
||||
|
||||
return { open: false, gate }
|
||||
}
|
||||
|
||||
module.exports = { observe, check, describe, phrase, variablesIn, STALL_MS }
|
||||
@@ -17,7 +17,7 @@
|
||||
// than preserved: a spec that silently carries `announcements` today is a spec
|
||||
// whose author believes announcements work, and the later phase that gives the
|
||||
// key meaning would inherit a corpus of unvalidated ones. The refusal list is the
|
||||
// changelog — Phase 4 added the recurrence shapes, Phase 5 adds a phase's
|
||||
// changelog — Phase 4 added the recurrence shapes, Phase 5 added a phase's
|
||||
// `advance`, Phase 10 adds `announcements`.
|
||||
//
|
||||
// **Phase 4 widened `schedule` from one shape to four** — `manual`, `once`,
|
||||
@@ -30,7 +30,8 @@
|
||||
|
||||
const registries = require('../modules/registries')
|
||||
const recurrence = require('./recurrence')
|
||||
const { checkLiteral } = require('../engagement/conditions')
|
||||
const conditionGrammar = require('../engagement/conditions')
|
||||
const { checkLiteral } = conditionGrammar
|
||||
|
||||
// A phase key is a slug: it is stored in `event_run_steps.phase`, it is what the
|
||||
// run console groups by, and it is what an operator reads in "phase 3 has not
|
||||
@@ -45,6 +46,31 @@ const MAX_PHASES = 40
|
||||
const MAX_STEPS_PER_PHASE = 100
|
||||
const MAX_STEPS = 500
|
||||
|
||||
// The two shapes a phase's `advance` may take (§E). There is deliberately no
|
||||
// third: a gate that never opens is held, made visible and left to an operator
|
||||
// (org lead, 2026-09-02), so there is no authored timeout and no disposition to
|
||||
// validate. Adding one later is one key and one branch, and this is the list a
|
||||
// reader should find it missing from.
|
||||
const ADVANCE_KINDS = ['after', 'on']
|
||||
|
||||
// `after: '30m'` — one integer and one unit, and nothing else. No `1h30m`, no
|
||||
// fractions: the whole reason a duration is a string here rather than the plain
|
||||
// integer seconds `core.wait` takes is that an operator proofreads it, and a
|
||||
// grammar that admits two spellings of ninety minutes is one an operator has to
|
||||
// parse rather than read.
|
||||
const AFTER_RE = /^(\d{1,6})(s|m|h|d)$/
|
||||
const AFTER_UNIT_SECONDS = { s: 1, m: 60, h: 3600, d: 86_400 }
|
||||
const MIN_AFTER_SECONDS = 1
|
||||
// A paste guard rather than a policy, in the spirit of MAX_PHASES: thirty days
|
||||
// is longer than any event this system is for, and a phase gate of ten years is
|
||||
// a typo that would otherwise hold a run — and its concurrency key — for ever.
|
||||
const MAX_AFTER_SECONDS = 30 * 86_400
|
||||
|
||||
// How many firings one `on` gate may wait for. Bounded for the reason MAX_LIST
|
||||
// is: it is authored into a JSON column, and "count: 100000" is a phase that
|
||||
// never advances written as one that eventually does.
|
||||
const MAX_ADVANCE_COUNT = 1000
|
||||
|
||||
// The four closed shapes of §E. `manual` is first because it is the default and
|
||||
// what an unscheduled draft carries; the other three are recurrences the runner
|
||||
// expands into occurrences ahead of time.
|
||||
@@ -158,6 +184,140 @@ function validateSchedule(kind, raw, errors) {
|
||||
return { kind: 'manual' }
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse `'30m'` into seconds, or answer null.
|
||||
*
|
||||
* Exported because the run console renders the same duration back and must not
|
||||
* grow a second opinion about what `'2h'` means.
|
||||
*/
|
||||
function parseAfter(raw) {
|
||||
const m = AFTER_RE.exec(String(raw ?? ''))
|
||||
if (!m) return null
|
||||
const seconds = Number(m[1]) * AFTER_UNIT_SECONDS[m[2]]
|
||||
if (seconds < MIN_AFTER_SECONDS || seconds > MAX_AFTER_SECONDS) return null
|
||||
return seconds
|
||||
}
|
||||
|
||||
/** Seconds back to the largest whole unit that expresses them exactly. */
|
||||
function formatAfter(seconds) {
|
||||
for (const unit of ['d', 'h', 'm']) {
|
||||
const size = AFTER_UNIT_SECONDS[unit]
|
||||
if (seconds % size === 0) return `${seconds / size}${unit}`
|
||||
}
|
||||
return `${seconds}s`
|
||||
}
|
||||
|
||||
/**
|
||||
* Check a phase's `advance` gate and answer the normalised form of it.
|
||||
*
|
||||
* Returns `null` for a phase with no gate — the common case, and the behaviour
|
||||
* every phase had before Phase 5: it advances when its steps go terminal and on
|
||||
* nothing else. A gate is an ADDITIONAL condition, never a replacement, so a
|
||||
* phase whose steps are still running is not advanced by a satisfied gate.
|
||||
*
|
||||
* **The duration is normalised the way `days` is** — `'120m'` is stored as
|
||||
* `'2h'` — because the spec is diffed between versions, and two spellings of one
|
||||
* delay differing as JSON is a version history that reports edits nobody made.
|
||||
*
|
||||
* **`where` is validated against the trigger's DECLARATION, at save, with the
|
||||
* offending variable named.** This is the whole trap of this phase, and it is
|
||||
* `engagement/conditions.js`'s own argument one system across: a predicate that
|
||||
* silently reads `undefined` is a phase that silently never advances, and the
|
||||
* night you find out is the night of the event.
|
||||
*
|
||||
* **A trigger nobody registers makes the gate DORMANT, not invalid.** Same rule
|
||||
* as a step naming an action no installed module declares: it saves, so
|
||||
* uninstalling a module is not destructive to an author's work, and it refuses
|
||||
* to publish, because a version runs are pinned to must not wait on a trigger
|
||||
* that can never fire.
|
||||
*/
|
||||
function validateAdvance(raw, path, errors) {
|
||||
if (raw === undefined || raw === null) return null
|
||||
if (!isPlainObject(raw)) {
|
||||
errors.push(`${path}: expected an object`)
|
||||
return null
|
||||
}
|
||||
|
||||
const keys = Object.keys(raw)
|
||||
const named = ADVANCE_KINDS.filter((k) => keys.includes(k))
|
||||
if (named.length !== 1) {
|
||||
errors.push(`${path}: expected exactly one of "after" or "on"`)
|
||||
return null
|
||||
}
|
||||
const kind = named[0]
|
||||
|
||||
if (kind === 'after') {
|
||||
const extra = keys.filter((k) => k !== 'after')
|
||||
if (extra.length) {
|
||||
errors.push(`${path}: unknown key(s) ${extra.join(', ')} for an "after" gate`)
|
||||
return null
|
||||
}
|
||||
const seconds = parseAfter(raw.after)
|
||||
if (seconds === null) {
|
||||
errors.push(
|
||||
`${path}.after: expected a duration like "30m" — a whole number of s, m, h or d, ` +
|
||||
`between ${MIN_AFTER_SECONDS}s and ${formatAfter(MAX_AFTER_SECONDS)}`,
|
||||
)
|
||||
return null
|
||||
}
|
||||
// Only the canonical string is stored. The seconds are re-derived by the one
|
||||
// caller that needs them (the runner, when it opens the gate) through the
|
||||
// exported `parseAfter`, rather than kept beside it as a second field two
|
||||
// versions of the spec could disagree about.
|
||||
return { after: formatAfter(seconds) }
|
||||
}
|
||||
|
||||
// `dormant` is in this list for the reason `actionVersion` and `dormant` are
|
||||
// in a step's — **validate must accept its own output.** A saved spec is
|
||||
// re-validated on every later save and again at publish, so a field the
|
||||
// validator itself added and then refused would make the second save of any
|
||||
// gated definition impossible. It is accepted and then RECOMPUTED below,
|
||||
// never trusted: dormancy is whether anybody registers that trigger right
|
||||
// now, not what was true when the spec was last written.
|
||||
const extra = keys.filter((k) => !['on', 'where', 'count', 'dormant'].includes(k))
|
||||
if (extra.length) {
|
||||
errors.push(`${path}: unknown key(s) ${extra.join(', ')} for an "on" gate`)
|
||||
return null
|
||||
}
|
||||
|
||||
const triggerId = raw.on
|
||||
if (typeof triggerId !== 'string' || !triggerId) {
|
||||
errors.push(`${path}.on: expected a trigger id`)
|
||||
return null
|
||||
}
|
||||
|
||||
let count = 1
|
||||
if (raw.count !== undefined && raw.count !== null) {
|
||||
if (!Number.isInteger(raw.count) || raw.count < 1 || raw.count > MAX_ADVANCE_COUNT) {
|
||||
errors.push(`${path}.count: expected a whole number between 1 and ${MAX_ADVANCE_COUNT}`)
|
||||
return null
|
||||
}
|
||||
count = raw.count
|
||||
}
|
||||
|
||||
const declaration = registries.eventTrigger(triggerId)
|
||||
if (!declaration) {
|
||||
// Dormant, exactly as an unregistered action is. `where` is carried through
|
||||
// unvalidated and unnormalised — there is no declaration to check it
|
||||
// against, and dropping it would silently delete an author's predicate the
|
||||
// moment a module was uninstalled.
|
||||
return { on: triggerId, where: raw.where ?? null, count, dormant: true }
|
||||
}
|
||||
|
||||
const checked = conditionGrammar.validate(declaration, raw.where ?? null)
|
||||
if (!checked.ok) {
|
||||
// The grammar paths its own errors from the root token `conditions`; this
|
||||
// re-roots them at the phase so an author reading five of them at once can
|
||||
// tell which phase each belongs to. The text after the path — the part that
|
||||
// names the variable — is the grammar's, unchanged.
|
||||
checked.errors.forEach((e) =>
|
||||
errors.push(`${path}.where${e.startsWith('conditions') ? e.slice('conditions'.length) : `: ${e}`}`),
|
||||
)
|
||||
return null
|
||||
}
|
||||
return { on: triggerId, where: checked.conditions, count, dormant: false }
|
||||
}
|
||||
|
||||
/**
|
||||
* Check one authored param object against an action's declared params.
|
||||
*
|
||||
@@ -269,11 +429,13 @@ function validate(raw, { knownActionIds = [] } = {}) {
|
||||
errors.push(`${path}: expected an object`)
|
||||
return
|
||||
}
|
||||
const extra = Object.keys(rawPhase).filter((k) => !['key', 'label', 'steps'].includes(k))
|
||||
const extra = Object.keys(rawPhase).filter((k) => !['key', 'label', 'steps', 'advance'].includes(k))
|
||||
if (extra.length) {
|
||||
errors.push(`${path}: unknown key(s) ${extra.join(', ')} (a phase gains "advance" in Phase 5)`)
|
||||
errors.push(`${path}: unknown key(s) ${extra.join(', ')}`)
|
||||
}
|
||||
|
||||
const advance = validateAdvance(rawPhase.advance, `${path}.advance`, errors)
|
||||
|
||||
const key = rawPhase.key
|
||||
if (typeof key !== 'string' || !PHASE_KEY.test(key) || key.length > MAX_PHASE_KEY) {
|
||||
errors.push(`${path}.key: bad phase key "${key}"`)
|
||||
@@ -366,7 +528,11 @@ function validate(raw, { knownActionIds = [] } = {}) {
|
||||
})
|
||||
})
|
||||
|
||||
phases.push({ key, label: rawPhase.label, steps })
|
||||
// `advance` is omitted rather than written as null when there is no gate:
|
||||
// the overwhelming majority of phases have none, and a spec full of
|
||||
// `"advance": null` is a diff between two versions that says something
|
||||
// changed about every phase the first time one phase gained a gate.
|
||||
phases.push(advance ? { key, label: rawPhase.label, steps, advance } : { key, label: rawPhase.label, steps })
|
||||
})
|
||||
|
||||
if (totalSteps > MAX_STEPS) errors.push(`spec: at most ${MAX_STEPS} steps in one definition`)
|
||||
@@ -386,10 +552,18 @@ const actionIdsIn = (spec) =>
|
||||
* SAVING (that is what makes an uninstall non-destructive), and it must stop
|
||||
* them PUBLISHING, because publishing is what makes a version a thing runs are
|
||||
* pinned to and a run cannot dispatch a verb nobody registers.
|
||||
*
|
||||
* **A phase's advance gate is dormant on the same rule** (Phase 5), and it is in
|
||||
* the same list because it fails for the same reason and the message already
|
||||
* reads correctly for both: a version that waits on a trigger nothing can emit
|
||||
* is a run that would never leave that phase.
|
||||
*/
|
||||
function publishable(spec) {
|
||||
const dormant = (spec?.phases || [])
|
||||
.flatMap((p) => (p.steps || []).filter((s) => s.dormant).map((s) => s.actionId))
|
||||
const phases = spec?.phases || []
|
||||
const dormant = [
|
||||
...phases.flatMap((p) => (p.steps || []).filter((s) => s.dormant).map((s) => s.actionId)),
|
||||
...phases.filter((p) => p.advance?.dormant).map((p) => p.advance.on),
|
||||
]
|
||||
return dormant.length ? { ok: false, dormant: [...new Set(dormant)] } : { ok: true, dormant: [] }
|
||||
}
|
||||
|
||||
@@ -405,8 +579,13 @@ module.exports = {
|
||||
actionIdsIn,
|
||||
emptySpec,
|
||||
defaultOnFailure,
|
||||
parseAfter,
|
||||
formatAfter,
|
||||
PHASE_KEY,
|
||||
SCHEDULE_KINDS,
|
||||
ADVANCE_KINDS,
|
||||
MAX_ADVANCE_COUNT,
|
||||
MAX_AFTER_SECONDS,
|
||||
ON_FAILURE,
|
||||
ON_FAILURE_BY_RISK,
|
||||
MAX_PHASES,
|
||||
|
||||
187
server/src/model/events/eventPhaseGates.db.js
Normal file
187
server/src/model/events/eventPhaseGates.db.js
Normal file
@@ -0,0 +1,187 @@
|
||||
// ── event_run_phase_gates — SQL only ───────────────────────────────────────
|
||||
//
|
||||
// EVENTS.md §E, and Phase 5 of EVENTS_PLAN.md. A phase used to advance on one
|
||||
// fact — every step terminal — and that fact lives in `event_run_steps`. An
|
||||
// advance CONDITION is a second fact, and it is the only one in this feature
|
||||
// that is not derivable from a row somebody already wrote: `{ on:
|
||||
// 'uo.champ.boss_up', count: 3 }` counts things that happen between one tick and
|
||||
// the next, and the runner is not running when they happen. A gate row is where
|
||||
// a firing is counted at the moment it fires.
|
||||
//
|
||||
// **Two writers, and they are not the same process leg.** The RUNNER opens a
|
||||
// gate (at phase entry) and closes an `after` one (when its deadline passes);
|
||||
// the EMIT PATH increments and closes an `on` one. Everything here is therefore
|
||||
// written as a single guarded statement rather than a read-then-write, which is
|
||||
// the same argument `event_run_budget`'s conditional increment makes one phase
|
||||
// early and the same one `runsDb.transition` makes for a status.
|
||||
//
|
||||
// **Nothing here throws at the emit path.** `observe` is called from inside a
|
||||
// game-event handler by way of `ctx.events.emit`, exactly as `engine.dispatch`
|
||||
// is, and a database problem of core's must not become a module's control flow.
|
||||
// The catch lives in `events/gates.js`; this file is the statements.
|
||||
|
||||
const { query } = require('../../utils/db')
|
||||
const { parseJson } = require('./eventJson')
|
||||
|
||||
const hydrate = (row) =>
|
||||
row && {
|
||||
...row,
|
||||
conditions: parseJson(row.conditions, null),
|
||||
last_event: parseJson(row.last_event, null),
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a phase's gate. **INSERT IGNORE against `uq_evgate_phase`**, so a process
|
||||
* that died between entering a phase and getting here opens no second gate on
|
||||
* the next tick — the idempotence `materialisePhase` has, for the same reason.
|
||||
*
|
||||
* Answers whether a row was created, which is what lets the caller log
|
||||
* `phase.entered`'s gate detail exactly once.
|
||||
*/
|
||||
async function open({ runId, phase, kind, afterSeconds = null, triggerId = null, conditions = null, needed = 1, now = new Date() }) {
|
||||
// `due_at` is computed here, once, from the moment the phase was entered —
|
||||
// never re-derived on a later tick from a `now` that has moved. A deadline
|
||||
// recomputed every fifteen seconds is a deadline that never arrives.
|
||||
const dueAt = kind === 'after' ? new Date(now.getTime() + afterSeconds * 1000) : null
|
||||
const result = await query(
|
||||
`INSERT IGNORE INTO event_run_phase_gates
|
||||
(run_id, phase, kind, after_seconds, trigger_id, conditions, needed, entered_at, due_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
[
|
||||
runId,
|
||||
phase,
|
||||
kind,
|
||||
afterSeconds,
|
||||
triggerId,
|
||||
conditions === null ? null : JSON.stringify(conditions),
|
||||
needed,
|
||||
now,
|
||||
dueAt,
|
||||
],
|
||||
)
|
||||
return Number(result?.affectedRows || 0) === 1
|
||||
}
|
||||
|
||||
/** One run's gate for one phase, or null. */
|
||||
const forPhase = async (runId, phase) =>
|
||||
hydrate(
|
||||
(
|
||||
await query('SELECT * FROM event_run_phase_gates WHERE run_id = ? AND phase = ? LIMIT 1', [
|
||||
runId,
|
||||
phase,
|
||||
])
|
||||
)[0] || null,
|
||||
)
|
||||
|
||||
/** Every gate a run has ever opened, oldest first — what the run console reads. */
|
||||
const listForRun = async (runId) =>
|
||||
(
|
||||
await query('SELECT * FROM event_run_phase_gates WHERE run_id = ? ORDER BY entered_at, id', [
|
||||
runId,
|
||||
])
|
||||
).map(hydrate)
|
||||
|
||||
/**
|
||||
* Every OPEN gate waiting on one trigger, with the run's status and phase.
|
||||
*
|
||||
* This is the emit path's only query and the one index in this feature on a hot
|
||||
* path. The join is what keeps a gate belonging to a cancelled run from counting
|
||||
* for ever: a run that will never advance again must stop tallying, and its
|
||||
* row's `satisfied_at` is not what says so.
|
||||
*
|
||||
* **`paused` counts.** The world does not stop because an operator paused the
|
||||
* console, and discarding firings that arrived during a pause would make pause a
|
||||
* destructive control — the tally an operator came back to would be lower than
|
||||
* the one they left, with nothing recording the difference.
|
||||
*/
|
||||
const openForTrigger = async (triggerId) =>
|
||||
(
|
||||
await query(
|
||||
`SELECT g.* FROM event_run_phase_gates g
|
||||
JOIN event_runs r ON r.id = g.run_id
|
||||
WHERE g.trigger_id = ? AND g.satisfied_at IS NULL
|
||||
AND r.status IN ('running','paused')
|
||||
AND r.current_phase = g.phase
|
||||
LIMIT 200`,
|
||||
[triggerId],
|
||||
)
|
||||
).map(hydrate)
|
||||
|
||||
/**
|
||||
* Count one matching firing, and close the gate if that was the last one needed.
|
||||
*
|
||||
* **One statement, with the threshold inside it.** Two emits arriving together
|
||||
* each add one and exactly one of them crosses `needed`; a read-then-write would
|
||||
* let both see 2 of 3 and neither satisfy, or both satisfy and advance a phase
|
||||
* twice. `WHERE satisfied_at IS NULL` is what makes a late arrival a no-op
|
||||
* rather than a tally that keeps climbing after the phase moved on.
|
||||
*
|
||||
* Answers `{ counted, satisfied }` read back from the row, so the caller logs
|
||||
* the tally the database actually holds rather than the one it predicted.
|
||||
*/
|
||||
async function count(gateId, { lastEvent = null, now = new Date() } = {}) {
|
||||
// **THE INCREMENT MUST BE LAST, and this is not style.** MariaDB evaluates an
|
||||
// UPDATE's SET assignments LEFT TO RIGHT, each one seeing the values already
|
||||
// assigned by the ones before it — which is a documented departure from
|
||||
// standard SQL, and it is invisible in a stub. With `tally = tally + 1` first,
|
||||
// the CASE that follows reads the ALREADY-INCREMENTED tally, so `tally + 1 >=
|
||||
// needed` is really `new + 1 >= needed` and a gate needing two firings closes
|
||||
// on the first. Written this way, both CASEs see the old tally and say exactly
|
||||
// what they read as. `eventRunnerSql.test.js` is what catches a reorder, and
|
||||
// it is what caught this one.
|
||||
const result = await query(
|
||||
`UPDATE event_run_phase_gates
|
||||
SET satisfied_at = CASE WHEN tally + 1 >= needed THEN ? ELSE NULL END,
|
||||
satisfied_by = CASE WHEN tally + 1 >= needed THEN 'condition' ELSE NULL END,
|
||||
last_event = ?,
|
||||
last_event_at = ?,
|
||||
tally = tally + 1
|
||||
WHERE id = ? AND satisfied_at IS NULL`,
|
||||
[now, lastEvent === null ? null : JSON.stringify(lastEvent), now, gateId],
|
||||
)
|
||||
if (Number(result?.affectedRows || 0) !== 1) return { counted: false, satisfied: false }
|
||||
const row = await byId(gateId)
|
||||
return { counted: true, satisfied: Boolean(row?.satisfied_at), tally: row?.tally ?? null }
|
||||
}
|
||||
|
||||
/**
|
||||
* Record that a firing was seen and did NOT match.
|
||||
*
|
||||
* Only `last_event_at` and `last_event` move: the tally is what the phase is
|
||||
* waiting on, and a near miss is not progress. It is recorded at all because
|
||||
* "the boss did spawn, in the wrong region" and "no boss has spawned" are
|
||||
* different answers to the operator's question, and only this column can tell
|
||||
* them apart on a screen.
|
||||
*/
|
||||
async function noteNearMiss(gateId, { lastEvent = null, now = new Date() } = {}) {
|
||||
const result = await query(
|
||||
`UPDATE event_run_phase_gates
|
||||
SET last_event = ?, last_event_at = ?
|
||||
WHERE id = ? AND satisfied_at IS NULL`,
|
||||
[lastEvent === null ? null : JSON.stringify(lastEvent), now, gateId],
|
||||
)
|
||||
return Number(result?.affectedRows || 0) === 1
|
||||
}
|
||||
|
||||
/**
|
||||
* Close a gate for a reason that is not a matching firing: `'elapsed'` when an
|
||||
* `after` deadline passed, `'forced'` when a human pressed advance.
|
||||
*
|
||||
* Guarded on `satisfied_at IS NULL` like everything else here, so a force that
|
||||
* races the tick that would have opened the gate anyway loses harmlessly and the
|
||||
* log records whichever actually happened rather than both.
|
||||
*/
|
||||
async function satisfy(gateId, by, { userId = null, now = new Date() } = {}) {
|
||||
const result = await query(
|
||||
`UPDATE event_run_phase_gates
|
||||
SET satisfied_at = ?, satisfied_by = ?, forced_by = ?
|
||||
WHERE id = ? AND satisfied_at IS NULL`,
|
||||
[now, by, userId, gateId],
|
||||
)
|
||||
return Number(result?.affectedRows || 0) === 1
|
||||
}
|
||||
|
||||
const byId = async (id) =>
|
||||
hydrate((await query('SELECT * FROM event_run_phase_gates WHERE id = ? LIMIT 1', [id]))[0] || null)
|
||||
|
||||
module.exports = { open, forPhase, byId, listForRun, openForTrigger, count, noteNearMiss, satisfy }
|
||||
@@ -7,15 +7,18 @@
|
||||
// needs no control, and a run that paused on a failed world write is the one
|
||||
// that does.
|
||||
//
|
||||
// **Two of §I's six run-level controls are deliberately not here.**
|
||||
// `advance` — force a phase forward — has no honest meaning yet: a phase today
|
||||
// advances when its steps go terminal, and the per-step skip already does that
|
||||
// one step at a time. Phase 5 is what gives a phase an `advance` CONDITION, and
|
||||
// that is the first moment "force it anyway" means something an operator could
|
||||
// predict. `cleanup` needs Phase 8's resource ledger; there is nothing to
|
||||
// revert, so cancel takes `{ reason }` and gains `cleanup` when there is
|
||||
// something for it to do. Both are absent rather than inert, which is the
|
||||
// posture Phase 1 set and Phase 2 kept.
|
||||
// **`advance` is the seventh, and it arrived in Phase 5 rather than Phase 3
|
||||
// because that is when it started meaning something.** A phase used to advance
|
||||
// when its steps went terminal and on nothing else, so "force it anyway" named
|
||||
// no state an operator could be in; a phase with a gate can wait for a boss that
|
||||
// will never spawn, and then it names exactly one. It is the other half of the
|
||||
// diagnosis panel: a screen that explains why a phase has not started, beside a
|
||||
// control that does something about it.
|
||||
//
|
||||
// **One of §I's controls is still not here.** `cleanup` needs Phase 8's resource
|
||||
// ledger; there is nothing to revert, so cancel takes `{ reason }` and gains
|
||||
// `cleanup` when there is something for it to do. Absent rather than inert,
|
||||
// which is the posture Phase 1 set and every phase since has kept.
|
||||
//
|
||||
// **Every control is guarded on the status it may act from, and the guard is a
|
||||
// WHERE clause rather than a read-then-write.** A run console rendered thirty
|
||||
@@ -33,6 +36,8 @@
|
||||
const runsDb = require('./eventRuns.db')
|
||||
const stepsDb = require('./eventRunSteps.db')
|
||||
const logDb = require('./eventRunLog.db')
|
||||
const gatesDb = require('./eventPhaseGates.db')
|
||||
const gates = require('../../events/gates')
|
||||
|
||||
const MAX_REASON = 500
|
||||
|
||||
@@ -168,6 +173,67 @@ async function cancel(runId, { reason } = {}, userId = null) {
|
||||
return { ok: true, run: await runsDb.getById(run.id), cancelledSteps: closed }
|
||||
}
|
||||
|
||||
/**
|
||||
* Force a phase forward: open its gate without the condition that would have.
|
||||
*
|
||||
* **It is legal only when the phase is actually waiting on a gate**, and the
|
||||
* three refusals are the whole design. A run that is not `running` is not
|
||||
* waiting on anything (409 naming what it is). A phase with no gate advances on
|
||||
* its steps and always has, so forcing it would be a control that duplicated the
|
||||
* runner rather than overriding it. And a phase whose steps have not all gone
|
||||
* terminal is not being held by its gate — it is being held by a step, and the
|
||||
* step-level skip is the honest control for that, one step at a time. A force
|
||||
* that swept past pending steps would be a cancel of half a phase under a button
|
||||
* labelled advance.
|
||||
*
|
||||
* **It satisfies the gate and stops.** The next tick advances the run, exactly
|
||||
* as it does after `resume` — the phase transition, the next phase's
|
||||
* materialisation, its own gate and the log lines are one sequence in
|
||||
* `advanceRun`, and a second copy of it here would be a second opinion about
|
||||
* what a phase boundary is. The response says which phase was released, so the
|
||||
* console can say so before the tick lands.
|
||||
*/
|
||||
async function advancePhase(runId, { reason } = {}, userId = null) {
|
||||
const run = await loadRun(runId)
|
||||
if (!run) return { ok: false, status: 404, errors: ['no such run'] }
|
||||
if (run.status !== 'running') return conflict(`a ${run.status} run has no phase to advance`)
|
||||
if (!run.current_phase) return conflict('this run has not entered a phase yet')
|
||||
|
||||
const gate = await gatesDb.forPhase(run.id, run.current_phase)
|
||||
if (!gate) return conflict(`phase "${run.current_phase}" has no advance condition; skip its steps instead`)
|
||||
if (gate.satisfied_at) return conflict(`phase "${run.current_phase}" is already past its advance condition`)
|
||||
|
||||
const openStep = await stepsDb.nextOpenStep(run.id, run.current_phase)
|
||||
if (openStep) {
|
||||
return conflict(
|
||||
`phase "${run.current_phase}" is waiting on step ${openStep.seq} (${openStep.action_id}), not on its advance condition`,
|
||||
)
|
||||
}
|
||||
|
||||
const note = clean(reason)
|
||||
if (!(await gatesDb.satisfy(gate.id, 'forced', { userId }))) {
|
||||
return conflict('this phase stopped waiting on its advance condition')
|
||||
}
|
||||
|
||||
const described = gates.describe(gate)
|
||||
await logDb.write({
|
||||
runId: run.id,
|
||||
kind: 'phase.advanced',
|
||||
phase: run.current_phase,
|
||||
detail: {
|
||||
because: 'forced',
|
||||
control: 'advance',
|
||||
by: userId,
|
||||
reason: note,
|
||||
waitedSeconds: described.elapsedSeconds,
|
||||
...(gate.kind === 'on'
|
||||
? { trigger: gate.trigger_id, seen: gate.tally, needed: gate.needed }
|
||||
: { after: gate.after_seconds }),
|
||||
},
|
||||
})
|
||||
return { ok: true, run: await runsDb.getById(run.id), phase: run.current_phase }
|
||||
}
|
||||
|
||||
// ── Step-level ────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
@@ -293,4 +359,4 @@ async function retryStep(runId, stepId, options = {}, userId = null) {
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { pause, resume, cancel, confirmStep, skipStep, retryStep }
|
||||
module.exports = { pause, resume, cancel, advancePhase, confirmStep, skipStep, retryStep }
|
||||
|
||||
@@ -33,6 +33,13 @@ const KINDS = [
|
||||
'step.retry', // a step failed transiently and will be attempted again
|
||||
'step.parked', // a step is waiting on a human and nothing is holding it
|
||||
'phase.completed', // every step of a phase reached a terminal status
|
||||
// Phase 5's four. `condition.evaluated` is written for BOTH outcomes (§
|
||||
// Observability), and the non-matching one is the more valuable of the two on
|
||||
// the night: "the boss did spawn, in Britain" and "no boss has spawned" are
|
||||
// different answers to the same question and look identical without it.
|
||||
'phase.gate', // a phase opened an advance gate, with what it waits for
|
||||
'condition.evaluated', // a firing was tested against a gate, matched or not
|
||||
'phase.advanced', // a gate opened: on a firing, on its deadline, or forced
|
||||
]
|
||||
|
||||
const hydrate = (row) => row && { ...row, detail: parseJson(row.detail, null) }
|
||||
|
||||
@@ -23,6 +23,16 @@ const hydrate = (row) => row && { ...row, params: parseJson(row.params, null), r
|
||||
// only if every path a run can take reaches one of them.
|
||||
const TERMINAL = ['completed', 'cancelled', 'failed', 'missed']
|
||||
|
||||
// §E's health values, worst last. Health is a HIGH-WATER MARK in this system —
|
||||
// nothing has ever cleared `degraded`, because a run whose announcement landed
|
||||
// on the second attempt did have trouble and that stays true for the rest of its
|
||||
// life — and `setHealth` enforces that rather than leaving it to every caller to
|
||||
// remember. `FIELD()` gives the same order inside the WHERE clause, 1-indexed,
|
||||
// which is what makes the guard one statement rather than a read and a write.
|
||||
const HEALTH_ORDER = ['ok', 'degraded', 'stalled']
|
||||
const HEALTH_RANK = Object.fromEntries(HEALTH_ORDER.map((h, i) => [h, i + 1]))
|
||||
const HEALTH_SQL_ORDER = HEALTH_ORDER.map((h) => `'${h}'`).join(', ')
|
||||
|
||||
// `waiting_steps` is the count of PARKED steps: `running` with a NULL lease, the
|
||||
// pair `park()` alone produces, which means a cue waiting on a human. It is a
|
||||
// correlated subquery on an admin list bounded at 500 rows rather than a column,
|
||||
@@ -359,11 +369,20 @@ const statusOf = async (id) => {
|
||||
* same degradation does not restamp `updated_at`.
|
||||
*/
|
||||
async function setHealth(id, health) {
|
||||
const result = await query('UPDATE event_runs SET health = ? WHERE id = ? AND health <> ?', [
|
||||
health,
|
||||
id,
|
||||
health,
|
||||
])
|
||||
// **Escalation only, and this is the guard rather than a convention.** Health
|
||||
// has always been a high-water mark here — `degraded` is never cleared,
|
||||
// because a run whose announcement landed on the second attempt DID have
|
||||
// trouble and that stays true — and Phase 5 gave the column a second writer
|
||||
// for `stalled`. Without a rank, a step that retried after a stall would
|
||||
// quietly demote `stalled` to `degraded` and a run that waited ninety minutes
|
||||
// on a boss that never came would end its life claiming it merely wobbled.
|
||||
const rank = HEALTH_RANK[health]
|
||||
if (!rank) return false
|
||||
const result = await query(
|
||||
`UPDATE event_runs SET health = ?
|
||||
WHERE id = ? AND FIELD(health, ${HEALTH_SQL_ORDER}) < ?`,
|
||||
[health, id, rank],
|
||||
)
|
||||
return Number(result?.affectedRows || 0) === 1
|
||||
}
|
||||
|
||||
|
||||
@@ -21,6 +21,8 @@
|
||||
const db = require('./eventRuns.db')
|
||||
const stepsDb = require('./eventRunSteps.db')
|
||||
const logDb = require('./eventRunLog.db')
|
||||
const gatesDb = require('./eventPhaseGates.db')
|
||||
const gates = require('../../events/gates')
|
||||
const definitionsDb = require('./eventDefinitions.db')
|
||||
const versionsDb = require('./eventVersions.db')
|
||||
|
||||
@@ -140,12 +142,30 @@ async function create(
|
||||
return { ok: true, created: true, run: await db.getById(runId) }
|
||||
}
|
||||
|
||||
/** A run, its steps and its status counts — what the run console reads. */
|
||||
/**
|
||||
* A run, its steps, its status counts and its phase gates — the run console.
|
||||
*
|
||||
* The gates arrive already DESCRIBED rather than as rows (Phase 5): the panel's
|
||||
* whole value is that it reads the way the condition builder reads, and those
|
||||
* words come from `engagement/conditions.js`'s own operator labels. Rendering
|
||||
* them in the browser would be a second implementation of a grammar the server
|
||||
* owns, and the first clause the two spelled differently would meet its operator
|
||||
* at two in the morning.
|
||||
*
|
||||
* Every gate the run has opened is returned, not only the current phase's. A
|
||||
* completed phase's gate answers "how long did phase 2 actually wait, and what
|
||||
* released it" — which is the same question as the live one, asked afterwards.
|
||||
*/
|
||||
async function detail(runId) {
|
||||
const run = await db.getById(runId)
|
||||
if (!run) return null
|
||||
const [steps, counts] = await Promise.all([stepsDb.listForRun(runId), stepsDb.statusCounts(runId)])
|
||||
return { run, steps, counts }
|
||||
const [steps, counts, gateRows] = await Promise.all([
|
||||
stepsDb.listForRun(runId),
|
||||
stepsDb.statusCounts(runId),
|
||||
gatesDb.listForRun(runId),
|
||||
])
|
||||
const now = new Date()
|
||||
return { run, steps, counts, gates: gateRows.map((g) => gates.describe(g, now)) }
|
||||
}
|
||||
|
||||
module.exports = { create, detail, renderConcurrencyKey }
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
|
||||
const registries = require('../../../modules/registries')
|
||||
const spec = require('../../../events/spec')
|
||||
const conditionGrammar = require('../../../engagement/conditions')
|
||||
const definitionsDb = require('../../../model/events/eventDefinitions.db')
|
||||
const definitions = require('../../../model/events/eventDefinitions.model')
|
||||
const versionsDb = require('../../../model/events/eventVersions.db')
|
||||
@@ -146,11 +147,40 @@ exports.catalog = (_req, res) => {
|
||||
onFailure: spec.ON_FAILURE,
|
||||
onFailureByRisk: spec.ON_FAILURE_BY_RISK,
|
||||
scheduleKinds: spec.SCHEDULE_KINDS,
|
||||
// **The trigger catalog is served here too, and not borrowed from
|
||||
// `/admin/engagement/triggers`** (Phase 5). §C's claim is that the trigger
|
||||
// catalog a module already ships IS the catalog of things that can advance a
|
||||
// phase — so it is the same registry, read twice. What differs is who may
|
||||
// read it: the engagement route is `adminOnly`, and event definitions are
|
||||
// authored by `admin` AND `editor`. Pointing this editor at that route would
|
||||
// have left an editor writing a trigger id from memory into a field the save
|
||||
// path then refused.
|
||||
//
|
||||
// Each declaration is reduced to what the gate form needs — id, label and
|
||||
// the variables a `where` may name. Everything else on a trigger (its
|
||||
// audience, its ceiling, its subject key) is about who gets MAILED, which is
|
||||
// a different question and not this screen's.
|
||||
triggers: registries.allTriggers().map((t) => ({
|
||||
id: t.id,
|
||||
label: t.label,
|
||||
description: t.description,
|
||||
owner: t.owner,
|
||||
variables: (t.variables || []).map((v) => ({
|
||||
name: v.name,
|
||||
type: v.type,
|
||||
required: v.required,
|
||||
description: v.description,
|
||||
})),
|
||||
})),
|
||||
operators: conditionGrammar.vocabulary(),
|
||||
advanceKinds: spec.ADVANCE_KINDS,
|
||||
limits: {
|
||||
maxPhases: spec.MAX_PHASES,
|
||||
maxStepsPerPhase: spec.MAX_STEPS_PER_PHASE,
|
||||
maxSteps: spec.MAX_STEPS,
|
||||
defaultBudgetMs: registries.DEFAULT_BUDGET_MS,
|
||||
maxAdvanceCount: spec.MAX_ADVANCE_COUNT,
|
||||
maxAfterSeconds: spec.MAX_AFTER_SECONDS,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -255,6 +285,10 @@ exports.getRun = async (req, res) => {
|
||||
run: shapeRun(found.run),
|
||||
steps: found.steps.map(shapeStep),
|
||||
counts: found.counts,
|
||||
// Already rendered in the condition builder's own words (Phase 5). See
|
||||
// `eventRuns.model.detail` for why the sentence is built here and not in
|
||||
// the browser.
|
||||
gates: found.gates,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -437,6 +471,20 @@ exports.resumeRun = async (req, res) => {
|
||||
return res.json({ run: shapeRun(result.run) })
|
||||
}
|
||||
|
||||
/** POST /api/v1/admin/events/runs/:runId/advance */
|
||||
exports.advanceRunPhase = async (req, res) => {
|
||||
const runId = asId(req.params.runId)
|
||||
if (!runId) return res.status(400).json({ error: 'bad run id' })
|
||||
const result = await controls.advancePhase(runId, { reason: req.body?.reason }, req.user.id)
|
||||
if (!result.ok) return res.status(result.status || 409).json({ errors: result.errors })
|
||||
await activity.log({
|
||||
req,
|
||||
action: 'event.run.advanced',
|
||||
detail: { runId, phase: result.phase, reason: req.body?.reason || null },
|
||||
})
|
||||
return res.json({ run: shapeRun(result.run), phase: result.phase })
|
||||
}
|
||||
|
||||
/** POST /api/v1/admin/events/runs/:runId/cancel */
|
||||
exports.cancelRun = async (req, res) => {
|
||||
const runId = asId(req.params.runId)
|
||||
|
||||
@@ -45,9 +45,9 @@ eventsRouter.get(
|
||||
'/catalog',
|
||||
// #swagger.tags = ['Admin · Events']
|
||||
// #swagger.summary = 'List every registered event action, with its param schema, risk class and reversibility'
|
||||
// #swagger.description = 'Served from the module registries, not from a table: an action is declared in code by core or by an installed module, so this is whatever registered on this boot, and an uninstalled module simply stops appearing. Core always declares core.announce, core.wait and core.cue. Also carries the closed vocabularies the authoring form renders — risk classes, reversibility classes, param types, failure dispositions and the spec size limits — so the editor offers exactly the set the save path checks against.'
|
||||
// #swagger.description = 'Served from the module registries, not from a table: an action is declared in code by core or by an installed module, so this is whatever registered on this boot, and an uninstalled module simply stops appearing. Core always declares core.announce, core.wait and core.cue. Also carries the closed vocabularies the authoring form renders — risk classes, reversibility classes, param types, failure dispositions and the spec size limits — so the editor offers exactly the set the save path checks against. Phase 5 added `triggers` and `operators`: the trigger catalog a module already ships IS the catalog of things a phase can advance on, and it is served here rather than borrowed from /admin/engagement/triggers because that route is admin-only while an event definition is authored by admin AND editor. Each trigger is reduced to its id, label and declared variables — a trigger's audience and ceiling are about who gets mailed, which is not this screen's question.'
|
||||
// #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }]
|
||||
/* #swagger.responses[200] = { description: 'The registered actions and the vocabularies over them', content: { "application/json": { schema: { type: "object", properties: { actions: { type: "array", items: { type: "object", additionalProperties: true } }, risks: { type: "array", items: { type: "string" } }, reversible: { type: "array", items: { type: "string" } }, paramTypes: { type: "array", items: { type: "string" } }, onFailure: { type: "array", items: { type: "string" } }, onFailureByRisk: { type: "object", additionalProperties: true }, scheduleKinds: { type: "array", items: { type: "string" } }, limits: { type: "object", additionalProperties: true } } } } } } */
|
||||
/* #swagger.responses[200] = { description: 'The registered actions and triggers, and the vocabularies over them', content: { "application/json": { schema: { type: "object", properties: { actions: { type: "array", items: { type: "object", additionalProperties: true } }, triggers: { type: "array", items: { type: "object", additionalProperties: true } }, operators: { type: "array", items: { type: "object", additionalProperties: true } }, risks: { type: "array", items: { type: "string" } }, reversible: { type: "array", items: { type: "string" } }, paramTypes: { type: "array", items: { type: "string" } }, onFailure: { type: "array", items: { type: "string" } }, onFailureByRisk: { type: "object", additionalProperties: true }, scheduleKinds: { type: "array", items: { type: "string" } }, advanceKinds: { type: "array", items: { type: "string" } }, limits: { type: "object", additionalProperties: true } } } } } } */
|
||||
/* #swagger.responses[403] = { description: 'Not staff', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */
|
||||
controller.catalog,
|
||||
)
|
||||
@@ -146,9 +146,9 @@ eventsRouter.get(
|
||||
'/runs/:runId',
|
||||
// #swagger.tags = ['Admin · Events']
|
||||
// #swagger.summary = 'One run: its status, health, cleanup state and every step with its params and idempotency key'
|
||||
// #swagger.description = 'The run console. `counts` summarises the step list by status. Steps carry the idempotency key core minted at materialisation — stable across every attempt, which is what lets the game side recognise a repeat.'
|
||||
// #swagger.description = 'The run console. `counts` summarises the step list by status. Steps carry the idempotency key core minted at materialisation — stable across every attempt, which is what lets the game side recognise a repeat. `gates` is the diagnosis panel (Phase 5): one entry per phase that authored an advance condition, already rendered in the condition builder's own words — `gte` as "is at least", `present` as "is present" — with the tally, how long it has waited, and the last related firing whether or not it matched. A phase is waiting on its gate only once every one of its steps is terminal; `stalled` means an `on` gate has waited past EVENT_PHASE_STALL_MS, which is visibility and never a timeout — nothing advances a phase but its condition or a human.'
|
||||
// #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }]
|
||||
/* #swagger.responses[200] = { description: 'The run, its steps and the status counts', content: { "application/json": { schema: { type: "object", properties: { run: { type: "object", additionalProperties: true }, steps: { type: "array", items: { type: "object", additionalProperties: true } }, counts: { type: "object", additionalProperties: true } } } } } } */
|
||||
/* #swagger.responses[200] = { description: 'The run, its steps, the status counts and the phase gates', content: { "application/json": { schema: { type: "object", properties: { run: { type: "object", additionalProperties: true }, steps: { type: "array", items: { type: "object", additionalProperties: true } }, counts: { type: "object", additionalProperties: true }, gates: { type: "array", items: { type: "object", additionalProperties: true } } } } } } } */
|
||||
/* #swagger.responses[404] = { description: 'No such run', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */
|
||||
controller.getRun,
|
||||
)
|
||||
@@ -175,9 +175,11 @@ eventsRouter.get(
|
||||
// read consistent, with one role owning both buttons, would behave badly in
|
||||
// exactly the case the moderator role exists for.
|
||||
//
|
||||
// `advance` and `cleanup` from the § API surface table are not here: the first
|
||||
// has no honest meaning until Phase 5 gives a phase an advance condition, the
|
||||
// second has no resource ledger to work over until Phase 8.
|
||||
// `advance` joined them in Phase 5, which is when it started meaning something:
|
||||
// a phase with an advance condition can wait on a boss that never spawns, and
|
||||
// that is the one state "force it anyway" names. `cleanup` from the § API
|
||||
// surface table is still not here — it has no resource ledger to work over until
|
||||
// Phase 8.
|
||||
|
||||
eventsRouter.post(
|
||||
'/runs/:runId/pause',
|
||||
@@ -220,6 +222,20 @@ eventsRouter.post(
|
||||
controller.cancelRun,
|
||||
)
|
||||
|
||||
eventsRouter.post(
|
||||
'/runs/:runId/advance',
|
||||
// #swagger.tags = ['Admin · Events']
|
||||
// #swagger.summary = 'Force the current phase past its advance condition'
|
||||
// #swagger.description = 'The other half of the diagnosis panel: a screen that says why a phase has not started, beside the control that does something about it. Legal only while the phase is genuinely waiting on its gate, and the three refusals are the design — a run that is not `running` is waiting on nothing; a phase with no advance condition already advances on its steps; and a phase with a step still open is held by that step, not by its gate, so the step-level skip is the honest control. Satisfies the gate and stops: the next tick performs the phase transition, exactly as it does after resume, so there is only ever one implementation of what a phase boundary is. The log records `because: forced` with the actor, the reason and how long the phase had waited.'
|
||||
// #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }]
|
||||
/* #swagger.requestBody = { required: false, content: { "application/json": { schema: { type: "object", properties: { reason: { type: "string", description: "Why the condition was overridden. Recorded in the run log with the actor." } } } } } } */
|
||||
/* #swagger.responses[200] = { description: 'The run, and the phase that was released', content: { "application/json": { schema: { type: "object", properties: { run: { type: "object", additionalProperties: true }, phase: { type: "string" } } } } } } */
|
||||
/* #swagger.responses[409] = { description: 'The phase is not waiting on an advance condition', content: { "application/json": { schema: { type: "object", properties: { errors: { type: "array", items: { type: "string" } } } } } } } */
|
||||
/* #swagger.responses[403] = { description: 'Not an admin or moderator', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */
|
||||
liveControl,
|
||||
controller.advanceRunPhase,
|
||||
)
|
||||
|
||||
eventsRouter.post(
|
||||
'/runs/:runId/steps/:stepId/confirm',
|
||||
// #swagger.tags = ['Admin · Events']
|
||||
|
||||
@@ -22,6 +22,7 @@
|
||||
|
||||
const registries = require('../modules/registries')
|
||||
const engine = require('../engagement/engine')
|
||||
const eventGates = require('../events/gates')
|
||||
const scopedPrefs = require('../engagement/scopedPrefs')
|
||||
const createLogger = require('./logger')
|
||||
|
||||
@@ -276,6 +277,20 @@ function emit(owner, triggerId, envelope = {}) {
|
||||
// await the delivery decision, and the tests use it directly.
|
||||
engine.dispatch(event).catch((err) => log.error('dispatch rejected', { trigger: triggerId, message: err.message }))
|
||||
|
||||
// **The trigger stream's second subscriber** (EVENTS.md §E, Phase 5). An event
|
||||
// run whose phase is waiting on `{ on: '<triggerId>', count: n }` counts this
|
||||
// firing here, at the moment it fires, because nothing observable survives to
|
||||
// the runner's next tick. Same seam, same posture: not awaited, never allowed
|
||||
// to reject, and it knows nothing about who emitted.
|
||||
//
|
||||
// It is a SECOND subscriber and not a leg of `dispatch` because the two
|
||||
// decide different things — who gets told, and whether a phase may proceed —
|
||||
// and neither must be able to fail the other. A rules lookup that throws must
|
||||
// not lose the count, and a gate write that throws must not lose the mail.
|
||||
eventGates
|
||||
.observe(event)
|
||||
.catch((err) => log.error('gate observation rejected', { trigger: triggerId, message: err.message }))
|
||||
|
||||
return { ok: true, event }
|
||||
}
|
||||
|
||||
|
||||
@@ -12,6 +12,14 @@
|
||||
// 3. **advance** — claim each due run and move it through its phases
|
||||
// 4. **prune** — the `event_run_log` retention sweep, on its own long clock
|
||||
//
|
||||
// **What a phase advances on, as of Phase 5.** Every step terminal, and — if the
|
||||
// phase authored one — its GATE open as well. The gate is an ADDITIONAL
|
||||
// condition and never a replacement: a phase whose steps are still running is
|
||||
// not advanced by a boss that spawned early. `{ after: '30m' }` is closed by
|
||||
// this file when its deadline passes; `{ on: '<trigger>', count: n }` is closed
|
||||
// by the EMIT PATH, because a firing between two ticks is not observable from
|
||||
// either of them. See `events/gates.js` for that division.
|
||||
//
|
||||
// **What "materialise" means, and why it is two halves.** Phase 4 completed it.
|
||||
// The first half EXPANDS: every `ready` definition's recurrence is computed in
|
||||
// its own IANA zone and every occurrence inside a fourteen-day horizon becomes a
|
||||
@@ -55,7 +63,10 @@ const logDb = require('../model/events/eventRunLog.db')
|
||||
const versionsDb = require('../model/events/eventVersions.db')
|
||||
const definitionsDb = require('../model/events/eventDefinitions.db')
|
||||
const runsModel = require('../model/events/eventRuns.model')
|
||||
const gatesDb = require('../model/events/eventPhaseGates.db')
|
||||
const recurrence = require('../events/recurrence')
|
||||
const gates = require('../events/gates')
|
||||
const spec = require('../events/spec')
|
||||
const registries = require('../modules/registries')
|
||||
const { dispatchStep } = require('../events/dispatch')
|
||||
const log = require('./logger')('event-runner')
|
||||
@@ -267,6 +278,89 @@ async function drainStep(run, step, now, carry = {}) {
|
||||
return applyFailure(run, step, result.error)
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a phase's advance gate, if it authored one.
|
||||
*
|
||||
* Called at phase entry, immediately after `materialisePhase` and BEFORE the
|
||||
* transition that makes the phase current — which is the safe order rather than
|
||||
* the tidy one. A process that died between the transition and this call would
|
||||
* leave a phase whose gate does not exist, and a missing gate does not hold a
|
||||
* phase: it advances on its steps alone, silently ignoring the condition its
|
||||
* author wrote. Opening first risks only a row for a phase this tick did not win,
|
||||
* which the winner's INSERT IGNORE then finds already correct.
|
||||
*/
|
||||
async function openGate(runId, phase, now) {
|
||||
const advance = phase?.advance
|
||||
if (!advance) return false
|
||||
|
||||
const created =
|
||||
advance.after !== undefined
|
||||
? await gatesDb.open({
|
||||
runId,
|
||||
phase: phase.key,
|
||||
kind: 'after',
|
||||
// Re-derived from the authored string rather than stored beside it:
|
||||
// `events/spec.js` owns what `'2h'` means, and this is the one caller
|
||||
// that needs the number.
|
||||
afterSeconds: spec.parseAfter(advance.after),
|
||||
now,
|
||||
})
|
||||
: await gatesDb.open({
|
||||
runId,
|
||||
phase: phase.key,
|
||||
kind: 'on',
|
||||
triggerId: advance.on,
|
||||
conditions: advance.where ?? null,
|
||||
needed: advance.count || 1,
|
||||
now,
|
||||
})
|
||||
|
||||
if (created) {
|
||||
await logDb.write({
|
||||
runId,
|
||||
kind: 'phase.gate',
|
||||
phase: phase.key,
|
||||
detail:
|
||||
advance.after !== undefined
|
||||
? { kind: 'after', after: advance.after }
|
||||
: { kind: 'on', trigger: advance.on, needed: advance.count || 1, where: gates.phrase(advance.where ?? null) },
|
||||
})
|
||||
}
|
||||
return created
|
||||
}
|
||||
|
||||
/**
|
||||
* A run whose phase has waited past `EVENT_PHASE_STALL_MS` is `stalled`.
|
||||
*
|
||||
* §E's third health value, and the first thing in this system ever to write it.
|
||||
* It is VISIBILITY and not a timeout: nothing advances, nothing fails, and a
|
||||
* human decides — which is the org lead's answer of 2026-09-02 and the reason
|
||||
* there is no authored deadline in the spec. What it must not be is quiet,
|
||||
* because a held run also holds its concurrency key, so every later occurrence
|
||||
* of the same definition goes `missed` behind it.
|
||||
*
|
||||
* `setHealth` only ever escalates, so this cannot undo a `degraded` a retry
|
||||
* earned, and the `run.health` line is written once because `setHealth` answers
|
||||
* whether it changed anything.
|
||||
*/
|
||||
async function noteStall(run, gate, now) {
|
||||
const described = gates.describe(gate, now)
|
||||
if (!described?.stalled) return false
|
||||
if (!(await runsDb.setHealth(run.id, 'stalled'))) return false
|
||||
await logDb.write({
|
||||
runId: run.id,
|
||||
kind: 'run.health',
|
||||
phase: gate.phase,
|
||||
detail: {
|
||||
to: 'stalled',
|
||||
because: `waiting ${described.elapsedSeconds}s on ${gate.trigger_id}`,
|
||||
seen: gate.tally,
|
||||
needed: gate.needed,
|
||||
},
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Advance one claimed run as far as it will go this tick.
|
||||
*
|
||||
@@ -297,6 +391,7 @@ async function advanceRun(run, now) {
|
||||
// died between the claim and here uneventful.
|
||||
const first = phases[0]
|
||||
await stepsDb.materialisePhase(run.id, first.key, first.steps || [])
|
||||
await openGate(run.id, first, now)
|
||||
if (!(await runsDb.transition(run.id, 'starting', 'running', { phase: first.key }))) return 'taken'
|
||||
phaseKey = first.key
|
||||
await logDb.write({ runId: run.id, kind: 'run.status', phase: first.key, detail: { from: 'starting', to: 'running' } })
|
||||
@@ -342,7 +437,39 @@ async function advanceRun(run, now) {
|
||||
return outcome === 'taken' ? 'taken' : 'stopped'
|
||||
}
|
||||
|
||||
// Every step of this phase is terminal.
|
||||
// Every step of this phase is terminal — which is the whole of the advance
|
||||
// test for a phase with no gate, and half of it for a phase with one.
|
||||
const { open, gate } = await gates.check(run.id, phaseKey, now)
|
||||
if (!open) {
|
||||
await noteStall(run, gate, now)
|
||||
// **Nothing is logged per tick here, deliberately.** The gate row IS the
|
||||
// state — the tally, the deadline, the last related event — and the run
|
||||
// console reads it directly. A `phase.waiting` line every fifteen seconds
|
||||
// would bury `condition.evaluated`, which is the line that actually says
|
||||
// something happened.
|
||||
return 'waiting'
|
||||
}
|
||||
// **`phase.advanced` is written by whoever made the DECISION, and a forced
|
||||
// gate's decision was not this tick's.** The emit path closes an `on` gate
|
||||
// and logs only `condition.evaluated`, so the line for that one is the
|
||||
// runner's; `gates.check` closes an `after` gate here, so that one is too.
|
||||
// A human closing it through the advance control already wrote the line,
|
||||
// with the actor and the reason — things this tick does not have — and a
|
||||
// second line from here made the console show the phase advancing twice,
|
||||
// the less informative one last. Found in the live walk.
|
||||
if (gate && gate.satisfied_by !== 'forced') {
|
||||
await logDb.write({
|
||||
runId: run.id,
|
||||
kind: 'phase.advanced',
|
||||
phase: phaseKey,
|
||||
detail: {
|
||||
because: gate.satisfied_by,
|
||||
waitedSeconds: Math.max(0, Math.round((new Date(gate.satisfied_at) - new Date(gate.entered_at)) / 1000)),
|
||||
...(gate.kind === 'on' ? { trigger: gate.trigger_id, seen: gate.tally, needed: gate.needed } : { after: gate.after_seconds }),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
await logDb.write({ runId: run.id, kind: 'phase.completed', phase: phaseKey, detail: { index: phaseIndex } })
|
||||
|
||||
const next = phases[phaseIndex + 1]
|
||||
@@ -358,6 +485,7 @@ async function advanceRun(run, now) {
|
||||
}
|
||||
|
||||
await stepsDb.materialisePhase(run.id, next.key, next.steps || [])
|
||||
await openGate(run.id, next, now)
|
||||
if (carry.holdUntil) {
|
||||
// `seq > -1` is the first step of the phase just created. Applied after
|
||||
// materialisation because that is the first moment there is a row to hold.
|
||||
|
||||
Reference in New Issue
Block a user