vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-03 06:31:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(gateway): own health snapshot collection

Browse Source
This commit is contained in:
Vincent Koc
2026-07-29 16:57:50 +08:00
parent 46f629882b
commit 0211407d8a
16 changed files with 733 additions and 670 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/commands/health.delivery-queue.test.ts
View File

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

14
src/commands/health.plugins.test.ts
View File

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

27
src/commands/health.snapshot.test.ts
View File

@@ -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<string, unknown> = {};
let testStore: Record<string, { updatedAt?: number }> = {};
@@ -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<typeof collectGatewayHealthSnapshot>[0],
"audience" | "probe"
> & {
includeSensitive?: boolean;
probe?: boolean;
};
let getHealthSnapshot: (params?: LegacyHealthSnapshotParams) => Promise<HealthSummary>;
let buildTelegramHealthSummaryForTest = buildTelegramHealthSummary;
let probeTelegramAccountForTestOverride:
| ((account: TelegramHealthAccount, timeoutMs: number) => Promise<Record<string, unknown>>)
@@ -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,

626
src/commands/health.ts
View File

@@ -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<string, unknown> | 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<DeliveryQueueHealthSummary["ingressFailed"]> = [];
try {
ingressFailed = countFailedChannelIngressQueueEntries().map((queue) => {
const entry: NonNullable<DeliveryQueueHealthSummary["ingressFailed"]>[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<string>();
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<typeof listSessionEntriesReadOnly>;
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<boolean> {
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<HealthSummary> {
const timeoutMs = params?.timeoutMs;
const cfg = await readRuntimeHealthConfig();
const { defaultAgentId, ordered } = resolveAgentOrder(cfg);
const channelBindings = buildChannelAccountBindings(cfg);
const sessionCache = new Map<string, HealthSummary["sessions"]>();
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<string, ChannelHealthSummary> = {};
const plugins = listReadOnlyChannelPluginsForConfig(cfg, {
includeSetupFallbackPlugins: false,
});
const channelOrder = plugins.map((plugin) => plugin.id);
const channelLabels: Record<string, string> = {};
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<string, ChannelAccountHealthSummary> = {};
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<string, unknown>) : 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<OpenClawConfig> {
const { readBestEffortConfig } = await loadConfigRuntime();
return await readBestEffortConfig();
}
async function readRuntimeHealthConfig(): Promise<OpenClawConfig> {
const { getRuntimeConfig } = await loadConfigRuntime();
return getRuntimeConfig();
}
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */

168
src/gateway/health/account-context.ts Normal file
View File

@@ -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<string, unknown> | 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<boolean> {
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,
};
}

427
src/gateway/health/collector.ts Normal file
View File

@@ -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<string, unknown>,
) => {
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<string>();
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<typeof listSessionEntriesReadOnly>;
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<HealthSummary> {
const cfg = await readRuntimeHealthConfig();
const { defaultAgentId, ordered } = resolveHealthAgentOrder(cfg);
const channelBindings = buildChannelAccountBindings(cfg);
const sessionCache = new Map<string, HealthSummary["sessions"]>();
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<string, ChannelHealthSummary> = {};
const plugins = listReadOnlyChannelPluginsForConfig(cfg, {
includeSetupFallbackPlugins: false,
});
const channelOrder = plugins.map((plugin) => plugin.id);
const channelLabels: Record<string, string> = {};
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<string, ChannelAccountHealthSummary> = {};
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<string, unknown>) : 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<OpenClawConfig> {
const { getRuntimeConfig } = await import("../../config/config.js");
return getRuntimeConfig();
}

60
src/gateway/health/delivery-queue.ts Normal file
View File

@@ -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<DeliveryQueueHealthSummary["ingressFailed"]> = [];
try {
ingressFailed = countFailedChannelIngressQueueEntries().map((queue) => {
const entry: NonNullable<DeliveryQueueHealthSummary["ingressFailed"]>[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 } : {}),
};
}

2
src/gateway/server-maintenance.test.ts
View File

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

2
src/gateway/server-maintenance.ts
View File

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

2
src/gateway/server-methods/health.ts
View File

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

2
src/gateway/server-node-events-types.ts
View File

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

2
src/gateway/server-node-events.test.ts
View File

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

52
src/gateway/server/health-state.test.ts
View File

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

11
src/gateway/server/health-state.ts
View File

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

2
src/gateway/test-helpers.maintenance-state.ts
View File

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

4
src/gateway/test-helpers.mocks.ts
View File

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