Timeline: durable alerts (open once, resolve when cleared)

changelog.d/timeline-alerts.added.md

@@ -0,0 +1 @@
+The Timeline plugin can raise durable alerts. `timeline/alerts` (read-only) folds the current window into an alert set: a Bot that fails most of its turns, a request left unanswered, or a handoff waiting too long opens ONE alert — opened once, not repeated every poll — and the alert resolves when the condition clears, so a panel or a routine sees what is wrong and since when.

docs/subsystems/timeline.md

@@ -27,6 +27,7 @@ Everything in `plugins/timeline/`: the recorder, the trace file, the read model
 | `plugins/timeline/src/recorder.js` | Feed events to compact trace records; rebuilds every field from a strict alphabet. |
 | `plugins/timeline/src/store.js` | The trace: single writer, write-tmp + fsync + rename, capped segments. |
 | `plugins/timeline/src/window.js` | Pure read model: records to lanes, turns, tools, arrows, edges, utilization, the token spend per Bot (#357) and per-Bot outcomes and recent failures (#471). |
+| `plugins/timeline/src/alerts.js` | Durable alerts (#357/#469): pure `createAlertTracker` opening a condition once and resolving it when it clears. |
 | `plugins/timeline/client.js` | Browser entry (`@flybot/timeline`): the 2 s poller exposed as `ctx.flybot.timeline`, the geometry functions (`layoutSwimlanes`, `layoutMap`, `waitRows`, `utilizationRows`) and the panel (swimlanes, who waits on whom, utilization), opened from its `flybot.sidebar.section` row, the only entry (not in the + menu, #269). |
 | `plugins/timeline/src/sanitize.js` | The cleaners for ids, labels, codes, times, counts. |
 | `plugins/timeline/src/otel.js` | OTLP export: maps trace records to OpenTelemetry GenAI spans (`spansFrom`, `otlpPayload`) and a cursor exporter (`createExporter`). Off unless `config.exporter` is set. |
@@ -36,6 +37,7 @@ Everything in `plugins/timeline/`: the recorder, the trace file, the read model
 - `plugins/timeline/index.js` — loaded as `@flybot/timeline`.
 - Endpoint `timeline/window` (wire name `flybot/timeline/window`), registered `{ read: true }` so a read-only host still answers. Body `{ windowMs?, to? }`; window 30 s to 24 h, default 15 min.
 - Endpoint `timeline/export` (#357, read-only): the current trace mapped to one OTLP/HTTP JSON payload (`resourceSpans`), the same mapping the timer exporter uses. Ids only unless the body sets `names: true` (#483); `serviceName` overrides the default. The panel or an automation can pull it on demand instead of waiting for the next flush.
+- Endpoint `timeline/alerts` (#357/#469, read-only): folds the current window into a durable alert set. A condition that holds opens ONE alert (`errors`, `waiting`, `stuck`) and it resolves when the condition clears; the answer is `{ open, resolved, changed }`, so a panel or a routine sees what is wrong and since when without re-deriving it every poll. Body `{ windowMs?, to? }`.
 
 ## Invariants
 

plugins/timeline/src/alerts.js

@@ -0,0 +1,98 @@
+// Durable alerts over the timeline window (#357/#469): a condition that stays true raises ONE
+// alert (opened once, not every poll), and the alert resolves when the condition clears. The
+// tracker is a small state machine the caller feeds each window; a panel or a routine reads the
+// open set to see what is wrong and since when, instead of re-deriving it from the drawing on
+// every request. Pure state over a window answer; it writes nothing and touches no Session.
+const RESOLVED_KEPT = 50
+
+/** The conditions a window is checked for. Each entry: a Bot predicate, kind and a detail line. */
+const RULES = [
+  {
+    kind: 'errors',
+    // A Bot that fails most of its turns (enough of them to mean something).
+    of(bot) { return bot.turns >= 3 && bot.errors / bot.turns >= 0.5 ? `${bot.errors} of ${bot.turns} turns failed` : undefined },
+  },
+  {
+    kind: 'waiting',
+    // A Bot whose sent request has not been answered.
+    of(bot) { return bot.waiting > 0 ? `${bot.waiting} request(s) unanswered` : undefined },
+  },
+  {
+    kind: 'stuck',
+    // A Bot involved in a request that has waited too long (from the longest open wait per side).
+    of(bot, context) { const waited = context.stuckBy.get(bot.id); return waited !== undefined ? `a request has waited ${Math.round(waited / 60000)}m` : undefined },
+  },
+]
+
+const keyOf = (botId, kind) => `${botId}\u0000${kind}`
+
+/**
+ * Builds the tracker. `evaluate(window)` folds one window into the alert set and resolves
+ * `{ open, resolved, changed }`; `changed` names the alerts that opened or resolved this call.
+ * A resolved alert is kept for a short while so a reader sees it cleared.
+ */
+export function createAlertTracker({ now = Date.now, stuckMs = 300_000 } = {}) {
+  /** @type {Map<string, {id:string, kind:string, botId:string, name:string, since:number, detail:string}>} */
+  const active = new Map()
+  /** @type {{id:string, kind:string, botId:string, name:string, since:number, detail:string, at:number}[]} */
+  const resolved = []
+
+  const conditionsOf = (window) => {
+    const lanes = new Map((window.lanes ?? []).map(lane => [lane.id, lane]))
+    const waitingBy = new Map()
+    for (const edge of window.edges ?? []) {
+      if (edge?.unanswered > 0 && typeof edge.from === 'string') waitingBy.set(edge.from, (waitingBy.get(edge.from) ?? 0) + edge.unanswered)
+    }
+    const stuckBy = new Map()
+    for (const arrow of window.exchanges ?? []) {
+      if (arrow?.waiting !== true) continue
+      const waited = Math.max(0, now() - (arrow.t ?? now()))
+      if (waited < stuckMs) continue
+      for (const side of [arrow.from, arrow.to]) if (typeof side === 'string') stuckBy.set(side, Math.max(stuckBy.get(side) ?? 0, waited))
+    }
+    const bots = (window.utilization ?? []).map(use => ({
+      id: use.lane,
+      name: lanes.get(use.lane)?.name ?? use.lane,
+      turns: use.turns ?? 0,
+      errors: use.errors ?? 0,
+      running: use.running === true,
+      waiting: waitingBy.get(use.lane) ?? 0,
+    }))
+    return { bots, context: { stuckBy } }
+  }
+
+  function evaluate(window = {}) {
+    const { bots, context } = conditionsOf(window)
+    const seen = new Set()
+    const changed = []
+    const at = now()
+    for (const bot of bots) {
+      for (const rule of RULES) {
+        const detail = rule.of(bot, context)
+        if (detail === undefined) continue
+        const id = keyOf(bot.id, rule.kind)
+        seen.add(id)
+        const found = active.get(id)
+        if (found === undefined) {
+          const alert = { id, kind: rule.kind, botId: bot.id, name: bot.name, since: at, detail }
+          active.set(id, alert)
+          changed.push({ event: 'open', ...alert })
+        } else if (found.detail !== detail || found.name !== bot.name) {
+          active.set(id, { ...found, detail, name: bot.name })
+          changed.push({ event: 'update', ...active.get(id) })
+        }
+      }
+    }
+    for (const [id, alert] of [...active]) {
+      if (seen.has(id)) continue
+      active.delete(id)
+      const record = { ...alert, at }
+      resolved.unshift(record)
+      if (resolved.length > RESOLVED_KEPT) resolved.length = RESOLVED_KEPT
+      changed.push({ event: 'resolve', ...record })
+    }
+    return { open: [...active.values()], resolved: resolved.slice(0, RESOLVED_KEPT), changed }
+  }
+
+  return { evaluate, open: () => [...active.values()] }
+}

plugins/timeline/src/timeline.js

@@ -3,9 +3,11 @@ import { createRecorder } from './recorder.js'
 import { openTrace, TraceLocked } from './store.js'
 import { buildWindow, clampWindow } from './window.js'
 import { createExporter, otlpPayload, spansFrom } from './otel.js'
+import { createAlertTracker } from './alerts.js'
 
 export const ENDPOINT = 'timeline/window'
 export const EXPORT_ENDPOINT = 'timeline/export'
+export const ALERTS_ENDPOINT = 'timeline/alerts'
 
 export function createTimeline(adapter, { dir, persist = true, now = Date.now, exporter, ...limits } = {}) {
   let trace
@@ -49,12 +51,18 @@ export function createTimeline(adapter, { dir, persist = true, now = Date.now, e
   // sends another home's names by accident.
   const exportWindow = (body = {}) => otlpPayload(spansFrom(trace.records(), { names: body?.names === true ? trace.names() : {} }), { serviceName: body?.serviceName })
   const offExport = adapter.handle(EXPORT_ENDPOINT, exportWindow)
+  // Durable alerts (#357/#469): each call folds the current window into the alert set, opening a
+  // condition once and resolving it when it clears. Read-only.
+  const alertTracker = createAlertTracker({ now })
+  const alerts = (body = {}) => alertTracker.evaluate(window(body))
+  const offAlerts = adapter.handle(ALERTS_ENDPOINT, alerts)
   return {
-    trace, recorder, window, exportNow, exportWindow,
+    trace, recorder, window, exportNow, exportWindow, alerts,
     dispose() {
       if (exportTimer !== undefined) clearInterval(exportTimer)
       offEndpoint()
       offExport()
+      offAlerts()
       off()
       trace.close()
     },

plugins/timeline/test/alerts.test.js

@@ -0,0 +1,66 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { createAlertTracker } from '../src/alerts.js'
+
+const win = (utilization, extra = {}) => ({
+  lanes: [{ id: 'a', kind: 'bot', name: 'Chief', main: true }, { id: 'b', kind: 'bot', name: 'Writer' }],
+  utilization, edges: [], exchanges: [], ...extra,
+})
+const failing = [{ lane: 'a', windowMs: 1000, busyMs: 100, turns: 4, errors: 3, running: false }]
+
+test('an alert opens once, is not repeated while it holds, and resolves when the condition clears', () => {
+  const tracker = createAlertTracker({ now: () => 1000 })
+  const first = tracker.evaluate(win(failing))
+  assert.deepEqual(first.changed.map(change => [change.event, change.kind, change.botId]), [['open', 'errors', 'a']])
+  assert.equal(first.open.length, 1)
+  assert.deepEqual(first.open[0], { id: 'a\u0000errors', kind: 'errors', botId: 'a', name: 'Chief', since: 1000, detail: '3 of 4 turns failed' })
+  // The same condition on the next window is not re-announced.
+  const second = tracker.evaluate(win(failing))
+  assert.deepEqual(second.changed, [])
+  assert.equal(second.open.length, 1)
+  // Cleared: it resolves and moves to the recent list.
+  const cleared = tracker.evaluate(win([{ lane: 'a', windowMs: 1000, busyMs: 100, turns: 4, errors: 0, running: false }]))
+  assert.deepEqual(cleared.changed.map(change => [change.event, change.kind]), [['resolve', 'errors']])
+  assert.equal(cleared.open.length, 0)
+  assert.equal(cleared.resolved.length, 1)
+  assert.equal(cleared.resolved[0].botId, 'a')
+})
+
+test('a changed detail updates the alert without reopening it', () => {
+  const tracker = createAlertTracker({ now: () => 1000 })
+  tracker.evaluate(win(failing))
+  const worse = tracker.evaluate(win([{ lane: 'a', windowMs: 1000, busyMs: 100, turns: 6, errors: 5, running: false }]))
+  assert.deepEqual(worse.changed.map(change => [change.event, change.detail]), [['update', '5 of 6 turns failed']])
+  assert.equal(worse.open.length, 1)
+})
+
+test('an unanswered request and a too-long wait each raise their own alert', () => {
+  const clock = 1_000_000
+  const tracker = createAlertTracker({ now: () => clock, stuckMs: 300_000 })
+  const opened = tracker.evaluate(win([
+    { lane: 'a', windowMs: 1000, busyMs: 100, turns: 4, errors: 3, running: false },
+    { lane: 'b', windowMs: 1000, busyMs: 0, turns: 0, errors: 0, running: false },
+  ], {
+    edges: [{ from: 'b', to: 'a', unanswered: 2 }],
+    exchanges: [{ id: 'x1', from: 'a', to: 'b', waiting: true, t: clock - 600_000 }],
+  }))
+  const kinds = opened.open.map(alert => `${alert.botId}:${alert.kind}`).sort()
+  assert.deepEqual(kinds, ['a:errors', 'a:stuck', 'b:stuck', 'b:waiting'])
+  const stuck = opened.open.find(alert => alert.kind === 'stuck')
+  assert.equal(stuck.detail, 'a request has waited 10m')
+})
+
+test('a short wait is not an alert', () => {
+  const clock = 1_000_000
+  const tracker = createAlertTracker({ now: () => clock, stuckMs: 300_000 })
+  const result = tracker.evaluate(win([{ lane: 'a', windowMs: 1000, busyMs: 0, turns: 0, errors: 0, running: false }], { exchanges: [{ id: 'x1', from: 'a', to: 'b', waiting: true, t: clock - 60_000 }] }))
+  assert.deepEqual(result.open, [])
+})
+
+test('an empty window opens nothing and resolves what it held', () => {
+  const tracker = createAlertTracker({ now: () => 1000 })
+  tracker.evaluate(win(failing))
+  const empty = tracker.evaluate({})
+  assert.deepEqual(empty.open, [])
+  assert.equal(empty.resolved.length, 1)
+})