// ── Starting a run from a trigger: the emit path's third subscriber ──────── // // Rust PLAN_REDESIGNS §14.3.12 (D370), and EVENTS.md §E "Starting a run from a // trigger". *"When anyone destroys Bradley on Main, start Launch Site // Lockdown"*; *"when a champion's boss appears, start Champion's Call"*. Before // this a trigger could only advance a phase of a run already going (`gates.js`) // or send a notice (`engine.dispatch`); now a definition's published version may // carry `starts`, and a firing that matches one creates a run of it, now. // // **Same seam and posture as the other two subscribers**: called from // `engagementEmit.emit`, never awaited, never allowed to reject. Whether some // event wanted to start is core's business, and a failure of core's must not // become the module's control flow. // // **Four rules decide whether a matching firing starts anything:** // // 1. The version must have passed its dry run. A trigger start is unattended, // as a scheduled one is, and §K's gate falls on the path where nobody is. // 2. One run at a time per definition. A firing while a run of it is going // starts nothing and is counted. // 3. A cooldown after the last trigger-started run (default 30 minutes, 0 to // 7 days). A firing inside it starts nothing and is counted. Rules 2 and 3 // are what stop twelve road Bradleys dying in a minute starting twelve runs. // 4. Core's own `event.*` triggers never start a run (refused at save, and // skipped here as well), so a run starting can never start another. // // Rules 2 and 3 are ONE conditional UPDATE (`eventTriggerStarts.db.claim`); the // occurrence key `UNIQUE (definition_id, scope, scheduled_for)` with // `scheduled_for` = the firing's time is what makes one firing reported twice // (two instances ingesting one feed) create one run. const definitionsDb = require('../model/events/eventDefinitions.db') const startsDb = require('../model/events/eventTriggerStarts.db') const runsModel = require('../model/events/eventRuns.model') const conditions = require('../engagement/conditions') const registries = require('../modules/registries') const spec = require('./spec') const gates = require('./gates') const log = require('../utils/logger')('event-starts') // What of a firing's data is kept on the run. The run row is admin-only, and it // is the answer to "why did this run happen", so the trigger's DECLARED // variables are kept and nothing else a module happened to send. Bounded, so // one oversized payload cannot make the insert the thing that fails. const MAX_ORIGIN_BYTES = 4096 /** The firing's time, in whole seconds (the column's precision), never in the future. */ function firingTime(occurredAt, now) { const parsed = occurredAt ? new Date(occurredAt) : now const at = Number.isNaN(parsed.getTime()) || parsed > now ? now : parsed return new Date(Math.floor(at.getTime() / 1000) * 1000) } const sameSecond = (a, b) => a && b && Math.floor(new Date(a).getTime() / 1000) === Math.floor(new Date(b).getTime() / 1000) /** The firing as the run will remember it. */ function originOf(event, declaration, start, at) { const declared = (declaration?.variables || []).map((v) => v.name) const data = event.data || {} let variables = Object.fromEntries(declared.filter((n) => n in data).map((n) => [n, data[n]])) if (Buffer.byteLength(JSON.stringify(variables), 'utf8') > MAX_ORIGIN_BYTES) { // Too big to keep whole: keep the ones the start's condition named, which // are the reason it matched, and say the rest were dropped. const named = gates.variablesIn(start.where) variables = Object.fromEntries(named.filter((n) => n in data).map((n) => [n, data[n]])) variables._truncated = true } return { kind: 'trigger', triggerId: event.triggerId, label: declaration?.label || event.triggerId, occurredAt: at.toISOString(), subject: event.subject ?? null, // The start's condition in the builder's own words, so the console can say // what it matched on without re-reading the version. matched: gates.phrase(start.where), variables, } } /** * Why a firing that matched did not get the claim. A duplicate of the firing * that started the last run is not counted: it is the same thing happening once, * reported twice. */ async function whyNot(definitionId, at, now) { const state = await startsDb.forDefinition(definitionId) if (state && sameSecond(state.last_started_at, at)) return 'duplicate' if (await startsDb.isGoing(definitionId, now)) return 'running' // A live claim is another firing starting it this instant: the run is going in // every sense but the row. if (state?.claimed_at && now - new Date(state.claimed_at) < startsDb.CLAIM_STALE_MS) return 'running' return 'cooldown' } /** One definition, one firing. Answers what happened, for the summary and the tests. */ async function consider(definition, event, at, now) { const starts = (definition.version_spec?.starts || []).filter((s) => s.on === event.triggerId && !s.dormant) const start = starts.find((s) => conditions.evaluate(s.where, event.data || {})) if (!start) return 'unmatched' if (now - at > Number(definition.grace_seconds || 0) * 1000) { // A firing older than the grace window would make a run the missed sweep // takes on its next tick. Said, and not started. log.warn('a firing too old to start a run', { definition: definition.id, trigger: event.triggerId, at }) return 'stale' } if (!definition.version_verified_at) { await startsDb.ignore(definition.id, 'unverified', now) log.warn('trigger start held: the published version has never been verified', { definition: definition.id, title: definition.title, trigger: event.triggerId, }) return 'unverified' } const cooldownSeconds = spec.parseCooldown(start.cooldown ?? spec.DEFAULT_COOLDOWN) ?? 0 if (!(await startsDb.claim(definition.id, { at, cooldownSeconds, now }))) { const reason = await whyNot(definition.id, at, now) if (reason !== 'duplicate') await startsDb.ignore(definition.id, reason, now) return reason } const declaration = registries.eventTrigger(event.triggerId) let result try { result = await runsModel.create( definition.id, // Scope is empty, as a scheduled occurrence's is (org lead, 2026-09-02): a // scope registry does not exist, and the condition is what narrows a start // to one server. { scope: '', scheduledFor: at, source: 'trigger', origin: originOf(event, declaration, start, at) }, null, ) } catch (err) { await startsDb.release(definition.id) throw err } if (!result.ok) { await startsDb.release(definition.id) if (result.code === 'unverified') await startsDb.ignore(definition.id, 'unverified', now) log.warn('trigger start refused', { definition: definition.id, errors: result.errors }) return 'refused' } if (!result.created) { // The occurrence already existed: this firing was reported twice, and the // claimant that made the run has recorded it. Nothing to count. await startsDb.release(definition.id) return 'duplicate' } await startsDb.started(definition.id, { at, runId: result.run.id }) log.info('run started by a trigger', { definition: definition.id, run: result.run.id, trigger: event.triggerId }) return 'started' } /** * The trigger stream's third subscriber. * * The common answer is one indexed-by-nothing query over the ready definitions * returning no rows: a deployment has tens of definitions, not thousands, and a * JSON column cannot carry an index. If that ever changes, a side table of * (trigger, definition) written at publish is the fix. */ async function observe(event, { now = new Date() } = {}) { const summary = { definitions: 0, started: 0, ignored: 0 } try { if (!event?.triggerId || event.triggerId.startsWith(spec.CORE_EVENT_PREFIX)) return summary const candidates = await definitionsDb.findStartingOn(event.triggerId) summary.definitions = candidates.length if (!candidates.length) return summary const at = firingTime(event.occurredAt, now) for (const definition of candidates) { try { const outcome = await consider(definition, event, at, now) if (outcome === 'started') summary.started += 1 if (['running', 'cooldown', 'unverified'].includes(outcome)) summary.ignored += 1 } catch (err) { log.error('trigger start failed', { definition: definition.id, trigger: event.triggerId, message: err.message }) } } } catch (err) { log.error('trigger start observation failed', { trigger: event?.triggerId, message: err.message }) } return summary } /** * The editor's line about a definition's trigger starts, from the state row. * Null when no trigger has ever tried to start it. */ function describeState(row) { if (!row) return null return { lastStartedAt: row.last_started_at || null, lastRunId: row.last_run_id || null, ignoredRunning: Number(row.ignored_running || 0), ignoredCooldown: Number(row.ignored_cooldown || 0), ignoredUnverified: Number(row.ignored_unverified || 0), lastIgnoredAt: row.last_ignored_at || null, lastIgnoredReason: row.last_ignored_reason || null, } } module.exports = { observe, consider, describeState, firingTime, originOf, MAX_ORIGIN_BYTES }