Push/daemon presence: background digest + /alerts settings page
Converts the app from 'dashboard you visit' to 'system that finds you.'
Critical for the phone-first staffing shop that won't open a URL —
the system reaches out when something matters.
Daemon:
- Starts once per Bun process (guarded via globalThis sentinel)
- Default interval 15 min (configurable, min 1, max 1440)
- On each cycle, buildDigest() compares current state against prior
snapshot persisted in mcp-server/data/notification_state.json
- Events detected:
- risk_escalation: role moved to tight or critical (was ok/watch)
- deadline_approaching: staffing window falls within warn window
(default 7 days) AND deadline date differs from prior
- memory_growth: playbook_memory entries grew by >= 5 since last run
Channels (all opt-out individually via config):
- console: always on, logged to journalctl -u lakehouse-agent
- file: always on, appends JSONL to mcp-server/data/notifications.jsonl
- webhook: optional, POSTs {text, digest} to configured URL
(Slack incoming-webhook / Discord webhook / any custom endpoint)
Digest format (human-readable, fits in a Slack message):
LAKEHOUSE DIGEST — 2026-04-20 23:24
3 staffing deadlines within window:
• Production Worker — 2d to 2026-04-23 · demand 724
• Maintenance Tech — 4d to 2026-04-25 · demand 32
• Electrician — 5d to 2026-04-26 · demand 34
+779 new playbooks (total 779, 2204 endorsed names)
snapshot: 0 critical · 0 tight · $275,599,326 pipeline
/alerts page:
- Current status table (daemon state, interval, webhook, last run)
- Config form: enable toggle, interval, deadline warn window, webhook
URL + label (saved to data/notification_config.json)
- 'Fire a test digest now' button — force a cycle without waiting
- Recent digests panel shows the last 10 dispatches with full text
End-to-end verified live:
- Daemon armed successfully on startup
- First-run digest dispatched to console + file in <1s
- Events detected correctly: 3 deadlines within 7 days from real
Chicago permit data; 779 playbook entries surfaced as memory growth
- Digest text format is Slack-pastable
- Dispatch records appear in /alerts recent list
TDZ caveat: startAlertsDaemon() invocation moved to end of module so
all const/let in the alerts block evaluate before daemon reads them.
Previously failed with 'Cannot access X before initialization' when
the call lived near the top of the file. Nav added to all 6 pages:
Dashboard · Walkthrough · Architecture · Spec · Onboard · Alerts.
This commit is contained in:
root
@@ -660,6 +660,51 @@ async function main() {
|
||||
});
|
||||
}
|
||||
|
||||
// Alerts — push/daemon settings page + config API + test-fire.
|
||||
if (url.pathname === "/alerts") {
|
||||
return new Response(Bun.file(import.meta.dir + "/alerts.html"), {
|
||||
headers: { ...cors, "Content-Type": "text/html" },
|
||||
});
|
||||
}
|
||||
if (url.pathname === "/alerts/config") {
|
||||
if (req.method === "GET") {
|
||||
const cfg = await loadAlertsConfig();
|
||||
const state = await loadAlertsState();
|
||||
return ok({ config: cfg, state: { last_run_at: state.last_run_at } });
|
||||
}
|
||||
if (req.method === "POST") {
|
||||
const b = await json();
|
||||
const prev = await loadAlertsConfig();
|
||||
const next: AlertsConfig = {
|
||||
enabled: b.enabled ?? prev.enabled,
|
||||
interval_minutes: Math.max(1, Number(b.interval_minutes ?? prev.interval_minutes)),
|
||||
webhook_url: typeof b.webhook_url === "string" ? b.webhook_url.trim() || undefined : prev.webhook_url,
|
||||
webhook_label: typeof b.webhook_label === "string" ? b.webhook_label : prev.webhook_label,
|
||||
deadline_warn_days: Math.max(1, Number(b.deadline_warn_days ?? prev.deadline_warn_days)),
|
||||
};
|
||||
await saveAlertsConfig(next);
|
||||
return ok({ saved: true, config: next,
|
||||
note: "Interval change requires server restart to apply. Current running interval unchanged this cycle." });
|
||||
}
|
||||
}
|
||||
if (url.pathname === "/alerts/fire" && req.method === "POST") {
|
||||
const cfg = await loadAlertsConfig();
|
||||
const d = await buildDigest();
|
||||
if (!d) return ok({ fired: false, reason: "no events since last run" });
|
||||
const res = await dispatchDigest(d, cfg);
|
||||
return ok({ fired: true, channels: res.channels, errors: res.errors, digest: d });
|
||||
}
|
||||
if (url.pathname === "/alerts/recent" && req.method === "GET") {
|
||||
const f = Bun.file(ALERTS_LOG_PATH);
|
||||
if (!(await f.exists())) return ok({ entries: [] });
|
||||
const text = await f.text();
|
||||
const lines = text.split("\n").filter(l => l.trim());
|
||||
const last = lines.slice(-10).reverse();
|
||||
const entries: any[] = [];
|
||||
for (const l of last) { try { entries.push(JSON.parse(l)); } catch {} }
|
||||
return ok({ entries });
|
||||
}
|
||||
|
||||
// Onboard ingest — forwards multipart/form-data correctly to
|
||||
// the Rust gateway /ingest/file. The generic /api/* passthrough
|
||||
// can't handle multipart because it reads as text and forwards
|
||||
@@ -1481,6 +1526,256 @@ async function removeFromClientBlacklist(client: string, worker_id: string): Pro
|
||||
return { removed, total: filtered.length };
|
||||
}
|
||||
|
||||
// ─── Push daemon (alerts) ───────────────────────────────────────────────
|
||||
// Background interval that detects notification-worthy events, assembles
|
||||
// a digest, and dispatches to configured channels. Converts the app from
|
||||
// "dashboard you visit" to "system that finds you" — essential for the
|
||||
// phone-first shop that won't remember to open a URL.
|
||||
|
||||
const ALERTS_CFG_PATH = `${import.meta.dir}/data/notification_config.json`;
|
||||
const ALERTS_STATE_PATH = `${import.meta.dir}/data/notification_state.json`;
|
||||
const ALERTS_LOG_PATH = `${import.meta.dir}/data/notifications.jsonl`;
|
||||
|
||||
interface AlertsConfig {
|
||||
enabled: boolean;
|
||||
interval_minutes: number;
|
||||
webhook_url?: string;
|
||||
webhook_label?: string;
|
||||
deadline_warn_days: number;
|
||||
}
|
||||
interface AlertsState {
|
||||
last_run_at?: string;
|
||||
last_forecast_by_role?: Record<string, { risk: string; coverage_pct: number; earliest_staffing_deadline: string }>;
|
||||
last_playbook_entries?: number;
|
||||
last_digest?: any;
|
||||
}
|
||||
|
||||
async function loadAlertsConfig(): Promise<AlertsConfig> {
|
||||
const f = Bun.file(ALERTS_CFG_PATH);
|
||||
if (!(await f.exists())) {
|
||||
return { enabled: true, interval_minutes: 15, deadline_warn_days: 7 };
|
||||
}
|
||||
try { return await f.json() as AlertsConfig; }
|
||||
catch { return { enabled: true, interval_minutes: 15, deadline_warn_days: 7 }; }
|
||||
}
|
||||
async function saveAlertsConfig(c: AlertsConfig): Promise<void> {
|
||||
await Bun.write(ALERTS_CFG_PATH, JSON.stringify(c, null, 2));
|
||||
}
|
||||
async function loadAlertsState(): Promise<AlertsState> {
|
||||
const f = Bun.file(ALERTS_STATE_PATH);
|
||||
if (!(await f.exists())) return {};
|
||||
try { return await f.json() as AlertsState; } catch { return {}; }
|
||||
}
|
||||
async function saveAlertsState(s: AlertsState): Promise<void> {
|
||||
await Bun.write(ALERTS_STATE_PATH, JSON.stringify(s, null, 2));
|
||||
}
|
||||
|
||||
// Build a digest by diffing current state against last-observed state.
|
||||
// Returns null if there's nothing worth sending.
|
||||
async function buildDigest(): Promise<any | null> {
|
||||
const cfg = await loadAlertsConfig();
|
||||
const state = await loadAlertsState();
|
||||
|
||||
// Pull current snapshots in parallel. /intelligence/staffing_forecast
|
||||
// is a BUN route (our localhost), not on the Rust gateway — reach it
|
||||
// via in-process fetch. /vectors/playbook_memory/stats is on the
|
||||
// gateway and gets there via api().
|
||||
const bunPort = process.env.PORT || "3700";
|
||||
const [forecast, memStats] = await Promise.all([
|
||||
fetch(`http://localhost:${bunPort}/intelligence/staffing_forecast`, {
|
||||
method: "POST", headers: { "Content-Type": "application/json" }, body: "{}"
|
||||
}).then(r => r.json()).catch(() => null as any),
|
||||
api("GET", "/vectors/playbook_memory/stats").catch(() => null as any),
|
||||
]);
|
||||
|
||||
const events: any[] = [];
|
||||
|
||||
// Event: role risk status changed (new critical/tight)
|
||||
const currentByRole: Record<string, any> = {};
|
||||
const priorByRole = state.last_forecast_by_role || {};
|
||||
if (forecast && Array.isArray(forecast.forecast)) {
|
||||
for (const f of forecast.forecast) {
|
||||
currentByRole[f.role] = {
|
||||
risk: f.risk,
|
||||
coverage_pct: f.coverage_pct,
|
||||
earliest_staffing_deadline: f.earliest_staffing_deadline,
|
||||
};
|
||||
const prior = priorByRole[f.role];
|
||||
const rank: Record<string, number> = { ok: 0, watch: 1, tight: 2, critical: 3 };
|
||||
if (!prior || (rank[f.risk] ?? 0) > (rank[prior.risk] ?? 0)) {
|
||||
// Risk got worse (or new role we haven't seen)
|
||||
if (f.risk === "critical" || f.risk === "tight") {
|
||||
events.push({
|
||||
kind: "risk_escalation",
|
||||
role: f.role,
|
||||
risk: f.risk,
|
||||
coverage_pct: f.coverage_pct,
|
||||
demand: f.demand_workers,
|
||||
available: f.bench_available,
|
||||
prior_risk: prior?.risk ?? null,
|
||||
});
|
||||
}
|
||||
}
|
||||
// Event: staffing deadline within N days that wasn't there before
|
||||
const d = f.days_to_deadline;
|
||||
if (d !== undefined && d >= 0 && d <= cfg.deadline_warn_days) {
|
||||
const priorD = prior?.earliest_staffing_deadline;
|
||||
if (priorD !== f.earliest_staffing_deadline) {
|
||||
events.push({
|
||||
kind: "deadline_approaching",
|
||||
role: f.role,
|
||||
days_to_deadline: d,
|
||||
date: f.earliest_staffing_deadline,
|
||||
demand: f.demand_workers,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Event: playbook memory grew significantly since last check
|
||||
const nowEntries = memStats?.entries ?? 0;
|
||||
const priorEntries = state.last_playbook_entries ?? 0;
|
||||
const grewBy = nowEntries - priorEntries;
|
||||
if (grewBy >= 5) {
|
||||
events.push({
|
||||
kind: "memory_growth",
|
||||
new_entries: grewBy,
|
||||
total_entries: nowEntries,
|
||||
total_endorsed_names: memStats?.total_names_endorsed ?? 0,
|
||||
});
|
||||
}
|
||||
|
||||
// Only return a digest if there's something to say. First-ever run is
|
||||
// a special case: surface the snapshot as a "welcome" digest.
|
||||
const isFirstRun = !state.last_run_at;
|
||||
if (events.length === 0 && !isFirstRun) return null;
|
||||
|
||||
const digest = {
|
||||
generated_at: new Date().toISOString(),
|
||||
is_first_run: isFirstRun,
|
||||
events,
|
||||
snapshot: {
|
||||
forecast_roles: Object.keys(currentByRole).length,
|
||||
critical: forecast?.critical_roles ?? 0,
|
||||
tight: forecast?.tight_roles ?? 0,
|
||||
playbook_entries: nowEntries,
|
||||
permits_30d: forecast?.permit_count ?? 0,
|
||||
construction_pipeline_usd: forecast?.total_cost ?? 0,
|
||||
},
|
||||
};
|
||||
|
||||
// Persist the updated state for next diff
|
||||
await saveAlertsState({
|
||||
last_run_at: digest.generated_at,
|
||||
last_forecast_by_role: currentByRole,
|
||||
last_playbook_entries: nowEntries,
|
||||
last_digest: digest,
|
||||
});
|
||||
|
||||
return digest;
|
||||
}
|
||||
|
||||
function formatDigestText(d: any): string {
|
||||
const lines: string[] = [];
|
||||
lines.push(`LAKEHOUSE DIGEST — ${d.generated_at.slice(0, 16).replace("T", " ")}`);
|
||||
lines.push("");
|
||||
if (d.is_first_run) {
|
||||
lines.push(`[initial snapshot] · ${d.snapshot.forecast_roles} roles tracked · `
|
||||
+ `${d.snapshot.playbook_entries} playbooks in memory · `
|
||||
+ `${d.snapshot.permits_30d} permits last 30d`);
|
||||
lines.push("");
|
||||
}
|
||||
const risk = d.events.filter((e: any) => e.kind === "risk_escalation");
|
||||
if (risk.length) {
|
||||
lines.push(`${risk.length} role${risk.length !== 1 ? "s" : ""} escalated to ${risk.map((r: any) => r.risk).filter((v: string, i: number, a: string[]) => a.indexOf(v) === i).join("/")}:`);
|
||||
for (const e of risk.slice(0, 5)) {
|
||||
lines.push(` • ${e.role} — coverage ${e.coverage_pct}% (${e.available}/${e.demand})${e.prior_risk ? ` · was ${e.prior_risk}` : " · new"}`);
|
||||
}
|
||||
lines.push("");
|
||||
}
|
||||
const dead = d.events.filter((e: any) => e.kind === "deadline_approaching");
|
||||
if (dead.length) {
|
||||
lines.push(`${dead.length} staffing deadline${dead.length !== 1 ? "s" : ""} within window:`);
|
||||
for (const e of dead.slice(0, 5)) {
|
||||
lines.push(` • ${e.role} — ${e.days_to_deadline}d to ${e.date} · demand ${e.demand}`);
|
||||
}
|
||||
lines.push("");
|
||||
}
|
||||
const mem = d.events.filter((e: any) => e.kind === "memory_growth");
|
||||
for (const e of mem) {
|
||||
lines.push(`+${e.new_entries} new playbooks (total ${e.total_entries}, ${e.total_endorsed_names} endorsed names)`);
|
||||
}
|
||||
lines.push(`snapshot: ${d.snapshot.critical} critical · ${d.snapshot.tight} tight · `
|
||||
+ `$${(d.snapshot.construction_pipeline_usd || 0).toLocaleString("en-US", { maximumFractionDigits: 0 })} pipeline`);
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
async function dispatchDigest(d: any, cfg: AlertsConfig): Promise<{ channels: string[]; errors: string[] }> {
|
||||
const channels: string[] = [];
|
||||
const errors: string[] = [];
|
||||
const text = formatDigestText(d);
|
||||
|
||||
// Channel 1: console
|
||||
console.log(`[alerts] ${text.split("\n").join(" | ")}`);
|
||||
channels.push("console");
|
||||
|
||||
// Channel 2: JSONL file (always-on audit)
|
||||
try {
|
||||
await Bun.write(ALERTS_LOG_PATH,
|
||||
(await Bun.file(ALERTS_LOG_PATH).exists() ? await Bun.file(ALERTS_LOG_PATH).text() : "")
|
||||
+ JSON.stringify({ at: d.generated_at, text, digest: d }) + "\n"
|
||||
);
|
||||
channels.push("file");
|
||||
} catch (e: any) { errors.push(`file: ${e.message}`); }
|
||||
|
||||
// Channel 3: webhook (opt-in)
|
||||
if (cfg.webhook_url) {
|
||||
try {
|
||||
const r = await fetch(cfg.webhook_url, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ text, digest: d }),
|
||||
});
|
||||
if (r.ok) channels.push("webhook");
|
||||
else errors.push(`webhook ${r.status}: ${(await r.text()).slice(0, 200)}`);
|
||||
} catch (e: any) { errors.push(`webhook: ${e.message}`); }
|
||||
}
|
||||
|
||||
return { channels, errors };
|
||||
}
|
||||
|
||||
// Background daemon — kicked off once on module init. Guard via a
|
||||
// globalThis sentinel so the startAlertsDaemon() call from near the
|
||||
// top of the file (before this block evaluates) doesn't hit a temporal
|
||||
// dead zone on a let/const binding.
|
||||
async function startAlertsDaemon() {
|
||||
const g = globalThis as any;
|
||||
if (g.__lakehouse_alerts_armed) return;
|
||||
g.__lakehouse_alerts_armed = true;
|
||||
const cfg = await loadAlertsConfig();
|
||||
if (!cfg.enabled) {
|
||||
console.log("[alerts] daemon disabled via config");
|
||||
return;
|
||||
}
|
||||
const ms = Math.max(60, cfg.interval_minutes * 60) * 1000;
|
||||
console.log(`[alerts] daemon armed · interval ${cfg.interval_minutes}min · webhook ${cfg.webhook_url ? "configured" : "disabled"}`);
|
||||
// Fire once shortly after startup, then on interval.
|
||||
setTimeout(runAlertsOnce, 10_000);
|
||||
setInterval(runAlertsOnce, ms);
|
||||
}
|
||||
async function runAlertsOnce() {
|
||||
try {
|
||||
const cfg = await loadAlertsConfig();
|
||||
if (!cfg.enabled) return;
|
||||
const d = await buildDigest();
|
||||
if (!d) return;
|
||||
await dispatchDigest(d, cfg);
|
||||
} catch (e: any) {
|
||||
console.error(`[alerts] cycle error: ${e.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Seed playbook_memory from a filled contract so the next hybrid query
|
||||
// ranks against it. Used by both runWeekSimulation (per-day) and the /log
|
||||
// endpoint (per manual logging). Fail-soft — seeding is best-effort.
|
||||
@@ -1622,3 +1917,11 @@ async function runWeekSimulation() {
|
||||
|
||||
return { days: results, summary };
|
||||
}
|
||||
|
||||
// Kick off the push/alerts daemon once per process. Placed at the END of
|
||||
// the module so all const/let declarations in the alerts block (paths,
|
||||
// helpers, etc.) have evaluated before the daemon reads them. Calling
|
||||
// from earlier in the file would hit a temporal dead zone on these
|
||||
// bindings.
|
||||
startAlertsDaemon().catch(e => console.error(`[alerts] startup error: ${e.message}`));
|
||||
|
||||
|
||||
Reference in New Issue
Block a user