improve(health): surface dead-lettered delivery queue entries (#99842)

* improve(health): surface dead-lettered delivery queue entries

Deliveries that exhaust retries are moved to the failed status in the
SQLite delivery queue for diagnostics, but no health surface ever read
them back: openclaw health stayed all-green while messages sat
dead-lettered, visible only in gateway logs. Add a per-queue failed
count accessor and report dead-lettered entries in the health snapshot
(JSON field plus a warning line), following the existing plugin and
context-engine health section pattern. Observer-only: no retry, purge,
or behavior change.

* test(health): cover snapshot and CLI dead-letter reporting

Add getHealthSnapshot coverage against an isolated state dir, an
openclaw health text-output test for the warning line, exact
oldest-failure timestamp assertions via fake timers, and a debug log
when the delivery-queue health read fails.

* fix(health): recompute dead-letter counts for cached gateway responses

The gateway health handler can serve a cached snapshot for up to a
refresh interval, so a delivery dead-lettered after the cache was
filled stayed hidden from openclaw health. Recompute the delivery
queue summary in mergeCachedHealthRuntimeState alongside the existing
context-engine and model-pricing live merges, and cover the cached
handler path with a regression test.

src/commands/health.snapshot.test.ts

@@ -552,6 +552,39 @@ describe("getHealthSnapshot", () => {
     ]);
   });
 
+  it("includes dead-lettered delivery queue entries in the health snapshot", async () => {
+    testConfig = { session: { store: "/tmp/x" } };
+    testStore = {};
+    setActivePluginRegistry(createTestRegistry([]));
+    const tmpStateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-health-dq-"));
+    const previousStateDir = process.env.OPENCLAW_STATE_DIR;
+    process.env.OPENCLAW_STATE_DIR = tmpStateDir;
+    try {
+      const { moveDeliveryQueueEntryToFailed, upsertDeliveryQueueEntry } =
+        await import("../infra/delivery-queue-sqlite.js");
+      const clean = await getHealthSnapshot({ timeoutMs: 10, probe: false });
+      expect(clean.deliveryQueues).toBeUndefined();
+
+      upsertDeliveryQueueEntry({
+        queueName: "outbound",
+        entry: { id: "dead-1", enqueuedAt: 1_000, retryCount: 5 },
+      });
+      moveDeliveryQueueEntryToFailed("outbound", "dead-1");
+
+      const snap = await getHealthSnapshot({ timeoutMs: 10, probe: false });
+      expect(snap.deliveryQueues?.failed).toEqual([
+        { queueName: "outbound", count: 1, oldestFailedAt: expect.any(Number) },
+      ]);
+    } finally {
+      if (previousStateDir === undefined) {
+        delete process.env.OPENCLAW_STATE_DIR;
+      } else {
+        process.env.OPENCLAW_STATE_DIR = previousStateDir;
+      }
+      fs.rmSync(tmpStateDir, { recursive: true, force: true });
+    }
+  });
+
   it("skips telegram probe when not configured", async () => {
     testConfig = { session: { store: "/tmp/x" } };
     testStore = {

src/commands/health.test.ts

@@ -10,6 +10,7 @@ import { formatHealthCheckFailure } from "./health-format.js";
 import type { HealthSummary } from "./health.js";
 import {
   formatContextEngineHealthLine,
+  formatDeliveryQueueHealthLine,
   formatHealthChannelLines,
   formatModelPricingHealthLine,
   healthCommand,
@@ -184,6 +185,26 @@ describe("healthCommand", () => {
     expect(parsed.sessions.count).toBe(1);
   });
 
+  it("prints the delivery queue warning line when the gateway reports dead-letters", async () => {
+    const snapshot = createHealthSummary({
+      channels: {},
+      channelOrder: [],
+      channelLabels: {},
+    });
+    snapshot.deliveryQueues = {
+      failed: [{ queueName: "outbound", count: 2, oldestFailedAt: Date.now() - 7_200_000 }],
+    };
+    callGatewayMock.mockResolvedValueOnce(snapshot);
+
+    await healthCommand({ json: false, timeoutMs: 1000, config: {} }, runtime as never);
+
+    expect(runtime.exit).not.toHaveBeenCalled();
+    const output = stripAnsi(runtime.log.mock.calls.map((c) => String(c[0])).join("\n"));
+    expect(output).toContain(
+      "Delivery queue: warning (dead-lettered entries — outbound: 2; oldest 2h ago)",
+    );
+  });
+
   it("prints the rich text summary and verbose gateway details", async () => {
     const recent = [
       { key: "main", updatedAt: Date.now() - 60_000, age: 60_000 },
@@ -486,6 +507,36 @@ describe("formatContextEngineHealthLine", () => {
   });
 });
 
+describe("formatDeliveryQueueHealthLine", () => {
+  it("summarizes dead-lettered delivery queue entries with the oldest age", () => {
+    const summary = createHealthSummary({
+      channels: {},
+      channelOrder: [],
+      channelLabels: {},
+    });
+    summary.deliveryQueues = {
+      failed: [
+        { queueName: "outbound", count: 3, oldestFailedAt: 90_000 },
+        { queueName: "session", count: 1 },
+      ],
+    };
+
+    expect(formatDeliveryQueueHealthLine(summary, 7_290_000)).toBe(
+      "Delivery queue: warning (dead-lettered entries — outbound: 3, session: 1; oldest 2h ago)",
+    );
+  });
+
+  it("returns null when no dead-lettered entries are reported", () => {
+    const summary = createHealthSummary({
+      channels: {},
+      channelOrder: [],
+      channelLabels: {},
+    });
+
+    expect(formatDeliveryQueueHealthLine(summary)).toBeNull();
+  });
+});
+
 describe("formatHealthCheckFailure", () => {
   it("keeps non-rich output stable", () => {
     const err = new Error("gateway closed (1006 abnormal closure): no close reason");

src/commands/health.ts

@@ -36,8 +36,10 @@ import { getGatewayModelPricingHealth } from "../gateway/model-pricing-cache-sta
 import { isGatewayModelPricingEnabled } from "../gateway/model-pricing-config.js";
 import type { ChannelRuntimeSnapshot } from "../gateway/server-channel-runtime.types.js";
 import { info } from "../globals.js";
+import { countFailedDeliveryQueueEntries } from "../infra/delivery-queue-sqlite.js";
 import { isTruthyEnvValue } from "../infra/env.js";
 import { formatErrorMessage } from "../infra/errors.js";
+import { formatDurationHuman } from "../infra/format-time/format-duration.js";
 import { resolveHeartbeatSummaryForAgent } from "../infra/heartbeat-summary.js";
 import { getActivePluginRegistry } from "../plugins/runtime.js";
 import { buildChannelAccountBindings, resolvePreferredAccountId } from "../routing/bindings.js";
@@ -54,6 +56,7 @@ import type {
   ChannelAccountHealthSummary,
   ChannelHealthSummary,
   ContextEngineHealthSummary,
+  DeliveryQueueHealthSummary,
   HealthSummary,
   PluginHealthErrorSummary,
   PluginHealthSummary,
@@ -242,6 +245,46 @@ export function formatContextEngineHealthLine(summary: HealthSummary): string |
   return `Context engine: warning (${quarantined.length} quarantined; downgraded to legacy: ${engines})`;
 }
 
+/** Builds dead-lettered delivery queue health; shared with cached gateway responses. */
+export function buildDeliveryQueueHealthSummary(): DeliveryQueueHealthSummary | undefined {
+  // Dead-lettered deliveries are retained in SQLite for diagnostics but had no
+  // health surface; a storage read failure must not take health down with it.
+  try {
+    const 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;
+    });
+    return failed.length > 0 ? { failed } : undefined;
+  } catch (error) {
+    debugHealth("delivery queue health read failed", error);
+    return undefined;
+  }
+}
+
+/** Formats dead-lettered delivery queue entries for text health output. */
+export function formatDeliveryQueueHealthLine(
+  summary: HealthSummary,
+  now = Date.now(),
+): string | null {
+  const failed = summary.deliveryQueues?.failed ?? [];
+  if (failed.length === 0) {
+    return null;
+  }
+  const counts = failed.map((queue) => `${queue.queueName}: ${queue.count}`).join(", ");
+  const oldest = failed
+    .map((queue) => queue.oldestFailedAt)
+    .filter((value): value is number => typeof value === "number");
+  const oldestNote =
+    oldest.length > 0 ? `; oldest ${formatDurationHuman(now - Math.min(...oldest))} ago` : "";
+  return `Delivery queue: warning (dead-lettered entries — ${counts}${oldestNote})`;
+}
+
 const resolveHeartbeatSummary = (cfg: OpenClawConfig, agentId: string) =>
   resolveHeartbeatSummaryForAgent(cfg, agentId);
 
@@ -658,6 +701,7 @@ export async function getHealthSnapshot(params?: {
 
   const pluginHealth = buildPluginHealthSummary();
   const contextEngineHealth = buildContextEngineHealthSummary();
+  const deliveryQueueHealth = buildDeliveryQueueHealthSummary();
   const summary: HealthSummary = {
     ok: true,
     ts: Date.now(),
@@ -665,6 +709,7 @@ export async function getHealthSnapshot(params?: {
     ...(params?.eventLoop ? { eventLoop: params.eventLoop } : {}),
     ...(pluginHealth ? { plugins: pluginHealth } : {}),
     ...(contextEngineHealth ? { contextEngines: contextEngineHealth } : {}),
+    ...(deliveryQueueHealth ? { deliveryQueues: deliveryQueueHealth } : {}),
     modelPricing: getGatewayModelPricingHealth({ enabled: isGatewayModelPricingEnabled(cfg) }),
     channels,
     channelOrder,
@@ -900,6 +945,10 @@ export async function healthCommand(
     if (contextEngineLine) {
       runtime.log(styleHealthChannelLine(contextEngineLine, rich));
     }
+    const deliveryQueueLine = formatDeliveryQueueHealthLine(summary);
+    if (deliveryQueueLine) {
+      runtime.log(styleHealthChannelLine(deliveryQueueLine, rich));
+    }
     for (const plugin of displayPlugins) {
       const channelSummary = summary.channels?.[plugin.id];
       if (!channelSummary || channelSummary.linked !== true) {

src/commands/health.types.ts

@@ -55,6 +55,15 @@ export type ContextEngineHealthSummary = {
   quarantined: ContextEngineHealthQuarantineSummary[];
 };
 
+/** Dead-lettered delivery queue entries surfaced in health output. */
+export type DeliveryQueueHealthSummary = {
+  failed: Array<{
+    queueName: string;
+    count: number;
+    oldestFailedAt?: number;
+  }>;
+};
+
 /** Optional model pricing cache health reported by the gateway. */
 type ModelPricingHealthSummary =
   import("../gateway/model-pricing-cache-state.js").GatewayModelPricingHealth;
@@ -67,6 +76,7 @@ export type HealthSummary = {
   eventLoop?: import("../gateway/server/event-loop-health.js").GatewayEventLoopHealth;
   plugins?: PluginHealthSummary;
   contextEngines?: ContextEngineHealthSummary;
+  deliveryQueues?: DeliveryQueueHealthSummary;
   modelPricing?: ModelPricingHealthSummary;
   channels: Record<string, ChannelHealthSummary>;
   channelOrder: string[];

src/gateway/server-methods/health.ts

@@ -2,6 +2,7 @@
 // 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 type { ChannelHealthSummary, HealthSummary } from "../../commands/health.types.js";
 import { getStatusSummary } from "../../commands/status.js";
 import { listContextEngineQuarantines } from "../../context-engine/registry.js";
@@ -92,7 +93,15 @@ function mergeCachedHealthRuntimeState(params: {
   cached: HealthSummary;
   eventLoop?: HealthSummary["eventLoop"];
 }): HealthSummary {
-  const { contextEngines: _cachedContextEngines, ...cached } = params.cached;
+  const {
+    contextEngines: _cachedContextEngines,
+    deliveryQueues: _cachedDeliveryQueues,
+    ...cached
+  } = params.cached;
+  // Dead-letter counts are cheap SQLite reads; recompute them like context
+  // engines so a delivery that failed after the cache was filled is not hidden
+  // for a refresh interval.
+  const deliveryQueues = buildDeliveryQueueHealthSummary();
   const quarantinedContextEngines: NonNullable<HealthSummary["contextEngines"]>["quarantined"] = [];
   for (const entry of listContextEngineQuarantines()) {
     const summary: NonNullable<HealthSummary["contextEngines"]>["quarantined"][number] = {
@@ -112,6 +121,7 @@ function mergeCachedHealthRuntimeState(params: {
     ...(quarantinedContextEngines.length > 0
       ? { contextEngines: { quarantined: quarantinedContextEngines } }
       : {}),
+    ...(deliveryQueues ? { deliveryQueues } : {}),
     modelPricing: getGatewayModelPricingHealth({
       enabled: params.cached.modelPricing?.state !== "disabled",
     }),

src/gateway/server-methods/server-methods.test.ts

@@ -4895,6 +4895,73 @@ describe("gateway healthHandlers.health cache freshness", () => {
     }
   });
 
+  it("merges live dead-lettered delivery queue counts into cached health responses", async () => {
+    const tmpStateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-health-cached-dq-"));
+    const previousStateDir = process.env.OPENCLAW_STATE_DIR;
+    process.env.OPENCLAW_STATE_DIR = tmpStateDir;
+    try {
+      const { moveDeliveryQueueEntryToFailed, upsertDeliveryQueueEntry } =
+        await import("../../infra/delivery-queue-sqlite.js");
+      // The cached snapshot was built before this delivery dead-lettered.
+      const cached = {
+        ok: true,
+        ts: Date.now(),
+        durationMs: 1,
+        channels: {},
+        channelOrder: [],
+        channelLabels: {},
+        heartbeatSeconds: 0,
+        defaultAgentId: "main",
+        agents: [],
+        sessions: { path: "/tmp/sessions.json", count: 0, recent: [] },
+      };
+      upsertDeliveryQueueEntry({
+        queueName: "outbound",
+        entry: { id: "dead-1", enqueuedAt: 1_000, retryCount: 5 },
+      });
+      moveDeliveryQueueEntryToFailed("outbound", "dead-1");
+
+      const respond = vi.fn();
+      const refreshHealthSnapshot = vi.fn().mockResolvedValue(cached);
+
+      await healthHandlers.health({
+        req: {} as never,
+        params: {} as never,
+        respond: respond as never,
+        context: {
+          getHealthCache: () => cached,
+          refreshHealthSnapshot,
+          getRuntimeSnapshot: () => ({ channels: {}, channelAccounts: {} }),
+          logHealth: { error: vi.fn() },
+        } as never,
+        client: { connect: { role: "operator", scopes: ["operator.read"] } } as never,
+        isWebchatConnect: () => false,
+      });
+
+      const payload = mockCallArg(respond, 0, 1) as
+        | {
+            deliveryQueues?: {
+              failed?: Array<{ queueName?: string; count?: number; oldestFailedAt?: number }>;
+            };
+          }
+        | undefined;
+      expect(payload?.deliveryQueues?.failed).toHaveLength(1);
+      expect(payload?.deliveryQueues?.failed?.[0]).toMatchObject({
+        queueName: "outbound",
+        count: 1,
+      });
+      expect(typeof payload?.deliveryQueues?.failed?.[0]?.oldestFailedAt).toBe("number");
+      expect(mockCallArg(respond, 0, 3)).toEqual({ cached: true });
+    } finally {
+      if (previousStateDir === undefined) {
+        delete process.env.OPENCLAW_STATE_DIR;
+      } else {
+        process.env.OPENCLAW_STATE_DIR = previousStateDir;
+      }
+      fs.rmSync(tmpStateDir, { recursive: true, force: true });
+    }
+  });
+
   it("refreshes cached health when a runtime account is missing from the cached account summary", async () => {
     const cached = {
       ok: true,

src/infra/delivery-queue-sqlite.test.ts

@@ -1,9 +1,10 @@
 // Validates SQLite delivery queue inflate guards against corrupted entry_json.
 import fs from "node:fs";
 import path from "node:path";
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { openOpenClawStateDatabase } from "../state/openclaw-state-db.js";
 import {
+  countFailedDeliveryQueueEntries,
   deleteDeliveryQueueEntry,
   loadDeliveryQueueEntries,
   loadDeliveryQueueEntry,
@@ -160,3 +161,63 @@ describe("delivery-queue-sqlite corrupt JSON resilience", () => {
     });
   });
 });
+
+describe("countFailedDeliveryQueueEntries", () => {
+  let tmpDir: string;
+  let stateDir: string;
+
+  beforeEach(() => {
+    tmpDir = fs.mkdtempSync(path.join(resolvePreferredOpenClawTmpDir(), "openclaw-dq-count-"));
+    stateDir = path.join(tmpDir, "state");
+    fs.mkdirSync(stateDir, { recursive: true });
+  });
+
+  afterEach(() => {
+    fs.rmSync(tmpDir, { recursive: true, force: true });
+  });
+
+  function enqueue(queueName: string, id: string, enqueuedAt: number) {
+    upsertDeliveryQueueEntry({
+      queueName,
+      entry: { id, enqueuedAt, retryCount: 0 },
+      stateDir,
+    });
+  }
+
+  it("returns an empty list when nothing is dead-lettered", () => {
+    enqueue("outbound", "pending-1", 1_000);
+
+    expect(countFailedDeliveryQueueEntries(stateDir)).toEqual([]);
+  });
+
+  it("counts dead-lettered entries per queue with the oldest failure timestamp", () => {
+    enqueue("outbound", "dead-1", 1_000);
+    enqueue("outbound", "dead-2", 2_000);
+    enqueue("outbound", "still-pending", 3_000);
+    enqueue("session", "dead-3", 4_000);
+    vi.useFakeTimers();
+    try {
+      vi.setSystemTime(50_000);
+      moveDeliveryQueueEntryToFailed("outbound", "dead-1", stateDir);
+      vi.setSystemTime(60_000);
+      moveDeliveryQueueEntryToFailed("outbound", "dead-2", stateDir);
+      vi.setSystemTime(70_000);
+      moveDeliveryQueueEntryToFailed("session", "dead-3", stateDir);
+    } finally {
+      vi.useRealTimers();
+    }
+
+    const counts = countFailedDeliveryQueueEntries(stateDir);
+
+    expect(counts).toHaveLength(2);
+    const outbound = counts.find((queue) => queue.queueName === "outbound");
+    expect(outbound?.count).toBe(2);
+    expect(outbound?.oldestFailedAt).toBe(50_000);
+    const session = counts.find((queue) => queue.queueName === "session");
+    expect(session?.count).toBe(1);
+    expect(session?.oldestFailedAt).toBe(70_000);
+    expect(loadDeliveryQueueEntries("outbound", stateDir).map((entry) => entry.id)).toEqual([
+      "still-pending",
+    ]);
+  });
+});

src/infra/delivery-queue-sqlite.ts

@@ -242,6 +242,41 @@ export function updateDeliveryQueueEntry(
   upsertDeliveryQueueEntry({ queueName, entry: update(current), stateDir });
 }
 
+/** Dead-lettered entry counts for one queue namespace. */
+export type FailedDeliveryQueueCount = {
+  queueName: string;
+  count: number;
+  oldestFailedAt: number | null;
+};
+
+/** Count dead-lettered (failed) entries per queue namespace for health reporting. */
+export function countFailedDeliveryQueueEntries(stateDir?: string): FailedDeliveryQueueCount[] {
+  const database = openStateDatabase(stateDir);
+  const queueDb = getNodeSqliteKysely<DeliveryQueueDatabase>(database.db);
+  const rows = executeSqliteQuerySync(
+    database.db,
+    queueDb
+      .selectFrom("delivery_queue_entries")
+      .select((eb) => [
+        "queue_name",
+        eb.fn.countAll().as("failed_count"),
+        eb.fn.min("failed_at").as("oldest_failed_at"),
+      ])
+      .where("status", "=", "failed")
+      .groupBy("queue_name")
+      .orderBy("queue_name", "asc"),
+  ).rows as Array<{
+    queue_name: string;
+    failed_count: number | bigint;
+    oldest_failed_at: number | bigint | null;
+  }>;
+  return rows.map((row) => ({
+    queueName: row.queue_name,
+    count: Number(row.failed_count),
+    oldestFailedAt: row.oldest_failed_at == null ? null : Number(row.oldest_failed_at),
+  }));
+}
+
 /** Mark a pending delivery queue entry as failed for later diagnostics. */
 export function moveDeliveryQueueEntryToFailed(
   queueName: string,