// Generate a deployment-scoped dashboard and local-only alert provisioning. // Same expressions serve the shipped reference and the short fault qualification. import { createHash } from "node:crypto"; import { writeFile } from "node:fs/promises"; import { resolve, join } from "node:path"; import { fileURLToPath } from "node:url"; export function build(options = {}) { const o = { deployment: "default", expectedSlots: 2, freshSeconds: 45, queueSeconds: 120, pendingSeconds: 60, intervalSeconds: 10, errorWindowSeconds: 300, ...options, }; if (!/^[A-Za-z0-9_-]{1,64}$/.test(o.deployment)) throw Error("Invalid deployment identifier"); for (const key of [ "expectedSlots", "freshSeconds", "queueSeconds", "pendingSeconds", "intervalSeconds", "errorWindowSeconds", ]) if (!Number.isInteger(o[key]) || o[key] < 1 || o[key] > 86400) throw Error(`Invalid ${key}`); if (o.intervalSeconds % 10) throw Error("Grafana evaluation interval must be a multiple of 10 seconds"); const id = `ow-${createHash("sha256").update(o.deployment).digest("hex").slice(0, 12)}`; // Stable per mounted reference file: Grafana cannot change a provisioned // dashboard's UID in place when deployment configuration is regenerated. const dashboardId = "openworker-operations"; const select = (m) => `(${m} and on(job,instance) target_info{job="openworker",openworker_deployment="${o.deployment}"})`; const fresh = (m) => `(${select(m)} > time() - ${o.freshSeconds})`; const heartbeat = fresh( "openworker_server_sampler_heartbeat_timestamp_seconds", ); const stateFresh = `(${fresh("openworker_server_state_observed_timestamp_seconds")} and on(job,instance) (${select("openworker_server_database_healthy")} == 1) and on(job,instance) ${heartbeat})`; const state = (m) => `max(${select(`openworker_server_${m}`)} and on(job,instance) ${stateFresh})`; const worker = (m) => `sum(${select(`openworker_worker_${m}`)} and on(job,instance) ${fresh("openworker_worker_heartbeat_timestamp_seconds")})`; const errors = (operation) => `(sum((${select(`openworker_operation_last_error_timestamp_seconds{openworker_operation="${operation}"}`)} > bool time() - ${o.errorWindowSeconds})) or vector(0))`; const q = { monitoringMissing: `(absent(${heartbeat}) or vector(0))`, databaseUnhealthy: `min(${select("openworker_server_database_healthy")} and on(job,instance) ${heartbeat})`, active: state("sessions_active"), queued: state("turns_queued"), approvals: state("approvals_pending"), questions: state("questions_pending"), queueAge: state("oldest_queued_seconds"), stalled: state("sessions_stalled"), unknown: state("unknown_recent"), capacity: `(${worker("capacity")} or vector(0))`, occupied: worker("active"), stateAge: `time() - max(${select("openworker_server_state_observed_timestamp_seconds")})`, stateMissing: `(absent(${stateFresh}) or vector(0))`, modelErrors: errors("model.attempt"), toolErrors: errors("tool.execute"), drops: `(sum(delta(openworker_telemetry_dropped[${o.errorWindowSeconds}s]) and on(job,instance) target_info{job="openworker",openworker_deployment="${o.deployment}"}) or vector(0))`, exportFailures: `(sum(delta(openworker_telemetry_export_failures[${o.errorWindowSeconds}s]) and on(job,instance) target_info{job="openworker",openworker_deployment="${o.deployment}"}) or vector(0))`, }; const definitions = [ [ "monitoring", "Telemetry missing", q.monitoringMissing, ">", 0, "critical", "Check sampler/API process, collector and network. Missing data is unknown, not an empty queue.", ], [ "database", "Database observer unavailable", q.databaseUnhealthy, "<", 1, "critical", "Check PostgreSQL reachability and bounded query latency. Restart the failed sampler/API after repair; this alert does not prove every API request failed.", ], [ "state", "Durable state observation stale", q.stateMissing, ">", 0, "warning", "Do not trust queue or approval counts. Check observer, database and telemetry delivery.", ], [ "capacity", "Worker capacity below target", q.capacity, "<", o.expectedSlots, "critical", "Inspect worker supervision, database and OpenShell. Restart only after checking durable lease and unknown-effect state. Capacity is advertised, not an end-to-end readiness check.", ], [ "queue", "Queued work aging", q.queueAge, ">", o.queueSeconds, "warning", "Check compatible pool capacity, permissions and parked work. Do not blindly resubmit turns.", ], [ "stalled", "Potentially stalled runs", q.stalled, ">", 0, "warning", "Inspect durable progress and long-running tools. Waiting approvals/questions are excluded. Never auto-retry an uncertain external effect.", ], [ "unknown", "Recent uncertain effects", q.unknown, ">", 0, "critical", "Inspect execution records and reconcile the external system before retrying. Clearing means no recent transitions, NOT that earlier effects were resolved.", ], [ "model", "Observed model errors", q.modelErrors, ">", 0, "warning", "Inspect sanitized traces/provider health and retry outcomes. Counts processes with a recent exported error, not errors or billing. The first error is detectable without a counter baseline.", ], [ "tool", "Observed tool errors", q.toolErrors, ">", 0, "warning", "Inspect durable tool result and environment health; distinguish a safe failure from an uncertain effect.", ], [ "drops", "Telemetry records dropped", q.drops, ">", 0, "warning", "Inspect collector latency/backpressure. The agent is not blocked by telemetry; exported evidence may be incomplete.", ], [ "export", "Telemetry export failures", q.exportFailures, ">", 0, "warning", "Check collector and TLS configuration. During complete export failure rely on the missing-telemetry alert instead.", ], ]; const rules = definitions.map( ([name, title, expr, cmp, threshold, severity, description]) => ({ uid: `${id}-${name}`, title: `${o.deployment} · ${title}`, condition: "B", data: [ { refId: "A", datasourceUid: "prometheus", relativeTimeRange: { from: 600, to: 0 }, model: { refId: "A", expr, instant: true, range: false, intervalMs: 1000, maxDataPoints: 43200, }, }, { refId: "B", datasourceUid: "__expr__", relativeTimeRange: { from: 0, to: 0 }, model: { refId: "B", type: "math", expression: `$A ${cmp} ${threshold}`, }, }, ], noDataState: "OK", execErrState: "Error", for: `${o.pendingSeconds}s`, annotations: { summary: title, description, runbook: "deploy/reference/OPERATIONS.md", __dashboardUid__: dashboardId, __panelId__: "13", }, labels: { service: "openworker", deployment: o.deployment, severity }, isPaused: false, }), ); // Grafana file provisioning cannot create a contact point without receivers. // An always-muted, loopback-only placeholder prevents external notifications. // This policy belongs ONLY in the owned single-tenant reference Grafana. const alerts = { apiVersion: 1, groups: [ { orgId: 1, name: `${id}-operations`, folder: "OpenWorker", interval: `${o.intervalSeconds}s`, rules, }, ], contactPoints: [ { orgId: 1, name: "OpenWorker local only", receivers: [ { uid: "ow-local-muted", type: "webhook", disableResolveMessage: true, settings: { url: "http://127.0.0.1:9" }, }, ], }, ], muteTimes: [ { orgId: 1, name: "openworker-local-muted", time_intervals: [ { times: [{ start_time: "00:00", end_time: "24:00" }] }, ], }, ], policies: [ { orgId: 1, receiver: "OpenWorker local only", group_by: ["alertname", "deployment"], routes: [ { receiver: "OpenWorker local only", mute_time_intervals: ["openworker-local-muted"], }, ], }, ], }; const panels = []; const stat = (title, expr, x, y, description, unit = "short") => panels.push({ id: panels.length + 1, title, type: "stat", description, datasource: { type: "prometheus", uid: "prometheus" }, gridPos: { x, y, w: 6, h: 4 }, targets: [{ refId: "A", expr, instant: true }], fieldConfig: { defaults: { unit, noValue: "Unknown", color: { mode: "fixed", fixedColor: "blue" }, }, overrides: [], }, options: { reduceOptions: { calcs: ["lastNotNull"], values: false }, colorMode: "value", graphMode: "none", textMode: "auto", }, }); stat( "Active sessions", q.active, 0, 0, "Current active turns; paused/terminal sessions excluded. Max across fresh API replicas, not a sum.", ); stat( "Queued turns", q.queued, 6, 0, "Durable queued submissions, including work waiting behind a parked turn.", ); stat( "Awaiting approval", q.approvals, 12, 0, "Unanswered approval interactions for the current turn.", ); stat( "Awaiting answer", q.questions, 18, 0, "Unanswered questions for the current turn.", ); stat( "Fresh worker slots", q.capacity, 0, 4, `Expected ${o.expectedSlots}. Only heartbeats newer than ${o.freshSeconds}s count. A heartbeat is not proof of healthy execution.`, ); stat( "Occupied slots", q.occupied, 6, 4, "Includes startup, active work and draining. Not a tool count.", ); stat( "Oldest queued turn", q.queueAge, 12, 4, "Time since durable submission, including queued work behind an approval.", "s", ); stat( "Potential stalls", q.stalled, 18, 4, "Active sessions without a recent durable event; legitimate quiet tools can trigger this. Waiting interactions excluded.", ); stat( "Recent uncertain transitions", q.unknown, 0, 8, "Count of completion/recovery transitions with unknown effects within configured observer window; not unresolved-effect inventory.", ); stat( "Database observer healthy", q.databaseUnhealthy, 6, 8, "1 = all fresh observers healthy; 0 = at least one failed. Missing = unknown, not a database-wide health assertion.", ); stat( "State observation age", q.stateAge, 12, 8, "Age of latest successful PG snapshot. Requires synchronized clocks.", "s", ); stat( "Telemetry missing (1 = yes)", q.monitoringMissing, 18, 8, "Detects absence even if the collector retains old samples.", ); panels.push({ id: 13, title: "Operational alerts", type: "alertlist", gridPos: { x: 0, y: 12, w: 24, h: 8 }, options: { viewMode: "list", groupMode: "default", maxItems: 20, sortOrder: 1, dashboardAlerts: true, alertName: "", stateFilter: { firing: true, pending: true, noData: true, error: true, normal: false, }, showOptions: "current", }, }); panels.push({ id: 14, title: "Observed errors and telemetry loss", type: "timeseries", gridPos: { x: 0, y: 20, w: 24, h: 7 }, description: "Processes with a recent exported model/tool error; telemetry-loss curves are rolling gauge deltas and can miss initial drops/resets. Absence is not proof of zero failures.", datasource: { type: "prometheus", uid: "prometheus" }, targets: [ ["Processes with model errors", q.modelErrors], ["Processes with tool errors", q.toolErrors], ["Dropped records", q.drops], ["Export failures", q.exportFailures], ].map(([name, expr], i) => ({ refId: String.fromCharCode(65 + i), expr, legendFormat: name, })), fieldConfig: { defaults: { min: 0, noValue: "Unknown" }, overrides: [] }, }); const dashboard = { uid: dashboardId, title: `OpenWorker · ${o.deployment} · Operations`, schemaVersion: 39, version: 1, editable: false, tags: ["openworker", "operations"], timezone: "browser", refresh: "10s", time: { from: "now-30m", to: "now" }, panels, links: [ { title: "Harness traces and operation metrics (all deployments)", url: "/d/openworker-harness", type: "link", }, ], }; return { options: o, id, queries: q, dashboard, alerts }; } export async function write(root, options) { const result = build(options); for (const [name, value] of [ ["operations-dashboard.json", result.dashboard], ["operations-alerts.json", result.alerts], ]) await writeFile(join(root, name), JSON.stringify(value, null, 2) + "\n", { mode: 0o600, }); return result; } if ( process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url) ) { const [root, deployment, slots] = process.argv.slice(2); if (!root || !deployment) throw Error( "Usage: node operations.mjs STATE_DIR DEPLOYMENT [EXPECTED_SLOTS]", ); await write(resolve(root), { deployment, expectedSlots: Number(slots ?? 2) }); }