diff --git a/docs/alert_automation_plugin.md b/docs/alert_automation_plugin.md index a25f830..c9df858 100644 --- a/docs/alert_automation_plugin.md +++ b/docs/alert_automation_plugin.md @@ -182,8 +182,29 @@ JSON/body limits and full-batch intake are enforced before acknowledgement. Repeated deliveries share the same incident key. `both` mode can reconcile source state through polling while using webhook events for prompt intake. +The authenticated HTTP handler validates and persists the batch, then wakes +work under the context owned by the native service lifecycle. It does not run +the Agent under the inbound HTTP client's Gateway scopes, and it does not grant +that client `operator.write`. This separation is required for native subagent +dispatch from a plugin-authenticated Webhook route. + +In Webhook-only mode, source recovery also clears the confirmed imported alert +from the Mosaic UI. The service retries that reconciliation on its maintenance +timer without fetching Alertmanager. A newer unresolved occurrence with the +same fingerprint prevents an older recovery from clearing its UI entry. +`pollSeconds` still controls periodic queue/recovery maintenance in this mode; +new Webhooks wake the service immediately and do not wait for a source poll. + ## Task state, report integrity and shutdown +The recipient prompt hook reads its report binding and integrity-checked full +report from the durable state directory. Agent hook registries can be created +without starting the Gateway service; follow-up context must not depend on a +service instance held in one plugin registration's memory. Each applied context +has a witness with the actual input prompt, receiving session, incident and full +report hash/context. If a native channel turn does not expose a run ID, the +witness has a separate generated identifier and a null native run ID. + The queue uses fingerprint **and** alert start time, matching the legacy worker. An alert firing again later is a new occurrence. Polling silence/inhibition does not imply resolution. Webhook `resolved` events and complete polling snapshots diff --git a/helm/mosaic-stack/files/agent-documentation.tar.gz b/helm/mosaic-stack/files/agent-documentation.tar.gz index 284573e..04ce464 100644 Binary files a/helm/mosaic-stack/files/agent-documentation.tar.gz and b/helm/mosaic-stack/files/agent-documentation.tar.gz differ diff --git a/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/core.ts b/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/core.ts index ab1df76..2b7b5c0 100644 --- a/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/core.ts +++ b/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/core.ts @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 import { createHash } from "node:crypto"; +import { AsyncResource } from "node:async_hooks"; import { mkdirSync, readFileSync, @@ -271,6 +272,56 @@ export function matches(alert: Json, labels: Record) { ); } +function reportContextFromDatabase( + db: DatabaseSync, + root: string, + sessionKey: string, + prompt = "", +) { + const records = db + .prepare( + "SELECT * FROM report_contexts WHERE session_key=? OR substr(?,1,length(session_key)+1)=session_key||':' ORDER BY created_at DESC", + ) + .all(sessionKey, sessionKey) as Json[]; + const selected = + records.find( + (r) => prompt.includes(r.incident_id) || prompt.includes(r.run_id), + ) ?? records[0]; + if (!selected) return undefined; + const report = readFileSync( + path.join(root, "incidents", selected.incident_id, "published-report.txt"), + "utf8", + ); + if (sha256(report) !== selected.report_sha256) + throw new Error("Stored report context integrity mismatch"); + return { ...selected, report }; +} + +/** Agent hook registries need not own a started Gateway service instance. */ +export function readStoredReportContext( + root: string, + sessionKey: string, + prompt = "", +) { + const filename = path.join(root, "state.sqlite3"); + if (!existsSync(filename)) return undefined; + const db = new DatabaseSync(filename, { readOnly: true }); + try { + db.exec("PRAGMA busy_timeout=5000"); + if ( + !db + .prepare( + "SELECT name FROM sqlite_schema WHERE type='table' AND name='report_contexts'", + ) + .get() + ) + return undefined; + return reportContextFromDatabase(db, root, sessionKey, prompt); + } finally { + db.close(); + } +} + /** Same incident/event schema as the legacy worker, permitting an explicit stopped-worker migration. */ export class Store { root: string; @@ -414,23 +465,7 @@ export class Store { .run(id, sessionKey, runId, sha256(report), now()); } reportContext(sessionKey: string, prompt = "") { - const records = this.db - .prepare( - "SELECT * FROM report_contexts WHERE session_key=? OR substr(?,1,length(session_key)+1)=session_key||':' ORDER BY created_at DESC", - ) - .all(sessionKey, sessionKey) as Json[]; - const selected = - records.find( - (r) => prompt.includes(r.incident_id) || prompt.includes(r.run_id), - ) ?? records[0]; - if (!selected) return undefined; - const report = readFileSync( - path.join(this.directory(selected.incident_id), "published-report.txt"), - "utf8", - ); - if (sha256(report) !== selected.report_sha256) - throw new Error("Stored report context integrity mismatch"); - return { ...selected, report }; + return reportContextFromDatabase(this.db, this.root, sessionKey, prompt); } close() { this.db.close(); @@ -598,6 +633,9 @@ export class AlertService { timer: ReturnType | undefined; polling = false; processing: Promise | undefined; + // Created by the native service lifecycle, before any inbound HTTP request. + // Background work must retain service ownership rather than a webhook caller's scopes. + workContext = new AsyncResource("mosaic-alert-automation-service"); logger: { info: (text: string) => void; error: (text: string) => void }; constructor( config: Config, @@ -630,20 +668,24 @@ export class AlertService { } start() { if (this.timer) return; - const tick = () => { + this.timer = setInterval(() => this.wake(), this.config.pollSeconds * 1000); + this.timer.unref(); + this.wake(); + } + wake() { + if (this.controller.signal.aborted) return; + this.workContext.runInAsyncScope(() => { void this.tick().catch((error) => this.logger.error(`alert-automation tick: ${error.name}`), ); - }; - this.timer = setInterval(tick, this.config.pollSeconds * 1000); - this.timer.unref(); - tick(); + }); } async tick() { if (this.polling || this.controller.signal.aborted) return; this.polling = true; try { if (this.config.sourceMode !== "webhook") await this.poll(); + else await this.reconcileResolvedUi(); this.kick(); } finally { this.polling = false; @@ -681,12 +723,24 @@ export class AlertService { at: now(), selected: selected.length, }); + await this.reconcileResolvedUi(new Set(matching.map((a) => a.fingerprint))); + } + async reconcileResolvedUi(presentFingerprints?: Set) { + const config = this.config; + const present = + presentFingerprints ?? + new Set( + this.store + .rows() + .filter((row) => !row.resolved_at) + .map((row) => row.fingerprint), + ); if (config.mosaicUrl) for (const row of this.store.rows()) { if ( !row.resolved_at || row.ui_cleared_at || - matching.some((a) => a.fingerprint === row.fingerprint) + present.has(row.fingerprint) ) continue; if ( @@ -749,7 +803,7 @@ export class AlertService { ), { at: now(), payload: body }, ); - this.kick(); + this.wake(); return { accepted: selected.length }; } kick() { @@ -1024,6 +1078,7 @@ export class AlertService { this.controller.abort(); if (this.timer) clearInterval(this.timer); this.timer = undefined; + this.workContext.emitDestroy(); // Native Agent runs may outlive shutdown. No new investigation or send is permitted afterwards. if (this.processing) void this.processing.finally(() => this.store.close()); else this.store.close(); diff --git a/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/plugin.ts b/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/plugin.ts index a7318e3..41a2821 100644 --- a/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/plugin.ts +++ b/helm/mosaic-stack/files/openclaw-seed/extensions/alert-automation/plugin.ts @@ -1,10 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -import { timingSafeEqual } from "node:crypto"; +import { randomUUID, timingSafeEqual } from "node:crypto"; import path from "node:path"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk"; -import { AlertService, normalizeConfig, save } from "./core.ts"; +import { + AlertService, + normalizeConfig, + readStoredReportContext, + save, +} from "./core.ts"; import { nativeRuntime } from "./runtime.ts"; export function registerAlertAutomation(api: OpenClawPluginApi) { @@ -35,8 +40,12 @@ export function registerAlertAutomation(api: OpenClawPluginApi) { // This restores the complete report on every human follow-up, rather than // relying on a transcript-only delivery mirror or an extra synthetic ACK turn. api.on("before_prompt_build", (event, ctx) => { - if (!service || !ctx.sessionKey) return; - const binding = service.store.reportContext( + if (!ctx.sessionKey) return; + const root = + service?.store.root ?? + path.join(api.runtime.state.resolveStateDir(), config.stateDirectory); + const binding = readStoredReportContext( + root, ctx.sessionKey, event.prompt ?? "", ); @@ -49,24 +58,30 @@ export function registerAlertAutomation(api: OpenClawPluginApi) { `Incident: ${binding.incident_id}\nRun ID: ${binding.run_id}\nReport SHA256: ${binding.report_sha256}\n\n` + binding.report + "\n[End MOSAIC automated alert report context]"; - if (ctx.runId) - save( - path.join( - service.store.directory(binding.incident_id), - "context-use", - ctx.runId.replace(/[^a-zA-Z0-9_-]/g, "_") + ".json", - ), - { - at: new Date().toISOString(), - sessionKey: ctx.sessionKey, - runId: ctx.runId, - reportIncidentId: binding.incident_id, - completeReportSha256: binding.report_sha256, - completeReportChars: binding.report.length, - mechanism: "before_prompt_build.prependContext", - completeContext: prependContext, - }, - ); + const witnessId = ctx.runId ?? randomUUID(); + save( + path.join( + root, + "incidents", + binding.incident_id, + "context-use", + witnessId.replace(/[^a-zA-Z0-9_-]/g, "_") + ".json", + ), + { + at: new Date().toISOString(), + sessionKey: ctx.sessionKey, + runId: ctx.runId ?? null, + witnessId, + inputPrompt: event.prompt ?? "", + serviceStartedInThisRegistration: Boolean(service), + lookupMode: "durable-readonly", + reportIncidentId: binding.incident_id, + completeReportSha256: binding.report_sha256, + completeReportChars: binding.report.length, + mechanism: "before_prompt_build.prependContext", + completeContext: prependContext, + }, + ); return { prependContext }; }); api.registerGatewayMethod( diff --git a/helm/mosaic-stack/tests/alert-automation.test.mjs b/helm/mosaic-stack/tests/alert-automation.test.mjs index 107fe19..b40f126 100644 --- a/helm/mosaic-stack/tests/alert-automation.test.mjs +++ b/helm/mosaic-stack/tests/alert-automation.test.mjs @@ -2,8 +2,16 @@ // SPDX-License-Identifier: Apache-2.0 import assert from "node:assert/strict"; +import { AsyncLocalStorage } from "node:async_hooks"; import { execFileSync } from "node:child_process"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { + existsSync, + mkdtempSync, + readFileSync, + readdirSync, + rmSync, + writeFileSync, +} from "node:fs"; import http from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -17,6 +25,7 @@ import { sha256, requestJson, completedReport, + save, } from "../files/openclaw-seed/extensions/alert-automation/core.ts"; import { nativeRuntime, @@ -605,9 +614,10 @@ test("HTTP transport records complete responses but never follows a credential-b ); }); -test("native plugin enforces webhook auth and restores full context on multiple turns and service restart", async (t) => { +test("native plugin owns webhook jobs independently of HTTP scope and restores report context", async (t) => { const root = directory(t), token = "unit-test-webhook-bearer-token-only"; + const requestScope = new AsyncLocalStorage(); const previous = process.env.MOSAIC_ALERT_WEBHOOK_TOKEN; process.env.MOSAIC_ALERT_WEBHOOK_TOKEN = token; t.after(() => { @@ -642,6 +652,8 @@ test("native plugin enforces webhook auth and restores full context on multiple runtime: { subagent: { async run(p) { + if (requestScope.getStore()?.httpClient) + throw new Error("missing scope: operator.write"); prompt = p.message; return { runId: "native" }; }, @@ -679,7 +691,9 @@ test("native plugin enforces webhook auth and restores full context on multiple lifecycle.start({ stateDir: root, config: api.config, logger }); t.after(() => lifecycle.stop()); const endpoint = await server(t, (req, res) => { - void route.handler(req, res); + requestScope.run({ httpClient: true, scopes: [] }, () => { + void route.handler(req, res); + }); }); const body = JSON.stringify({ alerts: [alert({ status: "firing" })] }); const response = await fetch(endpoint + route.path, { @@ -734,3 +748,104 @@ test("native plugin enforces webhook auth and restores full context on multiple await new Promise((resolve) => setTimeout(resolve, 30)); assert.equal(messages.length, 1); }); + +test("webhook-only recovery clears the imported UI alert without polling Alertmanager", async (t) => { + const root = directory(t); + let deleted = 0; + const ui = await server(t, (req, res) => { + if (req.method === "DELETE") { + deleted++; + assert.equal( + req.url, + "/api/alerts?id=alertmanager%3A1234567890abcdef", + ); + } + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ success: true })); + }); + const env = { + MOSAIC_ALERT_WEBHOOK_TOKEN: "test-only-webhook-token-not-a-real-credential", + MOSAIC_UI_MACHINE_TOKEN: "test-only-ui-token", + }; + const cfg = normalizeConfig({ + sourceMode: "webhook", + mosaicUrl: ui, + delivery: { + channel: "slack", + target: "channel:C_TEST", + sessionKey: "agent:default:slack:channel:c_test", + }, + }, env); + const rt = runtime(); + const service = new AlertService(cfg, root, rt, logger, env); + t.after(() => service.stop()); + service.start(); + const source = alert({ status: "firing" }); + service.ingestWebhook({ alerts: [source] }); + const id = incidentKey(source); + for (let i = 0; i < 100 && service.store.get(id)?.state !== "sent"; i++) + await new Promise((resolve) => setTimeout(resolve, 5)); + assert.equal(service.store.get(id).state, "sent"); + service.ingestWebhook({ alerts: [{ ...source, status: "resolved" }] }); + for (let i = 0; i < 100 && !service.store.get(id)?.ui_cleared_at; i++) + await new Promise((resolve) => setTimeout(resolve, 5)); + assert.ok(service.store.get(id).resolved_at); + assert.ok(service.store.get(id).ui_cleared_at); + assert.equal(deleted, 1); + assert.deepEqual(rt.counters(), { runs: 1, sends: 1 }); + assert.equal(existsSync(join(root, "last-poll.json")), false); +}); + +test("an agent-only hook registration reads the latest durable report without starting the Gateway service", (t) => { + const root = directory(t); + const cfg = config(); + const state = join(root, cfg.stateDirectory); + const store = new Store(state); + const source = alert(); + const id = incidentKey(source); + store.ingest([source], cfg); + const report = + "FRESH REPORT: evidence belongs to fresh-report-run, not the old conversation."; + // The real publisher has already persisted this file and binding. + save(join(store.directory(id), "published-report.txt"), report); + store.bindReport(id, cfg.delivery.sessionKey, "fresh-report-run", report); + store.close(); + let hook; + const api = { + pluginConfig: { alertmanagerUrl: cfg.alertmanagerUrl, delivery: cfg.delivery }, + config: {}, + logger, + runtime: { state: { resolveStateDir: () => root } }, + registerService() {}, + registerGatewayMethod() {}, + on(name, handler) { + assert.equal(name, "before_prompt_build"); + hook = handler; + }, + }; + registerAlertAutomation(api); + // Native Agent hook registries do not run the Gateway service lifecycle. + // Native channel turns can also omit ctx.runId. + const result = hook( + { prompt: "What is the report Run ID?" }, + { sessionKey: cfg.delivery.sessionKey }, + ); + assert.ok(result?.prependContext.includes(report)); + assert.ok(result.prependContext.includes("Run ID: fresh-report-run")); + const paths = readdirSync(join(state, "incidents", id, "context-use")); + assert.equal(paths.length, 1); + const witness = JSON.parse( + readFileSync(join(state, "incidents", id, "context-use", paths[0]), "utf8"), + ); + assert.equal(witness.runId, null); + assert.equal(witness.reportIncidentId, id); + assert.equal(witness.inputPrompt, "What is the report Run ID?"); + assert.equal(witness.completeReportSha256, sha256(report)); + assert.equal( + hook( + { prompt: "Same Run ID?" }, + { sessionKey: "agent:other:slack:channel:c_test" }, + ), + undefined, + ); +});