From 0211407d8a2c2733daeef7e830a6ed9105e02fac Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:57:50 +0800 Subject: [PATCH] refactor(gateway): own health snapshot collection --- src/commands/health.delivery-queue.test.ts | 2 +- src/commands/health.plugins.test.ts | 14 +- src/commands/health.snapshot.test.ts | 27 +- src/commands/health.ts | 626 +----------------- src/gateway/health/account-context.ts | 168 +++++ src/gateway/health/collector.ts | 427 ++++++++++++ src/gateway/health/delivery-queue.ts | 60 ++ src/gateway/server-maintenance.test.ts | 2 +- src/gateway/server-maintenance.ts | 2 +- src/gateway/server-methods/health.ts | 2 +- src/gateway/server-node-events-types.ts | 2 +- src/gateway/server-node-events.test.ts | 2 +- src/gateway/server/health-state.test.ts | 52 +- src/gateway/server/health-state.ts | 11 +- src/gateway/test-helpers.maintenance-state.ts | 2 +- src/gateway/test-helpers.mocks.ts | 4 +- 16 files changed, 733 insertions(+), 670 deletions(-) create mode 100644 src/gateway/health/account-context.ts create mode 100644 src/gateway/health/collector.ts create mode 100644 src/gateway/health/delivery-queue.ts diff --git a/src/commands/health.delivery-queue.test.ts b/src/commands/health.delivery-queue.test.ts index 02333797eea7..2a7f9e8f69ae 100644 --- a/src/commands/health.delivery-queue.test.ts +++ b/src/commands/health.delivery-queue.test.ts @@ -20,7 +20,7 @@ vi.mock("../channels/message/ingress-queue.js", async (importOriginal) => { }; }); -const { buildDeliveryQueueHealthSummary } = await import("./health.js"); +const { buildDeliveryQueueHealthSummary } = await import("../gateway/health/delivery-queue.js"); describe("buildDeliveryQueueHealthSummary", () => { beforeEach(() => { diff --git a/src/commands/health.plugins.test.ts b/src/commands/health.plugins.test.ts index 604371f142c8..89032035df49 100644 --- a/src/commands/health.plugins.test.ts +++ b/src/commands/health.plugins.test.ts @@ -12,9 +12,9 @@ const tempPaths: string[] = []; let setActivePluginRegistry: typeof import("../plugins/runtime.js").setActivePluginRegistry; let setActiveDegradedPlugins: typeof import("../plugins/runtime-degraded-state.js").setActiveDegradedPlugins; let createTestRegistry: typeof import("../test-utils/channel-plugins.js").createTestRegistry; -let getHealthSnapshot: typeof import("./health.js").getHealthSnapshot; +let collectGatewayHealthSnapshot: typeof import("../gateway/health/collector.js").collectGatewayHealthSnapshot; -describe("getHealthSnapshot plugin state", () => { +describe("collectGatewayHealthSnapshot plugin state", () => { beforeAll(async () => { vi.doMock("../config/config.js", () => ({ getRuntimeConfig: () => testConfig, @@ -35,12 +35,12 @@ describe("getHealthSnapshot plugin state", () => { import("../plugins/runtime.js"), import("../plugins/runtime-degraded-state.js"), import("../test-utils/channel-plugins.js"), - import("./health.js"), + import("../gateway/health/collector.js"), ]); setActivePluginRegistry = pluginsRuntime.setActivePluginRegistry; setActiveDegradedPlugins = degradedState.setActiveDegradedPlugins; createTestRegistry = channelTestUtils.createTestRegistry; - getHealthSnapshot = health.getHealthSnapshot; + collectGatewayHealthSnapshot = health.collectGatewayHealthSnapshot; }); afterEach(() => { @@ -93,7 +93,11 @@ describe("getHealthSnapshot plugin state", () => { }, ]); - const snap = await getHealthSnapshot({ timeoutMs: 10, probe: false }); + const snap = await collectGatewayHealthSnapshot({ + audience: "admin", + timeoutMs: 10, + probe: false, + }); expect(Value.Check(SnapshotSchema.properties.health, snap)).toBe(true); expect(snap.plugins?.unavailable).toEqual([ diff --git a/src/commands/health.snapshot.test.ts b/src/commands/health.snapshot.test.ts index 660067a5a7a9..ee549f4ec557 100644 --- a/src/commands/health.snapshot.test.ts +++ b/src/commands/health.snapshot.test.ts @@ -5,10 +5,11 @@ import path from "node:path"; import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import type { ChannelAccountSnapshot } from "../channels/plugins/types.public.js"; import type { ChannelPlugin } from "../channels/plugins/types.public.js"; +import type { collectGatewayHealthSnapshot } from "../gateway/health/collector.js"; +import type { HealthSummary } from "../gateway/health/types.js"; import { createPluginRecord } from "../plugins/status.test-fixtures.js"; import { MAX_TIMER_TIMEOUT_MS } from "../shared/number-coercion.js"; import { createOpenClawTestState } from "../test-utils/openclaw-test-state.js"; -import type { HealthSummary } from "./health.js"; let testConfig: Record = {}; let testStore: Record = {}; @@ -19,7 +20,15 @@ let setActivePluginRegistry: typeof import("../plugins/runtime.js").setActivePlu let setActiveDegradedPlugins: typeof import("../plugins/runtime-degraded-state.js").setActiveDegradedPlugins; let createChannelTestPluginBase: typeof import("../test-utils/channel-plugins.js").createChannelTestPluginBase; let createTestRegistry: typeof import("../test-utils/channel-plugins.js").createTestRegistry; -let getHealthSnapshot: typeof import("./health.js").getHealthSnapshot; +type LegacyHealthSnapshotParams = Omit< + Parameters[0], + "audience" | "probe" +> & { + includeSensitive?: boolean; + probe?: boolean; +}; + +let getHealthSnapshot: (params?: LegacyHealthSnapshotParams) => Promise; let buildTelegramHealthSummaryForTest = buildTelegramHealthSummary; let probeTelegramAccountForTestOverride: | ((account: TelegramHealthAccount, timeoutMs: number) => Promise>) @@ -91,15 +100,23 @@ async function loadFreshHealthModulesForTest() { import("../plugins/runtime.js"), import("../plugins/runtime-degraded-state.js"), import("../test-utils/channel-plugins.js"), - import("./health.js"), + import("../gateway/health/collector.js"), ]); + const collectSnapshot = health.collectGatewayHealthSnapshot; return { setActivePluginRegistry: pluginsRuntime.setActivePluginRegistry, setActiveDegradedPlugins: pluginDegradedState.setActiveDegradedPlugins, createChannelTestPluginBase: channelTestUtils.createChannelTestPluginBase, createTestRegistry: channelTestUtils.createTestRegistry, - getHealthSnapshot: health.getHealthSnapshot, + getHealthSnapshot: (params: LegacyHealthSnapshotParams = {}) => { + const { includeSensitive, probe, ...rest } = params; + return collectSnapshot({ + ...rest, + audience: includeSensitive === false ? "public" : "admin", + probe: probe !== false, + }); + }, }; } @@ -458,7 +475,7 @@ function createIMessageHealthPlugin(): HealthTestPlugin { }; } -describe("getHealthSnapshot", () => { +describe("collectGatewayHealthSnapshot", () => { beforeAll(async () => { ({ setActivePluginRegistry, diff --git a/src/commands/health.ts b/src/commands/health.ts index f4fa00bd474b..626544ca463f 100644 --- a/src/commands/health.ts +++ b/src/commands/health.ts @@ -1,27 +1,13 @@ -import { expectDefined } from "@openclaw/normalization-core"; /** Collects and renders gateway health for channels, agents, plugins, and sessions. */ -import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; import { asNullableRecord } from "@openclaw/normalization-core/record-coerce"; import { styleHealthChannelLine } from "../../packages/terminal-core/src/health-style.js"; import { isRich } from "../../packages/terminal-core/src/theme.js"; -import { listAgentEntries, resolveDefaultAgentId } from "../agents/agent-scope.js"; -import { inspectChannelAccount } from "../channels/account-inspection.js"; -import { redactChannelStatusSummaryBaseUrl } from "../channels/account-snapshot-fields.js"; -import { - resolveChannelAccountConfigured, - resolveChannelAccountEnabled, -} from "../channels/account-summary.js"; -import { countFailedChannelIngressQueueEntries } from "../channels/message/ingress-queue.js"; import { resolveChannelDefaultAccountId } from "../channels/plugins/helpers.js"; import { listReadOnlyChannelPluginsForConfig } from "../channels/plugins/read-only.js"; -import { buildChannelAccountSnapshotFromAccount } from "../channels/plugins/status.js"; -import type { ChannelPlugin } from "../channels/plugins/types.plugin.js"; -import type { ChannelAccountSnapshot } from "../channels/plugins/types.public.js"; import { probeGatewayStatus } from "../cli/daemon-cli/probe.js"; import { withProgress } from "../cli/progress.js"; import { resolveStorePath } from "../config/sessions/paths.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; -import { listContextEngineQuarantines } from "../context-engine/registry.js"; import { buildGatewayConnectionDetails, buildGatewayProbeConnectionDetails, @@ -29,39 +15,20 @@ import { formatGatewayTransportErrorJson, isGatewayCredentialsRequiredError, } from "../gateway/call.js"; -import { - DEFAULT_CHANNEL_CONNECT_GRACE_MS, - DEFAULT_CHANNEL_STALE_EVENT_THRESHOLD_MS, - evaluateChannelHealth, -} from "../gateway/channel-health-policy.js"; -import type { GatewayHotReloadStatus } from "../gateway/config-reload-status.types.js"; import { isGatewaySecretRefUnavailableError } from "../gateway/credentials.js"; -import type { - AgentHealthSummary, - ChannelAccountHealthSummary, - ChannelHealthSummary, - ContextEngineHealthSummary, - DeliveryQueueHealthSummary, - HealthSummary, - PluginHealthErrorSummary, - PluginHealthSummary, -} from "../gateway/health/types.js"; -import type { ChannelRuntimeSnapshot } from "../gateway/server-channel-runtime.types.js"; +import { resolveHealthAccountContext } from "../gateway/health/account-context.js"; +import { + buildHealthSessionSummary as buildSessionSummary, + resolveHealthAgentOrder as resolveAgentOrder, +} from "../gateway/health/collector.js"; +import type { AgentHealthSummary, HealthSummary } from "../gateway/health/types.js"; import { info } from "../globals.js"; -import { countFailedDeliveryQueueEntries } from "../infra/delivery-queue-sqlite.js"; import { isDiagnosticFlagEnabled } from "../infra/diagnostic-flags.js"; import { formatErrorMessage } from "../infra/errors.js"; import { formatDurationHuman } from "../infra/format-time/format-duration.js"; import { resolveHeartbeatSummaryForAgent } from "../infra/heartbeat-summary.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; -import { - degradedPluginMatchesRoot, - listActiveDegradedPlugins, - toPublicPluginVerificationDiagnostic, -} from "../plugins/runtime-degraded-state.js"; -import { getActivePluginRegistry } from "../plugins/runtime.js"; import { buildChannelAccountBindings, resolvePreferredAccountId } from "../routing/bindings.js"; -import { normalizeAgentId } from "../routing/session-key.js"; import { type RuntimeEnv, writeRuntimeJson } from "../runtime.js"; import { buildCredentialsRequiredHealthDiagnostic, @@ -136,45 +103,6 @@ export async function emitReachableGatewayAuthDiagnostic(params: { const loadConfigRuntime = async () => await import("../config/config.js"); -const PUBLIC_IMESSAGE_FULL_DISK_ACCESS_ERROR = - "imsg cannot access ~/Library/Messages/chat.db. Grant Full Disk Access to the Gateway/launcher process and restart Gateway."; - -const redactIMessageProbeErrorMessage = (message: string): string => { - const trimmed = message.trim(); - if (!trimmed) { - return ""; - } - return trimmed.replaceAll( - /\/Users\/[^/\s]+\/Library\/Messages\/chat\.db/g, - "~/Library/Messages/chat.db", - ); -}; - -const buildNonSensitiveProbeFailure = ( - channelId: string, - probe: unknown, -): Record | undefined => { - const record = asNullableRecord(probe); - if (channelId !== "imessage" || !record || record.ok !== false) { - return undefined; - } - if (typeof record.error !== "string") { - return undefined; - } - - // Preserve the actionable Full Disk Access failure while stripping the local - // username path before health leaves the gateway. - const error = redactIMessageProbeErrorMessage(record.error); - if ( - !/\bimsg\b/i.test(error) || - !error.includes("~/Library/Messages/chat.db") || - !/\bFull Disk Access\b/i.test(error) - ) { - return undefined; - } - return { ok: false, error: PUBLIC_IMESSAGE_FULL_DISK_ACCESS_ERROR }; -}; - const formatDurationParts = (ms: number): string => { if (!Number.isFinite(ms)) { return "unknown"; @@ -218,23 +146,6 @@ function formatEventLoopHealthLine(summary: HealthSummary): string | null { }`; } -function buildContextEngineHealthSummary(): ContextEngineHealthSummary | undefined { - const quarantined: ContextEngineHealthSummary["quarantined"] = []; - for (const entry of listContextEngineQuarantines()) { - const summary: ContextEngineHealthSummary["quarantined"][number] = { - engineId: entry.engineId, - operation: entry.operation, - reason: entry.reason, - failedAt: entry.failedAt.getTime(), - }; - if (entry.owner) { - summary.owner = entry.owner; - } - quarantined.push(summary); - } - return quarantined.length > 0 ? { quarantined } : undefined; -} - /** Formats context engine quarantine state for text health output. */ export function formatContextEngineHealthLine(summary: HealthSummary): string | null { const quarantined = summary.contextEngines?.quarantined ?? []; @@ -245,54 +156,6 @@ export function formatContextEngineHealthLine(summary: HealthSummary): string | return `Context engine: warning (${quarantined.length} quarantined; downgraded to legacy: ${engines})`; } -/** Builds dead-lettered inbound and outbound queue health for cached gateway responses. */ -export function buildDeliveryQueueHealthSummary(): DeliveryQueueHealthSummary | undefined { - // Queue health reads are diagnostic; a storage failure must not take the - // gateway health endpoint down with it. - let failed: DeliveryQueueHealthSummary["failed"] = []; - try { - failed = countFailedDeliveryQueueEntries().map((queue) => { - const entry: DeliveryQueueHealthSummary["failed"][number] = { - queueName: queue.queueName, - count: queue.count, - }; - if (queue.oldestFailedAt != null) { - entry.oldestFailedAt = queue.oldestFailedAt; - } - return entry; - }); - } catch (error) { - debugHealth(undefined, "outbound delivery queue health read failed", { - error: formatErrorMessage(error), - }); - } - let ingressFailed: NonNullable = []; - try { - ingressFailed = countFailedChannelIngressQueueEntries().map((queue) => { - const entry: NonNullable[number] = { - channelId: queue.channelId, - accountId: queue.accountId, - count: queue.count, - }; - if (queue.oldestFailedAt != null) { - entry.oldestFailedAt = queue.oldestFailedAt; - } - return entry; - }); - } catch (error) { - debugHealth(undefined, "channel ingress queue health read failed", { - error: formatErrorMessage(error), - }); - } - if (failed.length === 0 && ingressFailed.length === 0) { - return undefined; - } - return { - failed, - ...(ingressFailed.length > 0 ? { ingressFailed } : {}), - }; -} - /** Formats dead-lettered delivery queue entries for text health output. */ export function formatDeliveryQueueHealthLine( summary: HealthSummary, @@ -328,477 +191,6 @@ export function formatConfigReloadHealthLine(summary: HealthSummary): string | n const resolveHeartbeatSummary = (cfg: OpenClawConfig, agentId: string) => resolveHeartbeatSummaryForAgent(cfg, agentId); -const resolveAgentOrder = (cfg: OpenClawConfig) => { - const defaultAgentId = resolveDefaultAgentId(cfg); - const entries = listAgentEntries(cfg); - const seen = new Set(); - const ordered: Array<{ id: string; name?: string }> = []; - - for (const entry of entries) { - if (!entry || typeof entry !== "object") { - continue; - } - if (typeof entry.id !== "string" || !entry.id.trim()) { - continue; - } - const id = normalizeAgentId(entry.id); - if (!id || seen.has(id)) { - continue; - } - seen.add(id); - ordered.push({ id, name: typeof entry.name === "string" ? entry.name : undefined }); - } - - if (!seen.has(defaultAgentId)) { - ordered.unshift({ id: defaultAgentId }); - } - - if (ordered.length === 0) { - ordered.push({ id: defaultAgentId }); - } - - return { defaultAgentId, ordered }; -}; - -const buildSessionSummary = async (storePath: string, agentId?: string) => { - const { listSessionEntriesReadOnly } = await import("../config/sessions/session-accessor.js"); - const { isTransientSqliteError } = await import("../infra/unhandled-rejections.js"); - let listed: ReturnType; - try { - listed = listSessionEntriesReadOnly({ - ...(agentId ? { agentId } : {}), - storePath, - }); - } catch (error) { - if (!isTransientSqliteError(error)) { - throw error; - } - // Health is best-effort: an empty snapshot beats failing on a transient lock. - listed = []; - } - const sessions = listed - .filter(({ sessionKey }) => sessionKey !== "global" && sessionKey !== "unknown") - .map(({ sessionKey, entry }) => ({ key: sessionKey, updatedAt: entry?.updatedAt ?? 0 })) - .toSorted((a, b) => b.updatedAt - a.updatedAt); - const recent = sessions.slice(0, 5).map((s) => ({ - key: s.key, - updatedAt: s.updatedAt || null, - age: s.updatedAt ? Date.now() - s.updatedAt : null, - })); - return { - path: storePath, - count: sessions.length, - recent, - } satisfies HealthSummary["sessions"]; -}; - -function buildPluginHealthSummary(): PluginHealthSummary | undefined { - const registry = getActivePluginRegistry(); - const degradedPlugins = listActiveDegradedPlugins(); - const unavailable = degradedPlugins - .map(({ pluginId, state, diagnostic }) => ({ - id: pluginId, - state, - diagnostic: toPublicPluginVerificationDiagnostic(diagnostic), - })) - .toSorted((left, right) => left.id.localeCompare(right.id)); - const loaded = (registry?.plugins ?? []) - .filter((plugin) => plugin.status === "loaded") - .map((plugin) => plugin.id) - .toSorted((left, right) => left.localeCompare(right)); - const errors = (registry?.plugins ?? []) - .filter( - (plugin) => - plugin.status === "error" && - !degradedPlugins.some( - (degraded) => - plugin.id === degraded.pluginId && - plugin.failurePhase === "validation" && - plugin.activationReason === `configured-unavailable: ${degraded.diagnostic.reason}` && - Boolean(plugin.rootDir) && - degradedPluginMatchesRoot(degraded, plugin.rootDir ?? ""), - ), - ) - .map((plugin) => { - const error: PluginHealthErrorSummary = { - id: plugin.id, - origin: plugin.origin, - activated: plugin.activated === true, - error: plugin.error ?? "unknown plugin load error", - }; - if (plugin.activationSource) { - error.activationSource = plugin.activationSource; - } - if (plugin.activationReason) { - error.activationReason = plugin.activationReason; - } - if (plugin.failurePhase) { - error.failurePhase = plugin.failurePhase; - } - return error; - }) - .toSorted((left, right) => left.id.localeCompare(right.id)); - if (loaded.length === 0 && errors.length === 0 && unavailable.length === 0) { - return undefined; - } - return { loaded, errors, unavailable }; -} - -function readBooleanField(value: unknown, key: string): boolean | undefined { - const record = asNullableRecord(value); - if (!record) { - return undefined; - } - return typeof record[key] === "boolean" ? record[key] : undefined; -} - -const hasAccountValue = (account: unknown): boolean => account !== null && account !== undefined; - -function resolveProbeAccountEnabled(params: { - plugin: ChannelPlugin; - cfg: OpenClawConfig; - accountId: string; - account: unknown; - diagnostics: string[]; -}): boolean { - const fallback = readBooleanField(params.account, "enabled") ?? true; - try { - return resolveChannelAccountEnabled({ - plugin: params.plugin, - account: params.account, - cfg: params.cfg, - }); - } catch (error) { - params.diagnostics.push( - `${params.plugin.id}:${params.accountId}: failed to evaluate enabled state (${formatErrorMessage(error)}).`, - ); - return fallback; - } -} - -async function resolveProbeAccountConfigured(params: { - plugin: ChannelPlugin; - cfg: OpenClawConfig; - accountId: string; - account: unknown; - diagnostics: string[]; -}): Promise { - const fallback = readBooleanField(params.account, "configured") ?? true; - try { - return await resolveChannelAccountConfigured({ - plugin: params.plugin, - account: params.account, - cfg: params.cfg, - readAccountConfiguredField: true, - }); - } catch (error) { - params.diagnostics.push( - `${params.plugin.id}:${params.accountId}: failed to evaluate configured state (${formatErrorMessage(error)}).`, - ); - return fallback; - } -} - -async function resolveHealthAccountContext(params: { - plugin: ChannelPlugin; - cfg: OpenClawConfig; - accountId: string; -}): Promise<{ - probeAccount: unknown; - snapshotAccount: unknown; - enabled: boolean; - configured: boolean; - diagnostics: string[]; -}> { - const diagnostics: string[] = []; - let account: unknown; - try { - account = params.plugin.config.resolveAccount(params.cfg, params.accountId); - } catch (error) { - diagnostics.push( - `${params.plugin.id}:${params.accountId}: failed to resolve account (${formatErrorMessage(error)}).`, - ); - } - let inspectedAccount: unknown; - try { - inspectedAccount = await inspectChannelAccount(params); - } catch (error) { - diagnostics.push( - `${params.plugin.id}:${params.accountId}: failed to inspect account (${formatErrorMessage(error)}).`, - ); - } - - const probeAccount = hasAccountValue(account) ? account : inspectedAccount; - if (!hasAccountValue(probeAccount)) { - return { - probeAccount: {}, - snapshotAccount: {}, - enabled: false, - configured: false, - diagnostics, - }; - } - const snapshotAccount = hasAccountValue(inspectedAccount) ? inspectedAccount : probeAccount; - - const enabled = resolveProbeAccountEnabled({ - plugin: params.plugin, - cfg: params.cfg, - accountId: params.accountId, - account: probeAccount, - diagnostics, - }); - const configured = await resolveProbeAccountConfigured({ - plugin: params.plugin, - cfg: params.cfg, - accountId: params.accountId, - account: probeAccount, - diagnostics, - }); - - return { - probeAccount, - snapshotAccount, - enabled, - configured, - diagnostics, - }; -} - -/** Builds the gateway-side health snapshot for channels, agents, plugins, and sessions. */ -export async function getHealthSnapshot(params?: { - timeoutMs?: number; - probe?: boolean; - includeSensitive?: boolean; - runtimeSnapshot?: ChannelRuntimeSnapshot; - eventLoop?: HealthSummary["eventLoop"]; - configReloadHotReloadStatus?: GatewayHotReloadStatus; -}): Promise { - const timeoutMs = params?.timeoutMs; - const cfg = await readRuntimeHealthConfig(); - const { defaultAgentId, ordered } = resolveAgentOrder(cfg); - const channelBindings = buildChannelAccountBindings(cfg); - const sessionCache = new Map(); - const agents: AgentHealthSummary[] = []; - for (const entry of ordered) { - const storePath = resolveStorePath(cfg.session?.store, { agentId: entry.id }); - const sessionCacheKey = `${storePath}\0${entry.id}`; - const sessions = - sessionCache.get(sessionCacheKey) ?? (await buildSessionSummary(storePath, entry.id)); - sessionCache.set(sessionCacheKey, sessions); - agents.push({ - agentId: entry.id, - name: entry.name, - isDefault: entry.id === defaultAgentId, - heartbeat: resolveHeartbeatSummary(cfg, entry.id), - sessions, - }); - } - const defaultAgent = agents.find((agent) => agent.isDefault) ?? agents[0]; - const heartbeatSeconds = defaultAgent?.heartbeat.everyMs - ? Math.round(defaultAgent.heartbeat.everyMs / 1000) - : 0; - const sessions = - defaultAgent?.sessions ?? - (await buildSessionSummary( - resolveStorePath(cfg.session?.store, { agentId: defaultAgentId }), - defaultAgentId, - )); - - const start = Date.now(); - const cappedTimeout = resolveTimerTimeoutMs(timeoutMs, DEFAULT_TIMEOUT_MS, 50); - const doProbe = params?.probe !== false; - const includeSensitive = params?.includeSensitive !== false; - const channels: Record = {}; - const plugins = listReadOnlyChannelPluginsForConfig(cfg, { - includeSetupFallbackPlugins: false, - }); - const channelOrder = plugins.map((plugin) => plugin.id); - const channelLabels: Record = {}; - - for (const plugin of plugins) { - channelLabels[plugin.id] = plugin.meta.label ?? plugin.id; - const accountIds = plugin.config.listAccountIds(cfg); - const defaultAccountId = resolveChannelDefaultAccountId({ - plugin, - cfg, - accountIds, - }); - const boundAccounts = channelBindings.get(plugin.id)?.get(defaultAgentId) ?? []; - const preferredAccountId = resolvePreferredAccountId({ - accountIds, - defaultAccountId, - boundAccounts, - }); - const boundAccountIdsAll = Array.from( - new Set(Array.from(channelBindings.get(plugin.id)?.values() ?? []).flat()), - ); - const accountIdsToProbe = Array.from( - new Set( - [preferredAccountId, defaultAccountId, ...accountIds, ...boundAccountIdsAll].filter( - (value) => value && value.trim(), - ), - ), - ); - // Probe preferred/default/bound accounts first, but include all configured - // accounts so verbose health can explain account-specific failures. - debugHealth(cfg, "channel", { - id: plugin.id, - accountIds, - defaultAccountId, - boundAccounts, - preferredAccountId, - accountIdsToProbe, - }); - const accountSummaries: Record = {}; - - for (const accountId of accountIdsToProbe) { - const { probeAccount, snapshotAccount, enabled, configured, diagnostics } = - await resolveHealthAccountContext({ - plugin, - cfg, - accountId, - }); - if (diagnostics.length > 0) { - debugHealth(cfg, "account.diagnostics", { channel: plugin.id, accountId, diagnostics }); - } - - let probe: unknown; - let lastProbeAt: number | null = null; - if (enabled && configured && doProbe && plugin.status?.probeAccount) { - try { - probe = await plugin.status.probeAccount({ - account: probeAccount, - timeoutMs: cappedTimeout, - cfg, - }); - lastProbeAt = Date.now(); - } catch (err) { - probe = { ok: false, error: formatErrorMessage(err) }; - lastProbeAt = Date.now(); - } - } - - const probeRecord = - probe && typeof probe === "object" ? (probe as Record) : null; - const bot = - probeRecord && typeof probeRecord.bot === "object" - ? (probeRecord.bot as { username?: string | null }) - : null; - if (bot?.username) { - debugHealth(cfg, "probe.bot", { channel: plugin.id, accountId, username: bot.username }); - } - - const runtimeSnapshot = - params?.runtimeSnapshot?.channelAccounts[plugin.id]?.[accountId] ?? - (accountId === defaultAccountId ? params?.runtimeSnapshot?.channels[plugin.id] : undefined); - const nonSensitiveProbeFailure = buildNonSensitiveProbeFailure(plugin.id, probe); - const snapshotProbe = includeSensitive ? probe : nonSensitiveProbeFailure; - const snapshot: ChannelAccountSnapshot = await buildChannelAccountSnapshotFromAccount({ - plugin, - cfg, - accountId, - account: snapshotAccount, - runtime: runtimeSnapshot, - probe: snapshotProbe, - enabledFallback: enabled, - configuredFallback: configured, - }); - if (lastProbeAt) { - snapshot.lastProbeAt = lastProbeAt; - } - const health = evaluateChannelHealth(snapshot, { - channelId: plugin.id, - now: Date.now(), - staleEventThresholdMs: DEFAULT_CHANNEL_STALE_EVENT_THRESHOLD_MS, - channelConnectGraceMs: DEFAULT_CHANNEL_CONNECT_GRACE_MS, - }); - if (!health.healthy) { - snapshot.healthState = health.reason; - } - - const summary = plugin.status?.buildChannelSummary - ? await plugin.status.buildChannelSummary({ - account: probeAccount, - cfg, - defaultAccountId: accountId, - snapshot, - }) - : undefined; - // Summary hooks overlay the safe snapshot, so reapply URL redaction after the final merge. - const record = redactChannelStatusSummaryBaseUrl( - summary && typeof summary === "object" - ? ({ ...snapshot, ...summary } as ChannelAccountHealthSummary) - : ({ ...snapshot, accountId, configured } satisfies ChannelAccountHealthSummary), - ); - if (record.configured === undefined) { - record.configured = configured; - } - if (includeSensitive && record.probe === undefined && probe !== undefined) { - record.probe = probe; - } - if (!includeSensitive) { - const summaryProbeFailure = buildNonSensitiveProbeFailure(plugin.id, record.probe); - const safeProbeFailure = summaryProbeFailure ?? nonSensitiveProbeFailure; - if (safeProbeFailure) { - record.probe = safeProbeFailure; - } else { - delete record.probe; - } - } - if (record.lastProbeAt === undefined && lastProbeAt) { - record.lastProbeAt = lastProbeAt; - } - record.accountId = accountId; - accountSummaries[accountId] = record; - } - - const defaultSummary = - accountSummaries[preferredAccountId] ?? - accountSummaries[defaultAccountId] ?? - accountSummaries[accountIdsToProbe[0] ?? preferredAccountId]; - const fallbackSummary = - defaultSummary ?? - accountSummaries[ - expectDefined(Object.keys(accountSummaries)[0], "object.keys(account summaries) entry at 0") - ]; - if (fallbackSummary) { - channels[plugin.id] = { - ...fallbackSummary, - accounts: accountSummaries, - } satisfies ChannelHealthSummary; - } - } - - const pluginHealth = buildPluginHealthSummary(); - const contextEngineHealth = buildContextEngineHealthSummary(); - const deliveryQueueHealth = buildDeliveryQueueHealthSummary(); - const summary: HealthSummary = { - ok: true, - ts: Date.now(), - durationMs: Date.now() - start, - ...(params?.eventLoop ? { eventLoop: params.eventLoop } : {}), - ...(pluginHealth ? { plugins: pluginHealth } : {}), - ...(contextEngineHealth ? { contextEngines: contextEngineHealth } : {}), - ...(deliveryQueueHealth ? { deliveryQueues: deliveryQueueHealth } : {}), - ...(params?.configReloadHotReloadStatus - ? { configReload: { hotReloadStatus: params.configReloadHotReloadStatus } } - : {}), - channels, - channelOrder, - channelLabels, - heartbeatSeconds, - defaultAgentId, - agents, - sessions: { - path: sessions.path, - count: sessions.count, - recent: sessions.recent, - }, - }; - - return summary; -} - /** Runs the `openclaw health` command against the gateway and renders JSON or text. */ export async function healthCommand( opts: { @@ -1126,9 +518,3 @@ async function readBestEffortHealthConfig(): Promise { const { readBestEffortConfig } = await loadConfigRuntime(); return await readBestEffortConfig(); } - -async function readRuntimeHealthConfig(): Promise { - const { getRuntimeConfig } = await loadConfigRuntime(); - return getRuntimeConfig(); -} -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/gateway/health/account-context.ts b/src/gateway/health/account-context.ts new file mode 100644 index 000000000000..522e84b51753 --- /dev/null +++ b/src/gateway/health/account-context.ts @@ -0,0 +1,168 @@ +import { asNullableRecord } from "@openclaw/normalization-core/record-coerce"; +import { inspectChannelAccount } from "../../channels/account-inspection.js"; +import { + resolveChannelAccountConfigured, + resolveChannelAccountEnabled, +} from "../../channels/account-summary.js"; +import type { ChannelPlugin } from "../../channels/plugins/types.plugin.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { formatErrorMessage } from "../../infra/errors.js"; + +const PUBLIC_IMESSAGE_FULL_DISK_ACCESS_ERROR = + "imsg cannot access ~/Library/Messages/chat.db. Grant Full Disk Access to the Gateway/launcher process and restart Gateway."; + +const redactIMessageProbeErrorMessage = (message: string): string => { + const trimmed = message.trim(); + if (!trimmed) { + return ""; + } + return trimmed.replaceAll( + /\/Users\/[^/\s]+\/Library\/Messages\/chat\.db/g, + "~/Library/Messages/chat.db", + ); +}; + +export function buildNonSensitiveProbeFailure( + channelId: string, + probe: unknown, +): Record | undefined { + const record = asNullableRecord(probe); + if (channelId !== "imessage" || !record || record.ok !== false) { + return undefined; + } + if (typeof record.error !== "string") { + return undefined; + } + + // Preserve the actionable Full Disk Access failure while stripping the local + // username path before health leaves the gateway. + const error = redactIMessageProbeErrorMessage(record.error); + if ( + !/\bimsg\b/i.test(error) || + !error.includes("~/Library/Messages/chat.db") || + !/\bFull Disk Access\b/i.test(error) + ) { + return undefined; + } + return { ok: false, error: PUBLIC_IMESSAGE_FULL_DISK_ACCESS_ERROR }; +} + +function readBooleanField(value: unknown, key: string): boolean | undefined { + const record = asNullableRecord(value); + if (!record) { + return undefined; + } + return typeof record[key] === "boolean" ? record[key] : undefined; +} + +const hasAccountValue = (account: unknown): boolean => account !== null && account !== undefined; + +function resolveProbeAccountEnabled(params: { + plugin: ChannelPlugin; + cfg: OpenClawConfig; + accountId: string; + account: unknown; + diagnostics: string[]; +}): boolean { + const fallback = readBooleanField(params.account, "enabled") ?? true; + try { + return resolveChannelAccountEnabled({ + plugin: params.plugin, + account: params.account, + cfg: params.cfg, + }); + } catch (error) { + params.diagnostics.push( + `${params.plugin.id}:${params.accountId}: failed to evaluate enabled state (${formatErrorMessage(error)}).`, + ); + return fallback; + } +} + +async function resolveProbeAccountConfigured(params: { + plugin: ChannelPlugin; + cfg: OpenClawConfig; + accountId: string; + account: unknown; + diagnostics: string[]; +}): Promise { + const fallback = readBooleanField(params.account, "configured") ?? true; + try { + return await resolveChannelAccountConfigured({ + plugin: params.plugin, + account: params.account, + cfg: params.cfg, + readAccountConfiguredField: true, + }); + } catch (error) { + params.diagnostics.push( + `${params.plugin.id}:${params.accountId}: failed to evaluate configured state (${formatErrorMessage(error)}).`, + ); + return fallback; + } +} + +export async function resolveHealthAccountContext(params: { + plugin: ChannelPlugin; + cfg: OpenClawConfig; + accountId: string; +}): Promise<{ + probeAccount: unknown; + snapshotAccount: unknown; + enabled: boolean; + configured: boolean; + diagnostics: string[]; +}> { + const diagnostics: string[] = []; + let account: unknown; + try { + account = params.plugin.config.resolveAccount(params.cfg, params.accountId); + } catch (error) { + diagnostics.push( + `${params.plugin.id}:${params.accountId}: failed to resolve account (${formatErrorMessage(error)}).`, + ); + } + let inspectedAccount: unknown; + try { + inspectedAccount = await inspectChannelAccount(params); + } catch (error) { + diagnostics.push( + `${params.plugin.id}:${params.accountId}: failed to inspect account (${formatErrorMessage(error)}).`, + ); + } + + const probeAccount = hasAccountValue(account) ? account : inspectedAccount; + if (!hasAccountValue(probeAccount)) { + return { + probeAccount: {}, + snapshotAccount: {}, + enabled: false, + configured: false, + diagnostics, + }; + } + const snapshotAccount = hasAccountValue(inspectedAccount) ? inspectedAccount : probeAccount; + + const enabled = resolveProbeAccountEnabled({ + plugin: params.plugin, + cfg: params.cfg, + accountId: params.accountId, + account: probeAccount, + diagnostics, + }); + const configured = await resolveProbeAccountConfigured({ + plugin: params.plugin, + cfg: params.cfg, + accountId: params.accountId, + account: probeAccount, + diagnostics, + }); + + return { + probeAccount, + snapshotAccount, + enabled, + configured, + diagnostics, + }; +} diff --git a/src/gateway/health/collector.ts b/src/gateway/health/collector.ts new file mode 100644 index 000000000000..67712eae1a54 --- /dev/null +++ b/src/gateway/health/collector.ts @@ -0,0 +1,427 @@ +import { expectDefined } from "@openclaw/normalization-core"; +import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; +import { listAgentEntries, resolveDefaultAgentId } from "../../agents/agent-scope.js"; +import { redactChannelStatusSummaryBaseUrl } from "../../channels/account-snapshot-fields.js"; +import { resolveChannelDefaultAccountId } from "../../channels/plugins/helpers.js"; +import { listReadOnlyChannelPluginsForConfig } from "../../channels/plugins/read-only.js"; +import { buildChannelAccountSnapshotFromAccount } from "../../channels/plugins/status.js"; +import type { ChannelAccountSnapshot } from "../../channels/plugins/types.public.js"; +import { resolveStorePath } from "../../config/sessions/paths.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { listContextEngineQuarantines } from "../../context-engine/registry.js"; +import { isDiagnosticFlagEnabled } from "../../infra/diagnostic-flags.js"; +import { formatErrorMessage } from "../../infra/errors.js"; +import { resolveHeartbeatSummaryForAgent } from "../../infra/heartbeat-summary.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; +import { + degradedPluginMatchesRoot, + listActiveDegradedPlugins, + toPublicPluginVerificationDiagnostic, +} from "../../plugins/runtime-degraded-state.js"; +import { getActivePluginRegistry } from "../../plugins/runtime.js"; +import { buildChannelAccountBindings, resolvePreferredAccountId } from "../../routing/bindings.js"; +import { normalizeAgentId } from "../../routing/session-key.js"; +import { + DEFAULT_CHANNEL_CONNECT_GRACE_MS, + DEFAULT_CHANNEL_STALE_EVENT_THRESHOLD_MS, + evaluateChannelHealth, +} from "../channel-health-policy.js"; +import type { GatewayHotReloadStatus } from "../config-reload-status.types.js"; +import type { ChannelRuntimeSnapshot } from "../server-channel-runtime.types.js"; +import { buildNonSensitiveProbeFailure, resolveHealthAccountContext } from "./account-context.js"; +import { buildDeliveryQueueHealthSummary } from "./delivery-queue.js"; +import type { + AgentHealthSummary, + ChannelAccountHealthSummary, + ChannelHealthSummary, + ContextEngineHealthSummary, + HealthSummary, + PluginHealthErrorSummary, + PluginHealthSummary, +} from "./types.js"; + +const DEFAULT_HEALTH_TIMEOUT_MS = 10_000; +const healthLog = createSubsystemLogger("health"); + +export type HealthSnapshotAudience = "public" | "admin"; + +const debugHealth = ( + cfg: OpenClawConfig | undefined, + message: string, + meta?: Record, +) => { + if (isDiagnosticFlagEnabled("health", cfg)) { + healthLog.info(message, meta); + } +}; + +function buildContextEngineHealthSummary(): ContextEngineHealthSummary | undefined { + const quarantined: ContextEngineHealthSummary["quarantined"] = []; + for (const entry of listContextEngineQuarantines()) { + const summary: ContextEngineHealthSummary["quarantined"][number] = { + engineId: entry.engineId, + operation: entry.operation, + reason: entry.reason, + failedAt: entry.failedAt.getTime(), + }; + if (entry.owner) { + summary.owner = entry.owner; + } + quarantined.push(summary); + } + return quarantined.length > 0 ? { quarantined } : undefined; +} + +const resolveHeartbeatSummary = (cfg: OpenClawConfig, agentId: string) => + resolveHeartbeatSummaryForAgent(cfg, agentId); + +export function resolveHealthAgentOrder(cfg: OpenClawConfig) { + const defaultAgentId = resolveDefaultAgentId(cfg); + const entries = listAgentEntries(cfg); + const seen = new Set(); + const ordered: Array<{ id: string; name?: string }> = []; + + for (const entry of entries) { + if (!entry || typeof entry !== "object") { + continue; + } + if (typeof entry.id !== "string" || !entry.id.trim()) { + continue; + } + const id = normalizeAgentId(entry.id); + if (!id || seen.has(id)) { + continue; + } + seen.add(id); + ordered.push({ id, name: typeof entry.name === "string" ? entry.name : undefined }); + } + + if (!seen.has(defaultAgentId)) { + ordered.unshift({ id: defaultAgentId }); + } + if (ordered.length === 0) { + ordered.push({ id: defaultAgentId }); + } + + return { defaultAgentId, ordered }; +} + +export async function buildHealthSessionSummary(storePath: string, agentId?: string) { + const { listSessionEntriesReadOnly } = await import("../../config/sessions/session-accessor.js"); + const { isTransientSqliteError } = await import("../../infra/unhandled-rejections.js"); + let listed: ReturnType; + try { + listed = listSessionEntriesReadOnly({ + ...(agentId ? { agentId } : {}), + storePath, + }); + } catch (error) { + if (!isTransientSqliteError(error)) { + throw error; + } + // Health is best-effort: an empty snapshot beats failing on a transient lock. + listed = []; + } + const sessions = listed + .filter(({ sessionKey }) => sessionKey !== "global" && sessionKey !== "unknown") + .map(({ sessionKey, entry }) => ({ key: sessionKey, updatedAt: entry?.updatedAt ?? 0 })) + .toSorted((a, b) => b.updatedAt - a.updatedAt); + const recent = sessions.slice(0, 5).map((session) => ({ + key: session.key, + updatedAt: session.updatedAt || null, + age: session.updatedAt ? Date.now() - session.updatedAt : null, + })); + return { + path: storePath, + count: sessions.length, + recent, + } satisfies HealthSummary["sessions"]; +} + +function buildPluginHealthSummary(): PluginHealthSummary | undefined { + const registry = getActivePluginRegistry(); + const degradedPlugins = listActiveDegradedPlugins(); + const unavailable = degradedPlugins + .map(({ pluginId, state, diagnostic }) => ({ + id: pluginId, + state, + diagnostic: toPublicPluginVerificationDiagnostic(diagnostic), + })) + .toSorted((left, right) => left.id.localeCompare(right.id)); + const loaded = (registry?.plugins ?? []) + .filter((plugin) => plugin.status === "loaded") + .map((plugin) => plugin.id) + .toSorted((left, right) => left.localeCompare(right)); + const errors = (registry?.plugins ?? []) + .filter( + (plugin) => + plugin.status === "error" && + !degradedPlugins.some( + (degraded) => + plugin.id === degraded.pluginId && + plugin.failurePhase === "validation" && + plugin.activationReason === `configured-unavailable: ${degraded.diagnostic.reason}` && + Boolean(plugin.rootDir) && + degradedPluginMatchesRoot(degraded, plugin.rootDir ?? ""), + ), + ) + .map((plugin) => { + const error: PluginHealthErrorSummary = { + id: plugin.id, + origin: plugin.origin, + activated: plugin.activated === true, + error: plugin.error ?? "unknown plugin load error", + }; + if (plugin.activationSource) { + error.activationSource = plugin.activationSource; + } + if (plugin.activationReason) { + error.activationReason = plugin.activationReason; + } + if (plugin.failurePhase) { + error.failurePhase = plugin.failurePhase; + } + return error; + }) + .toSorted((left, right) => left.id.localeCompare(right.id)); + if (loaded.length === 0 && errors.length === 0 && unavailable.length === 0) { + return undefined; + } + return { loaded, errors, unavailable }; +} + +/** Collects the gateway-owned health snapshot for an explicit trust audience. */ +export async function collectGatewayHealthSnapshot(params: { + audience: HealthSnapshotAudience; + probe: boolean; + timeoutMs?: number; + runtimeSnapshot?: ChannelRuntimeSnapshot; + eventLoop?: HealthSummary["eventLoop"]; + configReloadHotReloadStatus?: GatewayHotReloadStatus; +}): Promise { + const cfg = await readRuntimeHealthConfig(); + const { defaultAgentId, ordered } = resolveHealthAgentOrder(cfg); + const channelBindings = buildChannelAccountBindings(cfg); + const sessionCache = new Map(); + const agents: AgentHealthSummary[] = []; + for (const entry of ordered) { + const storePath = resolveStorePath(cfg.session?.store, { agentId: entry.id }); + const sessionCacheKey = `${storePath}\0${entry.id}`; + const sessions = + sessionCache.get(sessionCacheKey) ?? (await buildHealthSessionSummary(storePath, entry.id)); + sessionCache.set(sessionCacheKey, sessions); + agents.push({ + agentId: entry.id, + name: entry.name, + isDefault: entry.id === defaultAgentId, + heartbeat: resolveHeartbeatSummary(cfg, entry.id), + sessions, + }); + } + const defaultAgent = agents.find((agent) => agent.isDefault) ?? agents[0]; + const heartbeatSeconds = defaultAgent?.heartbeat.everyMs + ? Math.round(defaultAgent.heartbeat.everyMs / 1000) + : 0; + const sessions = + defaultAgent?.sessions ?? + (await buildHealthSessionSummary( + resolveStorePath(cfg.session?.store, { agentId: defaultAgentId }), + defaultAgentId, + )); + + const start = Date.now(); + const cappedTimeout = resolveTimerTimeoutMs(params.timeoutMs, DEFAULT_HEALTH_TIMEOUT_MS, 50); + const includeSensitive = params.audience === "admin"; + const channels: Record = {}; + const plugins = listReadOnlyChannelPluginsForConfig(cfg, { + includeSetupFallbackPlugins: false, + }); + const channelOrder = plugins.map((plugin) => plugin.id); + const channelLabels: Record = {}; + + for (const plugin of plugins) { + channelLabels[plugin.id] = plugin.meta.label ?? plugin.id; + const accountIds = plugin.config.listAccountIds(cfg); + const defaultAccountId = resolveChannelDefaultAccountId({ + plugin, + cfg, + accountIds, + }); + const boundAccounts = channelBindings.get(plugin.id)?.get(defaultAgentId) ?? []; + const preferredAccountId = resolvePreferredAccountId({ + accountIds, + defaultAccountId, + boundAccounts, + }); + const boundAccountIdsAll = Array.from( + new Set(Array.from(channelBindings.get(plugin.id)?.values() ?? []).flat()), + ); + const accountIdsToProbe = Array.from( + new Set( + [preferredAccountId, defaultAccountId, ...accountIds, ...boundAccountIdsAll].filter( + (value) => value && value.trim(), + ), + ), + ); + // Probe preferred/default/bound accounts first, but include all configured + // accounts so verbose health can explain account-specific failures. + debugHealth(cfg, "channel", { + id: plugin.id, + accountIds, + defaultAccountId, + boundAccounts, + preferredAccountId, + accountIdsToProbe, + }); + const accountSummaries: Record = {}; + + for (const accountId of accountIdsToProbe) { + const { probeAccount, snapshotAccount, enabled, configured, diagnostics } = + await resolveHealthAccountContext({ + plugin, + cfg, + accountId, + }); + if (diagnostics.length > 0) { + debugHealth(cfg, "account.diagnostics", { channel: plugin.id, accountId, diagnostics }); + } + + let probe: unknown; + let lastProbeAt: number | null = null; + if (enabled && configured && params.probe && plugin.status?.probeAccount) { + try { + probe = await plugin.status.probeAccount({ + account: probeAccount, + timeoutMs: cappedTimeout, + cfg, + }); + lastProbeAt = Date.now(); + } catch (error) { + probe = { ok: false, error: formatErrorMessage(error) }; + lastProbeAt = Date.now(); + } + } + + const probeRecord = + probe && typeof probe === "object" ? (probe as Record) : null; + const bot = + probeRecord && typeof probeRecord.bot === "object" + ? (probeRecord.bot as { username?: string | null }) + : null; + if (bot?.username) { + debugHealth(cfg, "probe.bot", { channel: plugin.id, accountId, username: bot.username }); + } + + const runtimeSnapshot = + params.runtimeSnapshot?.channelAccounts[plugin.id]?.[accountId] ?? + (accountId === defaultAccountId ? params.runtimeSnapshot?.channels[plugin.id] : undefined); + const nonSensitiveProbeFailure = buildNonSensitiveProbeFailure(plugin.id, probe); + const snapshotProbe = includeSensitive ? probe : nonSensitiveProbeFailure; + const snapshot: ChannelAccountSnapshot = await buildChannelAccountSnapshotFromAccount({ + plugin, + cfg, + accountId, + account: snapshotAccount, + runtime: runtimeSnapshot, + probe: snapshotProbe, + enabledFallback: enabled, + configuredFallback: configured, + }); + if (lastProbeAt) { + snapshot.lastProbeAt = lastProbeAt; + } + const health = evaluateChannelHealth(snapshot, { + channelId: plugin.id, + now: Date.now(), + staleEventThresholdMs: DEFAULT_CHANNEL_STALE_EVENT_THRESHOLD_MS, + channelConnectGraceMs: DEFAULT_CHANNEL_CONNECT_GRACE_MS, + }); + if (!health.healthy) { + snapshot.healthState = health.reason; + } + + const summary = plugin.status?.buildChannelSummary + ? await plugin.status.buildChannelSummary({ + account: probeAccount, + cfg, + defaultAccountId: accountId, + snapshot, + }) + : undefined; + // Summary hooks overlay the safe snapshot, so reapply URL redaction after the final merge. + const record = redactChannelStatusSummaryBaseUrl( + summary && typeof summary === "object" + ? ({ ...snapshot, ...summary } as ChannelAccountHealthSummary) + : ({ ...snapshot, accountId, configured } satisfies ChannelAccountHealthSummary), + ); + if (record.configured === undefined) { + record.configured = configured; + } + if (includeSensitive && record.probe === undefined && probe !== undefined) { + record.probe = probe; + } + if (!includeSensitive) { + const summaryProbeFailure = buildNonSensitiveProbeFailure(plugin.id, record.probe); + const safeProbeFailure = summaryProbeFailure ?? nonSensitiveProbeFailure; + if (safeProbeFailure) { + record.probe = safeProbeFailure; + } else { + delete record.probe; + } + } + if (record.lastProbeAt === undefined && lastProbeAt) { + record.lastProbeAt = lastProbeAt; + } + record.accountId = accountId; + accountSummaries[accountId] = record; + } + + const defaultSummary = + accountSummaries[preferredAccountId] ?? + accountSummaries[defaultAccountId] ?? + accountSummaries[accountIdsToProbe[0] ?? preferredAccountId]; + const fallbackSummary = + defaultSummary ?? + accountSummaries[ + expectDefined(Object.keys(accountSummaries)[0], "object.keys(account summaries) entry at 0") + ]; + if (fallbackSummary) { + channels[plugin.id] = { + ...fallbackSummary, + accounts: accountSummaries, + } satisfies ChannelHealthSummary; + } + } + + const pluginHealth = buildPluginHealthSummary(); + const contextEngineHealth = buildContextEngineHealthSummary(); + const deliveryQueueHealth = buildDeliveryQueueHealthSummary(); + return { + ok: true, + ts: Date.now(), + durationMs: Date.now() - start, + ...(params.eventLoop ? { eventLoop: params.eventLoop } : {}), + ...(pluginHealth ? { plugins: pluginHealth } : {}), + ...(contextEngineHealth ? { contextEngines: contextEngineHealth } : {}), + ...(deliveryQueueHealth ? { deliveryQueues: deliveryQueueHealth } : {}), + ...(params.configReloadHotReloadStatus + ? { configReload: { hotReloadStatus: params.configReloadHotReloadStatus } } + : {}), + channels, + channelOrder, + channelLabels, + heartbeatSeconds, + defaultAgentId, + agents, + sessions: { + path: sessions.path, + count: sessions.count, + recent: sessions.recent, + }, + }; +} + +async function readRuntimeHealthConfig(): Promise { + const { getRuntimeConfig } = await import("../../config/config.js"); + return getRuntimeConfig(); +} diff --git a/src/gateway/health/delivery-queue.ts b/src/gateway/health/delivery-queue.ts new file mode 100644 index 000000000000..beb9baf70db7 --- /dev/null +++ b/src/gateway/health/delivery-queue.ts @@ -0,0 +1,60 @@ +import { countFailedChannelIngressQueueEntries } from "../../channels/message/ingress-queue.js"; +import { countFailedDeliveryQueueEntries } from "../../infra/delivery-queue-sqlite.js"; +import { isDiagnosticFlagEnabled } from "../../infra/diagnostic-flags.js"; +import { formatErrorMessage } from "../../infra/errors.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; +import type { DeliveryQueueHealthSummary } from "./types.js"; + +const healthLog = createSubsystemLogger("health"); + +const debugHealth = (message: string, error: unknown) => { + if (isDiagnosticFlagEnabled("health")) { + healthLog.info(message, { error: formatErrorMessage(error) }); + } +}; + +/** Builds dead-lettered inbound and outbound queue health for gateway snapshots. */ +export function buildDeliveryQueueHealthSummary(): DeliveryQueueHealthSummary | undefined { + // Queue health reads are diagnostic; a storage failure must not take the + // gateway health endpoint down with it. + let failed: DeliveryQueueHealthSummary["failed"] = []; + try { + failed = countFailedDeliveryQueueEntries().map((queue) => { + const entry: DeliveryQueueHealthSummary["failed"][number] = { + queueName: queue.queueName, + count: queue.count, + }; + if (queue.oldestFailedAt != null) { + entry.oldestFailedAt = queue.oldestFailedAt; + } + return entry; + }); + } catch (error) { + debugHealth("outbound delivery queue health read failed", error); + } + + let ingressFailed: NonNullable = []; + try { + ingressFailed = countFailedChannelIngressQueueEntries().map((queue) => { + const entry: NonNullable[number] = { + channelId: queue.channelId, + accountId: queue.accountId, + count: queue.count, + }; + if (queue.oldestFailedAt != null) { + entry.oldestFailedAt = queue.oldestFailedAt; + } + return entry; + }); + } catch (error) { + debugHealth("channel ingress queue health read failed", error); + } + + if (failed.length === 0 && ingressFailed.length === 0) { + return undefined; + } + return { + failed, + ...(ingressFailed.length > 0 ? { ingressFailed } : {}), + }; +} diff --git a/src/gateway/server-maintenance.test.ts b/src/gateway/server-maintenance.test.ts index 74bc50afb58d..1898ca2aac4a 100644 --- a/src/gateway/server-maintenance.test.ts +++ b/src/gateway/server-maintenance.test.ts @@ -2,7 +2,7 @@ // stale chat buffers, expired runs, health summaries, and timer disposal. import { afterEach, describe, expect, it, vi } from "vitest"; import { managedWorktrees } from "../agents/worktrees/service.js"; -import type { HealthSummary } from "../commands/health.js"; +import type { HealthSummary } from "./health/types.js"; const CURATOR_INITIAL_DELAY_MS = 5 * 60_000; const CURATOR_SWEEP_INTERVAL_MS = 24 * 60 * 60_000; import type { ChatAbortControllerEntry } from "./chat-abort.js"; diff --git a/src/gateway/server-maintenance.ts b/src/gateway/server-maintenance.ts index b255357a096f..d36eaf04a839 100644 --- a/src/gateway/server-maintenance.ts +++ b/src/gateway/server-maintenance.ts @@ -7,7 +7,6 @@ import { resolveWorktreeCleanupLimits, WORKTREE_GC_INTERVAL_MS, } from "../agents/worktrees/service.js"; -import type { HealthSummary } from "../commands/health.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { sweepStaleRunContexts } from "../infra/agent-events.js"; import { pruneOrphanedDeliveryQueueMedia } from "../infra/outbound/delivery-queue-media-spool.js"; @@ -21,6 +20,7 @@ import { } from "./chat-abort.js"; import type { QueuedChatTurnMap } from "./chat-queued-turns.js"; import { pruneStaleControlPlaneBuckets } from "./control-plane-rate-limit.js"; +import type { HealthSummary } from "./health/types.js"; import { chatAbortMarkerTimestampMs } from "./server-chat-state.js"; import type { ChatRunState } from "./server-chat-state.js"; import type { ChatRunEntry } from "./server-chat.js"; diff --git a/src/gateway/server-methods/health.ts b/src/gateway/server-methods/health.ts index 6bbcf43e40dc..b92d0fd89c20 100644 --- a/src/gateway/server-methods/health.ts +++ b/src/gateway/server-methods/health.ts @@ -2,10 +2,10 @@ // detecting stale channel runtime state against live gateway snapshots. import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js"; import type { ChannelAccountSnapshot } from "../../channels/plugins/types.public.js"; -import { buildDeliveryQueueHealthSummary } from "../../commands/health.js"; import { getStatusSummary } from "../../commands/status.js"; import { listContextEngineQuarantines } from "../../context-engine/registry.js"; import type { GatewayHotReloadStatus } from "../config-reload-status.types.js"; +import { buildDeliveryQueueHealthSummary } from "../health/delivery-queue.js"; import type { ChannelHealthSummary, HealthSummary } from "../health/types.js"; import type { ChannelRuntimeSnapshot } from "../server-channel-runtime.types.js"; import { HEALTH_REFRESH_INTERVAL_MS } from "../server-constants.js"; diff --git a/src/gateway/server-node-events-types.ts b/src/gateway/server-node-events-types.ts index ead29fbd2370..474988d5e7e3 100644 --- a/src/gateway/server-node-events-types.ts +++ b/src/gateway/server-node-events-types.ts @@ -2,8 +2,8 @@ // Defines the narrowed context and event envelope for node-originated handlers. import type { ModelCatalogEntry } from "../agents/model-catalog.js"; import type { CliDeps } from "../cli/deps.types.js"; -import type { HealthSummary } from "../commands/health.js"; import type { ChatAbortControllerEntry } from "./chat-abort.js"; +import type { HealthSummary } from "./health/types.js"; import type { ChatRunEntry, ChatRunRegistration } from "./server-chat.js"; import type { DedupeEntry } from "./server-shared.js"; diff --git a/src/gateway/server-node-events.test.ts b/src/gateway/server-node-events.test.ts index f840982cc936..21960cb7fda8 100644 --- a/src/gateway/server-node-events.test.ts +++ b/src/gateway/server-node-events.test.ts @@ -152,7 +152,7 @@ vi.mock("../infra/device-pairing.js", () => ({ updatePairedDevicePresence: updatePairedDevicePresenceMock, })); import type { CliDeps } from "../cli/deps.js"; -import type { HealthSummary } from "../commands/health.js"; +import type { HealthSummary } from "./health/types.js"; import type { NodeEventContext } from "./server-node-events-types.js"; import { handleNodeEvent } from "./server-node-events.js"; diff --git a/src/gateway/server/health-state.test.ts b/src/gateway/server/health-state.test.ts index 5e6ad29dcf29..1ca487285c97 100644 --- a/src/gateway/server/health-state.test.ts +++ b/src/gateway/server/health-state.test.ts @@ -1,21 +1,21 @@ // Health-state tests cover probe coalescing, sensitive snapshots, and broadcast version behavior. import { beforeEach, describe, expect, it, vi } from "vitest"; -import type { HealthSummary } from "../../commands/health.js"; +import type { HealthSummary } from "../health/types.js"; /** * Health-state cache tests covering coalescing, sensitive probes, and broadcasts. */ -const getHealthSnapshotMock = vi.hoisted(() => vi.fn()); +const collectGatewayHealthSnapshotMock = vi.hoisted(() => vi.fn()); -vi.mock("../../commands/health.js", () => ({ - getHealthSnapshot: getHealthSnapshotMock, +vi.mock("../health/collector.js", () => ({ + collectGatewayHealthSnapshot: collectGatewayHealthSnapshotMock, })); function healthSnapshotCallArg(index = 0) { - return getHealthSnapshotMock.mock.calls.at(index)?.at(0) as + return collectGatewayHealthSnapshotMock.mock.calls.at(index)?.at(0) as | { + audience?: "public" | "admin"; eventLoop?: unknown; - includeSensitive?: boolean; probe?: boolean; runtimeSnapshot?: unknown; configReloadHotReloadStatus?: unknown; @@ -44,8 +44,8 @@ function createHealthSummary(): HealthSummary { async function loadHealthState() { vi.resetModules(); - getHealthSnapshotMock.mockReset(); - getHealthSnapshotMock.mockResolvedValue(createHealthSummary()); + collectGatewayHealthSnapshotMock.mockReset(); + collectGatewayHealthSnapshotMock.mockResolvedValue(createHealthSummary()); return await import("./health-state.js"); } @@ -57,7 +57,7 @@ describe("refreshGatewayHealthSnapshot", () => { it("keeps refreshes coalesced while preserving the first probe intent", async () => { const healthState = await loadHealthState(); let resolveSnapshot: ((summary: HealthSummary) => void) | undefined; - getHealthSnapshotMock.mockImplementation( + collectGatewayHealthSnapshotMock.mockImplementation( () => new Promise((resolve) => { resolveSnapshot = resolve; @@ -67,10 +67,10 @@ describe("refreshGatewayHealthSnapshot", () => { const first = healthState.refreshGatewayHealthSnapshot({ probe: false }); const second = healthState.refreshGatewayHealthSnapshot({ probe: true }); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(1); - expect(getHealthSnapshotMock).toHaveBeenCalledWith({ + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(1); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledWith({ + audience: "public", probe: false, - includeSensitive: false, runtimeSnapshot: undefined, }); expect(Object.hasOwn(healthSnapshotCallArg() ?? {}, "eventLoop")).toBe(false); @@ -99,7 +99,7 @@ describe("refreshGatewayHealthSnapshot", () => { getEventLoopHealth: () => undefined, }); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); expect(healthSnapshotCallArg()?.eventLoop).toBe(eventLoop); expect(Object.hasOwn(healthSnapshotCallArg(1) ?? {}, "eventLoop")).toBe(false); }); @@ -116,7 +116,7 @@ describe("refreshGatewayHealthSnapshot", () => { getConfigReloaderHotReloadStatus: () => undefined, }); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); expect(healthSnapshotCallArg()?.configReloadHotReloadStatus).toBe("disabled"); expect(Object.hasOwn(healthSnapshotCallArg(1) ?? {}, "configReloadHotReloadStatus")).toBe( false, @@ -141,17 +141,17 @@ describe("refreshGatewayHealthSnapshot", () => { }, }); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); expect( - getHealthSnapshotMock.mock.calls + collectGatewayHealthSnapshotMock.mock.calls .map((_call, index) => healthSnapshotCallArg(index)?.probe) .toSorted((a, b) => Number(a) - Number(b)), ).toEqual([false, true]); expect( - getHealthSnapshotMock.mock.calls.map( - (_call, index) => healthSnapshotCallArg(index)?.includeSensitive, + collectGatewayHealthSnapshotMock.mock.calls.map( + (_call, index) => healthSnapshotCallArg(index)?.audience, ), - ).toEqual([false, false]); + ).toEqual(["public", "public"]); expect(healthSnapshotCallArg()?.runtimeSnapshot).toBe(runtimeSnapshot); expect(healthSnapshotCallArg(1)?.runtimeSnapshot).toBeUndefined(); }); @@ -161,7 +161,7 @@ describe("refreshGatewayHealthSnapshot", () => { const sensitiveSummary = createHealthSummary(); const safeSummary = createHealthSummary(); const broadcast = vi.fn(); - getHealthSnapshotMock + collectGatewayHealthSnapshotMock .mockResolvedValueOnce(sensitiveSummary) .mockResolvedValueOnce(safeSummary); healthState.setBroadcastHealthUpdate(broadcast); @@ -186,7 +186,7 @@ describe("refreshGatewayHealthSnapshot", () => { const safeSummary = createHealthSummary(); let resolveSensitive: (() => void) | undefined; let resolveSafe: (() => void) | undefined; - getHealthSnapshotMock + collectGatewayHealthSnapshotMock .mockImplementationOnce( () => new Promise((resolve) => { @@ -206,9 +206,9 @@ describe("refreshGatewayHealthSnapshot", () => { }); const safe = healthState.refreshGatewayHealthSnapshot({ probe: false }); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(2); - expect(healthSnapshotCallArg()?.includeSensitive).toBe(true); - expect(healthSnapshotCallArg(1)?.includeSensitive).toBe(false); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(healthSnapshotCallArg()?.audience).toBe("admin"); + expect(healthSnapshotCallArg(1)?.audience).toBe("public"); resolveSensitive?.(); resolveSafe?.(); @@ -224,7 +224,7 @@ describe("refreshGatewayHealthSnapshot", () => { ])("releases the $label refresh lane after rejection", async ({ includeSensitive }) => { const healthState = await loadHealthState(); const recovered = createHealthSummary(); - getHealthSnapshotMock + collectGatewayHealthSnapshotMock .mockRejectedValueOnce(new Error("snapshot failed")) .mockResolvedValueOnce(recovered); @@ -235,6 +235,6 @@ describe("refreshGatewayHealthSnapshot", () => { healthState.refreshGatewayHealthSnapshot({ probe: false, includeSensitive }), ).resolves.toBe(recovered); - expect(getHealthSnapshotMock).toHaveBeenCalledTimes(2); + expect(collectGatewayHealthSnapshotMock).toHaveBeenCalledTimes(2); }); }); diff --git a/src/gateway/server/health-state.ts b/src/gateway/server/health-state.ts index 2796bb78c7cd..8a34fac1ab4b 100644 --- a/src/gateway/server/health-state.ts +++ b/src/gateway/server/health-state.ts @@ -1,7 +1,6 @@ // Gateway health state builds snapshots, caches health probes, and broadcasts health/presence version changes. import type { Snapshot } from "../../../packages/gateway-protocol/src/index.js"; import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; -import { getHealthSnapshot, type HealthSummary } from "../../commands/health.js"; import { createConfigIO, getRuntimeConfig } from "../../config/io.js"; import { STATE_DIR } from "../../config/paths.js"; import { getRuntimeConfigAppliedHash } from "../../config/runtime-snapshot.js"; @@ -11,6 +10,8 @@ import { getUpdateAvailable } from "../../infra/update-startup.js"; import { normalizeMainKey } from "../../routing/session-key.js"; import { resolveGatewayAuth } from "../auth.js"; import type { GatewayHotReloadStatus } from "../config-reload-status.types.js"; +import { collectGatewayHealthSnapshot } from "../health/collector.js"; +import type { HealthSummary } from "../health/types.js"; import type { ChannelRuntimeSnapshot } from "../server-channel-runtime.types.js"; import type { GatewayEventLoopHealth } from "./event-loop-health.js"; @@ -30,7 +31,7 @@ export function buildGatewaySnapshot(opts?: { includeSensitive?: boolean }): Sna const presence = listSystemPresence(); const uptimeMs = Math.round(process.uptime() * 1000); const updateAvailable = getUpdateAvailable() ?? undefined; - // Health is async; caller should await getHealthSnapshot and replace later if needed. + // Health is async; the caller replaces this with the collected snapshot. const emptyHealth: Snapshot["health"] = {}; const snapshot: Snapshot = { presence, @@ -96,9 +97,9 @@ export async function refreshGatewayHealthSnapshot(opts?: { } const eventLoop = opts?.getEventLoopHealth?.(); const configReloadHotReloadStatus = opts?.getConfigReloaderHotReloadStatus?.(); - const snap = await getHealthSnapshot({ - probe: opts?.probe, - includeSensitive, + const snap = await collectGatewayHealthSnapshot({ + audience: includeSensitive ? "admin" : "public", + probe: opts?.probe !== false, runtimeSnapshot, ...(eventLoop ? { eventLoop } : {}), ...(configReloadHotReloadStatus ? { configReloadHotReloadStatus } : {}), diff --git a/src/gateway/test-helpers.maintenance-state.ts b/src/gateway/test-helpers.maintenance-state.ts index 83108d0f20a1..dce6355af069 100644 --- a/src/gateway/test-helpers.maintenance-state.ts +++ b/src/gateway/test-helpers.maintenance-state.ts @@ -1,6 +1,6 @@ // Gateway maintenance-state test helper. // Builds minimal timer/health/chat state for maintenance tests. -import type { HealthSummary } from "../commands/health.js"; +import type { HealthSummary } from "./health/types.js"; import { createChatRunState } from "./server-chat-state.js"; /** Create a Gateway maintenance-state stub with configurable health/presence versions. */ diff --git a/src/gateway/test-helpers.mocks.ts b/src/gateway/test-helpers.mocks.ts index 707174d3cab7..4a72ae6b35ae 100644 --- a/src/gateway/test-helpers.mocks.ts +++ b/src/gateway/test-helpers.mocks.ts @@ -241,8 +241,8 @@ vi.mock("/src/agents/embedded-agent-runner/runs.js", async () => { >("../agents/embedded-agent-runner/runs.js", { includeActiveCount: true }); }); -vi.mock("../commands/health.js", () => ({ - getHealthSnapshot: vi.fn().mockResolvedValue({ ok: true, stub: true }), +vi.mock("./health/collector.js", () => ({ + collectGatewayHealthSnapshot: vi.fn().mockResolvedValue({ ok: true, stub: true }), })); vi.mock("../commands/status.js", () => ({ getStatusSummary: vi.fn().mockResolvedValue({ ok: true }),