diff --git a/docs/observability.md b/docs/observability.md index f9f061592..746ee012a 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -10,9 +10,13 @@ This document describes runtime observability emitted by Agent Orchestrator. ## Emission Model -- **Structured logs**: JSON lines on stderr, controlled by `AO_LOG_LEVEL`. +- **Structured logs**: JSON lines on stderr when enabled by config or `AO_OBSERVABILITY_STDERR`. - Supported levels: `debug`, `info`, `warn`, `error`. - Default level: `warn` (production-safe, avoids high-volume info logs). + - Default stderr mirroring: disabled. + - Runtime env vars override YAML: + - `AO_LOG_LEVEL=info` + - `AO_OBSERVABILITY_STDERR=1` - **Durable snapshots**: process-local JSON snapshots under: - `~/.agent-orchestrator/{config-hash}-observability/processes/*.json` - **Aggregated view**: merged by project via: @@ -38,6 +42,7 @@ Counters are emitted per project and operation: - `lifecycle_poll` (`lifecycle.merge_cleanup.completed`) — auto-cleanup ran after a PR was detected as merged; session runtime + worktree + metadata were torn down - `lifecycle_poll` (`lifecycle.merge_cleanup.deferred`) — auto-cleanup is waiting for the agent to idle (or for the `mergeCleanupIdleGraceMs` window to elapse) before tearing down - `lifecycle_poll` (`lifecycle.merge_cleanup.failed`) — auto-cleanup threw during `sessionManager.kill()`; the session stays in `merged` so the next poll retries +- `notification_delivery` (`notification.deliver`) — per-target notifier dispatch success/failure, with event ID/type, priority, project/session IDs, target reference, plugin name, and delivery method - `api_request` (web API routes) - `sse_connect`, `sse_snapshot`, `sse_disconnect` - `websocket_connect`, `websocket_disconnect`, `websocket_error` (websocket servers) @@ -83,10 +88,11 @@ Health records provide current status and failure context per surface: - **Dashboard**: use **Copy debug info** in the hero toolbar (desktop) to copy `/api/observability` plus page URL, project scope, and correlation id to the clipboard for issue reports. The observability banner shows overall status, SSE stream state, last correlation id, and latest failure reason. - **API**: `/api/observability` returns merged per-project diagnostics (`overallStatus`, metrics, health, recent traces, session state). - **Terminal websocket health**: `/health` endpoints include active sessions and websocket/terminal health counters with last error/disconnect reasons. +- **Notifications**: successful deliveries update `notification_delivery` metrics and per-target health; failed/missing targets also appear in `ao events`. ## Rollout Notes -1. Deploy with default `AO_LOG_LEVEL=warn` to avoid noisy logs. +1. Deploy with default `observability.logLevel: warn` and `observability.stderr: false` to avoid noisy logs. 2. Validate `/api/observability` and dashboard banner in a canary environment. -3. If deeper triage is needed, temporarily raise `AO_LOG_LEVEL=info` (or `debug`), then revert to `warn`. +3. If deeper triage is needed, temporarily raise `AO_LOG_LEVEL=info` (or `debug`) and set `AO_OBSERVABILITY_STDERR=1`, then revert to defaults. 4. Monitor `lastFailureReason` and surface-level `reason` fields before enabling broader rollout. diff --git a/packages/cli/__tests__/lib/notify-test.test.ts b/packages/cli/__tests__/lib/notify-test.test.ts index 7e6ad1721..f93de3d18 100644 --- a/packages/cli/__tests__/lib/notify-test.test.ts +++ b/packages/cli/__tests__/lib/notify-test.test.ts @@ -1,9 +1,15 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; -import type { - Notifier, - OrchestratorConfig, - OrchestratorEvent, - PluginRegistry, +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdirSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { randomUUID } from "node:crypto"; +import { + closeDb, + readObservabilitySummary, + type Notifier, + type OrchestratorConfig, + type OrchestratorEvent, + type PluginRegistry, } from "@aoagents/ao-core"; import { addSinkNotifierConfig, @@ -15,8 +21,9 @@ import { } from "../../src/lib/notify-test.js"; function makeConfig(overrides: Partial = {}): OrchestratorConfig { + const testHome = process.env["HOME"] ?? process.env["USERPROFILE"] ?? tmpdir(); return { - configPath: "/tmp/agent-orchestrator.yaml", + configPath: join(testHome, "agent-orchestrator.yaml"), readyThresholdMs: 300_000, defaults: { runtime: "tmux", @@ -61,8 +68,33 @@ function makeRegistry(notifiers: Record | undefined>): } describe("notify test helper", () => { + let tempRoot: string; + let originalHome: string | undefined; + let originalUserProfile: string | undefined; + + beforeEach(() => { + tempRoot = join(tmpdir(), `ao-notify-test-${randomUUID()}`); + mkdirSync(tempRoot, { recursive: true }); + originalHome = process.env["HOME"]; + originalUserProfile = process.env["USERPROFILE"]; + process.env["HOME"] = tempRoot; + process.env["USERPROFILE"] = tempRoot; + }); + afterEach(() => { + closeDb(); vi.restoreAllMocks(); + if (originalHome === undefined) { + delete process.env["HOME"]; + } else { + process.env["HOME"] = originalHome; + } + if (originalUserProfile === undefined) { + delete process.env["USERPROFILE"]; + } else { + process.env["USERPROFILE"] = originalUserProfile; + } + rmSync(tempRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); }); it("builds realistic CI and PR template data", () => { @@ -105,13 +137,17 @@ describe("notify test helper", () => { slack: { name: "slack", notify }, }); - const result = await runNotifyTest(makeConfig(), registry, { to: ["alerts"] }); + const config = makeConfig(); + const result = await runNotifyTest(config, registry, { to: ["alerts"] }); expect(result.ok).toBe(true); expect(result.targets).toEqual([{ reference: "alerts", pluginName: "slack" }]); expect(registry.get).toHaveBeenCalledWith("notifier", "alerts"); expect(registry.get).toHaveBeenCalledWith("notifier", "slack"); expect(notify).toHaveBeenCalledTimes(1); + + const summary = readObservabilitySummary(config); + expect(summary.projects["demo"]?.metrics["notification_delivery"]?.success).toBe(1); }); it("resolves explicit routes through notificationRouting before defaults", () => { @@ -155,6 +191,7 @@ describe("notify test helper", () => { expect(result.ok).toBe(true); expect(result.deliveries[0]?.status).toBe("dry_run"); expect(notify).not.toHaveBeenCalled(); + expect(readObservabilitySummary(makeConfig()).projects["demo"]).toBeUndefined(); }); it("uses notifyWithActions when available", async () => { @@ -194,12 +231,18 @@ describe("notify test helper", () => { ops: { name: "ops", notify: passing }, }); - const result = await runNotifyTest(makeConfig(), registry, { route: "action" }); + const config = makeConfig(); + const result = await runNotifyTest(config, registry, { route: "action" }); expect(result.ok).toBe(false); expect(failing).toHaveBeenCalledTimes(1); expect(passing).toHaveBeenCalledTimes(1); expect(result.deliveries.map((delivery) => delivery.status)).toEqual(["failed", "sent"]); + const summary = readObservabilitySummary(config); + expect(summary.projects["demo"]?.metrics["notification_delivery"]).toMatchObject({ + success: 1, + failure: 1, + }); }); it("reports unresolved targets and no-target configs as failures", async () => { diff --git a/packages/cli/src/lib/config-instruction.ts b/packages/cli/src/lib/config-instruction.ts index f487d2065..e02095c44 100644 --- a/packages/cli/src/lib/config-instruction.ts +++ b/packages/cli/src/lib/config-instruction.ts @@ -20,6 +20,10 @@ terminalPort: 14800 # Optional terminal WebSocket port override directTerminalPort: 14801 # Optional direct terminal WebSocket port override readyThresholdMs: 300000 # Ms before "ready" becomes "idle" (default: 5 min) +observability: + logLevel: warn # debug | info | warn | error + stderr: false # Mirror structured logs to stderr + # ── Default plugins ───────────────────────────────────────────────── # These apply to all projects unless overridden per-project. diff --git a/packages/cli/src/lib/notify-test.ts b/packages/cli/src/lib/notify-test.ts index a9ad451ea..47b50f05a 100644 --- a/packages/cli/src/lib/notify-test.ts +++ b/packages/cli/src/lib/notify-test.ts @@ -5,6 +5,8 @@ import { buildPRStateNotificationData, buildReactionNotificationData, buildSessionTransitionNotificationData, + createProjectObserver, + recordNotificationDelivery, resolveNotifierTarget, type CICheck, type EventPriority, @@ -541,6 +543,7 @@ export async function runNotifyTest( const warnings: string[] = []; const errors: string[] = []; const deliveries: NotifyDeliveryResult[] = []; + const observer = request.dryRun ? null : createProjectObserver(config, "notify-test"); if (targets.length === 0) { errors.push( @@ -567,6 +570,18 @@ export async function runNotifyTest( if (!notifier) { const error = `${target.reference}: notifier plugin "${target.pluginName}" is not loaded`; errors.push(error); + if (observer) { + recordNotificationDelivery({ + observer, + event, + target, + outcome: "failure", + method: "notify", + reason: "notifier target not found", + failureKind: "target_missing", + recordActivityEvent: true, + }); + } deliveries.push({ reference: target.reference, pluginName: target.pluginName, @@ -588,6 +603,8 @@ export async function runNotifyTest( } try { + const method = + actions.length > 0 && notifier.notifyWithActions ? "notifyWithActions" : "notify"; if (actions.length > 0 && notifier.notifyWithActions) { await notifier.notifyWithActions(event, actions); deliveries.push({ @@ -612,14 +629,37 @@ export async function runNotifyTest( warning, }); } + if (observer) { + recordNotificationDelivery({ + observer, + event, + target, + outcome: "success", + method, + }); + } } catch (err) { const error = `${target.reference}: ${err instanceof Error ? err.message : String(err)}`; + const method = + actions.length > 0 && notifier.notifyWithActions ? "notifyWithActions" : "notify"; errors.push(error); + if (observer) { + recordNotificationDelivery({ + observer, + event, + target, + outcome: "failure", + method, + reason: err instanceof Error ? err.message : String(err), + failureKind: "delivery_failed", + recordActivityEvent: true, + }); + } deliveries.push({ reference: target.reference, pluginName: target.pluginName, status: "failed", - method: actions.length > 0 && notifier.notifyWithActions ? "notifyWithActions" : "notify", + method, error, }); } diff --git a/packages/core/__tests__/config.test.ts b/packages/core/__tests__/config.test.ts index 1bd68af56..041aac7ef 100644 --- a/packages/core/__tests__/config.test.ts +++ b/packages/core/__tests__/config.test.ts @@ -157,6 +157,9 @@ projects: globalConfigPath, [ "port: 4000", + "observability:", + " logLevel: info", + " stderr: true", "defaults:", " runtime: tmux", " agent: claude-code", @@ -185,6 +188,7 @@ projects: ); const config = loadConfig(globalConfigPath); + expect(config.observability).toEqual({ logLevel: "info", stderr: true }); expect(Object.keys(config.projects)).toEqual(["clean-project"]); expect(config.projects["clean-project"]).toBeDefined(); expect(config.degradedProjects["broken-project"]).toMatchObject({ diff --git a/packages/core/src/__tests__/config-validation.test.ts b/packages/core/src/__tests__/config-validation.test.ts index 1a6ce9c49..8dd359161 100644 --- a/packages/core/src/__tests__/config-validation.test.ts +++ b/packages/core/src/__tests__/config-validation.test.ts @@ -1247,3 +1247,61 @@ describe("Config Validation - Power Config", () => { ).toThrow(); }); }); + +describe("Config Validation - Observability Config", () => { + const baseConfig = { + projects: { + proj1: { + path: "/repos/test", + repo: "org/test", + defaultBranch: "main", + }, + }, + }; + + it("applies default observability config", () => { + const config = validateConfig(baseConfig); + + expect(config.observability).toEqual({ + logLevel: "warn", + stderr: false, + }); + }); + + it("accepts explicit observability settings", () => { + const config = validateConfig({ + ...baseConfig, + observability: { + logLevel: "info", + stderr: true, + }, + }); + + expect(config.observability).toEqual({ + logLevel: "info", + stderr: true, + }); + }); + + it("rejects invalid observability settings", () => { + expect(() => + validateConfig({ + ...baseConfig, + observability: { + logLevel: "verbose", + stderr: false, + }, + }), + ).toThrow(); + + expect(() => + validateConfig({ + ...baseConfig, + observability: { + logLevel: "warn", + stderr: "yes", + }, + }), + ).toThrow(); + }); +}); diff --git a/packages/core/src/__tests__/lifecycle-manager-instrumentation.test.ts b/packages/core/src/__tests__/lifecycle-manager-instrumentation.test.ts index 70b0b0296..d1d844899 100644 --- a/packages/core/src/__tests__/lifecycle-manager-instrumentation.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager-instrumentation.test.ts @@ -11,8 +11,8 @@ * test that proves the lifecycle check completes successfully even if * recordActivityEvent itself throws (B2 invariant). * - * Notifier instrumentation is intentionally omitted — the notifier subsystem - * is undergoing larger work and AE evidence there is not currently useful. + * Notification delivery instrumentation is covered in lifecycle-manager.test.ts + * because it needs real observability snapshots in addition to activity events. */ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; import type * as ActivityEventsModule from "../activity-events.js"; diff --git a/packages/core/src/__tests__/lifecycle-manager.test.ts b/packages/core/src/__tests__/lifecycle-manager.test.ts index 095680c00..e78146fa1 100644 --- a/packages/core/src/__tests__/lifecycle-manager.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager.test.ts @@ -2844,6 +2844,112 @@ describe("reactions", () => { expect(notifier.notify).toHaveBeenCalledWith( expect.objectContaining({ type: "merge.completed" }), ); + + const summary = readObservabilitySummary(config); + expect(summary.projects["my-app"]?.metrics["notification_delivery"]?.success).toBe(1); + expect( + summary.projects["my-app"]?.recentTraces.some( + (trace) => + trace.operation === "notification.deliver" && + trace.outcome === "success" && + trace.data?.["targetReference"] === "desktop", + ), + ).toBe(true); + }); + + it("records notifier delivery failures without interrupting lifecycle transitions", async () => { + const notifier = createMockNotifier(); + vi.mocked(notifier.notify).mockRejectedValue(new Error("webhook failed")); + const mockSCM = createMockSCM({ + getPRState: vi.fn().mockResolvedValue("merged"), + enrichSessionsPRBatch: mockBatchEnrichment({ state: "merged" }), + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + const lm = setupCheck("app-1", { + session: makeSession({ status: "approved", pr: makePR() }), + registry, + }); + + await lm.check("app-1"); + + expect(lm.getStates().get("app-1")).toBe("merged"); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "notifier", + kind: "notification.delivery_failed", + level: "warn", + data: expect.objectContaining({ + eventType: "merge.completed", + targetReference: "desktop", + targetPlugin: "desktop", + }), + }), + ); + + const summary = readObservabilitySummary(config); + expect(summary.projects["my-app"]?.metrics["notification_delivery"]?.failure).toBe(1); + expect(summary.projects["my-app"]?.health["notification.delivery.desktop"]?.status).toBe( + "warn", + ); + }); + + it("records missing notifier targets as delivery failures", async () => { + const mockSCM = createMockSCM({ + getPRState: vi.fn().mockResolvedValue("merged"), + enrichSessionsPRBatch: mockBatchEnrichment({ state: "merged" }), + }); + const configWithMissingNotifier: OrchestratorConfig = { + ...config, + notificationRouting: { + ...config.notificationRouting, + action: ["missing"], + }, + }; + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + return null; + }), + }; + + const lm = setupCheck("app-1", { + session: makeSession({ status: "approved", pr: makePR() }), + registry, + configOverride: configWithMissingNotifier, + }); + + await lm.check("app-1"); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "notifier", + kind: "notification.target_missing", + level: "warn", + data: expect.objectContaining({ + eventType: "merge.completed", + targetReference: "missing", + targetPlugin: "missing", + }), + }), + ); + + const summary = readObservabilitySummary(configWithMissingNotifier); + expect(summary.projects["my-app"]?.metrics["notification_delivery"]?.failure).toBe(1); }); it("resolves notifier aliases from notificationRouting before dispatch", async () => { diff --git a/packages/core/src/__tests__/observability.test.ts b/packages/core/src/__tests__/observability.test.ts index 3b034eb5a..78ca4fed3 100644 --- a/packages/core/src/__tests__/observability.test.ts +++ b/packages/core/src/__tests__/observability.test.ts @@ -159,6 +159,105 @@ describe("observability snapshot", () => { } }); + it("uses YAML observability log level and stderr settings", () => { + const originalLogLevel = process.env["AO_LOG_LEVEL"]; + const originalObservabilityStderr = process.env["AO_OBSERVABILITY_STDERR"]; + delete process.env["AO_LOG_LEVEL"]; + delete process.env["AO_OBSERVABILITY_STDERR"]; + config.observability = { logLevel: "warn", stderr: true }; + + const stderrSpy = vi.spyOn(process.stderr, "write").mockReturnValue(true); + + try { + const observer = createProjectObserver(config, "session-manager"); + observer.recordOperation({ + metric: "spawn", + operation: "notification.success-no-audit", + outcome: "success", + correlationId: "corr-info", + projectId: "my-app", + level: "info", + }); + observer.recordOperation({ + metric: "notification_delivery", + operation: "notification.deliver", + outcome: "failure", + correlationId: "corr-warn", + projectId: "my-app", + sessionId: "app-1", + reason: "notifier target not found", + level: "warn", + }); + + const auditDir = join(getObservabilityBaseDir(config.configPath), "processes"); + const auditFiles = readdirSync(auditDir).filter((fileName) => fileName.endsWith(".ndjson")); + const auditLog = auditFiles + .map((fileName) => readFileSync(join(auditDir, fileName), "utf-8")) + .join("\n"); + + expect(auditLog).toContain('"operation":"notification.deliver"'); + expect(auditLog).not.toContain("notification.success-no-audit"); + expect(stderrSpy).toHaveBeenCalledTimes(1); + expect(String(stderrSpy.mock.calls[0]?.[0])).toContain('"operation":"notification.deliver"'); + } finally { + stderrSpy.mockRestore(); + if (originalLogLevel === undefined) { + delete process.env["AO_LOG_LEVEL"]; + } else { + process.env["AO_LOG_LEVEL"] = originalLogLevel; + } + if (originalObservabilityStderr === undefined) { + delete process.env["AO_OBSERVABILITY_STDERR"]; + } else { + process.env["AO_OBSERVABILITY_STDERR"] = originalObservabilityStderr; + } + } + }); + + it("lets observability env vars override YAML settings", () => { + const originalLogLevel = process.env["AO_LOG_LEVEL"]; + const originalObservabilityStderr = process.env["AO_OBSERVABILITY_STDERR"]; + process.env["AO_LOG_LEVEL"] = "info"; + process.env["AO_OBSERVABILITY_STDERR"] = "0"; + config.observability = { logLevel: "error", stderr: true }; + + const stderrSpy = vi.spyOn(process.stderr, "write").mockReturnValue(true); + + try { + const observer = createProjectObserver(config, "session-manager"); + observer.recordOperation({ + metric: "notification_delivery", + operation: "notification.deliver", + outcome: "success", + correlationId: "corr-env", + projectId: "my-app", + sessionId: "app-1", + level: "info", + }); + + const auditDir = join(getObservabilityBaseDir(config.configPath), "processes"); + const auditFiles = readdirSync(auditDir).filter((fileName) => fileName.endsWith(".ndjson")); + const auditLog = auditFiles + .map((fileName) => readFileSync(join(auditDir, fileName), "utf-8")) + .join("\n"); + + expect(auditLog).toContain('"operation":"notification.deliver"'); + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + stderrSpy.mockRestore(); + if (originalLogLevel === undefined) { + delete process.env["AO_LOG_LEVEL"]; + } else { + process.env["AO_LOG_LEVEL"] = originalLogLevel; + } + if (originalObservabilityStderr === undefined) { + delete process.env["AO_OBSERVABILITY_STDERR"]; + } else { + process.env["AO_OBSERVABILITY_STDERR"] = originalObservabilityStderr; + } + } + }); + it("redacts sensitive observability payload fields before persisting them", () => { const observer = createProjectObserver(config, "session-manager"); diff --git a/packages/core/src/activity-events.ts b/packages/core/src/activity-events.ts index 6065c3d41..16ca7c8d7 100644 --- a/packages/core/src/activity-events.ts +++ b/packages/core/src/activity-events.ts @@ -20,6 +20,7 @@ export type ActivityEventSource = | "scm" | "runtime" | "agent" + | "notifier" | "reaction" | "report-watcher"; @@ -51,6 +52,9 @@ export type ActivityEventKind = | "session.auto_cleanup_failed" | "lifecycle.poll_failed" | "detecting.escalated" + // Notification delivery + | "notification.delivery_failed" + | "notification.target_missing" // Report watcher | "report_watcher.triggered"; diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index 37c59d3ce..ed119f511 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -198,6 +198,13 @@ const NotifierConfigSchema = z .passthrough() .superRefine((value, ctx) => validatePluginConfigFields(value, ctx, "Notifier")); +const ObservabilityConfigSchema = z + .object({ + logLevel: z.enum(["debug", "info", "warn", "error"]).default("warn"), + stderr: z.boolean().default(false), + }) + .default({}); + const AgentPermissionSchema = z .enum(["permissionless", "default", "auto-edit", "suggest", "skip"]) .default("permissionless") @@ -354,6 +361,7 @@ const OrchestratorConfigSchema = z.object({ readyThresholdMs: z.number().int().nonnegative().default(300_000), power: PowerConfigSchema, lifecycle: LifecycleConfigSchema, + observability: ObservabilityConfigSchema, defaults: DefaultPluginsSchema.default({}), plugins: z.array(InstalledPluginConfigSchema).default([]), dashboard: DashboardConfigSchema.optional(), @@ -844,6 +852,7 @@ function buildEffectiveConfigFromFlatLocalPath( terminalPort: globalConfig.terminalPort, directTerminalPort: globalConfig.directTerminalPort, readyThresholdMs: globalConfig.readyThresholdMs, + observability: globalConfig.observability, defaults: globalConfig.defaults, notifiers: globalConfig.notifiers, notificationRouting: globalConfig.notificationRouting, @@ -884,6 +893,7 @@ function buildEffectiveConfigFromGlobalConfigPath(configPath: string): LoadedCon terminalPort: globalConfig.terminalPort, directTerminalPort: globalConfig.directTerminalPort, readyThresholdMs: globalConfig.readyThresholdMs, + observability: globalConfig.observability, defaults: globalConfig.defaults, notifiers: globalConfig.notifiers, notificationRouting: globalConfig.notificationRouting, diff --git a/packages/core/src/global-config.ts b/packages/core/src/global-config.ts index cda4ca625..814ba9cb9 100644 --- a/packages/core/src/global-config.ts +++ b/packages/core/src/global-config.ts @@ -205,6 +205,13 @@ export const GlobalConfigSchema = z updateChannel: UpdateChannelSchema.optional().catch(undefined), /** Override path-based install detection. Optional. */ installMethod: InstallMethodOverrideSchema.optional().catch(undefined), + /** Structured observability defaults. Env vars still override at runtime. */ + observability: z + .object({ + logLevel: z.enum(["debug", "info", "warn", "error"]).default("warn"), + stderr: z.boolean().default(false), + }) + .optional(), /** Cross-project defaults — projects inherit when fields are omitted. */ defaults: z .object({ @@ -991,6 +998,8 @@ export function migrateToGlobalConfig(oldConfigPath: string, globalConfigPath?: newGlobal.directTerminalPort = parsed["directTerminalPort"] as number; if (parsed["readyThresholdMs"] !== null && parsed["readyThresholdMs"] !== undefined) newGlobal.readyThresholdMs = parsed["readyThresholdMs"] as number; + if (parsed["observability"] !== null && parsed["observability"] !== undefined) + newGlobal.observability = parsed["observability"] as GlobalConfig["observability"]; if (parsed["defaults"] !== null && parsed["defaults"] !== undefined) newGlobal.defaults = parsed["defaults"] as GlobalConfig["defaults"]; if (parsed["notifiers"] !== null && parsed["notifiers"] !== undefined) @@ -1081,6 +1090,10 @@ export function createDefaultGlobalConfig(): GlobalConfig { return { port: 3000, readyThresholdMs: 300_000, + observability: { + logLevel: "warn", + stderr: false, + }, defaults: { runtime: getDefaultRuntime(), agent: "claude-code", diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index ac209b615..e7af8c4d8 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -212,6 +212,10 @@ export { } from "./observability.js"; export { execGhObserved, getGhTraceFilePath } from "./gh-trace.js"; export { resolveNotifierTarget } from "./notifier-resolution.js"; +export { + recordNotificationDelivery, + sanitizeNotificationDeliveryReason, +} from "./notification-observability.js"; export { NOTIFICATION_DATA_SCHEMA_VERSION, buildCIFailureNotificationData, @@ -253,6 +257,12 @@ export type { ProjectObserver, } from "./observability.js"; export type { GhTraceContext, GhTraceEntry } from "./gh-trace.js"; +export type { + NotificationDeliveryFailureKind, + NotificationDeliveryMethod, + NotificationDeliveryTarget, + RecordNotificationDeliveryInput, +} from "./notification-observability.js"; // Feedback tools — contracts, validation, and report storage export { diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 9ababe14d..9fa1a928f 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -69,6 +69,7 @@ import { } from "./report-watcher.js"; import { createCorrelationId, createProjectObserver } from "./observability.js"; import { resolveNotifierTarget } from "./notifier-resolution.js"; +import { recordNotificationDelivery } from "./notification-observability.js"; import { resolveAgentSelection, resolveSessionRole } from "./agent-selection.js"; import { DETECTING_MAX_ATTEMPTS, @@ -2279,12 +2280,40 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan const notifier = registry.get("notifier", target.reference) ?? registry.get("notifier", target.pluginName); - if (notifier) { - try { - await notifier.notify(eventWithPriority); - } catch { - // Notifier failed — not much we can do - } + if (!notifier) { + recordNotificationDelivery({ + observer, + event: eventWithPriority, + target, + outcome: "failure", + method: "notify", + reason: "notifier target not found", + failureKind: "target_missing", + recordActivityEvent: true, + }); + continue; + } + + try { + await notifier.notify(eventWithPriority); + recordNotificationDelivery({ + observer, + event: eventWithPriority, + target, + outcome: "success", + method: "notify", + }); + } catch (err) { + recordNotificationDelivery({ + observer, + event: eventWithPriority, + target, + outcome: "failure", + method: "notify", + reason: err instanceof Error ? err.message : String(err), + failureKind: "delivery_failed", + recordActivityEvent: true, + }); } } } diff --git a/packages/core/src/notification-observability.ts b/packages/core/src/notification-observability.ts new file mode 100644 index 000000000..1105b8c4b --- /dev/null +++ b/packages/core/src/notification-observability.ts @@ -0,0 +1,153 @@ +import { recordActivityEvent, type ActivityEventKind } from "./activity-events.js"; +import { createCorrelationId, type ProjectObserver } from "./observability.js"; +import type { OrchestratorEvent } from "./types.js"; + +export type NotificationDeliveryMethod = "notify" | "notifyWithActions"; +export type NotificationDeliveryFailureKind = "delivery_failed" | "target_missing"; + +export interface NotificationDeliveryTarget { + reference: string; + pluginName: string; +} + +export interface RecordNotificationDeliveryInput { + observer: ProjectObserver; + event: OrchestratorEvent; + target: NotificationDeliveryTarget; + outcome: "success" | "failure"; + method?: NotificationDeliveryMethod; + reason?: string; + failureKind?: NotificationDeliveryFailureKind; + recordActivityEvent?: boolean; +} + +const TOKEN_PATTERNS: ReadonlyArray = [ + [/\bBearer\s+[A-Za-z0-9._~+/=-]{12,}\b/gi, "Bearer [redacted]"], + [/\bgh[pousr]_[A-Za-z0-9_]{20,}\b/g, "[redacted]"], + [/\bgithub_pat_[A-Za-z0-9_]{20,}\b/g, "[redacted]"], + [/\bsk-(?:ant-)?(?:proj-|svcacct-)?[A-Za-z0-9_-]{16,}\b/g, "[redacted]"], + [/\bxox[baprs]-[A-Za-z0-9-]{10,}\b/g, "[redacted]"], + [/\bAKIA[0-9A-Z]{16}\b/g, "[redacted]"], + [/\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b/g, "[redacted]"], + [ + /\b([A-Z][A-Z0-9_]*(?:TOKEN|PASSWORD|SECRET|AUTHORIZATION|COOKIE|API_KEY|APIKEY)[A-Z0-9_]*)=([^\s"'`]{6,})/g, + "$1=[redacted]", + ], +]; + +function redactCredentialUrls(input: string): string { + let result = input; + let offset = 0; + while (offset < result.length) { + const proto = result.indexOf("://", offset); + if (proto === -1) break; + if (proto < 4) { + offset = proto + 3; + continue; + } + + const schemeEnd = result.slice(Math.max(0, proto - 5), proto).toLowerCase(); + if (!schemeEnd.endsWith("http") && !schemeEnd.endsWith("https")) { + offset = proto + 3; + continue; + } + + let cursor = proto + 3; + let redacted = false; + while (cursor < result.length) { + const ch = result.charCodeAt(cursor); + if (ch <= 0x20 || ch === 0x2f) break; + if (ch === 0x40) { + result = result.slice(0, proto + 3).toLowerCase() + "[redacted]" + result.slice(cursor); + offset = proto + 3 + "[redacted]".length + 1; + redacted = true; + break; + } + cursor++; + } + if (!redacted) offset = proto + 3; + } + return result; +} + +export function sanitizeNotificationDeliveryReason(reason: string): string { + let cleaned = redactCredentialUrls(reason.replace(/\s+/g, " ").trim()); + for (const [pattern, replacement] of TOKEN_PATTERNS) { + cleaned = cleaned.replace(pattern, replacement); + } + return cleaned.length > 300 ? `${cleaned.slice(0, 297)}...` : cleaned; +} + +function notificationSurface(reference: string): string { + const suffix = reference.replace(/[^a-zA-Z0-9_-]+/g, "-").replace(/^-+|-+$/g, ""); + return `notification.delivery.${suffix || "target"}`; +} + +function activityKind(kind: NotificationDeliveryFailureKind): ActivityEventKind { + return kind === "target_missing" ? "notification.target_missing" : "notification.delivery_failed"; +} + +function safeRecordActivityFailure(input: RecordNotificationDeliveryInput, reason: string): void { + if (!input.failureKind) return; + try { + recordActivityEvent({ + projectId: input.event.projectId, + sessionId: input.event.sessionId, + source: "notifier", + kind: activityKind(input.failureKind), + level: "warn", + summary: + input.failureKind === "target_missing" + ? `notification target missing: ${input.target.reference}` + : `notification delivery failed: ${input.target.reference}`, + data: { + eventId: input.event.id, + eventType: input.event.type, + priority: input.event.priority, + targetReference: input.target.reference, + targetPlugin: input.target.pluginName, + deliveryMethod: input.method ?? "notify", + errorMessage: reason, + }, + }); + } catch { + // Activity events are diagnostic-only; notification delivery must not depend on them. + } +} + +export function recordNotificationDelivery(input: RecordNotificationDeliveryInput): void { + const reason = input.reason ? sanitizeNotificationDeliveryReason(input.reason) : undefined; + const data = { + eventId: input.event.id, + eventType: input.event.type, + priority: input.event.priority, + targetReference: input.target.reference, + targetPlugin: input.target.pluginName, + deliveryMethod: input.method ?? "notify", + }; + + input.observer.recordOperation({ + metric: "notification_delivery", + operation: "notification.deliver", + outcome: input.outcome, + correlationId: createCorrelationId("notification"), + projectId: input.event.projectId, + sessionId: input.event.sessionId, + reason, + data, + level: input.outcome === "failure" ? "warn" : "info", + }); + + input.observer.setHealth({ + surface: notificationSurface(input.target.reference), + status: input.outcome === "failure" ? "warn" : "ok", + projectId: input.event.projectId, + correlationId: createCorrelationId("notification-health"), + reason, + details: data, + }); + + if (input.outcome === "failure" && input.recordActivityEvent) { + safeRecordActivityFailure(input, reason ?? "notification delivery failed"); + } +} diff --git a/packages/core/src/observability.ts b/packages/core/src/observability.ts index 15f62efdd..3c9f4eb4f 100644 --- a/packages/core/src/observability.ts +++ b/packages/core/src/observability.ts @@ -24,6 +24,7 @@ export type ObservabilityMetricName = | "graphql_batch" | "kill" | "lifecycle_poll" + | "notification_delivery" | "restore" | "send" | "spawn" @@ -167,25 +168,43 @@ function sanitizeComponent(component: string): string { return component.replace(/[^a-zA-Z0-9_-]+/g, "-").replace(/^-+|-+$/g, "") || "component"; } -function getLogLevel(): ObservabilityLevel { - const raw = process.env["AO_LOG_LEVEL"]?.trim().toLowerCase(); - if (raw === "debug" || raw === "info" || raw === "warn" || raw === "error") { - return raw; +function parseLogLevel(raw: string | undefined): ObservabilityLevel | null { + const value = raw?.trim().toLowerCase(); + if (value === "debug" || value === "info" || value === "warn" || value === "error") { + return value; } - return "warn"; + return null; } -function shouldLog(level: ObservabilityLevel): boolean { - return LEVEL_ORDER[level] >= LEVEL_ORDER[getLogLevel()]; +function parseBooleanEnv(raw: string | undefined): boolean | null { + const value = raw?.trim().toLowerCase(); + if (!value) return null; + if (value === "1" || value === "true" || value === "yes" || value === "on") return true; + if (value === "0" || value === "false" || value === "no" || value === "off") return false; + return null; } -function shouldMirrorStructuredLogsToStderr(): boolean { - const raw = process.env["AO_OBSERVABILITY_STDERR"]?.trim().toLowerCase(); - return raw === "1" || raw === "true" || raw === "yes" || raw === "on"; +function getLogLevel(config: OrchestratorConfig): ObservabilityLevel { + const raw = process.env["AO_LOG_LEVEL"]?.trim().toLowerCase(); + return parseLogLevel(raw) ?? config.observability?.logLevel ?? "warn"; } -function emitStructuredLog(entry: Record, level: ObservabilityLevel): void { - if (!shouldMirrorStructuredLogsToStderr() || !shouldLog(level)) return; +function shouldLog(level: ObservabilityLevel, config: OrchestratorConfig): boolean { + return LEVEL_ORDER[level] >= LEVEL_ORDER[getLogLevel(config)]; +} + +function shouldMirrorStructuredLogsToStderr(config: OrchestratorConfig): boolean { + return ( + parseBooleanEnv(process.env["AO_OBSERVABILITY_STDERR"]) ?? config.observability?.stderr ?? false + ); +} + +function emitStructuredLog( + config: OrchestratorConfig, + entry: Record, + level: ObservabilityLevel, +): void { + if (!shouldMirrorStructuredLogsToStderr(config) || !shouldLog(level, config)) return; process.stderr.write(`${JSON.stringify({ ...entry, level })}\n`); } @@ -424,9 +443,9 @@ export function createProjectObserver( const snapshot = readSnapshot(filePath, normalizedComponent); updater(snapshot); writeSnapshot(config, snapshot); - if (logEntry && shouldLog(logEntry.level)) { + if (logEntry && shouldLog(logEntry.level, config)) { appendAuditLog(config, normalizedComponent, logEntry.payload, logEntry.level); - emitStructuredLog(logEntry.payload, logEntry.level); + emitStructuredLog(config, logEntry.payload, logEntry.level); } } catch (error) { const payload = { @@ -438,7 +457,7 @@ export function createProjectObserver( reason: error instanceof Error ? error.message : String(error), }; appendObservabilityFailure(config, payload); - emitStructuredLog(payload, "error"); + emitStructuredLog(config, payload, "error"); } } diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 1d5af5b3e..6db9a89e3 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -1334,6 +1334,13 @@ export interface LifecycleConfig { mergeCleanupIdleGraceMs: number; } +export interface ObservabilityConfig { + /** Minimum structured log level to persist/mirror. Defaults to "warn". */ + logLevel: ObservabilityLevel; + /** Mirror structured observability logs to stderr. Defaults to false. */ + stderr: boolean; +} + /** Top-level orchestrator configuration (from agent-orchestrator.yaml) */ export interface OrchestratorConfig { /** Optional JSON Schema hint for editor autocomplete/validation. */ @@ -1369,6 +1376,12 @@ export interface OrchestratorConfig { */ lifecycle?: LifecycleConfig; + /** + * Process observability settings. Populated with defaults by Zod when loaded + * from YAML, but optional for hand-constructed tests. + */ + observability?: ObservabilityConfig; + /** Default plugin selections */ defaults: DefaultPlugins; diff --git a/schema/config.schema.json b/schema/config.schema.json index a5b78af6e..cf4dd56cc 100644 --- a/schema/config.schema.json +++ b/schema/config.schema.json @@ -41,6 +41,9 @@ "lifecycle": { "$ref": "#/$defs/lifecycleConfig" }, + "observability": { + "$ref": "#/$defs/observabilityConfig" + }, "defaults": { "$ref": "#/$defs/defaultPlugins" }, @@ -513,6 +516,21 @@ } } }, + "observabilityConfig": { + "type": "object", + "additionalProperties": false, + "properties": { + "logLevel": { + "type": "string", + "enum": ["debug", "info", "warn", "error"], + "default": "warn" + }, + "stderr": { + "type": "boolean", + "default": false + } + } + }, "notificationRouting": { "type": "object", "default": {},