bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add notification delivery observability

This commit is contained in:
whoisasx 2026-05-19 04:04:00 +05:30
parent 2ce27ebdc4
commit 2f7e133179
18 changed files with 665 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
docs/observability.md
View File

@ -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.

61
packages/cli/__tests__/lib/notify-test.test.ts
View File

@ -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> = {}): 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<string, Partial<Notifier> | 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 () => {

4
packages/cli/src/lib/config-instruction.ts
View File

@ -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.

42
packages/cli/src/lib/notify-test.ts
View File

@ -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,
});
}

4
packages/core/__tests__/config.test.ts
View File

@ -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({

58
packages/core/src/__tests__/config-validation.test.ts
View File

@ -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();
});
});

4
packages/core/src/__tests__/lifecycle-manager-instrumentation.test.ts
View File

@ -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";

106
packages/core/src/__tests__/lifecycle-manager.test.ts
View File

@ -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 () => {

99
packages/core/src/__tests__/observability.test.ts
View File

@ -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");

4
packages/core/src/activity-events.ts
View File

@ -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";

10
packages/core/src/config.ts
View File

@ -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,

13
packages/core/src/global-config.ts
View File

@ -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",

10
packages/core/src/index.ts
View File

@ -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 {

41
packages/core/src/lifecycle-manager.ts
View File

@ -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>("notifier", target.reference) ??
registry.get<Notifier>("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,
});
}
}
}

153
packages/core/src/notification-observability.ts Normal file
View File

@ -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<readonly [RegExp, string]> = [
[/\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");
}
}

49
packages/core/src/observability.ts
View File

@ -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<string, unknown>, 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<string, unknown>,
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");
}
}

13
packages/core/src/types.ts
View File

@ -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;

18
schema/config.schema.json
View File

@ -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": {},